openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-05 15:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge f530914a54 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-07-01 22:14:43 -07:00
committed by GitHub
parent 7245964ad8 f530914a54
commit b0dc58a2c1
14 changed files with 872 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
codex-rs/app-server/tests/common/json_logging.rs
View File

@@ -1,11 +1,79 @@
use std::path::Path;
use std::process::Command;
use std::process::Stdio;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use serde_json::Value;
use serde_json::json;
use tokio::sync::Notify;
#[derive(Clone, Default)]
pub(crate) struct JsonLogCapture {
lines: Arc<Mutex<Vec<String>>>,
updated: Arc<Notify>,
}
impl JsonLogCapture {
pub(crate) fn record(&self, line: String) {
self.lines
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(line);
self.updated.notify_one();
}
pub(crate) async fn wait_for_event(&self, event_name: &str) -> Result<Value> {
let mut events = self.wait_for_events(event_name, /*count*/ 1).await?;
Ok(events.remove(0))
}
pub(crate) async fn wait_for_events(
&self,
event_name: &str,
count: usize,
) -> Result<Vec<Value>> {
let result = tokio::time::timeout(Duration::from_secs(10), async {
loop {
let updated = self.updated.notified();
let events = self
.events()?
.into_iter()
.filter(|event| event["fields"]["event.name"].as_str() == Some(event_name))
.collect::<Vec<_>>();
if events.len() >= count {
return Ok(events);
}
updated.await;
}
})
.await;
match result {
Ok(result) => result,
Err(_) => {
let lines = self
.lines
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.join("\n");
anyhow::bail!(
"timed out waiting for {count} JSON log event(s) named `{event_name}`; captured stderr:\n{lines}"
)
}
}
}
pub(crate) fn events(&self) -> Result<Vec<Value>> {
let lines = self
.lines
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
json_log_events(lines.iter().map(String::as_str))
}
}
pub fn app_server_json_shutdown_event(
binary: &str,
@@ -31,25 +99,39 @@ pub fn app_server_json_shutdown_event(
let stderr = String::from_utf8(output.stderr)?;
anyhow::ensure!(output.status.success(), "app-server failed: {stderr}");
let events = stderr
.lines()
.filter(|line| !line.is_empty())
.map(serde_json::from_str::<Value>)
.collect::<serde_json::Result<Vec<_>>>()
.with_context(|| format!("app-server stderr was not JSONL: {stderr}"))?;
let events = json_log_events(stderr.lines())
.with_context(|| format!("app-server stderr was not valid JSONL: {stderr}"))?;
let event = events
.iter()
.find(|event| event["fields"]["message"] == "processor task exited")
.context("missing INFO shutdown event in app-server JSON logs")?;
let timestamp = event["timestamp"]
.as_str()
.context("shutdown event did not include a timestamp")?;
chrono::DateTime::parse_from_rfc3339(timestamp)
.with_context(|| format!("shutdown event timestamp was not RFC 3339: {timestamp}"))?;
Ok(json!({
"level": event["level"],
"fields": event["fields"],
"target": event["target"],
}))
}
fn json_log_events<'a>(lines: impl IntoIterator<Item = &'a str>) -> Result<Vec<Value>> {
lines
.into_iter()
.filter(|line| !line.is_empty())
.map(|line| {
let event = serde_json::from_str::<Value>(line)
.with_context(|| format!("log line was not JSON: {line}"))?;
anyhow::ensure!(
event["level"].is_string()
&& event["fields"].is_object()
&& event["target"].is_string(),
"JSON log event did not include level, fields, and target: {line}"
);
let timestamp = event["timestamp"]
.as_str()
.with_context(|| format!("JSON log event did not include a timestamp: {line}"))?;
chrono::DateTime::parse_from_rfc3339(timestamp).with_context(|| {
format!("JSON log event timestamp was not RFC 3339: {timestamp}")
})?;
Ok(event)
})
.collect()
}

40
codex-rs/app-server/tests/common/test_app_server.rs
View File

