openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Report code mode host request durations (#41452)

Browse Source 
## Why

Code mode wall time should measure the host operation itself, without including
client-side response delays or idle time between requests.

## What changed

- Measure each execute, wait, and terminate request in the code mode host.
- Carry the duration through the stdio and gRPC protocols and use it for
  model-visible wall time.
- Emit a structured `codex.code_mode.host_timing` event correlated with the
  conversation, turn, tool call, and cell.

## Testing

- Cover successful and failed execution timing, delayed response reads,
  repeated waits, termination, and missing cells across stdio and gRPC.
- Verify timing survives protocol serialization and is reflected in app-server
  model output and structured telemetry.

GitOrigin-RevId: d24af30c3fc5820521b4beba1f9970714dad6482
This commit is contained in:
Adam Perry @ OpenAI
2026-08-29 02:52:58 +00:00
committed by copyberry
parent 75388bf321
commit 48e22a5fa0
34 changed files with 1398 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/code_mode_host.rs
+152
View File
@@ -5,8 +5,14 @@ use anyhow::Context;
use anyhow::Result;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use codex_app_server_protocol::ThreadDecrementElicitationParams;
use codex_app_server_protocol::ThreadDecrementElicitationResponse;
use codex_app_server_protocol::ThreadIncrementElicitationParams;
use codex_app_server_protocol::ThreadIncrementElicitationResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_features::Feature;
@@ -14,6 +20,7 @@ use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use test_case::test_case;
use tokio::io::AsyncBufReadExt;
use tokio::io::BufReader;
use tokio::process::Command;
@@ -24,6 +31,151 @@ const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 60);
#[cfg(not(any(target_os = "macos", windows)))]
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 10);
/// Model output and the structured timing consumed by Bridge describe the same
/// host operation over either transport, including missing-cell responses.
#[test_case("exec", "grpc"; "exec_grpc")]
#[test_case("wait", "grpc"; "wait_grpc")]
#[test_case("exec", "stdio"; "exec_stdio")]
#[test_case("wait", "stdio"; "wait_stdio")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn code_mode_model_output_uses_structured_host_timing(
tool_name: &str,
transport: &str,
) -> Result<()> {
let mut host = match transport {
"grpc" => Some(
Command::new(codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?)
.args(["--listen", "grpc://127.0.0.1:0"])
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.kill_on_drop(/*kill_on_drop*/ true)
.spawn()?,
),
"stdio" => None,
_ => anyhow::bail!("unknown host transport: {transport}"),
};
let host_url = if let Some(host) = host.as_mut() {
let stdout = host.stdout.take().context("host stdout unavailable")?;
Some(
timeout(
DEFAULT_READ_TIMEOUT,
BufReader::new(stdout).lines().next_line(),
)
.await??
.context("host exited before publishing its URL")?,
)
} else {
None
};
let model_server = responses::start_mock_server().await;
let invocation = if tool_name == "exec" {
responses::ev_custom_tool_call(
"timed-call",
"exec",
"await new Promise(resolve => setTimeout(resolve, 100)); text('timed');",
)
} else {
responses::ev_function_call("timed-call", "wait", r#"{"cell_id":"missing"}"#)
};
responses::mount_sse_once(
&model_server,
responses::sse(vec![
responses::ev_response_created("resp-1"),
invocation,
responses::ev_completed("resp-1"),
]),
)
.await;
let follow_up = responses::mount_sse_once(
&model_server,
responses::sse(vec![
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-2"),
]),
)
.await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&model_server.uri())
.enable_feature(Feature::CodeModeOnly)
.write(codex_home.path())?;
let mut builder = TestAppServer::builder()
.with_codex_home(codex_home.path())
.with_json_logging("codex_code_mode::timing=info,codex_core::tools::parallel=info");
if let Some(host_url) = host_url.as_deref() {
builder = builder.with_args(&["--code-mode-host", host_url]);
}
let mut app_server = builder.build_initialized().await?;
let thread = app_server
.start_thread(ThreadStartParams::default())
.await?;
let increment_id = app_server
.send_request(
"thread/increment_elicitation",
Some(serde_json::to_value(ThreadIncrementElicitationParams {
thread_id: thread.thread.id.clone(),
})?),
)
.await?;
let _: ThreadIncrementElicitationResponse = app_server.read_response(increment_id).await?;
let start_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread.thread.id.clone(),
input: vec![UserInput::Text {
text: "run the timed operation".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse = app_server.read_response(start_id).await?;
let timing = app_server
.wait_for_json_log_event("codex.code_mode.host_timing")
.await?;
// Hold only the app-server after the host outcome, so local elapsed time
// cannot round to the same displayed duration as the host measurement.
tokio::time::sleep(Duration::from_millis(/*millis*/ 300)).await;
let decrement_id = app_server
.send_request(
"thread/decrement_elicitation",
Some(serde_json::to_value(ThreadDecrementElicitationParams {
thread_id: thread.thread.id.clone(),
})?),
)
.await?;
let _: ThreadDecrementElicitationResponse = app_server.read_response(decrement_id).await?;
let completed: TurnCompletedNotification =
app_server.read_notification("turn/completed").await?;
assert_eq!(completed.turn.status, TurnStatus::Completed);
let fields = &timing["fields"];
let code_mode_host_duration_ns = fields["code_mode_host_duration_ns"]
.as_u64()
.context("missing host timing")?;
assert_eq!(fields["conversation_id"], thread.thread.id);
assert_eq!(fields["turn_id"], completed.turn.id);
assert_eq!(fields["call_id"], "timed-call");
assert_eq!(fields["tool_name"], tool_name);
let request = follow_up.single_request();
let output = if tool_name == "exec" {
request.custom_tool_call_output("timed-call")
} else {
request.function_call_output("timed-call")
};
let status = if tool_name == "exec" {
"Script completed"
} else {
"Script failed"
};
let seconds = Duration::from_nanos(code_mode_host_duration_ns).as_secs_f32();
let seconds = (seconds * 10.0).round() / 10.0;
assert_eq!(
output["output"][0]["text"],
format!("{status}\nWall time {seconds:.1} seconds\nOutput:\n")
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn app_server_shares_flag_selected_grpc_code_mode_host_across_threads() -> Result<()> {
let host_program = codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?;
codex-rs/code-mode-host/src/grpc/conversions.rs
+23 -8
View File
@@ -3,6 +3,7 @@ use codex_code_mode_protocol::CodeModeToolKind;
use codex_code_mode_protocol::ExecuteRequest;
use codex_code_mode_protocol::FunctionCallOutputContentItem;
use codex_code_mode_protocol::ImageDetail;
use codex_code_mode_protocol::MissingCodeModeHostDuration;
use codex_code_mode_protocol::RuntimeResponse;
use codex_code_mode_protocol::ToolDefinition;
use codex_code_mode_protocol::WaitOutcome;
@@ -79,51 +80,65 @@ fn json_field(value: Option<Vec<u8>>, field: &str) -> Result<Option<JsonValue>,
.transpose()
}
pub(super) fn execution_outcome(response: RuntimeResponse) -> proto::ExecutionOutcome {
let (cell_id, content_items, outcome) = match response {
/// Preserves the response's timing; the host handler must record it first.
pub(super) fn execution_outcome(
response: RuntimeResponse,
) -> Result<proto::ExecutionOutcome, MissingCodeModeHostDuration> {
let (cell_id, content_items, outcome, code_mode_host_duration) = match response {
RuntimeResponse::Yielded {
cell_id,
content_items,
code_mode_host_duration,
} => (
cell_id,
content_items,
proto::execution_outcome::Outcome::Yielded(proto::ExecutionYielded {}),
code_mode_host_duration,
),
RuntimeResponse::Terminated {
cell_id,
content_items,
code_mode_host_duration,
} => (
cell_id,
content_items,
proto::execution_outcome::Outcome::Terminated(proto::ExecutionTerminated {}),
code_mode_host_duration,
),
RuntimeResponse::Result {
cell_id,
content_items,
error_text,
code_mode_host_duration,
} => (
cell_id,
content_items,
proto::execution_outcome::Outcome::Completed(proto::ExecutionCompleted { error_text }),
code_mode_host_duration,
),
};
proto::ExecutionOutcome {
let code_mode_host_duration = code_mode_host_duration.ok_or(MissingCodeModeHostDuration)?;
Ok(proto::ExecutionOutcome {
cell_id: cell_id.to_string(),
content_items: content_items.into_iter().map(content_item).collect(),
outcome: Some(outcome),
}
code_mode_host_duration_ns: u64::try_from(code_mode_host_duration.as_nanos())
.unwrap_or(u64::MAX),
})
}
pub(super) fn wait_response(outcome: WaitOutcome) -> proto::WaitResponse {
pub(super) fn wait_response(
outcome: WaitOutcome,
) -> Result<proto::WaitResponse, MissingCodeModeHostDuration> {
let state = match outcome {
WaitOutcome::LiveCell(response) => {
proto::wait_response::State::LiveCell(execution_outcome(response))
proto::wait_response::State::LiveCell(execution_outcome(response)?)
}
WaitOutcome::MissingCell(response) => {
proto::wait_response::State::MissingCell(execution_outcome(response))
proto::wait_response::State::MissingCell(execution_outcome(response)?)
}
};
proto::WaitResponse { state: Some(state) }
Ok(proto::WaitResponse { state: Some(state) })
}
fn content_item(item: FunctionCallOutputContentItem) -> proto::ContentItem {
codex-rs/code-mode-host/src/grpc/conversions_tests.rs
+30 -2
View File
@@ -1,13 +1,18 @@
use std::time::Duration;
use codex_code_mode_protocol::CellId;
use codex_code_mode_protocol::FunctionCallOutputContentItem;
use codex_code_mode_protocol::ImageDetail;
use codex_code_mode_protocol::MissingCodeModeHostDuration;
use codex_code_mode_protocol::RuntimeResponse;
use codex_code_mode_protocol::WaitOutcome;
use codex_code_mode_protocol::grpc as proto;
use pretty_assertions::assert_eq;
use tonic::Code;
use super::execute_request;
use super::execution_outcome;
use super::wait_response;
#[test]
fn rejects_missing_names_unknown_tool_kinds_and_invalid_json_schemas() {
@@ -65,6 +70,7 @@ fn rejects_missing_names_unknown_tool_kinds_and_invalid_json_schemas() {
#[test]
fn maps_text_image_audio_and_terminal_error_without_losing_details() {
let outcome = execution_outcome(RuntimeResponse::Result {
code_mode_host_duration: Some(Duration::from_nanos(/*nanos*/ 123_456_789)),
cell_id: CellId::new("cell".to_string()),
content_items: vec![
FunctionCallOutputContentItem::InputText {
@@ -83,7 +89,8 @@ fn maps_text_image_audio_and_terminal_error_without_losing_details() {
assert_eq!(
outcome,
proto::ExecutionOutcome {
Ok(proto::ExecutionOutcome {
code_mode_host_duration_ns: 123_456_789,
cell_id: "cell".to_string(),
content_items: vec![
proto::ContentItem {
@@ -108,6 +115,27 @@ fn maps_text_image_audio_and_terminal_error_without_losing_details() {
error_text: Some("failed".to_string()),
},
)),
}
})
);
}
/// Encoding must not turn a missing request measurement into measured zero,
/// including when no live cell remains to supply output.
#[test]
fn grpc_encoding_rejects_untimed_runtime_output() {
let response = RuntimeResponse::Terminated {
cell_id: CellId::new("cell".to_string()),
content_items: Vec::new(),
code_mode_host_duration: None,
};
assert_eq!(
execution_outcome(response.clone()),
Err(MissingCodeModeHostDuration)
);
for outcome in [
WaitOutcome::LiveCell(response.clone()),
WaitOutcome::MissingCell(response),
] {
assert_eq!(wait_response(outcome), Err(MissingCodeModeHostDuration));
}
}
codex-rs/code-mode-host/src/grpc/mod.rs
+22 -7
View File
@@ -9,6 +9,7 @@ mod waits;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Instant;
use codex_code_mode_protocol::CellId;
use codex_code_mode_protocol::WaitRequest;
@@ -131,6 +132,7 @@ impl GrpcCodeModeHost {
request: proto::ExecuteRequest,
callback_traceparent: Option<String>,
) -> Result<Response<GrpcStream<proto::ExecuteEvent>>, Status> {
let received_at = Instant::now();
let session = self.state.session(&request.session_id)?;
validation::identifier(&request.execution_id, "execution ID")?;
let request_permit = self.state.request_permit()?;
@@ -177,10 +179,15 @@ impl GrpcCodeModeHost {
biased;
_ = sender.closed() => {}
response = started.initial_response() => {
let event = response.map(|response| proto::ExecuteEvent {
event: Some(proto::execute_event::Event::Outcome(
conversions::execution_outcome(response),
)),
// Freeze timing before conversion or transport backpressure.
let code_mode_host_duration = received_at.elapsed();
let event = response.and_then(|response| {
let response = response.with_code_mode_host_duration(code_mode_host_duration);
let outcome = conversions::execution_outcome(response)
.map_err(|error| error.to_string())?;
Ok(proto::ExecuteEvent {
event: Some(proto::execute_event::Event::Outcome(outcome)),
})
}).map_err(Status::internal);
let _ = sender.send(event).await;
}
@@ -218,6 +225,7 @@ impl GrpcCodeModeHost {
&self,
request: proto::WaitRequest,
) -> Result<Response<proto::WaitResponse>, Status> {
let received_at = Instant::now();
let session = self.state.session(&request.session_id)?;
validation::identifier(&request.cell_id, "cell ID")?;
validation::identifier(&request.wait_id, "wait ID")?;
@@ -239,7 +247,10 @@ impl GrpcCodeModeHost {
outcome.map_err(Status::failed_precondition)?
}
};
Ok(Response::new(conversions::wait_response(outcome)))
let outcome = outcome.with_code_mode_host_duration(received_at.elapsed());
let response = conversions::wait_response(outcome)
.map_err(|error| Status::internal(error.to_string()))?;
Ok(Response::new(response))
}
async fn cancel_wait_request(
@@ -257,11 +268,15 @@ impl GrpcCodeModeHost {
&self,
request: proto::TerminateRequest,
) -> Result<Response<proto::WaitResponse>, Status> {
let received_at = Instant::now();
let session = self.state.session(&request.session_id)?;
validation::identifier(&request.cell_id, "cell ID")?;
let _permit = self.state.request_permit()?;
let result = session.terminate(CellId::new(request.cell_id)).await?;
Ok(Response::new(conversions::wait_response(result)))
let outcome = session.terminate(CellId::new(request.cell_id)).await?;
let outcome = outcome.with_code_mode_host_duration(received_at.elapsed());
let response = conversions::wait_response(outcome)
.map_err(|error| Status::internal(error.to_string()))?;
Ok(Response::new(response))
}
}
codex-rs/code-mode-host/src/grpc/service_tests.rs
+10 -1
View File
@@ -91,11 +91,16 @@ async fn execute_stream_starts_immediately_and_wait_preserves_missing_cells() {
request.yield_time_ms = Some(60_000);
let (cell_id, mut execution) = execute_events(&host, request).await;
let yielded = execution.next().await.unwrap().unwrap();
let Some(proto::execute_event::Event::Outcome(yielded_outcome)) = &yielded.event else {
panic!("expected execution outcome");
};
assert_eq!(
execution.next().await.unwrap().unwrap(),
yielded,
proto::ExecuteEvent {
event: Some(proto::execute_event::Event::Outcome(
proto::ExecutionOutcome {
code_mode_host_duration_ns: yielded_outcome.code_mode_host_duration_ns,
cell_id: cell_id.clone(),
content_items: vec![proto::ContentItem {
item: Some(proto::content_item::Item::Text(proto::TextContent {
@@ -119,11 +124,15 @@ async fn execute_stream_starts_immediately_and_wait_preserves_missing_cells() {
.await
.expect("wait for completion")
.into_inner();
let Some(proto::wait_response::State::LiveCell(completed_outcome)) = &completed.state else {
panic!("expected live-cell wait outcome");
};
assert_eq!(
completed,
proto::WaitResponse {
state: Some(proto::wait_response::State::LiveCell(
proto::ExecutionOutcome {
code_mode_host_duration_ns: completed_outcome.code_mode_host_duration_ns,
cell_id: cell_id.clone(),
content_items: vec![proto::ContentItem {
item: Some(proto::content_item::Item::Text(proto::TextContent {
codex-rs/code-mode-host/src/host_tests.rs
+2
View File
@@ -7,6 +7,7 @@ use std::sync::atomic::AtomicBool;
use std::task::Context;
use std::task::Poll;
use std::time::Duration;
use std::time::Instant;
use codex_code_mode_protocol::CodeModeSessionCellExecutionLimits;
use codex_code_mode_protocol::host::Capability;
@@ -594,6 +595,7 @@ async fn active_cell_limit_rejects_execute_without_disconnecting() {
request: execute_request("text(\"hello\");"),
},
CancellationToken::new(),
Instant::now(),
)
.await;
codex-rs/code-mode-host/src/lib.rs
+19 -9
View File
@@ -7,6 +7,7 @@ use std::sync::PoisonError;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use anyhow::Context;
use anyhow::Result;
@@ -27,6 +28,7 @@ use codex_code_mode_protocol::host::RequestId;
use codex_code_mode_protocol::host::SESSION_RESOURCE_LIMITS_CAPABILITY;
use codex_code_mode_protocol::host::SessionId;
use codex_code_mode_protocol::host::SupportedProtocolVersions;
use codex_code_mode_protocol::host::WireWaitOutcome;
use codex_code_mode_runtime::InProcessCodeModeSession;
use tokio::io::AsyncRead;
use tokio::io::AsyncWrite;
@@ -306,6 +308,8 @@ impl HostState {
request_id: RequestId,
request: HostRequest,
) -> Result<(), anyhow::Error> {
// Include host admission and scheduling before the runtime observes the request.
let received_at = Instant::now();
let request_kind = RequestKind::from(&request);
let cancellation = self
.requests
@@ -324,7 +328,7 @@ impl HostState {
let request_task = self.request_tasks.spawn(async move {
let _permit = permit;
state
.handle_request(request_id, request, cancellation)
.handle_request(request_id, request, cancellation, received_at)
.await;
state.finish_request(request_id);
});
@@ -357,6 +361,7 @@ impl HostState {
request_id: RequestId,
request: HostRequest,
cancellation: CancellationToken,
received_at: Instant,
) {
if let HostRequest::Execute { request, .. } = &request {
tracing::Span::current().record("call_id", request.tool_call_id.as_str());
@@ -428,6 +433,7 @@ impl HostState {
request_id,
started,
active_cell_permit,
received_at,
);
let _ = initial_response_sent.await;
}
@@ -445,10 +451,12 @@ impl HostState {
_ = cancellation.cancelled() => {
Err("code-mode request cancelled".to_string())
}
result = session.wait(request.into()) => result.map(|outcome| {
HostResponse::WaitCompleted {
outcome: outcome.into(),
}
result = session.wait(request.into()) => result.and_then(|outcome| {
let outcome = outcome.with_code_mode_host_duration(received_at.elapsed());
Ok(HostResponse::WaitCompleted {
outcome: WireWaitOutcome::try_from(outcome)
.map_err(|error| error.to_string())?,
})
}),
}
}
@@ -461,10 +469,12 @@ impl HostState {
cell_id,
} => {
let result = match self.session(&session_id) {
Ok(session) => session.terminate(cell_id.into()).await.map(|outcome| {
HostResponse::WaitCompleted {
outcome: outcome.into(),
}
Ok(session) => session.terminate(cell_id.into()).await.and_then(|outcome| {
let outcome = outcome.with_code_mode_host_duration(received_at.elapsed());
Ok(HostResponse::WaitCompleted {
outcome: WireWaitOutcome::try_from(outcome)
.map_err(|error| error.to_string())?,
})
}),
Err(err) => Err(err),
};
codex-rs/code-mode-host/src/peer.rs
+32 -9
View File
@@ -5,6 +5,7 @@ use std::sync::Mutex as StdMutex;
use std::sync::PoisonError;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use std::time::Instant;
use codex_code_mode_protocol::CellId;
use codex_code_mode_protocol::StartedCell;
@@ -17,6 +18,7 @@ use codex_code_mode_protocol::host::MAX_PENDING_DELEGATE_CALLS;
use codex_code_mode_protocol::host::RequestId;
use codex_code_mode_protocol::host::SessionId;
use codex_code_mode_protocol::host::WireResult;
use codex_code_mode_protocol::host::WireRuntimeResponse;
use tokio::sync::Mutex;
use tokio::sync::Notify;
use tokio::sync::OwnedSemaphorePermit;
@@ -58,6 +60,11 @@ enum CellMessage {
Closed,
}
struct InitialRequest {
id: RequestId,
received_at: Instant,
}
impl HostPeer {
pub(super) fn new(outgoing_tx: mpsc::Sender<EncodedFrame>) -> Self {
Self {
@@ -97,11 +104,7 @@ impl HostPeer {
}
}
fn initial_response(
&self,
id: RequestId,
result: Result<codex_code_mode_protocol::host::WireRuntimeResponse, String>,
) {
fn initial_response(&self, id: RequestId, result: Result<WireRuntimeResponse, String>) {
let message = HostToClient::InitialResponse {
id,
result: WireResult::from_result(result),
@@ -213,6 +216,7 @@ impl HostPeer {
request_id: RequestId,
started: StartedCell,
active_cell_permit: OwnedSemaphorePermit,
received_at: Instant,
) -> oneshot::Receiver<()> {
let (initial_response_sent_tx, initial_response_sent_rx) = oneshot::channel();
let key = (session_id, started.cell_id.clone());
@@ -242,7 +246,10 @@ impl HostPeer {
drive_cell(
peer,
key,
request_id,
InitialRequest {
id: request_id,
received_at,
},
started,
messages_rx,
initial_response_sent_tx,
@@ -411,7 +418,7 @@ impl HostPeer {
async fn drive_cell(
peer: Arc<HostPeer>,
key: (SessionId, CellId),
request_id: RequestId,
request: InitialRequest,
started: StartedCell,
mut messages_rx: mpsc::Receiver<CellMessage>,
initial_response_sent_tx: oneshot::Sender<()>,
@@ -424,7 +431,15 @@ async fn drive_cell(
tokio::select! {
biased;
result = &mut initial_response => {
peer.initial_response(request_id, result.map(Into::into));
// Freeze the request duration before response conversion or delivery.
let code_mode_host_duration = request.received_at.elapsed();
peer.initial_response(
request.id,
result.and_then(|response| {
let response = response.with_code_mode_host_duration(code_mode_host_duration);
WireRuntimeResponse::try_from(response).map_err(|error| error.to_string())
}),
);
if let Some(initial_response_sent_tx) = initial_response_sent_tx.take() {
let _ = initial_response_sent_tx.send(());
}
@@ -448,7 +463,15 @@ async fn drive_cell(
};
if closed {
peer.initial_response(request_id, initial_response.await.map(Into::into));
let result = initial_response.await;
let code_mode_host_duration = request.received_at.elapsed();
peer.initial_response(
request.id,
result.and_then(|response| {
let response = response.with_code_mode_host_duration(code_mode_host_duration);
WireRuntimeResponse::try_from(response).map_err(|error| error.to_string())
}),
);
if let Some(initial_response_sent_tx) = initial_response_sent_tx.take() {
let _ = initial_response_sent_tx.send(());
}
codex-rs/code-mode-host/src/peer_tests.rs
+3
View File
@@ -1,5 +1,6 @@
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use codex_code_mode_protocol::CellId;
use codex_code_mode_protocol::RuntimeResponse;
@@ -38,11 +39,13 @@ async fn start_cell_reports_when_initial_response_is_enqueued() {
RequestId::new(/*value*/ 1),
started,
active_cell_permit,
Instant::now(),
);
assert_eq!(initial_response_sent.try_recv(), Err(TryRecvError::Empty));
response_tx
.send(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id.clone(),
content_items: Vec::new(),
error_text: None,
codex-rs/code-mode-host/tests/grpc.rs
+191 -108
View File
@@ -126,8 +126,13 @@ fn tool(name: &str) -> ToolDefinition {
}
}
fn text_response(cell: &str, value: &str) -> RuntimeResponse {
fn text_response(
cell: &str,
value: &str,
code_mode_host_duration: Option<Duration>,
) -> RuntimeResponse {
RuntimeResponse::Result {
code_mode_host_duration,
cell_id: cell_id(cell),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: value.to_string(),
@@ -207,9 +212,11 @@ async fn tcp_session_persists_values_and_forwards_tools_notifications_and_closur
.await
.map_err(anyhow::Error::msg)?;
let actual = execute(&session, request(r#"store("key", "persisted");"#)).await?;
assert_eq!(
execute(&session, request(r#"store("key", "persisted");"#)).await?,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: None,
@@ -221,9 +228,10 @@ async fn tcp_session_persists_values_and_forwards_tools_notifications_and_closur
);
callback.tool_call_id = "call-2".to_string();
callback.enabled_tools = vec![tool("echo")];
let actual = execute(&session, callback).await?;
assert_eq!(
execute(&session, callback).await?,
text_response("2", "output")
actual,
text_response("2", "output", actual.code_mode_host_duration())
);
timeout(TEST_TIMEOUT, delegate.notification_delivered.notified())
.await
@@ -346,9 +354,10 @@ async fn cancelling_execution_before_admission_keeps_the_session_usable() -> Res
.await
.context("cancelled execution leaked its remote cell")??;
let actual = execute(&session, request(r#"text("still alive");"#)).await?;
assert_eq!(
execute(&session, request(r#"text("still alive");"#)).await?,
text_response("2", "still alive")
actual,
text_response("2", "still alive", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
Ok(())
@@ -427,9 +436,10 @@ async fn dropping_a_started_cell_off_runtime_terminates_its_buffered_remote_exec
.contains(&abandoned_cell)
);
let actual = execute(&session, request(r#"text("still alive");"#)).await?;
assert_eq!(
execute(&session, request(r#"text("still alive");"#)).await?,
text_response("2", "still alive")
actual,
text_response("2", "still alive", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
Ok(())
@@ -469,9 +479,10 @@ async fn dropping_an_initial_response_terminates_its_pending_remote_execution()
.await
.context("dropping an initial response did not terminate its pending remote execution")?;
let actual = execute(&session, request(r#"text("still alive");"#)).await?;
assert_eq!(
execute(&session, request(r#"text("still alive");"#)).await?,
text_response("2", "still alive")
actual,
text_response("2", "still alive", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
Ok(())
@@ -490,22 +501,29 @@ async fn synchronous_delegate_panics_do_not_orphan_callbacks_or_close_the_sessio
r#"try { await tools.echo({}); text("unexpected"); } catch (_) { text("tool recovered"); }"#,
);
tool_panic.enabled_tools = vec![tool("echo")];
let actual = execute(&session, tool_panic).await?;
assert_eq!(
execute(&session, tool_panic).await?,
text_response("1", "tool recovered")
actual,
text_response("1", "tool recovered", actual.code_mode_host_duration())
);
let actual = execute(
&session,
request(r#"notify("panic"); text("notification recovered");"#),
)
.await?;
assert_eq!(
execute(
&session,
request(r#"notify("panic"); text("notification recovered");"#),
actual,
text_response(
"2",
"notification recovered",
actual.code_mode_host_duration()
)
.await?,
text_response("2", "notification recovered")
);
let actual = execute(&session, request(r#"text("still alive");"#)).await?;
assert_eq!(
execute(&session, request(r#"text("still alive");"#)).await?,
text_response("3", "still alive")
actual,
text_response("3", "still alive", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
@@ -524,13 +542,15 @@ async fn tool_delegate_self_cancellation_returns_an_error_without_hanging() -> R
let mut callback =
request(r#"try { await tools.echo({}); } catch (_) { text("tool recovered"); }"#);
callback.enabled_tools = vec![tool("echo")];
let actual = execute(&session, callback).await?;
assert_eq!(
execute(&session, callback).await?,
text_response("1", "tool recovered")
actual,
text_response("1", "tool recovered", actual.code_mode_host_duration())
);
let actual = execute(&session, request(r#"text("still alive");"#)).await?;
assert_eq!(
execute(&session, request(r#"text("still alive");"#)).await?,
text_response("2", "still alive")
actual,
text_response("2", "still alive", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
@@ -549,12 +569,14 @@ async fn concurrent_wait_rejects_without_displacing_the_active_observer() -> Res
pending.yield_time_ms = Some(/*value*/ 1);
let started = session.execute(pending).await.map_err(anyhow::Error::msg)?;
let running_cell = started.cell_id.clone();
let actual = started
.initial_response()
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
started
.initial_response()
.await
.map_err(anyhow::Error::msg)?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell.clone(),
content_items: Vec::new(),
}
@@ -583,22 +605,26 @@ async fn concurrent_wait_rejects_without_displacing_the_active_observer() -> Res
format!("exec cell {running_cell} already has an active observer")
);
let actual = timeout(TEST_TIMEOUT, first_wait)
.await
.context("active wait was displaced by the rejected observer")??
.map_err(anyhow::Error::msg)?;
assert_eq!(
timeout(TEST_TIMEOUT, first_wait)
.await
.context("active wait was displaced by the rejected observer")??
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell.clone(),
content_items: Vec::new(),
})
);
let actual = session
.terminate(running_cell.clone())
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
session
.terminate(running_cell.clone())
.await
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell,
content_items: Vec::new(),
})
@@ -620,12 +646,14 @@ async fn dropping_a_wait_retires_its_observer_before_the_next_wait() -> Result<(
pending.yield_time_ms = Some(/*value*/ 1);
let started = session.execute(pending).await.map_err(anyhow::Error::msg)?;
let running_cell = started.cell_id.clone();
let actual = started
.initial_response()
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
started
.initial_response()
.await
.map_err(anyhow::Error::msg)?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell.clone(),
content_items: Vec::new(),
}
@@ -642,28 +670,32 @@ async fn dropping_a_wait_retires_its_observer_before_the_next_wait() -> Result<(
first_wait.abort();
let _ = first_wait.await;
let actual = timeout(
TEST_TIMEOUT,
session.wait(WaitRequest {
cell_id: running_cell.clone(),
yield_time_ms: 1,
}),
)
.await
.context("replacement wait did not observe cancellation retirement")?
.map_err(anyhow::Error::msg)?;
assert_eq!(
timeout(
TEST_TIMEOUT,
session.wait(WaitRequest {
cell_id: running_cell.clone(),
yield_time_ms: 1,
}),
)
.await
.context("replacement wait did not observe cancellation retirement")?
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell.clone(),
content_items: Vec::new(),
})
);
let actual = session
.terminate(running_cell.clone())
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
session
.terminate(running_cell.clone())
.await
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell,
content_items: Vec::new(),
})
@@ -683,9 +715,11 @@ async fn dropping_a_session_off_runtime_retires_its_active_cells() -> Result<()>
.map_err(anyhow::Error::msg)?;
let mut pending = request("await new Promise(() => {});");
pending.yield_time_ms = Some(/*value*/ 1);
let actual = execute(&session, pending).await?;
assert_eq!(
execute(&session, pending).await?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -744,9 +778,10 @@ async fn large_unary_tool_completion_does_not_block_an_independent_session() ->
.context("large tool callback did not start")??
.forget();
let actual = execute(&fast_session, request(r#"text("fast-before");"#)).await?;
assert_eq!(
execute(&fast_session, request(r#"text("fast-before");"#)).await?,
text_response("1", "fast-before")
actual,
text_response("1", "fast-before", actual.code_mode_host_duration())
);
delegate.release.add_permits(/*n*/ 1);
@@ -754,12 +789,24 @@ async fn large_unary_tool_completion_does_not_block_an_independent_session() ->
let fast_response = execute(&fast_session, request(r#"text("fast-during");"#));
let (slow_response, fast_response) = tokio::join!(slow_response, fast_response);
assert_eq!(fast_response?, text_response("2", "fast-during"));
let actual = fast_response?;
assert_eq!(
slow_response
.context("large unary tool response did not complete")?
.map_err(anyhow::Error::msg)?,
text_response("1", "8388608")
actual,
text_response("2", "fast-during", actual.code_mode_host_duration())
);
let actual = slow_response
.context("large unary tool response did not complete")?
.map_err(anyhow::Error::msg)?;
assert_eq!(
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "8388608".to_string(),
}],
error_text: None,
}
);
slow_session.shutdown().await.map_err(anyhow::Error::msg)?;
fast_session.shutdown().await.map_err(anyhow::Error::msg)?;
@@ -791,18 +838,27 @@ async fn single_subscription_processes_slow_and_fast_tools_concurrently() -> Res
let mut fast = request(r#"const result = await tools.fast({}); text(result.value);"#);
fast.enabled_tools = vec![tool("fast")];
let actual = execute(&session, fast).await?;
assert_eq!(
execute(&session, fast).await?,
text_response("2", "isolated")
actual,
text_response("2", "isolated", actual.code_mode_host_duration())
);
delegate.release.add_permits(/*n*/ 1);
let actual = timeout(TEST_TIMEOUT, slow_cell.initial_response())
.await
.context("slow tool did not finish")?
.map_err(anyhow::Error::msg)?;
assert_eq!(
timeout(TEST_TIMEOUT, slow_cell.initial_response())
.await
.context("slow tool did not finish")?
.map_err(anyhow::Error::msg)?,
text_response("1", "8388608")
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "8388608".to_string(),
}],
error_text: None,
}
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
Ok(())
@@ -835,50 +891,59 @@ async fn sessions_enforce_independent_yield_limits() -> Result<()> {
let mut pending = request("await new Promise(() => {});");
pending.yield_time_ms = Some(/*value*/ 60_000);
let actual = execute(&limited, pending).await?;
assert_eq!(
execute(&limited, pending).await?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
}
);
let actual = timeout(
TEST_TIMEOUT,
limited.wait(WaitRequest {
cell_id: cell_id("1"),
yield_time_ms: 60_000,
}),
)
.await
.context("session yield limit did not bound an explicit wait")?
.map_err(anyhow::Error::msg)?;
assert_eq!(
timeout(
TEST_TIMEOUT,
limited.wait(WaitRequest {
cell_id: cell_id("1"),
yield_time_ms: 60_000,
}),
)
.await
.context("session yield limit did not bound an explicit wait")?
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
})
);
let actual = execute(
&other,
request(r#"await new Promise(resolve => setTimeout(resolve, 25)); text("isolated");"#),
)
.await?;
assert_eq!(
execute(
&other,
request(r#"await new Promise(resolve => setTimeout(resolve, 25)); text("isolated");"#),
)
.await?,
text_response("1", "isolated")
actual,
text_response("1", "isolated", actual.code_mode_host_duration())
);
let actual = limited
.terminate(cell_id("1"))
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
limited
.terminate(cell_id("1"))
.await
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
})
);
let actual = execute(&limited, request("await new Promise(() => {});")).await?;
assert_eq!(
execute(&limited, request("await new Promise(() => {});")).await?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("2"),
content_items: Vec::new(),
}
@@ -1027,7 +1092,11 @@ async fn cached_session_recovers_after_a_remote_host_restarts() -> Result<()> {
let callback_cell_id = callback_cell_id.clone();
assert_eq!(
callback_response,
text_response(callback_cell_id.as_str(), "output")
text_response(
callback_cell_id.as_str(),
"output",
callback_response.code_mode_host_duration()
)
);
let concurrent_response = concurrent_response?;
let RuntimeResponse::Result {
@@ -1040,7 +1109,11 @@ async fn cached_session_recovers_after_a_remote_host_restarts() -> Result<()> {
let concurrent_cell_id = concurrent_cell_id.clone();
assert_eq!(
concurrent_response,
text_response(concurrent_cell_id.as_str(), "concurrent")
text_response(
concurrent_cell_id.as_str(),
"concurrent",
concurrent_response.code_mode_host_duration()
)
);
let mut replacement_cell_ids = [callback_cell_id.as_str(), concurrent_cell_id.as_str()];
replacement_cell_ids.sort_unstable();
@@ -1075,32 +1148,41 @@ async fn cached_session_recovers_after_a_remote_host_restarts() -> Result<()> {
let started = session.execute(pending).await.map_err(anyhow::Error::msg)?;
let replacement_cell_id = started.cell_id.clone();
assert_eq!(replacement_cell_id, cell_id("g2:3"));
let actual = started
.initial_response()
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
started.initial_response().await,
Ok(RuntimeResponse::Yielded {
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: replacement_cell_id.clone(),
content_items: Vec::new(),
})
}
);
let actual = session
.wait(WaitRequest {
cell_id: replacement_cell_id.clone(),
yield_time_ms: 1,
})
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
session
.wait(WaitRequest {
cell_id: replacement_cell_id.clone(),
yield_time_ms: 1,
})
.await
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: replacement_cell_id.clone(),
content_items: Vec::new(),
})
);
let actual = session
.terminate(replacement_cell_id.clone())
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
session
.terminate(replacement_cell_id.clone())
.await
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: replacement_cell_id,
content_items: Vec::new(),
})
@@ -1140,9 +1222,10 @@ async fn unix_socket_endpoints_execute_code_mode_cells() -> Result<()> {
.create_session(Arc::new(NoopCodeModeSessionDelegate))
.await
.map_err(anyhow::Error::msg)?;
let actual = execute(&session, request(r#"text("unix socket")"#)).await?;
assert_eq!(
execute(&session, request(r#"text("unix socket")"#)).await?,
text_response("1", "unix socket")
actual,
text_response("1", "unix socket", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
}
codex-rs/code-mode-host/tests/grpc_notifications.rs
+45 -23
View File
@@ -102,8 +102,13 @@ fn request(source: &str) -> ExecuteRequest {
}
}
fn text_response(cell: &str, value: &str) -> RuntimeResponse {
fn text_response(
cell: &str,
value: &str,
code_mode_host_duration: Option<Duration>,
) -> RuntimeResponse {
RuntimeResponse::Result {
code_mode_host_duration,
cell_id: cell_id(cell),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: value.to_string(),
@@ -149,11 +154,12 @@ async fn completed_cells_drain_pending_notifications_before_completion() -> Resu
.forget();
assert!(!completion.is_finished());
delegate.release.add_permits(/*n*/ 1);
let actual = timeout(TEST_TIMEOUT, completion)
.await
.context("completed cell did not finish after notification delivery")???;
assert_eq!(
timeout(TEST_TIMEOUT, completion)
.await
.context("completed cell did not finish after notification delivery")???,
text_response("1", "done")
actual,
text_response("1", "done", actual.code_mode_host_duration())
);
timeout(TEST_TIMEOUT, delegate.delivered.acquire())
.await
@@ -165,9 +171,10 @@ async fn completed_cells_drain_pending_notifications_before_completion() -> Resu
.context("completed cell was not retired")??
.forget();
let actual = execute(&session, request(r#"text("still alive");"#)).await?;
assert_eq!(
execute(&session, request(r#"text("still alive");"#)).await?,
text_response("2", "still alive")
actual,
text_response("2", "still alive", actual.code_mode_host_duration())
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
@@ -185,9 +192,11 @@ async fn completed_waits_drain_pending_notifications_before_returning() -> Resul
.map_err(anyhow::Error::msg)?;
let pending = request(r#"yield_control(); notify("notice"); text("done");"#);
let cell = session.execute(pending).await.map_err(anyhow::Error::msg)?;
let actual = cell.initial_response().await.map_err(anyhow::Error::msg)?;
assert_eq!(
cell.initial_response().await.map_err(anyhow::Error::msg)?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -209,11 +218,19 @@ async fn completed_waits_drain_pending_notifications_before_returning() -> Resul
.forget();
assert!(!completion.is_finished());
delegate.release.add_permits(/*n*/ 1);
let actual = timeout(TEST_TIMEOUT, completion)
.await
.context("wait did not finish after notification delivery")???;
assert_eq!(
timeout(TEST_TIMEOUT, completion)
.await
.context("wait did not finish after notification delivery")???,
WaitOutcome::LiveCell(text_response("1", "done"))
actual,
WaitOutcome::LiveCell(RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
}],
error_text: None,
})
);
timeout(TEST_TIMEOUT, delegate.delivered.acquire())
.await
@@ -242,19 +259,23 @@ async fn termination_cancels_pending_notifications() -> Result<()> {
.await
.context("notification did not start")??
.forget();
let actual = cell.initial_response().await.map_err(anyhow::Error::msg)?;
assert_eq!(
cell.initial_response().await.map_err(anyhow::Error::msg)?,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
}
);
let actual = session
.terminate(cell_id("1"))
.await
.map_err(anyhow::Error::msg)?;
assert_eq!(
session
.terminate(cell_id("1"))
.await
.map_err(anyhow::Error::msg)?,
actual,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -282,13 +303,14 @@ async fn oversized_notification_text_is_delivered_unchanged() -> Result<()> {
.await
.map_err(anyhow::Error::msg)?;
let actual = execute(
&session,
request(r#"notify("🦀".repeat(512)); text("done");"#),
)
.await?;
assert_eq!(
execute(
&session,
request(r#"notify("🦀".repeat(512)); text("done");"#),
)
.await?,
text_response("1", "done")
actual,
text_response("1", "done", actual.code_mode_host_duration())
);
timeout(TEST_TIMEOUT, delegate.notification_delivered.notified())
.await
codex-rs/code-mode-host/tests/grpc_timing.rs
+286
View File
@@ -0,0 +1,286 @@
//! Checks that remote timing measures each observation, not client delays or cell lifetime.
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use anyhow::Context;
use anyhow::Result;
use codex_code_mode::CodeModeSessionProvider;
use codex_code_mode::ExecuteRequest;
use codex_code_mode::GrpcCodeModeSessionProvider;
use codex_code_mode::NoopCodeModeSessionDelegate;
use codex_code_mode::ProcessOwnedCodeModeSessionProvider;
use codex_code_mode::RuntimeResponse;
use codex_code_mode::WaitOutcome;
use codex_code_mode::WaitRequest;
use codex_code_mode_protocol::grpc;
use codex_code_mode_protocol::grpc::code_mode_host_client::CodeModeHostClient;
use pretty_assertions::assert_eq;
use tokio::time::sleep;
use tokio::time::timeout;
#[path = "support/host.rs"]
mod host;
#[expect(dead_code, reason = "Other suites use the shared cell_id helper.")]
#[path = "support/recording_delegate.rs"]
mod recording_delegate;
use host::HostHarness;
use recording_delegate::RecordingDelegate;
const TEST_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 20);
/// Successful and failed JavaScript include computation and timers in their host timing.
/// Waiting to read a finished execution must not increase that measurement.
#[tokio::test]
async fn execution_timing_includes_javascript_but_excludes_delayed_reads() -> Result<()> {
let host = HostHarness::start("grpc://127.0.0.1:0").await?;
let mut client = CodeModeHostClient::connect(host.endpoint).await?;
let mut events = client
.open_session(grpc::OpenSessionRequest {
cell_execution_limits: None,
})
.await?
.into_inner();
let opened = timeout(TEST_TIMEOUT, events.message())
.await??
.context("session ended")?;
let Some(grpc::session_event::Event::Opened(opened)) = opened.event else {
anyhow::bail!("expected session opening event");
};
let endings = ["", "throw new Error('timed failure');"];
for (index, suffix) in endings.into_iter().enumerate() {
let observed = Instant::now();
let mut execution = client
.execute(grpc::ExecuteRequest {
session_id: opened.session_id.clone(),
execution_id: format!("execution-{index}"),
tool_call_id: format!("call-{index}"),
source: format!(
"const until = Date.now() + 150; while (Date.now() < until) {{}} \
await new Promise(resolve => setTimeout(resolve, 150)); {suffix}"
),
enabled_tools: Vec::new(),
yield_time_ms: Some(/*value*/ 5_000),
max_output_tokens: Some(/*value*/ 1_000),
})
.await?
.into_inner();
let started = timeout(TEST_TIMEOUT, execution.message())
.await??
.context("execution ended")?;
let Some(grpc::execute_event::Event::Started(started)) = started.event else {
anyhow::bail!("expected execution admission");
};
let closed = timeout(TEST_TIMEOUT, events.message())
.await??
.context("session ended")?;
assert!(
matches!(closed.event, Some(grpc::session_event::Event::CellClosed(closed))
if closed.cell_id == started.cell_id)
);
// CellClosed confirms the JS finished before introducing a client-only delay.
sleep(Duration::from_millis(/*millis*/ 600)).await;
let result = timeout(TEST_TIMEOUT, execution.message())
.await??
.context("execution ended")?;
let Some(grpc::execute_event::Event::Outcome(outcome)) = result.event else {
anyhow::bail!("expected execution outcome");
};
let duration = Duration::from_nanos(outcome.code_mode_host_duration_ns);
assert!(duration >= Duration::from_millis(/*millis*/ 250));
assert!(duration + Duration::from_millis(/*millis*/ 300) <= observed.elapsed());
let Some(grpc::execution_outcome::Outcome::Completed(completed)) = outcome.outcome else {
anyhow::bail!("expected completed execution");
};
if suffix.is_empty() {
assert_eq!(completed.error_text, None);
} else {
assert!(
completed
.error_text
.context("expected JS failure")?
.contains("timed failure")
);
}
}
Ok(())
}
/// The process-owned host freezes successful and failed execution timing before
/// the client reads its initial response, including JavaScript computation and timers.
#[tokio::test]
async fn stdio_execution_timing_includes_javascript_but_excludes_delayed_reads() -> Result<()> {
let provider = ProcessOwnedCodeModeSessionProvider::with_host_program(
codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?,
);
let delegate = Arc::new(RecordingDelegate::default());
let session = provider
.create_session(delegate.clone())
.await
.map_err(anyhow::Error::msg)?;
let endings = ["", "throw new Error('timed failure');"];
for (index, suffix) in endings.into_iter().enumerate() {
let observed = Instant::now();
let started = session
.execute(ExecuteRequest {
tool_call_id: format!("call-{index}"),
source: format!(
"const until = Date.now() + 150; while (Date.now() < until) {{}} \
await new Promise(resolve => setTimeout(resolve, 150)); \
notify('finished'); {suffix}"
),
enabled_tools: Vec::new(),
yield_time_ms: Some(/*value*/ 5_000),
max_output_tokens: Some(/*value*/ 1_000),
})
.await
.map_err(anyhow::Error::msg)?;
timeout(TEST_TIMEOUT, delegate.notification_delivered.notified())
.await
.context("JS did not report completion of its timed work")?;
// Delay only after the host has finished the computation and timer.
sleep(Duration::from_millis(/*millis*/ 600)).await;
let response = timeout(TEST_TIMEOUT, started.initial_response())
.await?
.map_err(anyhow::Error::msg)?;
let duration = response
.code_mode_host_duration()
.context("missing stdio exec timing")?;
assert!(duration >= Duration::from_millis(/*millis*/ 250));
assert!(duration + Duration::from_millis(/*millis*/ 300) <= observed.elapsed());
let RuntimeResponse::Result { error_text, .. } = response else {
anyhow::bail!("expected completed execution");
};
if suffix.is_empty() {
assert_eq!(error_text, None);
} else {
assert!(
error_text
.context("expected JS failure")?
.contains("timed failure")
);
}
}
session.shutdown().await.map_err(anyhow::Error::msg)?;
Ok(())
}
/// Every wait and termination measures its own request, including missing-cell outcomes.
/// Earlier execution and background time must not be charged again by later observations.
#[tokio::test]
async fn observation_timing_excludes_previous_requests_and_background_time() -> Result<()> {
let host = HostHarness::start("grpc://127.0.0.1:0").await?;
let providers: [Arc<dyn CodeModeSessionProvider>; 2] = [
Arc::new(ProcessOwnedCodeModeSessionProvider::with_host_program(
codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?,
)),
Arc::new(GrpcCodeModeSessionProvider::new(host.endpoint)),
];
for provider in providers {
let session = provider
.create_session(Arc::new(NoopCodeModeSessionDelegate))
.await
.map_err(anyhow::Error::msg)?;
let started = session
.execute(ExecuteRequest {
tool_call_id: "call-1".to_string(),
source: "await new Promise(() => {});".to_string(),
enabled_tools: Vec::new(),
yield_time_ms: Some(/*value*/ 200),
max_output_tokens: Some(/*value*/ 1_000),
})
.await
.map_err(anyhow::Error::msg)?;
let cell_id = started.cell_id.clone();
let initial = started
.initial_response()
.await
.map_err(anyhow::Error::msg)?;
let duration = initial
.code_mode_host_duration()
.context("missing exec timing")?;
assert!(duration >= Duration::from_millis(/*millis*/ 150));
let yielded = RuntimeResponse::Yielded {
cell_id: cell_id.clone(),
content_items: Vec::new(),
code_mode_host_duration: Some(duration),
};
assert_eq!(initial, yielded);
for _ in 0..2 {
sleep(Duration::from_millis(/*millis*/ 300)).await;
let observed = Instant::now();
let outcome = session
.wait(WaitRequest {
cell_id: cell_id.clone(),
yield_time_ms: 50,
})
.await
.map_err(anyhow::Error::msg)?;
let duration = outcome
.code_mode_host_duration()
.context("missing wait timing")?;
assert!(duration >= Duration::from_millis(/*millis*/ 40));
assert!(
duration <= observed.elapsed(),
"wait included time before its request: {duration:?}"
);
assert_eq!(
outcome,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
cell_id: cell_id.clone(),
content_items: Vec::new(),
code_mode_host_duration: Some(duration),
})
);
}
let observed = Instant::now();
let terminated = session
.terminate(cell_id.clone())
.await
.map_err(anyhow::Error::msg)?;
let duration = terminated
.code_mode_host_duration()
.context("missing termination timing")?;
assert!(duration <= observed.elapsed());
assert_eq!(
terminated,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
cell_id: cell_id.clone(),
content_items: Vec::new(),
code_mode_host_duration: Some(duration),
})
);
let observed = Instant::now();
let missing = session
.wait(WaitRequest {
cell_id: cell_id.clone(),
yield_time_ms: 50,
})
.await
.map_err(anyhow::Error::msg)?;
let duration = missing
.code_mode_host_duration()
.context("missing absent-cell timing")?;
assert!(duration <= observed.elapsed());
assert_eq!(
missing,
WaitOutcome::MissingCell(RuntimeResponse::Result {
error_text: Some(format!("exec cell {cell_id} not found")),
cell_id,
content_items: Vec::new(),
code_mode_host_duration: Some(duration),
})
);
session.shutdown().await.map_err(anyhow::Error::msg)?;
}
Ok(())
}
codex-rs/code-mode-host/tests/stdio.rs
+91 -54
View File
@@ -256,22 +256,25 @@ async fn session_execution_limits_are_isolated_on_a_shared_process_host() {
assert_eq!(
response,
RuntimeResponse::Yielded {
code_mode_host_duration: response.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
}
);
let actual = tokio::time::timeout(
Duration::from_secs(5),
limited.wait(WaitRequest {
cell_id: cell_id("1"),
yield_time_ms: 60_000,
}),
)
.await
.expect("session limit should bound explicit waits")
.expect("wait for yielded cell");
assert_eq!(
tokio::time::timeout(
Duration::from_secs(5),
limited.wait(WaitRequest {
cell_id: cell_id("1"),
yield_time_ms: 60_000,
}),
)
.await
.expect("session limit should bound explicit waits")
.expect("wait for yielded cell"),
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -299,9 +302,11 @@ async fn remote_session_persists_values_forwards_delegates_and_controls_cells()
.await
.expect("create remote session");
let actual = execute(&session, execute_request(r#"store("key", "persisted");"#)).await;
assert_eq!(
execute(&session, execute_request(r#"store("key", "persisted");"#),).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: None,
@@ -324,9 +329,11 @@ text(result.value);
input_schema: None,
output_schema: None,
}];
let actual = execute(&session, callback_request).await;
assert_eq!(
execute(&session, callback_request).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("2"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "output".to_string(),
@@ -352,32 +359,38 @@ text(result.value);
let mut pending_request = execute_request("await new Promise(() => {});");
pending_request.tool_call_id = "call-3".to_string();
pending_request.yield_time_ms = Some(1);
let actual = execute(&session, pending_request).await;
assert_eq!(
execute(&session, pending_request).await,
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("3"),
content_items: Vec::new(),
}
);
let actual = session
.wait(WaitRequest {
cell_id: cell_id("3"),
yield_time_ms: 1,
})
.await
.expect("wait for cell");
assert_eq!(
session
.wait(WaitRequest {
cell_id: cell_id("3"),
yield_time_ms: 1,
})
.await
.expect("wait for cell"),
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("3"),
content_items: Vec::new(),
})
);
let actual = session
.terminate(cell_id("3"))
.await
.expect("terminate cell");
assert_eq!(
session
.terminate(cell_id("3"))
.await
.expect("terminate cell"),
actual,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("3"),
content_items: Vec::new(),
})
@@ -403,9 +416,11 @@ async fn dropping_long_wait_releases_observer_before_next_wait() {
request.yield_time_ms = Some(1);
let started = session.execute(request).await.expect("start execution");
let running_cell_id = started.cell_id.clone();
let actual = started.initial_response().await.expect("initial response");
assert_eq!(
started.initial_response().await.expect("initial response"),
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell_id.clone(),
content_items: Vec::new(),
}
@@ -425,18 +440,20 @@ async fn dropping_long_wait_releases_observer_before_next_wait() {
first_wait.abort();
let _ = first_wait.await;
let actual = tokio::time::timeout(
Duration::from_secs(2),
session.wait(WaitRequest {
cell_id: running_cell_id.clone(),
yield_time_ms: 1,
}),
)
.await
.expect("second wait timeout")
.expect("second wait");
assert_eq!(
tokio::time::timeout(
Duration::from_secs(2),
session.wait(WaitRequest {
cell_id: running_cell_id.clone(),
yield_time_ms: 1,
})
)
.await
.expect("second wait timeout")
.expect("second wait"),
actual,
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell_id.clone(),
content_items: Vec::new(),
})
@@ -493,9 +510,11 @@ return;
let started = session.execute(request).await.expect("start execution");
let running_cell_id = started.cell_id.clone();
let actual = started.initial_response().await.expect("initial response");
assert_eq!(
started.initial_response().await.expect("initial response"),
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell_id.clone(),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "hello world".to_string(),
@@ -544,12 +563,14 @@ return;
CallbackEvent::CellClosed(running_cell_id.clone()),
]
);
let actual = wait_task
.await
.expect("wait task")
.expect("wait for terminal response");
assert_eq!(
wait_task
.await
.expect("wait task")
.expect("wait for terminal response"),
actual,
WaitOutcome::LiveCell(RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: running_cell_id,
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "hello".to_string(),
@@ -579,9 +600,11 @@ async fn oversized_execute_request_does_not_close_the_shared_host() {
"unexpected error: {error}"
);
let actual = execute(&session, execute_request(r#"text("still alive");"#)).await;
assert_eq!(
execute(&session, execute_request(r#"text("still alive");"#)).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "still alive".to_string(),
@@ -621,9 +644,11 @@ try {{
));
oversized_argument.enabled_tools = vec![tool("big_argument")];
oversized_argument.yield_time_ms = Some(60_000);
let actual = execute_to_terminal(&session, oversized_argument).await;
assert_eq!(
execute_to_terminal(&session, oversized_argument).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "argument rejected".to_string(),
@@ -643,9 +668,11 @@ try {
);
oversized_result.enabled_tools = vec![tool("big_result")];
oversized_result.yield_time_ms = Some(60_000);
let actual = execute_to_terminal(&session, oversized_result).await;
assert_eq!(
execute_to_terminal(&session, oversized_result).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("2"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "result rejected".to_string(),
@@ -654,9 +681,11 @@ try {
}
);
let actual = execute(&session, execute_request(r#"text("still alive");"#)).await;
assert_eq!(
execute(&session, execute_request(r#"text("still alive");"#)).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("3"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "still alive".to_string(),
@@ -691,9 +720,11 @@ async fn oversized_initial_response_does_not_close_the_shared_host() {
"unexpected error: {error}"
);
let actual = execute(&session, execute_request(r#"text("still alive");"#)).await;
assert_eq!(
execute(&session, execute_request(r#"text("still alive");"#)).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("2"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "still alive".to_string(),
@@ -755,12 +786,14 @@ async fn child_process_loss_cleans_up_and_rebuilds_the_shared_host() {
.await
.expect("start first cell");
let cell_a = started_a.cell_id.clone();
let actual = started_a
.initial_response()
.await
.expect("first initial response");
assert_eq!(
started_a
.initial_response()
.await
.expect("first initial response"),
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_a.clone(),
content_items: Vec::new(),
}
@@ -777,12 +810,14 @@ async fn child_process_loss_cleans_up_and_rebuilds_the_shared_host() {
.await
.expect("start second cell");
let cell_b = started_b.cell_id.clone();
let actual = started_b
.initial_response()
.await
.expect("second initial response");
assert_eq!(
started_b
.initial_response()
.await
.expect("second initial response"),
actual,
RuntimeResponse::Yielded {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_b.clone(),
content_items: Vec::new(),
}
@@ -864,9 +899,11 @@ async fn child_process_loss_cleans_up_and_rebuilds_the_shared_host() {
.await
.expect("unrelated session cleanup timeout");
let actual = execute(&session_b, execute_request(r#"text("replacement");"#)).await;
assert_eq!(
execute(&session_b, execute_request(r#"text("replacement");"#)).await,
actual,
RuntimeResponse::Result {
code_mode_host_duration: actual.code_mode_host_duration(),
cell_id: cell_id("g2:1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "replacement".to_string(),
codex-rs/code-mode-protocol/src/grpc/codex.code_mode.v1.proto
+6
View File
@@ -198,6 +198,12 @@ message ExecutionOutcome {
string cell_id = 1;
repeated ContentItem content_items = 2;
// Elapsed monotonic time from receipt of this Execute, Wait, or Terminate
// request until its outcome is ready, before serialization and delivery.
// Includes nested tool waits, but not background time between requests.
// Always supplied by the host; zero is a valid measurement.
uint64 code_mode_host_duration_ns = 6;
oneof outcome {
ExecutionYielded yielded = 3;
ExecutionTerminated terminated = 4;
codex-rs/code-mode-protocol/src/host/host_tests.rs
+11 -1
View File
@@ -463,6 +463,7 @@ fn host_to_client_v1_variants_are_pinned() {
request_id(/*value*/ 3),
HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Yielded {
code_mode_host_duration_ns: 0,
cell_id: cell_id("cell-1"),
content_items: content_items(),
}),
@@ -474,6 +475,7 @@ fn host_to_client_v1_variants_are_pinned() {
"Yielded": {
"cell_id": "cell-1",
"content_items": content_items_json(),
"code_mode_host_duration_ns": 0,
},
},
},
@@ -483,6 +485,7 @@ fn host_to_client_v1_variants_are_pinned() {
request_id(/*value*/ 4),
HostResponse::WaitCompleted {
outcome: WireWaitOutcome::MissingCell(WireRuntimeResponse::Result {
code_mode_host_duration_ns: 0,
cell_id: cell_id("missing-cell"),
content_items: Vec::new(),
error_text: Some("cell not found".to_string()),
@@ -496,6 +499,7 @@ fn host_to_client_v1_variants_are_pinned() {
"cell_id": "missing-cell",
"content_items": [],
"error_text": "cell not found",
"code_mode_host_duration_ns": 0,
},
},
},
@@ -540,6 +544,7 @@ fn host_to_client_v1_variants_are_pinned() {
id: request_id(/*value*/ 7),
result: WireResult::Ok {
value: WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: cell_id("cell-1"),
content_items: Vec::new(),
},
@@ -551,7 +556,11 @@ fn host_to_client_v1_variants_are_pinned() {
"result": {
"status": "ok",
"value": {
"Terminated": { "cell_id": "cell-1", "content_items": [] },
"Terminated": {
"cell_id": "cell-1",
"content_items": [],
"code_mode_host_duration_ns": 0,
},
},
},
}),
@@ -798,6 +807,7 @@ fn every_nested_v1_object_rejects_unknown_fields() {
"Yielded": {
"cell_id": "cell-1",
"content_items": [],
"code_mode_host_duration_ns": 0,
"unexpected": true,
},
}))
codex-rs/code-mode-protocol/src/host/payload.rs
+63 -23
View File
@@ -1,4 +1,5 @@
use std::num::TryFromIntError;
use std::time::Duration;
use codex_protocol::ToolName;
use serde::Deserialize;
@@ -12,6 +13,7 @@ use crate::CodeModeToolKind;
use crate::ExecuteRequest;
use crate::FunctionCallOutputContentItem;
use crate::ImageDetail;
use crate::MissingCodeModeHostDuration;
use crate::RuntimeResponse;
use crate::ToolDefinition;
use crate::WaitOutcome;
@@ -312,51 +314,81 @@ impl From<WireContentItem> for FunctionCallOutputContentItem {
}
/// Runtime output returned over the V1 host connection.
///
/// Host time is required and covers this request, not the cell lifetime. The
/// app-server and host run at the same version; no negotiation is needed.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub enum WireRuntimeResponse {
Yielded {
cell_id: WireCellId,
content_items: Vec<WireContentItem>,
code_mode_host_duration_ns: u64,
},
Terminated {
cell_id: WireCellId,
content_items: Vec<WireContentItem>,
code_mode_host_duration_ns: u64,
},
Result {
cell_id: WireCellId,
content_items: Vec<WireContentItem>,
error_text: Option<String>,
code_mode_host_duration_ns: u64,
},
}
impl From<RuntimeResponse> for WireRuntimeResponse {
fn from(value: RuntimeResponse) -> Self {
match value {
impl TryFrom<RuntimeResponse> for WireRuntimeResponse {
type Error = MissingCodeModeHostDuration;
/// Preserves the response's timing; the host handler must record it first.
fn try_from(value: RuntimeResponse) -> Result<Self, Self::Error> {
Ok(match value {
RuntimeResponse::Yielded {
cell_id,
content_items,
} => Self::Yielded {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
},
code_mode_host_duration,
} => {
let code_mode_host_duration =
code_mode_host_duration.ok_or(MissingCodeModeHostDuration)?;
Self::Yielded {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
code_mode_host_duration_ns: u64::try_from(code_mode_host_duration.as_nanos())
.unwrap_or(u64::MAX),
}
}
RuntimeResponse::Terminated {
cell_id,
content_items,
} => Self::Terminated {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
},
code_mode_host_duration,
} => {
let code_mode_host_duration =
code_mode_host_duration.ok_or(MissingCodeModeHostDuration)?;
Self::Terminated {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
code_mode_host_duration_ns: u64::try_from(code_mode_host_duration.as_nanos())
.unwrap_or(u64::MAX),
}
}
RuntimeResponse::Result {
cell_id,
content_items,
error_text,
} => Self::Result {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
error_text,
},
}
code_mode_host_duration,
} => {
let code_mode_host_duration =
code_mode_host_duration.ok_or(MissingCodeModeHostDuration)?;
Self::Result {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
error_text,
code_mode_host_duration_ns: u64::try_from(code_mode_host_duration.as_nanos())
.unwrap_or(u64::MAX),
}
}
})
}
}
@@ -366,25 +398,31 @@ impl From<WireRuntimeResponse> for RuntimeResponse {
WireRuntimeResponse::Yielded {
cell_id,
content_items,
code_mode_host_duration_ns,
} => Self::Yielded {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
code_mode_host_duration: Some(Duration::from_nanos(code_mode_host_duration_ns)),
},
WireRuntimeResponse::Terminated {
cell_id,
content_items,
code_mode_host_duration_ns,
} => Self::Terminated {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
code_mode_host_duration: Some(Duration::from_nanos(code_mode_host_duration_ns)),
},
WireRuntimeResponse::Result {
cell_id,
content_items,
error_text,
code_mode_host_duration_ns,
} => Self::Result {
cell_id: cell_id.into(),
content_items: content_items.into_iter().map(Into::into).collect(),
error_text,
code_mode_host_duration: Some(Duration::from_nanos(code_mode_host_duration_ns)),
},
}
}
@@ -398,12 +436,14 @@ pub enum WireWaitOutcome {
MissingCell(WireRuntimeResponse),
}
impl From<WaitOutcome> for WireWaitOutcome {
fn from(value: WaitOutcome) -> Self {
match value {
WaitOutcome::LiveCell(response) => Self::LiveCell(response.into()),
WaitOutcome::MissingCell(response) => Self::MissingCell(response.into()),
}
impl TryFrom<WaitOutcome> for WireWaitOutcome {
type Error = MissingCodeModeHostDuration;
fn try_from(value: WaitOutcome) -> Result<Self, Self::Error> {
Ok(match value {
WaitOutcome::LiveCell(response) => Self::LiveCell(response.try_into()?),
WaitOutcome::MissingCell(response) => Self::MissingCell(response.try_into()?),
})
}
}
codex-rs/code-mode-protocol/src/lib.rs
+1
View File
@@ -30,6 +30,7 @@ pub use runtime::DEFAULT_MAX_OUTPUT_TOKENS_PER_EXEC_CALL;
pub use runtime::DEFAULT_WAIT_YIELD_TIME_MS;
pub use runtime::ExecuteRequest;
pub use runtime::ExecuteToPendingOutcome;
pub use runtime::MissingCodeModeHostDuration;
pub use runtime::RuntimeResponse;
pub use runtime::WaitOutcome;
pub use runtime::WaitRequest;
codex-rs/code-mode-protocol/src/runtime.rs
+94
View File
@@ -1,3 +1,7 @@
use std::error::Error;
use std::fmt;
use std::time::Duration;
use codex_protocol::ToolName;
use serde::Deserialize;
use serde::Serialize;
@@ -38,6 +42,29 @@ pub enum WaitOutcome {
MissingCell(RuntimeResponse),
}
impl WaitOutcome {
/// Returns timing for this wait or termination request, when supplied by its host.
pub fn code_mode_host_duration(&self) -> Option<Duration> {
match self {
Self::LiveCell(response) | Self::MissingCell(response) => {
response.code_mode_host_duration()
}
}
}
/// Records the enclosing host request's duration before wire conversion.
pub fn with_code_mode_host_duration(self, code_mode_host_duration: Duration) -> Self {
match self {
Self::LiveCell(response) => {
Self::LiveCell(response.with_code_mode_host_duration(code_mode_host_duration))
}
Self::MissingCell(response) => {
Self::MissingCell(response.with_code_mode_host_duration(code_mode_host_duration))
}
}
}
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
pub enum ExecuteToPendingOutcome {
Pending {
@@ -62,23 +89,86 @@ impl From<WaitOutcome> for RuntimeResponse {
}
}
/// Runtime output with optional timing for the host request that observed it.
///
/// The JavaScript session returns untimed output. The host handler records its
/// complete request duration before conversion; wire conversions preserve that
/// field and reject untimed output. Decoded host responses always contain timing,
/// including measured zero. Raw traces also retain timing when present.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub enum RuntimeResponse {
Yielded {
cell_id: CellId,
content_items: Vec<FunctionCallOutputContentItem>,
#[serde(skip_serializing_if = "Option::is_none")]
code_mode_host_duration: Option<Duration>,
},
Terminated {
cell_id: CellId,
content_items: Vec<FunctionCallOutputContentItem>,
#[serde(skip_serializing_if = "Option::is_none")]
code_mode_host_duration: Option<Duration>,
},
Result {
cell_id: CellId,
content_items: Vec<FunctionCallOutputContentItem>,
error_text: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
code_mode_host_duration: Option<Duration>,
},
}
impl RuntimeResponse {
/// Returns timing for this observation, excluding background work between requests.
pub fn code_mode_host_duration(&self) -> Option<Duration> {
match self {
Self::Yielded {
code_mode_host_duration,
..
}
| Self::Terminated {
code_mode_host_duration,
..
}
| Self::Result {
code_mode_host_duration,
..
} => *code_mode_host_duration,
}
}
/// Records the enclosing host request's duration before wire conversion.
pub fn with_code_mode_host_duration(mut self, code_mode_host_duration: Duration) -> Self {
match &mut self {
Self::Yielded {
code_mode_host_duration: value,
..
}
| Self::Terminated {
code_mode_host_duration: value,
..
}
| Self::Result {
code_mode_host_duration: value,
..
} => *value = Some(code_mode_host_duration),
}
self
}
}
/// An untimed runtime response cannot be encoded for delivery to a host client.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct MissingCodeModeHostDuration;
impl fmt::Display for MissingCodeModeHostDuration {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("code-mode response is missing host duration")
}
}
impl Error for MissingCodeModeHostDuration {}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct CodeModeNestedToolCall {
pub cell_id: CellId,
@@ -87,3 +177,7 @@ pub struct CodeModeNestedToolCall {
pub tool_kind: CodeModeToolKind,
pub input: Option<JsonValue>,
}
#[cfg(test)]
#[path = "runtime_tests.rs"]
mod tests;
codex-rs/code-mode-protocol/src/runtime_tests.rs
+107
View File
@@ -0,0 +1,107 @@
//! Regression coverage for timing metadata at serialization boundaries.
use std::time::Duration;
use pretty_assertions::assert_eq;
use super::MissingCodeModeHostDuration;
use super::RuntimeResponse;
use super::WaitOutcome;
use crate::CellId;
use crate::FunctionCallOutputContentItem;
use crate::host::WireRuntimeResponse;
use crate::host::WireWaitOutcome;
/// Raw runtime serialization preserves absent timing; stdio always carries a
/// measured duration, including zero, without changing the response payload.
#[test]
fn code_mode_host_duration_survives_runtime_and_stdio_serialization() {
let content_items = vec![FunctionCallOutputContentItem::InputText {
text: "output".to_string(),
}];
for response in [
RuntimeResponse::Yielded {
cell_id: CellId::new("yielded-cell".to_string()),
content_items: content_items.clone(),
code_mode_host_duration: None,
},
RuntimeResponse::Terminated {
cell_id: CellId::new("terminated-cell".to_string()),
content_items: content_items.clone(),
code_mode_host_duration: None,
},
RuntimeResponse::Result {
cell_id: CellId::new("completed-cell".to_string()),
content_items,
error_text: Some("execution failed".to_string()),
code_mode_host_duration: None,
},
] {
for duration in [
None,
Some(Duration::ZERO),
Some(Duration::from_nanos(/*nanos*/ 1_234_567_890)),
Some(Duration::from_nanos(u64::MAX)),
] {
let mut expected = response.clone();
match &mut expected {
RuntimeResponse::Yielded {
code_mode_host_duration,
..
}
| RuntimeResponse::Terminated {
code_mode_host_duration,
..
}
| RuntimeResponse::Result {
code_mode_host_duration,
..
} => *code_mode_host_duration = duration,
}
let payload = serde_json::to_value(&expected).expect("serialize response");
assert_eq!(
serde_json::from_value::<RuntimeResponse>(payload).expect("deserialize response"),
expected
);
if duration.is_some() {
let wire_payload = serde_json::to_value(
WireRuntimeResponse::try_from(expected.clone()).expect("timed response"),
)
.expect("serialize response over stdio");
assert_eq!(
RuntimeResponse::from(
serde_json::from_value::<WireRuntimeResponse>(wire_payload)
.expect("deserialize stdio response")
),
expected
);
}
}
}
}
/// Encoding must not turn a missing request measurement into measured zero,
/// including when no live cell remains to supply output.
#[test]
fn stdio_encoding_rejects_untimed_runtime_output() {
let response = RuntimeResponse::Terminated {
cell_id: CellId::new("cell".to_string()),
content_items: Vec::new(),
code_mode_host_duration: None,
};
assert_eq!(
WireRuntimeResponse::try_from(response.clone()),
Err(MissingCodeModeHostDuration)
);
for outcome in [
WaitOutcome::LiveCell(response.clone()),
WaitOutcome::MissingCell(response),
] {
assert_eq!(
WireWaitOutcome::try_from(outcome),
Err(MissingCodeModeHostDuration)
);
}
}
codex-rs/code-mode-runtime/src/service.rs
+4
View File
@@ -347,6 +347,7 @@ fn runtime_response(
runtime::CellEvent::Yielded { content_items } => Ok(RuntimeResponse::Yielded {
cell_id: cell_id.clone(),
content_items: content_items.into_iter().map(output_item).collect(),
code_mode_host_duration: None,
}),
runtime::CellEvent::Completed {
content_items,
@@ -355,10 +356,12 @@ fn runtime_response(
cell_id: cell_id.clone(),
content_items: content_items.into_iter().map(output_item).collect(),
error_text,
code_mode_host_duration: None,
}),
runtime::CellEvent::Terminated { content_items } => Ok(RuntimeResponse::Terminated {
cell_id: cell_id.clone(),
content_items: content_items.into_iter().map(output_item).collect(),
code_mode_host_duration: None,
}),
runtime::CellEvent::Pending { .. } => {
Err("cell returned a pending frontier unexpectedly".to_string())
@@ -391,6 +394,7 @@ fn missing_cell_response(cell_id: CellId) -> RuntimeResponse {
error_text: Some(format!("exec cell {cell_id} not found")),
cell_id,
content_items: Vec::new(),
code_mode_host_duration: None,
}
}
codex-rs/code-mode-runtime/src/service_contract_tests.rs
+12
View File
@@ -199,6 +199,7 @@ async fn yields_and_resumes() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "before".to_string(),
@@ -214,6 +215,7 @@ async fn yields_and_resumes() {
.await
.unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "after".to_string(),
@@ -261,6 +263,7 @@ text("after");
.unwrap(),
WaitToPendingOutcome::LiveCell(ExecuteToPendingOutcome::Completed(
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "after".to_string(),
@@ -284,6 +287,7 @@ async fn observed_natural_completion_wins_over_termination() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -318,6 +322,7 @@ async fn observed_natural_completion_wins_over_termination() {
assert_eq!(
service.terminate(cell_id("1")).await.unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
@@ -345,6 +350,7 @@ async fn termination_cancels_pending_callbacks_before_responding() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -352,6 +358,7 @@ async fn termination_cancels_pending_callbacks_before_responding() {
assert_eq!(
service.terminate(cell_id("1")).await.unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -415,6 +422,7 @@ async fn repeated_termination_is_rejected_while_callback_cleanup_is_pending() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -438,6 +446,7 @@ async fn repeated_termination_is_rejected_while_callback_cleanup_is_pending() {
assert_eq!(
first_termination.await.unwrap().unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -459,6 +468,7 @@ async fn second_observer_is_rejected_without_displacing_the_first() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -482,6 +492,7 @@ async fn second_observer_is_rejected_without_displacing_the_first() {
);
let terminated = RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
};
@@ -513,6 +524,7 @@ async fn natural_completion_cleans_up_callbacks_before_responding() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
codex-rs/code-mode-runtime/src/service_tests.rs
+37
View File
@@ -80,6 +80,7 @@ async fn execute_waits_for_nested_tool_during_yield_grace() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
@@ -118,6 +119,7 @@ async fn execute_and_wait_clamp_yield_grace_without_stopping_the_cell() {
assert_eq!(
initial_response.await.unwrap().unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -136,6 +138,7 @@ async fn execute_and_wait_clamp_yield_grace_without_stopping_the_cell() {
assert_eq!(
wait_response.await.unwrap().unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -153,6 +156,7 @@ async fn execute_and_wait_clamp_yield_grace_without_stopping_the_cell() {
assert_eq!(
completion.await.unwrap().unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
@@ -198,6 +202,7 @@ async fn wait_waits_for_nested_tool_during_yield_grace() {
assert_eq!(
response.unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
@@ -242,6 +247,7 @@ async fn zero_yield_limit_is_immediate_and_scoped_to_its_session() {
assert_eq!(
zero_response.await.unwrap().unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -258,6 +264,7 @@ async fn zero_yield_limit_is_immediate_and_scoped_to_its_session() {
assert_eq!(
zero_wait.await.unwrap().unwrap(),
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -268,6 +275,7 @@ async fn zero_yield_limit_is_immediate_and_scoped_to_its_session() {
assert_eq!(
limited_response.await.unwrap().unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -389,6 +397,7 @@ async fn synchronous_exit_returns_successfully() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "before".to_string(),
@@ -435,6 +444,7 @@ async fn stored_values_are_shared_between_cells_but_not_sessions() {
assert_eq!(
write_response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: None,
@@ -443,6 +453,7 @@ async fn stored_values_are_shared_between_cells_but_not_sessions() {
assert_eq!(
same_session,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("2"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "visible".to_string(),
@@ -453,6 +464,7 @@ async fn stored_values_are_shared_between_cells_but_not_sessions() {
assert_eq!(
other_session,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "undefined".to_string(),
@@ -476,6 +488,7 @@ async fn shutdown_interrupts_cpu_bound_cells() {
assert_eq!(
cell.initial_response().await.unwrap(),
RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
}
@@ -517,6 +530,7 @@ async fn execute_to_pending_returns_completed_for_synchronous_results() {
assert_eq!(
response,
ExecuteToPendingOutcome::Completed(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
@@ -558,6 +572,7 @@ async fn execute_to_pending_returns_once_the_runtime_is_quiescent() {
assert_eq!(
termination,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -598,6 +613,7 @@ await Promise.all([
assert_eq!(
termination,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -662,6 +678,7 @@ await Promise.all([
assert_eq!(
termination,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -725,6 +742,7 @@ await new Promise(() => {});
assert_eq!(
termination,
WaitOutcome::LiveCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
})
@@ -775,6 +793,7 @@ text("done");
resumed_response,
WaitToPendingOutcome::LiveCell(ExecuteToPendingOutcome::Completed(
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "done".to_string(),
@@ -920,6 +939,7 @@ async fn v8_console_is_not_exposed_on_global_this() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "false".to_string(),
@@ -960,6 +980,7 @@ text(value);
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "jeudi 2 janvier \u{e0} 03:04:05".to_string(),
@@ -999,6 +1020,7 @@ text(formatter.format(new Date("2025-01-02T03:04:05Z")));
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "jeudi 2 janvier \u{e0} 03:04:05".to_string(),
@@ -1034,6 +1056,7 @@ text(JSON.stringify(returnsUndefined));
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![
FunctionCallOutputContentItem::InputText {
@@ -1072,6 +1095,7 @@ async fn text_helper_serializes_objects() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: r#"{"json":true}"#.to_string(),
@@ -1115,6 +1139,7 @@ text(circular);
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: Some(error_text),
@@ -1149,6 +1174,7 @@ audio({
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![
FunctionCallOutputContentItem::InputAudio {
@@ -1184,6 +1210,7 @@ async fn audio_helper_rejects_non_data_urls() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: Some(
@@ -1220,6 +1247,7 @@ image({
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==".to_string(),
@@ -1253,6 +1281,7 @@ generatedImage({
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![
FunctionCallOutputContentItem::InputImage {
@@ -1294,6 +1323,7 @@ image(
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,AAA".to_string(),
@@ -1332,6 +1362,7 @@ image(
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==".to_string(),
@@ -1365,6 +1396,7 @@ image({
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: vec![FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,AAA".to_string(),
@@ -1400,6 +1432,7 @@ async fn image_helpers_reject_remote_urls() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: Some(
@@ -1434,6 +1467,7 @@ async fn image_helpers_reject_invalid_image_outputs() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: Some(
@@ -1468,6 +1502,7 @@ image({
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: Some("image detail must be one of: auto, low, high, original".to_string()),
@@ -1505,6 +1540,7 @@ image({
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: Some(
@@ -1529,6 +1565,7 @@ async fn wait_reports_missing_cell_separately_from_runtime_results() {
assert_eq!(
response,
WaitOutcome::MissingCell(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: cell_id("missing"),
content_items: Vec::new(),
error_text: Some("exec cell missing not found".to_string()),
codex-rs/code-mode-runtime/tests/jit.rs
+1
View File
@@ -29,6 +29,7 @@ async fn code_mode_runs_with_jit_disabled() {
assert_eq!(
response,
RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id,
content_items: Vec::new(),
error_text: None,
codex-rs/code-mode/src/grpc_session/conversion.rs
+15 -8
View File
@@ -1,3 +1,5 @@
use std::time::Duration;
use codex_code_mode_protocol::CellId;
use codex_code_mode_protocol::CodeModeNestedToolCall;
use codex_code_mode_protocol::CodeModeToolKind;
@@ -89,6 +91,7 @@ pub(super) fn tool_call(call: grpc::ToolCall) -> Result<CodeModeNestedToolCall,
}
pub(super) fn runtime_response(outcome: grpc::ExecutionOutcome) -> Result<RuntimeResponse, String> {
let code_mode_host_duration = Some(Duration::from_nanos(outcome.code_mode_host_duration_ns));
super::validate_identifier(&outcome.cell_id, "cell ID")?;
let cell_id = CellId::new(outcome.cell_id);
let content_items = outcome
@@ -96,24 +99,28 @@ pub(super) fn runtime_response(outcome: grpc::ExecutionOutcome) -> Result<Runtim
.into_iter()
.map(content_item)
.collect::<Result<Vec<_>, _>>()?;
match outcome
let response = match outcome
.outcome
.ok_or_else(|| "code-mode execution omitted its outcome".to_string())?
{
grpc::execution_outcome::Outcome::Yielded(_) => Ok(RuntimeResponse::Yielded {
grpc::execution_outcome::Outcome::Yielded(_) => RuntimeResponse::Yielded {
cell_id,
content_items,
}),
grpc::execution_outcome::Outcome::Terminated(_) => Ok(RuntimeResponse::Terminated {
code_mode_host_duration,
},
grpc::execution_outcome::Outcome::Terminated(_) => RuntimeResponse::Terminated {
cell_id,
content_items,
}),
grpc::execution_outcome::Outcome::Completed(completed) => Ok(RuntimeResponse::Result {
code_mode_host_duration,
},
grpc::execution_outcome::Outcome::Completed(completed) => RuntimeResponse::Result {
cell_id,
content_items,
error_text: completed.error_text,
}),
}
code_mode_host_duration,
},
};
Ok(response)
}
pub(super) fn wait_outcome(response: grpc::WaitResponse) -> Result<WaitOutcome, String> {
codex-rs/code-mode/src/grpc_session/conversion_tests.rs
+35
View File
@@ -11,6 +11,7 @@ use codex_code_mode_protocol::grpc;
use codex_protocol::ToolName;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::time::Duration;
use super::execute_request;
use super::runtime_response;
@@ -91,6 +92,7 @@ fn tool_call_decodes_structured_input_and_namespace() {
#[test]
fn runtime_response_decodes_mixed_content_items() {
let outcome = grpc::ExecutionOutcome {
code_mode_host_duration_ns: 0,
cell_id: "cell".to_string(),
content_items: vec![
grpc::ContentItem {
@@ -120,6 +122,7 @@ fn runtime_response_decodes_mixed_content_items() {
assert_eq!(
runtime_response(outcome),
Ok(RuntimeResponse::Result {
code_mode_host_duration: Some(Duration::ZERO),
cell_id: CellId::new("cell".to_string()),
content_items: vec![
FunctionCallOutputContentItem::InputText {
@@ -143,6 +146,7 @@ fn wait_outcome_preserves_missing_cell_state() {
let response = grpc::WaitResponse {
state: Some(grpc::wait_response::State::MissingCell(
grpc::ExecutionOutcome {
code_mode_host_duration_ns: 0,
cell_id: "missing".to_string(),
content_items: Vec::new(),
outcome: Some(grpc::execution_outcome::Outcome::Terminated(
@@ -155,6 +159,7 @@ fn wait_outcome_preserves_missing_cell_state() {
assert_eq!(
wait_outcome(response),
Ok(WaitOutcome::MissingCell(RuntimeResponse::Terminated {
code_mode_host_duration: Some(Duration::ZERO),
cell_id: CellId::new("missing".to_string()),
content_items: Vec::new(),
}))
@@ -164,6 +169,7 @@ fn wait_outcome_preserves_missing_cell_state() {
#[test]
fn oversized_response_cell_ids_are_rejected() {
let response = grpc::ExecutionOutcome {
code_mode_host_duration_ns: 0,
cell_id: "x".repeat(grpc::MAX_IDENTIFIER_BYTES + 1),
content_items: Vec::new(),
outcome: Some(grpc::execution_outcome::Outcome::Yielded(
@@ -183,6 +189,7 @@ fn oversized_response_cell_ids_are_rejected() {
#[test]
fn invalid_output_enums_and_missing_oneofs_are_rejected() {
let invalid_image = grpc::ExecutionOutcome {
code_mode_host_duration_ns: 0,
cell_id: "cell".to_string(),
content_items: vec![grpc::ContentItem {
item: Some(grpc::content_item::Item::Image(grpc::ImageContent {
@@ -198,3 +205,31 @@ fn invalid_output_enums_and_missing_oneofs_are_rejected() {
assert!(runtime_response(invalid_image).is_err());
assert!(wait_outcome(grpc::WaitResponse { state: None }).is_err());
}
/// Zero is a valid measurement, and decoding must not round nanoseconds to
/// milliseconds even at the limits of the wire representation.
#[test]
fn host_timing_preserves_zero_and_nanosecond_precision() {
for code_mode_host_duration_ns in [0, 1_234_567, u64::MAX] {
let outcome = grpc::ExecutionOutcome {
cell_id: "cell".to_string(),
content_items: Vec::new(),
code_mode_host_duration_ns,
outcome: Some(grpc::execution_outcome::Outcome::Yielded(
grpc::ExecutionYielded {},
)),
};
let expected = RuntimeResponse::Yielded {
cell_id: CellId::new("cell".to_string()),
content_items: Vec::new(),
code_mode_host_duration: Some(Duration::from_nanos(code_mode_host_duration_ns)),
};
assert_eq!(runtime_response(outcome.clone()).as_ref(), Ok(&expected));
assert_eq!(
wait_outcome(grpc::WaitResponse {
state: Some(grpc::wait_response::State::LiveCell(outcome)),
}),
Ok(WaitOutcome::LiveCell(expected))
);
}
}
codex-rs/code-mode/src/grpc_session/generation.rs
+6
View File
@@ -85,25 +85,31 @@ fn public_runtime_response(generation: u64, response: RuntimeResponse) -> Runtim
RuntimeResponse::Yielded {
cell_id,
content_items,
code_mode_host_duration,
} => RuntimeResponse::Yielded {
cell_id: public_cell_id(generation, &cell_id),
content_items,
code_mode_host_duration,
},
RuntimeResponse::Terminated {
cell_id,
content_items,
code_mode_host_duration,
} => RuntimeResponse::Terminated {
cell_id: public_cell_id(generation, &cell_id),
content_items,
code_mode_host_duration,
},
RuntimeResponse::Result {
cell_id,
content_items,
error_text,
code_mode_host_duration,
} => RuntimeResponse::Result {
cell_id: public_cell_id(generation, &cell_id),
content_items,
error_text,
code_mode_host_duration,
},
}
}
codex-rs/code-mode/src/grpc_session/generation_tests.rs
+38
View File
@@ -2,6 +2,7 @@ use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use codex_code_mode_protocol::CellId;
use codex_code_mode_protocol::CodeModeNestedToolCall;
@@ -171,6 +172,7 @@ async fn reconnected_execution_maps_started_and_initial_response_ids() {
let claimed = Arc::new(AtomicBool::new(false));
let initial_response_claimed = Arc::clone(&claimed);
let response = RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: wire_id.clone(),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "result".to_string(),
@@ -192,6 +194,7 @@ async fn reconnected_execution_maps_started_and_initial_response_ids() {
assert_eq!(
started.initial_response().await,
Ok(RuntimeResponse::Result {
code_mode_host_duration: None,
cell_id: public_id,
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "result".to_string(),
@@ -205,10 +208,12 @@ async fn reconnected_execution_maps_started_and_initial_response_ids() {
fn reconnected_wait_maps_live_and_missing_outcomes() {
let public_id = CellId::new("g2:42".to_string());
let yielded = RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: CellId::new("42".to_string()),
content_items: Vec::new(),
};
let terminated = RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: CellId::new("42".to_string()),
content_items: Vec::new(),
};
@@ -216,6 +221,7 @@ fn reconnected_wait_maps_live_and_missing_outcomes() {
assert_eq!(
public_wait_outcome(/*generation*/ 2, WaitOutcome::LiveCell(yielded)),
WaitOutcome::LiveCell(RuntimeResponse::Yielded {
code_mode_host_duration: None,
cell_id: public_id.clone(),
content_items: Vec::new(),
})
@@ -223,8 +229,40 @@ fn reconnected_wait_maps_live_and_missing_outcomes() {
assert_eq!(
public_wait_outcome(/*generation*/ 2, WaitOutcome::MissingCell(terminated)),
WaitOutcome::MissingCell(RuntimeResponse::Terminated {
code_mode_host_duration: None,
cell_id: public_id,
content_items: Vec::new(),
})
);
}
/// Replacing a host generation changes public cell IDs without losing the
/// operation's original duration on either the exec or wait path.
#[tokio::test]
async fn cell_id_remapping_preserves_code_mode_host_duration() {
let cell_id = CellId::new("42".to_string());
let code_mode_host_duration = Some(Duration::from_nanos(/*nanos*/ 1_234_567));
let response = RuntimeResponse::Yielded {
cell_id: cell_id.clone(),
content_items: Vec::new(),
code_mode_host_duration,
};
let initial = response.clone();
let started = StartedCell::from_future(cell_id, async move { Ok(initial) });
let expected = RuntimeResponse::Yielded {
cell_id: CellId::new("g2:42".to_string()),
content_items: Vec::new(),
code_mode_host_duration,
};
assert_eq!(
public_started_cell(/*generation*/ 2, started)
.initial_response()
.await
.as_ref(),
Ok(&expected),
);
assert_eq!(
public_wait_outcome(/*generation*/ 2, WaitOutcome::LiveCell(response)),
WaitOutcome::LiveCell(expected),
);
}
codex-rs/code-mode/src/remote_session/connection/driver/cell_ids.rs
+6
View File
@@ -60,25 +60,31 @@ pub(super) fn public_runtime_response(
RuntimeResponse::Yielded {
cell_id,
content_items,
code_mode_host_duration,
} => RuntimeResponse::Yielded {
cell_id: public_cell_id_from_protocol(generation, &cell_id),
content_items,
code_mode_host_duration,
},
RuntimeResponse::Terminated {
cell_id,
content_items,
code_mode_host_duration,
} => RuntimeResponse::Terminated {
cell_id: public_cell_id_from_protocol(generation, &cell_id),
content_items,
code_mode_host_duration,
},
RuntimeResponse::Result {
cell_id,
content_items,
error_text,
code_mode_host_duration,
} => RuntimeResponse::Result {
cell_id: public_cell_id_from_protocol(generation, &cell_id),
content_items,
error_text,
code_mode_host_duration,
},
}
}
codex-rs/code-mode/src/remote_session/connection/driver_tests.rs
+14
View File
@@ -773,6 +773,7 @@ async fn terminate_closes_cell_without_waiting_for_delegate_cleanup() {
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
}),
@@ -792,6 +793,7 @@ async fn terminate_closes_cell_without_waiting_for_delegate_cleanup() {
response_rx.await.expect("terminate reply"),
Ok(codex_code_mode_protocol::WaitOutcome::LiveCell(
codex_code_mode_protocol::RuntimeResponse::Terminated {
code_mode_host_duration: Some(Duration::ZERO),
cell_id: CellId::new("1".to_string()),
content_items: Vec::new(),
}
@@ -1096,6 +1098,7 @@ async fn mismatched_initial_response_fails_connection_and_closes_cell_once() {
id: RequestId::new(/*value*/ 2),
result: WireResult::Ok {
value: WireRuntimeResponse::Yielded {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("2".to_string()).into(),
content_items: Vec::new(),
},
@@ -1143,6 +1146,7 @@ async fn mismatched_wait_response_fails_connection() {
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Yielded {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("2".to_string()).into(),
content_items: Vec::new(),
}),
@@ -1187,6 +1191,7 @@ async fn mismatched_terminate_response_fails_connection() {
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::MissingCell(WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("2".to_string()).into(),
content_items: Vec::new(),
}),
@@ -1239,6 +1244,7 @@ async fn remote_wait_accepts_durations_longer_than_five_minutes() {
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Yielded {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
}),
@@ -1252,6 +1258,7 @@ async fn remote_wait_accepts_durations_longer_than_five_minutes() {
response_rx.await.expect("wait reply"),
Ok(codex_code_mode_protocol::WaitOutcome::LiveCell(
codex_code_mode_protocol::RuntimeResponse::Yielded {
code_mode_host_duration: Some(Duration::ZERO),
cell_id: CellId::new("1".to_string()),
content_items: Vec::new(),
}
@@ -1416,6 +1423,7 @@ async fn cancelled_wait_is_retired_before_next_wait_is_sent() {
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Yielded {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
}),
@@ -1429,6 +1437,7 @@ async fn cancelled_wait_is_retired_before_next_wait_is_sent() {
second_rx.await.expect("second wait reply"),
Ok(codex_code_mode_protocol::WaitOutcome::LiveCell(
codex_code_mode_protocol::RuntimeResponse::Yielded {
code_mode_host_duration: Some(Duration::ZERO),
cell_id: CellId::new("1".to_string()),
content_items: Vec::new(),
}
@@ -1492,6 +1501,7 @@ async fn abandoned_execute_is_tracked_and_terminated_after_admission() {
id: RequestId::new(/*value*/ 2),
result: WireResult::Ok {
value: WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
},
@@ -1506,6 +1516,7 @@ async fn abandoned_execute_is_tracked_and_terminated_after_admission() {
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
}),
@@ -1591,6 +1602,7 @@ async fn delivered_but_unclaimed_execute_is_terminated_when_the_caller_is_cancel
id: RequestId::new(/*value*/ 2),
result: WireResult::Ok {
value: WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
},
@@ -1606,6 +1618,7 @@ async fn delivered_but_unclaimed_execute_is_terminated_when_the_caller_is_cancel
result: WireResult::Ok {
value: HostResponse::WaitCompleted {
outcome: WireWaitOutcome::LiveCell(WireRuntimeResponse::Terminated {
code_mode_host_duration_ns: 0,
cell_id: CellId::new("1".to_string()).into(),
content_items: Vec::new(),
}),
@@ -1652,6 +1665,7 @@ async fn session_accepts_more_than_4096_cells_without_growing_a_tombstone_set()
id: RequestId::new(request_id),
result: WireResult::Ok {
value: WireRuntimeResponse::Yielded {
code_mode_host_duration_ns: 0,
cell_id: CellId::new(cell_id.clone()).into(),
content_items: Vec::new(),
},
codex-rs/core/src/tools/code_mode/execute_handler.rs
+7 -1
View File
@@ -109,6 +109,9 @@ impl CodeModeExecuteHandler {
.initial_response()
.await
.map_err(FunctionCallError::RespondToModel)?;
if let Some(code_mode_host_duration) = response.code_mode_host_duration() {
telemetry.record_code_mode_host_duration(code_mode_host_duration);
}
// Record the raw runtime boundary. The model-visible custom-tool output
// is produced by `handle_runtime_response` and later linked through
// `CodeCell.output_item_ids` in the reduced trace.
@@ -131,7 +134,10 @@ impl CodeModeExecuteHandler {
});
}
exec.session.services.elicitations.wait_until_clear().await;
handle_runtime_response(&exec, response, args.max_output_tokens, started_at)
let wall_time = response
.code_mode_host_duration()
.unwrap_or_else(|| started_at.elapsed());
handle_runtime_response(&exec, response, args.max_output_tokens, wall_time)
.await
.map_err(FunctionCallError::RespondToModel)
}
codex-rs/core/src/tools/code_mode/mod.rs
+4 -4
View File
@@ -251,7 +251,7 @@ pub(super) async fn handle_runtime_response(
exec: &ExecContext,
response: RuntimeResponse,
max_output_tokens: Option<usize>,
started_at: std::time::Instant,
wall_time: Duration,
) -> Result<FunctionToolOutput, String> {
let script_status = format_script_status(&response);
@@ -260,14 +260,14 @@ pub(super) async fn handle_runtime_response(
let mut content_items = into_function_call_output_content_items(content_items);
sanitize_runtime_image_detail(exec.turn.as_ref(), &mut content_items);
content_items = truncate_code_mode_result(content_items, max_output_tokens);
prepend_script_status(&mut content_items, &script_status, started_at.elapsed());
prepend_script_status(&mut content_items, &script_status, wall_time);
Ok(FunctionToolOutput::from_content(content_items, Some(true)))
}
RuntimeResponse::Terminated { content_items, .. } => {
let mut content_items = into_function_call_output_content_items(content_items);
sanitize_runtime_image_detail(exec.turn.as_ref(), &mut content_items);
content_items = truncate_code_mode_result(content_items, max_output_tokens);
prepend_script_status(&mut content_items, &script_status, started_at.elapsed());
prepend_script_status(&mut content_items, &script_status, wall_time);
Ok(FunctionToolOutput::from_content(content_items, Some(true)))
}
RuntimeResponse::Result {
@@ -284,7 +284,7 @@ pub(super) async fn handle_runtime_response(
});
}
content_items = truncate_code_mode_result(content_items, max_output_tokens);
prepend_script_status(&mut content_items, &script_status, started_at.elapsed());
prepend_script_status(&mut content_items, &script_status, wall_time);
Ok(FunctionToolOutput::from_content(
content_items,
Some(success),
codex-rs/core/src/tools/code_mode/telemetry.rs
+22
View File
@@ -3,6 +3,7 @@ use codex_analytics::CodeModeToolCallFact;
use codex_analytics::CodeModeToolCallStatus;
use codex_analytics::TurnAnalyticsMetadata;
use std::sync::Arc;
use std::time::Duration;
pub(super) struct CodeModeToolCallGuard {
analytics: AnalyticsEventsClient,
@@ -45,6 +46,27 @@ impl CodeModeToolCallGuard {
CodeModeToolCallStatus::Failed
};
}
pub(super) fn record_code_mode_host_duration(&self, duration: Duration) {
let Ok(code_mode_host_duration_ns) = u64::try_from(duration.as_nanos()) else {
return;
};
// Bridge joins this record to the outer tool-completion event. Emit it
// before the handler returns so consumers never need another timeout.
tracing::info!(
target: "codex_code_mode::timing",
{
event.name = "codex.code_mode.host_timing",
conversation_id = %self.thread_id,
turn_id = %self.turn_id,
call_id = %self.call_id,
cell_id = self.cell_id.as_deref(),
tool_name = self.tool_name,
code_mode_host_duration_ns,
},
"code-mode host operation completed"
);
}
}
impl Drop for CodeModeToolCallGuard {
codex-rs/core/src/tools/code_mode/wait_handler.rs
+8 -1
View File
@@ -94,6 +94,7 @@ impl CodeModeWaitHandler {
})?;
let exec = ExecContext { session, turn };
let started_at = std::time::Instant::now();
telemetry.cell_id = Some(args.cell_id.clone());
let cell_id = codex_code_mode::CellId::new(args.cell_id);
let wait_response = if args.terminate {
exec.session
@@ -152,8 +153,14 @@ impl CodeModeWaitHandler {
);
}
}
if let Some(code_mode_host_duration) = wait_response.code_mode_host_duration() {
telemetry.record_code_mode_host_duration(code_mode_host_duration);
}
exec.session.services.elicitations.wait_until_clear().await;
handle_runtime_response(&exec, wait_response.into(), args.max_tokens, started_at)
let wall_time = wait_response
.code_mode_host_duration()
.unwrap_or_else(|| started_at.elapsed());
handle_runtime_response(&exec, wait_response.into(), args.max_tokens, wall_time)
.await
.map_err(FunctionCallError::RespondToModel)
.map(boxed_tool_output)
codex-rs/core/src/tools/tool_dispatch_trace_tests.rs
+1
View File
@@ -100,6 +100,7 @@ impl codex_code_mode::CodeModeSession for MissingCellCodeModeSession {
Box::pin(async move {
Ok(codex_code_mode::WaitOutcome::MissingCell(
codex_code_mode::RuntimeResponse::Result {
code_mode_host_duration: None,
error_text: Some(format!("exec cell {cell_id} not found")),
cell_id,
content_items: Vec::new(),