@@ -123,6 +123,8 @@ use core_test_support::test_codex::TestEnv;
use core_test_support::test_codex::test_env;
use tokio::process::Command;
use crate::json_logging::JsonLogCapture;
pub struct TestAppServer {
next_request_id: AtomicI64,
/// Retain this child process until the client is dropped. The Tokio runtime
@@ -134,6 +136,7 @@ pub struct TestAppServer {
stdout: BufReader<ChildStdout>,
pending_messages: VecDeque<JSONRPCMessage>,
auto_env: Option<TestEnv>,
json_logs: JsonLogCapture,
}
pub const DEFAULT_CLIENT_NAME: &str = "codex-app-server-tests";
@@ -160,6 +163,14 @@ impl TestAppServer {
/// URL-based configuration, this helper rejects a `codex_home` containing
/// that file.
pub async fn new_with_auto_env(codex_home: &Path) -> anyhow::Result<Self> {
Self::new_with_auto_env_and_env(codex_home, &[]).await
}
/// Starts an auto-environment app server with child-process environment overrides.
pub async fn new_with_auto_env_and_env(
codex_home: &Path,
extra_env_overrides: &[(&str, Option<&str>)],
) -> anyhow::Result<Self> {
let environments_toml = codex_home.join("environments.toml");
ensure!(
!environments_toml
@@ -172,7 +183,7 @@ impl TestAppServer {
let auto_env = test_env().await?;
// Noise registry configuration takes precedence over the URL-based
// provider, so clear inherited values to keep the selection hermetic.
let env_overrides = [
let mut env_overrides = vec![
(
CODEX_EXEC_SERVER_URL_ENV_VAR,
auto_env.environment().exec_server_url(),
@@ -182,6 +193,7 @@ impl TestAppServer {
(CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR, None),
(CODEX_EXEC_SERVER_NOISE_CHATGPT_ACCOUNT_ID_ENV_VAR, None),
];
env_overrides.extend_from_slice(extra_env_overrides);
let mut app_server = Self::new_with_env(codex_home, &env_overrides).await?;
app_server.auto_env = Some(auto_env);
Ok(app_server)
@@ -202,6 +214,28 @@ impl TestAppServer {
})
}
/// Waits for a JSON stderr event whose structured `event.name` field matches.
pub async fn wait_for_json_log_event(
&self,
event_name: &str,
) -> anyhow::Result<serde_json::Value> {
self.json_logs.wait_for_event(event_name).await
}
/// Waits for the requested number of JSON stderr events with the same `event.name` field.
pub async fn wait_for_json_log_events(
&self,
event_name: &str,
count: usize,
) -> anyhow::Result<Vec<serde_json::Value>> {
self.json_logs.wait_for_events(event_name, count).await
}
/// Returns every stderr line parsed and validated as a JSON log event.
pub fn json_log_events(&self) -> anyhow::Result<Vec<serde_json::Value>> {
self.json_logs.events()
}
pub async fn new_without_managed_config(codex_home: &Path) -> anyhow::Result<Self> {
Self::new_with_env(codex_home, &[(DISABLE_MANAGED_CONFIG_ENV_VAR, Some("1"))]).await
}
@@ -322,10 +356,13 @@ impl TestAppServer {
// Forward child's stderr to our stderr so failures are visible even
// when stdout/stderr are captured by the test harness.
let json_logs = JsonLogCapture::default();
if let Some(stderr) = process.stderr.take() {
let json_logs = json_logs.clone();
let mut stderr_reader = BufReader::new(stderr).lines();
tokio::spawn(async move {
while let Ok(Some(line)) = stderr_reader.next_line().await {
json_logs.record(line.clone());
eprintln!("[mcp stderr] {line}");
}
});
@@ -337,6 +374,7 @@ impl TestAppServer {
stdout,
pending_messages: VecDeque::new(),
auto_env: None,
json_logs,
})
}

185
codex-rs/app-server/tests/suite/logging.rs
View File

@@ -1,8 +1,28 @@
use anyhow::Result;
use app_test_support::TestAppServer;
use app_test_support::app_server_json_shutdown_event;
use app_test_support::create_exec_command_sse_response;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::to_response;
use app_test_support::write_mock_responses_config_toml;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput;
use codex_features::Feature;
use core_test_support::skip_if_no_network;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::BTreeMap;
use tempfile::TempDir;
use tokio::time::Duration;
use tokio::time::timeout;
const READ_TIMEOUT: Duration = Duration::from_secs(10);
#[test]
fn standalone_app_server_emits_json_info_events() -> Result<()> {
@@ -25,3 +45,168 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn app_server_emits_structured_tool_timing_events() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = create_mock_responses_server_sequence(vec![
create_exec_command_sse_response("exec-call-1")?,
create_final_assistant_message_sse_response("done")?,
])
.await;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml(
codex_home.path(),
&server.uri(),
&BTreeMap::from([(Feature::UnifiedExec, true)]),
/*auto_compact_limit*/ 100_000,
/*requires_openai_auth*/ None,
"mock_provider",
"compact",
)?;
let mut app_server = TestAppServer::new_with_auto_env_and_env(
codex_home.path(),
&[
("LOG_FORMAT", Some("json")),
(
"RUST_LOG",
Some(
"warn,codex_core::tools::parallel=info,codex_core::turn_timing=info,codex_otel.log_only=info",
),
),
],
)
.await?;
timeout(READ_TIMEOUT, app_server.initialize()).await??;
let thread_start_id = app_server
.send_thread_start_request_with_auto_env(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let thread_start_response: JSONRPCResponse = timeout(
READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(thread_start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?;
let turn_start_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "run a command".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_start_response: JSONRPCResponse = timeout(
READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
)
.await??;
let TurnStartResponse { turn } = to_response(turn_start_response)?;
timeout(
READ_TIMEOUT,
app_server.read_stream_until_notification_message("turn/completed"),
)
.await??;
let tool_call = app_server
.wait_for_json_log_event("codex.tool_call")
.await?;
assert_eq!(tool_call["level"], "INFO");
assert_eq!(tool_call["target"], "codex_core::tools::parallel");
assert_eq!(tool_call["fields"]["message"], "tool call completed");
assert!(tool_call["fields"]["trace_id"].is_string());
assert_eq!(tool_call["fields"]["conversation.id"], thread.id);
assert_eq!(tool_call["fields"]["turn_id"], turn.id);
assert_eq!(tool_call["fields"]["tool_name"], "exec_command");
assert_eq!(tool_call["fields"]["call_id"], "exec-call-1");
assert_eq!(tool_call["fields"]["tool_source"], "direct");
assert_eq!(tool_call["fields"]["code_mode_cell_id"], "");
assert_eq!(tool_call["fields"]["code_mode_runtime_tool_call_id"], "");
assert_eq!(tool_call["fields"]["execution_started"], true);
assert_nonnegative_duration_fields(
&tool_call,
&[
"dispatch_duration_seconds",
"handler_duration_seconds",
"total_duration_seconds",
],
);
let dispatch_duration = duration_field(&tool_call, "dispatch_duration_seconds");
let handler_duration = duration_field(&tool_call, "handler_duration_seconds");
let total_duration = duration_field(&tool_call, "total_duration_seconds");
assert!(total_duration > 0.0);
assert!(
(dispatch_duration + handler_duration - total_duration).abs() < 0.000_001,
"dispatch and handler durations must sum to total duration: {tool_call}"
);
let tool_result = app_server
.wait_for_json_log_event("codex.tool_result")
.await?;
assert_eq!(tool_result["level"], "INFO");
assert_eq!(tool_result["target"], "codex_otel.log_only");
assert!(tool_result["fields"]["trace_id"].is_string());
assert_eq!(tool_result["fields"]["conversation.id"], thread.id);
assert_eq!(tool_result["fields"]["turn_id"], turn.id);
assert_eq!(tool_result["fields"]["tool_name"], "exec_command");
assert_eq!(tool_result["fields"]["call_id"], "exec-call-1");
assert_eq!(tool_result["fields"]["tool_source"], "direct");
assert!(
matches!(
tool_result["fields"]["success"].as_str(),
Some("true" | "false")
),
"success must be a boolean string: {tool_result}"
);
assert_nonnegative_duration_fields(&tool_result, &["duration_seconds"]);
let inferences = app_server
.wait_for_json_log_events("codex.inference", /*count*/ 2)
.await?;
assert_eq!(
inferences
.iter()
.map(|event| event["fields"]["inference_index"].as_u64())
.collect::<Vec<_>>(),
vec![Some(1), Some(2)]
);
for inference in inferences {
assert_eq!(inference["level"], "INFO");
assert_eq!(inference["target"], "codex_core::turn_timing");
assert_eq!(inference["fields"]["message"], "inference completed");
assert!(inference["fields"]["trace_id"].is_string());
assert_eq!(inference["fields"]["conversation.id"], thread.id);
assert_eq!(inference["fields"]["turn_id"], turn.id);
assert_eq!(inference["fields"]["model"], "mock-model");
assert_eq!(
inference["fields"]["provider_name"],
"Mock provider for test"
);
assert_eq!(inference["fields"]["result"], "success");
assert_nonnegative_duration_fields(&inference, &["duration_seconds"]);
}
Ok(())
}
fn assert_nonnegative_duration_fields(event: &serde_json::Value, fields: &[&str]) {
for field in fields {
let duration = duration_field(event, field);
assert!(duration >= 0.0, "{field} must be nonnegative: {event}");
}
}
fn duration_field(event: &serde_json::Value, field: &str) -> f64 {
event["fields"][field]
.as_f64()
.unwrap_or_else(|| panic!("{field} must be a JSON number: {event}"))
}

32
codex-rs/core/src/session/turn.rs
View File

@@ -65,6 +65,7 @@ use crate::tools::router::extension_tool_executors;
use crate::tools::spec_plan::search_tool_enabled;
use crate::tools::spec_plan::tool_suggest_enabled;
use crate::turn_diff_tracker::TurnDiffTracker;
use crate::turn_timing::InferenceTimingResult;
use crate::turn_timing::record_turn_ttft_metric;
use crate::util::error_or_panic;
use codex_analytics::AppInvocation;
@@ -1903,13 +1904,19 @@ async fn try_run_sampling_request(
auth_mode = sess.services.auth_manager.auth_mode(),
features = sess.features.enabled_features(),
);
let provider_info = turn_context.provider.info();
let inference_trace = sess.services.rollout_thread_trace.inference_trace_context(
turn_context.sub_id.as_str(),
turn_context.model_info.slug.as_str(),
turn_context.provider.info().name.as_str(),
provider_info.name.as_str(),
);
let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling();
let mut stream = client_session
let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(
&sess.thread_id,
&turn_context.sub_id,
&turn_context.model_info.slug,
&provider_info.name,
);
let stream_result = client_session
.stream(
prompt,
&turn_context.model_info,
@@ -1922,7 +1929,18 @@ async fn try_run_sampling_request(
)
.instrument(trace_span!("stream_request"))
.or_cancel(&cancellation_token)
.await??;
.await;
let mut stream = match stream_result {
Ok(Ok(stream)) => stream,
Ok(Err(err)) => {
sampling_timing_guard.finish_inference(InferenceTimingResult::Error);
return Err(err);
}
Err(codex_async_utils::CancelErr::Cancelled) => {
sampling_timing_guard.finish_inference(InferenceTimingResult::Cancelled);
return Err(CodexErr::TurnAborted);
}
};
let mut in_flight: FuturesOrdered<BoxFuture<'static, CodexResult<ResponseInputItem>>> =
FuturesOrdered::new();
let mut needs_follow_up = false;
@@ -2346,7 +2364,11 @@ async fn try_run_sampling_request(
}
}
};
drop(sampling_timing_guard);
sampling_timing_guard.finish_inference(match &outcome {
Ok(_) => InferenceTimingResult::Success,
Err(CodexErr::TurnAborted) => InferenceTimingResult::Cancelled,
Err(_) => InferenceTimingResult::Error,
});
flush_assistant_text_segments_all(
&sess,

82
codex-rs/core/src/tools/parallel.rs
View File

@@ -1,4 +1,5 @@
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Instant;
@@ -9,6 +10,7 @@ use tokio_util::either::Either;
use tokio_util::sync::CancellationToken;
use tokio_util::task::AbortOnDropHandle;
use tracing::Instrument;
use tracing::info;
use tracing::instrument;
use tracing::trace_span;
@@ -27,6 +29,16 @@ use crate::tools::router::ToolRouter;
use codex_protocol::error::CodexErr;
use codex_protocol::models::ResponseInputItem;
struct ToolCallTimingGuard {
started_at: Instant,
execution_started_at: Arc<OnceLock<Instant>>,
conversation_id: String,
turn_id: String,
call_id: String,
tool_name: codex_tools::ToolName,
source: ToolCallSource,
}
#[derive(Clone)]
pub(crate) struct ToolCallRuntime {
router: Arc<ToolRouter>,
@@ -96,6 +108,11 @@ impl ToolCallRuntime {
let invocation_cancellation_token = cancellation_token.clone();
let wait_for_runtime_cancellation = self.router.tool_waits_for_runtime_cancellation(&call);
let started = Instant::now();
let tool_call_timing_guard =
ToolCallTimingGuard::capture(started, &session.thread_id, &turn.sub_id, &call, &source);
let execution_started_at = tool_call_timing_guard
.as_ref()
.map(|timing| Arc::clone(&timing.execution_started_at));
let abort_session = Arc::clone(&session);
let abort_source = source.clone();
let abort_turn = Arc::clone(&turn);
@@ -119,6 +136,9 @@ impl ToolCallRuntime {
} else {
Either::Right(lock.write().await)
};
if let Some(execution_started_at) = execution_started_at {
let _ = execution_started_at.set(Instant::now());
}
router
.dispatch_tool_call_with_terminal_outcome(
@@ -135,6 +155,7 @@ impl ToolCallRuntime {
}));
async move {
let _tool_call_timing_guard = tool_call_timing_guard;
tokio::select! {
res = &mut handle => res.map_err(Self::tool_task_join_error)?,
_ = cancellation_token.cancelled() => {
@@ -237,6 +258,67 @@ impl ToolCallRuntime {
}
}
impl ToolCallTimingGuard {
fn capture(
started_at: Instant,
conversation_id: &impl std::fmt::Display,
turn_id: &str,
call: &ToolCall,
source: &ToolCallSource,
) -> Option<Self> {
if !tracing::enabled!(tracing::Level::INFO) {
return None;
}
Some(Self {
started_at,
execution_started_at: Arc::new(OnceLock::new()),
conversation_id: conversation_id.to_string(),
turn_id: turn_id.to_string(),
call_id: call.call_id.clone(),
tool_name: call.tool_name.clone(),
source: source.clone(),
})
}
}
impl Drop for ToolCallTimingGuard {
fn drop(&mut self) {
let completed_at = Instant::now();
info!(
event.name = "codex.tool_call",
trace_id = %codex_otel::current_span_trace_id().unwrap_or_default(),
conversation.id = %self.conversation_id,
turn_id = %self.turn_id,
tool_name = %self.tool_name,
call_id = %self.call_id,
tool_source = match &self.source {
ToolCallSource::Direct => "direct",
ToolCallSource::CodeMode { .. } => "code_mode",
},
code_mode_cell_id = match &self.source {
ToolCallSource::Direct => "",
ToolCallSource::CodeMode { cell_id, .. } => cell_id.as_str(),
},
code_mode_runtime_tool_call_id = match &self.source {
ToolCallSource::Direct => "",
ToolCallSource::CodeMode { runtime_tool_call_id, .. } => runtime_tool_call_id.as_str(),
},
execution_started = self.execution_started_at.get().is_some(),
dispatch_duration_seconds = self.execution_started_at.get().map_or_else(
|| completed_at.duration_since(self.started_at).as_secs_f64(),
|execution_started_at| execution_started_at.duration_since(self.started_at).as_secs_f64(),
),
handler_duration_seconds = self.execution_started_at.get().map_or(
0.0,
|execution_started_at| completed_at.duration_since(*execution_started_at).as_secs_f64(),
),
total_duration_seconds = completed_at.duration_since(self.started_at).as_secs_f64(),
"tool call completed"
);
}
}
#[cfg(test)]
mod tests {
use super::*;

54
codex-rs/core/src/tools/registry.rs
View File

@@ -14,6 +14,7 @@ use crate::sandbox_tags::permission_profile_policy_tag;
use crate::sandbox_tags::permission_profile_sandbox_tag;
use crate::session::turn_context::TurnContext;
use crate::tools::context::FunctionToolOutput;
use crate::tools::context::ToolCallSource;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
@@ -444,6 +445,8 @@ impl ToolRegistry {
None => {
let message = unsupported_tool_call_message(&invocation.payload, &tool_name);
let log_payload = invocation.payload.log_payload();
let extra_trace_fields =
tool_result_trace_fields(&invocation.turn.sub_id, &invocation.source, "", "");
otel.tool_result_with_tags(
tool_name_flat.as_ref(),
&call_id_owned,
@@ -452,7 +455,7 @@ impl ToolRegistry {
/*success*/ false,
&message,
&base_tool_result_tags,
/*extra_trace_fields*/ &[],
&extra_trace_fields,
);
let err = FunctionCallError::RespondToModel(message);
dispatch_trace.record_failed(&err);
@@ -463,18 +466,25 @@ impl ToolRegistry {
let telemetry_tags = tool.telemetry_tags(&invocation).await;
let mut tool_result_tags =
Vec::with_capacity(base_tool_result_tags.len() + telemetry_tags.len());
let mut extra_trace_fields = Vec::new();
let mut mcp_server = "";
let mut mcp_server_origin = "";
tool_result_tags.extend_from_slice(&base_tool_result_tags);
for (key, value) in &telemetry_tags {
if matches!(*key, "mcp_server" | "mcp_server_origin") {
extra_trace_fields.push((*key, value.as_str()));
} else {
tool_result_tags.push((*key, value.as_str()));
match *key {
"mcp_server" => mcp_server = value.as_str(),
"mcp_server_origin" => mcp_server_origin = value.as_str(),
_ => tool_result_tags.push((*key, value.as_str())),
}
}
if !tool.matches_kind(&invocation.payload) {
let message = format!("tool {tool_name} invoked with incompatible payload");
let log_payload = invocation.payload.log_payload();
let extra_trace_fields = tool_result_trace_fields(
&invocation.turn.sub_id,
&invocation.source,
mcp_server,
mcp_server_origin,
);
otel.tool_result_with_tags(
tool_name_flat.as_ref(),
&call_id_owned,
@@ -541,6 +551,12 @@ impl ToolRegistry {
let response_cell = tokio::sync::Mutex::new(None);
let invocation_for_tool = invocation.clone();
let log_payload = invocation.payload.log_payload();
let extra_trace_fields = tool_result_trace_fields(
&invocation.turn.sub_id,
&invocation.source,
mcp_server,
mcp_server_origin,
);
let result = otel
.log_tool_result_with_tags(
@@ -670,6 +686,32 @@ impl ToolRegistry {
}
}
fn tool_result_trace_fields<'a>(
turn_id: &'a str,
source: &'a ToolCallSource,
mcp_server: &'a str,
mcp_server_origin: &'a str,
) -> [(&'static str, &'a str); 6] {
let (tool_source, code_mode_cell_id, code_mode_runtime_tool_call_id) = match source {
ToolCallSource::Direct => ("direct", "", ""),
ToolCallSource::CodeMode {
cell_id,
runtime_tool_call_id,
} => ("code_mode", cell_id.as_str(), runtime_tool_call_id.as_str()),
};
[
("turn_id", turn_id),
("tool_source", tool_source),
("code_mode_cell_id", code_mode_cell_id),
(
"code_mode_runtime_tool_call_id",
code_mode_runtime_tool_call_id,
),
("mcp_server", mcp_server),
("mcp_server_origin", mcp_server_origin),
]
}
async fn notify_tool_finish_if_unclaimed(
invocation: &ToolInvocation,
terminal_outcome_reached: Option<&AtomicBool>,

80
codex-rs/core/src/turn_timing.rs
View File

@@ -10,6 +10,7 @@ use codex_otel::TURN_TTFM_DURATION_METRIC;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseItem;
use tokio::sync::Mutex;
use tracing::info;
use crate::ResponseEvent;
use crate::session::turn_context::TurnContext;
@@ -80,6 +81,25 @@ pub(crate) struct TurnProfileTimingGuard {
timing: Arc<TurnTimingState>,
phase: TurnProfilePhase,
active: bool,
inference_log_context: Option<InferenceLogContext>,
}
#[derive(Clone, Copy)]
pub(crate) enum InferenceTimingResult {
Success,
Cancelled,
Error,
}
struct InferenceLogContext {
started_at: Instant,
conversation_id: String,
turn_id: String,
model: String,
provider_name: String,
trace_id: String,
inference_index: u32,
result: InferenceTimingResult,
}
impl TurnTimingState {
@@ -118,12 +138,35 @@ impl TurnTimingState {
self.profile_state().complete(Instant::now())
}
pub(crate) fn begin_sampling(self: &Arc<Self>) -> TurnProfileTimingGuard {
let active = self.profile_state().begin_sampling(Instant::now());
pub(crate) fn begin_sampling(
self: &Arc<Self>,
conversation_id: &impl std::fmt::Display,
turn_id: &str,
model: &str,
provider_name: &str,
) -> TurnProfileTimingGuard {
let started_at = Instant::now();
let inference_index = self.profile_state().begin_sampling(started_at);
let active = inference_index.is_some();
let inference_log_context = if tracing::enabled!(tracing::Level::INFO) {
inference_index.map(|inference_index| InferenceLogContext {
started_at,
conversation_id: conversation_id.to_string(),
turn_id: turn_id.to_string(),
model: model.to_string(),
provider_name: provider_name.to_string(),
trace_id: codex_otel::current_span_trace_id().unwrap_or_default(),
inference_index,
result: InferenceTimingResult::Error,
})
} else {
None
};
TurnProfileTimingGuard {
timing: Arc::clone(self),
phase: TurnProfilePhase::Sampling,
active,
inference_log_context,
}
}
@@ -137,6 +180,7 @@ impl TurnTimingState {
timing: Arc::clone(self),
phase: TurnProfilePhase::ToolBlocking,
active,
inference_log_context: None,
}
}
@@ -166,6 +210,14 @@ impl TurnTimingState {
}
}
impl TurnProfileTimingGuard {
pub(crate) fn finish_inference(mut self, result: InferenceTimingResult) {
if let Some(log_context) = &mut self.inference_log_context {
log_context.result = result;
}
}
}
impl Drop for TurnProfileTimingGuard {
fn drop(&mut self) {
if self.active {
@@ -173,6 +225,24 @@ impl Drop for TurnProfileTimingGuard {
.profile_state()
.end_phase(Instant::now(), self.phase);
}
if let Some(log_context) = &self.inference_log_context {
info!(
event.name = "codex.inference",
trace_id = %log_context.trace_id,
conversation.id = %log_context.conversation_id,
turn_id = %log_context.turn_id,
model = %log_context.model,
provider_name = %log_context.provider_name,
inference_index = log_context.inference_index,
result = match log_context.result {
InferenceTimingResult::Success => "success",
InferenceTimingResult::Cancelled => "cancelled",
InferenceTimingResult::Error => "error",
},
duration_seconds = log_context.started_at.elapsed().as_secs_f64(),
"inference completed"
);
}
}
}
@@ -200,12 +270,12 @@ impl TurnProfileState {
};
}
fn begin_sampling(&mut self, now: Instant) -> bool {
fn begin_sampling(&mut self, now: Instant) -> Option<u32> {
if self.completed_profile.is_some()
|| self.started_at.is_none()
|| self.active_phase.is_some()
{
return false;
return None;
}
self.advance(now);
if self.seen_sampling {
@@ -214,7 +284,7 @@ impl TurnProfileState {
self.seen_sampling = true;
self.active_phase = Some(TurnProfilePhase::Sampling);
self.sampling_request_count = self.sampling_request_count.saturating_add(1);
true
Some(self.sampling_request_count)
}
fn record_sampling_retry(&mut self) {

2
codex-rs/core/tests/suite/otel.rs
View File

@@ -710,7 +710,7 @@ async fn turn_and_completed_response_spans_record_token_usage() {
);
}
#[tokio::test]
#[tokio::test(flavor = "current_thread")]
async fn handle_responses_span_records_response_kind_and_tool_name() {
let buffer: &'static Mutex<Vec<u8>> = Box::leak(Box::new(Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()

4
codex-rs/exec-server/src/local_process.rs
View File

@@ -334,7 +334,7 @@ impl LocalProcess {
output_notify: Arc::clone(&output_notify),
open_streams: 2,
closed: false,
metrics: Some(self.inner.telemetry.process_started()),
metrics: Some(self.inner.telemetry.process_started(process_id.as_ref())),
termination_requested: false,
sandbox: prepared.sandbox,
sandbox_denied: false,
@@ -1306,7 +1306,7 @@ mod tests {
output_notify: Arc::clone(&output_notify),
open_streams: 2,
closed: false,
metrics: Some(backend.inner.telemetry.process_started()),
metrics: Some(backend.inner.telemetry.process_started(process_id.as_ref())),
termination_requested: false,
sandbox: SandboxType::None,
sandbox_denied: false,

24
codex-rs/exec-server/src/server/processor.rs
View File

@@ -128,6 +128,13 @@ async fn run_connection(
JsonRpcConnectionEvent::Message(message) => match message {
codex_exec_server_protocol::JSONRPCMessage::Request(request) => {
let request_started_at = Instant::now();
let request_log_context = telemetry.request_log_context(
&request.id,
request
.trace
.as_ref()
.and_then(|trace| trace.traceparent.as_deref()),
);
if let Some((method, route)) = router.request_route(request.method.as_str()) {
let request_span = request_span(method, &request);
let message = tokio::select! {
@@ -135,6 +142,7 @@ async fn run_connection(
_ = disconnected_rx.changed() => {
request_span.record("result", "disconnected");
telemetry.request_completed(
request_log_context.as_ref(),
method,
"disconnected",
request_started_at.elapsed(),
@@ -149,6 +157,7 @@ async fn run_connection(
{
request_span.record("result", "disconnected");
telemetry.request_completed(
request_log_context.as_ref(),
method,
"disconnected",
request_started_at.elapsed(),
@@ -156,7 +165,12 @@ async fn run_connection(
break;
}
request_span.record("result", result);
telemetry.request_completed(method, result, request_started_at.elapsed());
telemetry.request_completed(
request_log_context.as_ref(),
method,
result,
request_started_at.elapsed(),
);
drop(request_span);
} else {
let method = "unknown";
@@ -174,6 +188,7 @@ async fn run_connection(
{
request_span.record("result", "disconnected");
telemetry.request_completed(
request_log_context.as_ref(),
method,
"disconnected",
request_started_at.elapsed(),
@@ -181,7 +196,12 @@ async fn run_connection(
break;
}
request_span.record("result", "error");
telemetry.request_completed(method, "error", request_started_at.elapsed());
telemetry.request_completed(
request_log_context.as_ref(),
method,
"error",
request_started_at.elapsed(),
);
}
}
codex_exec_server_protocol::JSONRPCMessage::Notification(notification) => {

73
codex-rs/exec-server/src/telemetry.rs
View File

@@ -4,6 +4,7 @@ use std::time::Duration;
use std::time::Instant;
use codex_otel::MetricsClient;
use tracing::info;
use tracing::warn;
const CONNECTIONS_ACTIVE_METRIC: &str = "exec_server_connections_active";
@@ -96,10 +97,21 @@ pub(crate) struct ConnectionMetricGuard {
pub(crate) struct ProcessMetricGuard {
telemetry: ExecServerTelemetry,
log_context: Option<ProcessLogContext>,
started_at: Instant,
result: &'static str,
}
struct ProcessLogContext {
process_id: String,
trace_id: String,
}
pub(crate) struct RequestLogContext {
request_id: String,
traceparent: Option<String>,
}
impl ExecServerTelemetry {
pub fn new(metrics: MetricsClient) -> Self {
let active = Arc::new(Mutex::new(ActiveCounts::default()));
@@ -129,10 +141,22 @@ impl ExecServerTelemetry {
pub(crate) fn request_completed(
&self,
log_context: Option<&RequestLogContext>,
method: &'static str,
result: &'static str,
duration: Duration,
) {
if let Some(log_context) = log_context {
info!(
event.name = "codex.exec_server_request",
request_id = %log_context.request_id,
method,
result,
duration_seconds = duration.as_secs_f64(),
traceparent = log_context.traceparent.as_deref().unwrap_or(""),
"exec-server request completed"
);
}
self.with_inner(|inner| {
let tags = [("method", method), ("result", result)];
inner.counter(REQUESTS_TOTAL_METRIC, REQUESTS_TOTAL_DESCRIPTION, &tags);
@@ -145,6 +169,17 @@ impl ExecServerTelemetry {
});
}
pub(crate) fn request_log_context(
&self,
request_id: &impl std::fmt::Display,
traceparent: Option<&str>,
) -> Option<RequestLogContext> {
self.info_events_enabled().then(|| RequestLogContext {
request_id: request_id.to_string(),
traceparent: traceparent.map(str::to_string),
})
}
pub(crate) fn remote_registration_completed(&self, result: &'static str, duration: Duration) {
self.record_operation(REMOTE_REGISTRATION_METRICS, result, duration);
}
@@ -163,18 +198,37 @@ impl ExecServerTelemetry {
});
}
pub(crate) fn process_started(&self) -> ProcessMetricGuard {
pub(crate) fn process_started(&self, process_id: &str) -> ProcessMetricGuard {
self.with_inner(|inner| {
inner.adjust_process_count(/*delta*/ 1);
});
ProcessMetricGuard {
telemetry: self.clone(),
log_context: self.info_events_enabled().then(|| ProcessLogContext {
process_id: process_id.to_string(),
trace_id: codex_otel::current_span_trace_id().unwrap_or_default(),
}),
started_at: Instant::now(),
result: "unknown",
}
}
fn process_finished(&self, result: &'static str, duration: Duration) {
fn process_finished(
&self,
log_context: Option<&ProcessLogContext>,
result: &'static str,
duration: Duration,
) {
if let Some(log_context) = log_context {
info!(
event.name = "codex.exec_server_process",
process_id = %log_context.process_id,
trace_id = %log_context.trace_id,
result,
duration_seconds = duration.as_secs_f64(),
"exec-server process completed"
);
}
self.with_inner(|inner| {
inner.adjust_process_count(/*delta*/ -1);
inner.counter(
@@ -191,6 +245,10 @@ impl ExecServerTelemetry {
});
}
pub(crate) fn info_events_enabled(&self) -> bool {
tracing::enabled!(tracing::Level::INFO)
}
fn connection_finished(&self, transport: ConnectionTransport) {
self.with_inner(|inner| {
inner.adjust_connection_count(transport, /*delta*/ -1);
@@ -236,8 +294,11 @@ impl ProcessMetricGuard {
impl Drop for ProcessMetricGuard {
fn drop(&mut self) {
self.telemetry
.process_finished(self.result, self.started_at.elapsed());
self.telemetry.process_finished(
self.log_context.as_ref(),
self.result,
self.started_at.elapsed(),
);
}
}
@@ -338,3 +399,7 @@ fn register_active_gauge(
warn!(metric = name, "failed to register exec-server gauge");
}
}
#[cfg(test)]
#[path = "telemetry_tests.rs"]
mod tests;

170
codex-rs/exec-server/src/telemetry_tests.rs Normal file
View File

@@ -0,0 +1,170 @@
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use pretty_assertions::assert_eq;
use tracing::Event;
use tracing::Level;
use tracing::Subscriber;
use tracing::field::Field;
use tracing::field::Visit;
use tracing_subscriber::Layer;
use tracing_subscriber::layer::Context;
use tracing_subscriber::prelude::*;
use tracing_subscriber::registry::LookupSpan;
use super::ExecServerTelemetry;
#[derive(Clone, Debug, PartialEq, Eq)]
struct CapturedEvent {
level: Level,
target: String,
fields: BTreeMap<String, String>,
}
#[derive(Clone, Default)]
struct CaptureLayer {
events: Arc<Mutex<Vec<CapturedEvent>>>,
}
impl CaptureLayer {
fn events(&self) -> Vec<CapturedEvent> {
self.events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
}
impl<S> Layer<S> for CaptureLayer
where
S: Subscriber + for<'lookup> LookupSpan<'lookup>,
{
fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) {
let mut visitor = FieldVisitor::default();
event.record(&mut visitor);
self.events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(CapturedEvent {
level: *event.metadata().level(),
target: event.metadata().target().to_string(),
fields: visitor.fields,
});
}
}
#[derive(Default)]
struct FieldVisitor {
fields: BTreeMap<String, String>,
}
impl Visit for FieldVisitor {
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
self.fields
.insert(field.name().to_string(), format!("{value:?}"));
}
fn record_str(&mut self, field: &Field, value: &str) {
self.fields
.insert(field.name().to_string(), value.to_string());
}
fn record_f64(&mut self, field: &Field, value: f64) {
self.fields
.insert(field.name().to_string(), value.to_string());
}
}
#[test]
fn exec_server_timing_events_are_structured_info_logs() {
let capture = CaptureLayer::default();
let subscriber = tracing_subscriber::registry().with(capture.clone());
tracing::subscriber::with_default(subscriber, || {
tracing::callsite::rebuild_interest_cache();
let telemetry = ExecServerTelemetry::default();
let request_log_context = telemetry
.request_log_context(&"request-1", Some("00-trace-parent"))
.expect("INFO event should be enabled");
telemetry.request_completed(
Some(&request_log_context),
"process/start",
"success",
Duration::from_millis(42),
);
telemetry.process_started("process-1").finish("success");
});
let events = capture.events();
let request = events
.iter()
.find(|event| {
event
.fields
.get("event.name")
.is_some_and(|name| name == "codex.exec_server_request")
})
.expect("request timing event");
assert_eq!(request.level, Level::INFO);
assert_eq!(request.target, "codex_exec_server::telemetry");
assert_eq!(
request.fields,
BTreeMap::from([
("duration_seconds".to_string(), "0.042".to_string()),
(
"event.name".to_string(),
"codex.exec_server_request".to_string(),
),
(
"message".to_string(),
"exec-server request completed".to_string(),
),
("method".to_string(), "process/start".to_string()),
("request_id".to_string(), "request-1".to_string()),
("result".to_string(), "success".to_string()),
("traceparent".to_string(), "00-trace-parent".to_string()),
])
);
let process = events
.iter()
.find(|event| {
event
.fields
.get("event.name")
.is_some_and(|name| name == "codex.exec_server_process")
})
.expect("process timing event");
assert_eq!(process.level, Level::INFO);
assert_eq!(process.target, "codex_exec_server::telemetry");
assert_eq!(
process.fields.get("message").map(String::as_str),
Some("exec-server process completed")
);
assert_eq!(
process.fields.get("process_id").map(String::as_str),
Some("process-1")
);
assert_eq!(process.fields.get("trace_id").map(String::as_str), Some(""));
assert_eq!(
process.fields.get("result").map(String::as_str),
Some("success")
);
assert!(process.fields.contains_key("duration_seconds"));
}
#[test]
fn disabled_info_events_do_not_capture_process_log_values() {
let subscriber = tracing_subscriber::registry()
.with(tracing_subscriber::filter::filter_fn(|_metadata| false));
tracing::subscriber::with_default(subscriber, || {
tracing::callsite::rebuild_interest_cache();
let telemetry = ExecServerTelemetry::default();
assert!(!telemetry.info_events_enabled());
let process = telemetry.process_started("process-1");
assert!(process.log_context.is_none());
});
}

38
codex-rs/otel/src/events/session_telemetry.rs
View File

@@ -1062,8 +1062,16 @@ impl SessionTelemetry {
log_event!(
self,
event.name = "codex.tool_result",
trace_id = %crate::current_span_trace_id().unwrap_or_default(),
turn_id = "",
tool_name = %tool_name,
call_id = "",
tool_source = "",
code_mode_cell_id = "",
code_mode_runtime_tool_call_id = "",
arguments = "",
duration_ms = %Duration::ZERO.as_millis(),
duration_seconds = Duration::ZERO.as_secs_f64(),
success = %false,
output = %error,
mcp_server = "",
@@ -1072,8 +1080,15 @@ impl SessionTelemetry {
trace_event!(
self,
event.name = "codex.tool_result",
trace_id = %crate::current_span_trace_id().unwrap_or_default(),
turn_id = "",
tool_name = %tool_name,
call_id = "",
tool_source = "",
code_mode_cell_id = "",
code_mode_runtime_tool_call_id = "",
duration_ms = %Duration::ZERO.as_millis(),
duration_seconds = Duration::ZERO.as_secs_f64(),
success = %false,
output_length = error.len() as i64,
output_line_count = error.lines().count() as i64,
@@ -1101,33 +1116,42 @@ impl SessionTelemetry {
tags.extend_from_slice(extra_tags);
self.counter(TOOL_CALL_COUNT_METRIC, /*inc*/ 1, &tags);
self.record_duration(TOOL_CALL_DURATION_METRIC, duration, &tags);
let mcp_server = trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("");
let mcp_server_origin =
trace_field_value(extra_trace_fields, "mcp_server_origin").unwrap_or("");
log_event!(
self,
event.name = "codex.tool_result",
trace_id = %crate::current_span_trace_id().unwrap_or_default(),
turn_id = %trace_field_value(extra_trace_fields, "turn_id").unwrap_or(""),
tool_name = %tool_name,
call_id = %call_id,
tool_source = %trace_field_value(extra_trace_fields, "tool_source").unwrap_or(""),
code_mode_cell_id = %trace_field_value(extra_trace_fields, "code_mode_cell_id").unwrap_or(""),
code_mode_runtime_tool_call_id = %trace_field_value(extra_trace_fields, "code_mode_runtime_tool_call_id").unwrap_or(""),
arguments = %arguments,
duration_ms = %duration.as_millis(),
duration_seconds = duration.as_secs_f64(),
success = %success_str,
output = %output,
mcp_server = %mcp_server,
mcp_server_origin = %mcp_server_origin,
mcp_server = %trace_field_value(extra_trace_fields, "mcp_server").unwrap_or(""),
mcp_server_origin = %trace_field_value(extra_trace_fields, "mcp_server_origin").unwrap_or(""),
);
trace_event!(
self,
event.name = "codex.tool_result",
trace_id = %crate::current_span_trace_id().unwrap_or_default(),
turn_id = %trace_field_value(extra_trace_fields, "turn_id").unwrap_or(""),
tool_name = %tool_name,
call_id = %call_id,
tool_source = %trace_field_value(extra_trace_fields, "tool_source").unwrap_or(""),
code_mode_cell_id = %trace_field_value(extra_trace_fields, "code_mode_cell_id").unwrap_or(""),
code_mode_runtime_tool_call_id = %trace_field_value(extra_trace_fields, "code_mode_runtime_tool_call_id").unwrap_or(""),
duration_ms = %duration.as_millis(),
duration_seconds = duration.as_secs_f64(),
success = %success_str,
arguments_length = arguments.len() as i64,
output_length = output.len() as i64,
output_line_count = output.lines().count() as i64,
tool_origin = if mcp_server.is_empty() { "builtin" } else { "mcp" },
mcp_tool = !mcp_server.is_empty(),
tool_origin = if trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("").is_empty() { "builtin" } else { "mcp" },
mcp_tool = !trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("").is_empty(),
);
}

27
codex-rs/otel/tests/suite/otel_export_routing_policy.rs
View File

@@ -255,6 +255,10 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() {
&[
("mcp_server", "internal-mcp"),
("mcp_server_origin", "stdio"),
("turn_id", "turn-1"),
("tool_source", "code_mode"),
("code_mode_cell_id", "cell-1"),
("code_mode_runtime_tool_call_id", "runtime-call-1"),
],
);
});
@@ -286,6 +290,29 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() {
tool_log_attrs.get("mcp_server_origin").map(String::as_str),
Some("stdio")
);
assert_eq!(
tool_log_attrs.get("turn_id").map(String::as_str),
Some("turn-1")
);
assert_eq!(
tool_log_attrs.get("tool_source").map(String::as_str),
Some("code_mode")
);
assert_eq!(
tool_log_attrs.get("code_mode_cell_id").map(String::as_str),
Some("cell-1")
);
assert_eq!(
tool_log_attrs
.get("code_mode_runtime_tool_call_id")
.map(String::as_str),
Some("runtime-call-1")
);
assert_eq!(
tool_log_attrs.get("duration_seconds").map(String::as_str),
Some("0.042")
);
assert!(tool_log_attrs.contains_key("trace_id"));
let spans = span_exporter.get_finished_spans().expect("span export");
assert_eq!(spans.len(), 1);