mirror of
https://github.com/openai/codex.git
synced 2026-11-12 21:20:34 +00:00
telemetry: log structured direct tool-call timing (#30334)
## Why App-server deployments can consume structured JSON logs for operational measurements without requiring an OTEL exporter. Existing tool-result telemetry reports the handler outcome, but it does not separate time spent waiting to dispatch from time spent executing the handler. A compact completion event for the outer, direct tool call lets consumers measure those phases and correlate them with a conversation and turn. Code-mode calls are intentionally excluded so nested runtime calls do not create overlapping events that are easy to double-count. ## What changed - Added a [`ToolCallTimingGuard`](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/core/src/tools/parallel.rs#L32) around direct tool calls. Event-only strings and timing state are captured only when the `codex_core::tools::parallel` `INFO` target is enabled. - Added a [`codex.tool_call`](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/core/src/tools/parallel.rs#L313) completion event with conversation, turn, tool, call, trace, dispatch, handler, and total timing fields. - Recorded the execution-start marker after the dispatch lock is acquired. Event emission snapshots that marker once so a concurrently starting dispatch cannot produce internally inconsistent fields. - Limited the event to `ToolCallSource::Direct`; [unit coverage](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/core/src/tools/parallel.rs#L365) verifies code-mode calls are ignored. - Added [cancellation coverage](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/core/src/tools/parallel.rs#L408) that holds the execution gate and verifies a call cancelled before admission emits exactly one dispatch-only timing event. - Added reusable [`JsonLogCapture`](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/app-server/tests/common/json_logging.rs#L15) support, a [JSON-logging-specific `TestAppServer` constructor](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/app-server/tests/common/test_app_server.rs#L172), and an [end-to-end app-server test](https://github.com/openai/codex/blob/141110a73cff041c42fd51f47e363e83df1dde51/codex-rs/app-server/tests/suite/logging.rs#L52) that drives a direct `exec_command` through the public v2 JSON-RPC API and validates the emitted JSON event. Exec-server-specific request and process timing remains in the stacked PR #30901. ## Suggested logging filter ```bash LOG_FORMAT=json \ RUST_LOG='warn,codex_core::tools::parallel=info' \ codex app-server ``` ## Event example Identifier and timing values are illustrative. ### `codex.tool_call` ```json { "timestamp": "2026-06-27T03:45:20.443Z", "level": "INFO", "fields": { "message": "tool call completed", "event.name": "codex.tool_call", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "conversation.id": "67e55044-10b1-426f-9247-bb680e5fe0c8", "turn_id": "019f04f8-6ac2-78f1-8625-f04a6d35af18", "tool_name": "exec_command", "call_id": "call_7b8483", "tool_source": "direct", "execution_started": true, "dispatch_duration_ms": 12, "handler_duration_ms": 431, "total_duration_ms": 443 }, "target": "codex_core::tools::parallel" } ``` If execution never starts, `execution_started` is `false`, `handler_duration_ms` is `0`, and `dispatch_duration_ms` covers the full observed lifetime. If a duration cannot be represented as an unsigned 64-bit millisecond value, all three duration fields are omitted rather than populated with a sentinel that could corrupt downstream calculations. ## Test plan - `just test -p codex-core tool_call_timing_guard_ignores_code_mode_source` - `just test -p codex-core cancellation_before_dispatch_admission_logs_dispatch_only_timing` - `just test -p codex-app-server app_server_emits_structured_tool_call_timing_event` --- [//]: # (BEGIN SAPLING FOOTER) Stack created with [Sapling](https://sapling-scm.com). Best reviewed with [ReviewStack](https://reviewstack.dev/openai/codex/pull/30334). * __->__ #30334
This commit is contained in:
Michael Bolin
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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,31 @@ 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 that emits JSON logs.
|
||||
///
|
||||
/// `rust_log` is the value to use for the `RUST_LOG` environment variable.
|
||||
pub async fn new_with_auto_env_and_json_logging(
|
||||
codex_home: &Path,
|
||||
rust_log: impl Into<String>,
|
||||
) -> anyhow::Result<Self> {
|
||||
let rust_log = rust_log.into();
|
||||
Self::new_with_auto_env_and_env(
|
||||
codex_home,
|
||||
&[
|
||||
("LOG_FORMAT", Some("json")),
|
||||
("RUST_LOG", Some(rust_log.as_str())),
|
||||
],
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
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 +200,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 +210,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 +231,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 +373,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 +391,7 @@ impl TestAppServer {
|
||||
stdout,
|
||||
pending_messages: VecDeque::new(),
|
||||
auto_env: None,
|
||||
json_logs,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -1,8 +1,30 @@
|
||||
use anyhow::Context;
|
||||
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::Value;
|
||||
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 +47,126 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn app_server_emits_structured_tool_call_timing_event() -> 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_json_logging(
|
||||
codex_home.path(),
|
||||
"warn,codex_core::tools::parallel=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 mut tool_call = app_server
|
||||
.wait_for_json_log_event("codex.tool_call")
|
||||
.await?;
|
||||
let tool_call_object = tool_call
|
||||
.as_object_mut()
|
||||
.context("tool call log event must be an object")?;
|
||||
// JsonLogCapture already validates the timestamp as RFC 3339.
|
||||
tool_call_object
|
||||
.remove("timestamp")
|
||||
.context("tool call log event must include a timestamp")?;
|
||||
let fields = tool_call_object
|
||||
.get_mut("fields")
|
||||
.and_then(Value::as_object_mut)
|
||||
.context("tool call log event fields must be an object")?;
|
||||
let trace_id = fields
|
||||
.remove("trace_id")
|
||||
.context("tool call log event must include trace_id")?;
|
||||
anyhow::ensure!(trace_id.is_string(), "trace_id must be a string");
|
||||
let dispatch_duration_ms = fields
|
||||
.remove("dispatch_duration_ms")
|
||||
.and_then(|duration| duration.as_u64())
|
||||
.context("dispatch_duration_ms must be a nonnegative integer")?;
|
||||
let handler_duration_ms = fields
|
||||
.remove("handler_duration_ms")
|
||||
.and_then(|duration| duration.as_u64())
|
||||
.context("handler_duration_ms must be a nonnegative integer")?;
|
||||
let total_duration_ms = fields
|
||||
.remove("total_duration_ms")
|
||||
.and_then(|duration| duration.as_u64())
|
||||
.context("total_duration_ms must be a nonnegative integer")?;
|
||||
let accounted_duration_ms = dispatch_duration_ms
|
||||
.checked_add(handler_duration_ms)
|
||||
.context("dispatch and handler durations must not overflow")?;
|
||||
anyhow::ensure!(
|
||||
total_duration_ms >= accounted_duration_ms
|
||||
&& total_duration_ms - accounted_duration_ms <= 1,
|
||||
"dispatch and handler durations must account for total duration within integer truncation"
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
tool_call,
|
||||
json!({
|
||||
"level": "INFO",
|
||||
"fields": {
|
||||
"message": "tool call completed",
|
||||
"event.name": "codex.tool_call",
|
||||
"conversation.id": thread.id,
|
||||
"turn_id": turn.id,
|
||||
"tool_name": "exec_command",
|
||||
"call_id": "exec-call-1",
|
||||
"tool_source": "direct",
|
||||
"execution_started": true,
|
||||
},
|
||||
"target": "codex_core::tools::parallel",
|
||||
})
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -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,15 @@ 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,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct ToolCallRuntime {
|
||||
router: Arc<ToolRouter>,
|
||||
@@ -96,6 +107,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);
|
||||
@@ -112,13 +128,18 @@ impl ToolCallRuntime {
|
||||
);
|
||||
let abort_dispatch_span = dispatch_span.clone();
|
||||
|
||||
let mut handle: AbortOnDropHandle<Result<AnyToolResult, FunctionCallError>> =
|
||||
let mut dispatch_handle: AbortOnDropHandle<Result<AnyToolResult, FunctionCallError>> =
|
||||
AbortOnDropHandle::new(tokio::spawn(async move {
|
||||
let _guard = if supports_parallel {
|
||||
Either::Left(lock.read().await)
|
||||
} else {
|
||||
Either::Right(lock.write().await)
|
||||
};
|
||||
// Admission through the parallel-execution gate marks the end
|
||||
// of dispatch waiting and the start of handler execution.
|
||||
if let Some(execution_started_at) = execution_started_at {
|
||||
let _ = execution_started_at.set(Instant::now());
|
||||
}
|
||||
|
||||
router
|
||||
.dispatch_tool_call_with_terminal_outcome(
|
||||
@@ -135,28 +156,29 @@ 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)?,
|
||||
res = &mut dispatch_handle => res.map_err(Self::tool_task_join_error)?,
|
||||
_ = cancellation_token.cancelled() => {
|
||||
if terminal_outcome_reached.load(Ordering::Acquire) || handle.is_finished() {
|
||||
handle.await.map_err(Self::tool_task_join_error)?
|
||||
if terminal_outcome_reached.load(Ordering::Acquire) || dispatch_handle.is_finished() {
|
||||
dispatch_handle.await.map_err(Self::tool_task_join_error)?
|
||||
} else {
|
||||
let secs = started.elapsed().as_secs_f32().max(0.1);
|
||||
abort_dispatch_span.record("aborted", true);
|
||||
if wait_for_runtime_cancellation {
|
||||
if terminal_outcome_reached.swap(true, Ordering::AcqRel) {
|
||||
return handle.await.map_err(Self::tool_task_join_error)?;
|
||||
return dispatch_handle.await.map_err(Self::tool_task_join_error)?;
|
||||
}
|
||||
// The abort owns the terminal outcome; await only so
|
||||
// the runtime can finish process teardown.
|
||||
match handle.await {
|
||||
match dispatch_handle.await {
|
||||
Ok(_) => {}
|
||||
Err(err) if err.is_cancelled() => {}
|
||||
Err(err) => return Err(Self::tool_task_join_error(err)),
|
||||
}
|
||||
} else {
|
||||
handle.abort();
|
||||
match handle.await {
|
||||
dispatch_handle.abort();
|
||||
match dispatch_handle.await {
|
||||
Ok(result) => return result,
|
||||
Err(err) if err.is_cancelled() => {}
|
||||
Err(err) => return Err(Self::tool_task_join_error(err)),
|
||||
@@ -237,6 +259,88 @@ impl ToolCallRuntime {
|
||||
}
|
||||
}
|
||||
|
||||
impl ToolCallTimingGuard {
|
||||
fn capture(
|
||||
started_at: Instant,
|
||||
conversation_id: &impl std::fmt::Display,
|
||||
turn_id: &str,
|
||||
call: &ToolCall,
|
||||
source: &ToolCallSource,
|
||||
) -> Option<Self> {
|
||||
// Code-mode calls are nested within a direct code-mode tool call whose
|
||||
// timing already includes them. Suppress nested guards so consumers do
|
||||
// not mistake overlapping events for independent tool-call latency.
|
||||
if !matches!(source, ToolCallSource::Direct) || !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(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ToolCallTimingGuard {
|
||||
fn drop(&mut self) {
|
||||
let completed_at = Instant::now();
|
||||
// Snapshot once so a concurrently-starting dispatch cannot make one
|
||||
// event internally inconsistent.
|
||||
let execution_started_at = self
|
||||
.execution_started_at
|
||||
.get()
|
||||
.copied()
|
||||
.filter(|execution_started_at| *execution_started_at <= completed_at);
|
||||
let duration_ms = |duration: std::time::Duration| u64::try_from(duration.as_millis()).ok();
|
||||
let total_duration_ms = duration_ms(completed_at.duration_since(self.started_at));
|
||||
let dispatch_duration_ms = execution_started_at.map_or_else(
|
||||
|| total_duration_ms,
|
||||
|execution_started_at| {
|
||||
duration_ms(execution_started_at.duration_since(self.started_at))
|
||||
},
|
||||
);
|
||||
let handler_duration_ms = execution_started_at.map_or(Some(0), |execution_started_at| {
|
||||
duration_ms(completed_at.duration_since(execution_started_at))
|
||||
});
|
||||
|
||||
macro_rules! log_tool_call {
|
||||
($dispatch_duration_ms:expr, $handler_duration_ms:expr, $total_duration_ms:expr) => {
|
||||
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 = "direct",
|
||||
execution_started = execution_started_at.is_some(),
|
||||
dispatch_duration_ms = $dispatch_duration_ms,
|
||||
handler_duration_ms = $handler_duration_ms,
|
||||
total_duration_ms = $total_duration_ms,
|
||||
"tool call completed"
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
match (dispatch_duration_ms, handler_duration_ms, total_duration_ms) {
|
||||
(Some(dispatch_duration_ms), Some(handler_duration_ms), Some(total_duration_ms)) => {
|
||||
log_tool_call!(dispatch_duration_ms, handler_duration_ms, total_duration_ms);
|
||||
}
|
||||
_ => {
|
||||
log_tool_call!(
|
||||
tracing::field::Empty,
|
||||
tracing::field::Empty,
|
||||
tracing::field::Empty
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -255,6 +359,154 @@ mod tests {
|
||||
use pretty_assertions::assert_eq;
|
||||
use tokio::sync::Notify;
|
||||
use tokio::sync::oneshot;
|
||||
use tracing_test::internal::MockWriter;
|
||||
|
||||
#[test]
|
||||
fn tool_call_timing_guard_ignores_code_mode_source() {
|
||||
let subscriber = tracing_subscriber::fmt()
|
||||
.with_max_level(tracing::Level::INFO)
|
||||
.finish();
|
||||
tracing::subscriber::with_default(subscriber, || {
|
||||
let call = ToolCall {
|
||||
tool_name: codex_tools::ToolName::plain("test_tool"),
|
||||
call_id: "call-1".to_string(),
|
||||
payload: ToolPayload::Function {
|
||||
arguments: "{}".to_string(),
|
||||
},
|
||||
};
|
||||
let direct_guard = ToolCallTimingGuard::capture(
|
||||
Instant::now(),
|
||||
&"conversation-id",
|
||||
"turn-id",
|
||||
&call,
|
||||
&ToolCallSource::Direct,
|
||||
);
|
||||
assert!(
|
||||
direct_guard.is_some(),
|
||||
"direct tool calls should create a timing guard"
|
||||
);
|
||||
drop(direct_guard);
|
||||
|
||||
let code_mode_guard = ToolCallTimingGuard::capture(
|
||||
Instant::now(),
|
||||
&"conversation-id",
|
||||
"turn-id",
|
||||
&call,
|
||||
&ToolCallSource::CodeMode {
|
||||
cell_id: "cell-1".to_string(),
|
||||
runtime_tool_call_id: "runtime-call-1".to_string(),
|
||||
},
|
||||
);
|
||||
assert!(
|
||||
code_mode_guard.is_none(),
|
||||
"nested code-mode calls should not create overlapping timing events"
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancellation_before_dispatch_admission_logs_dispatch_only_timing() -> anyhow::Result<()>
|
||||
{
|
||||
let (session, turn_context) = crate::session::tests::make_session_and_context().await;
|
||||
let session = Arc::new(session);
|
||||
let turn_context = Arc::new(turn_context);
|
||||
let tool_name = codex_tools::ToolName::plain("test_tool");
|
||||
let handler = Arc::new(ImmediateHandler {
|
||||
tool_name: tool_name.clone(),
|
||||
}) as Arc<dyn CoreToolRuntime>;
|
||||
let step_context = StepContext::for_test(Arc::clone(&turn_context));
|
||||
let router = Arc::new(ToolRouter::from_parts(
|
||||
ToolRegistry::from_tools([handler]),
|
||||
Vec::new(),
|
||||
));
|
||||
let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new()));
|
||||
let runtime = ToolCallRuntime::new(router, session, step_context, tracker);
|
||||
let execution_gate = Arc::clone(&runtime.parallel_execution);
|
||||
let execution_gate_guard = execution_gate
|
||||
.try_write_owned()
|
||||
.expect("execution gate should be available before dispatch starts");
|
||||
let (release_execution_gate_tx, release_execution_gate_rx) = std::sync::mpsc::channel();
|
||||
let execution_gate_task = tokio::task::spawn_blocking(move || {
|
||||
let _execution_gate_guard = execution_gate_guard;
|
||||
release_execution_gate_rx
|
||||
.recv()
|
||||
.expect("test should release the execution gate");
|
||||
});
|
||||
|
||||
let buffer: &'static std::sync::Mutex<Vec<u8>> =
|
||||
Box::leak(Box::new(std::sync::Mutex::new(Vec::new())));
|
||||
let subscriber = tracing_subscriber::fmt()
|
||||
.with_ansi(false)
|
||||
.with_max_level(tracing::Level::INFO)
|
||||
.with_writer(MockWriter::new(buffer))
|
||||
.finish();
|
||||
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
|
||||
|
||||
let cancellation_token = CancellationToken::new();
|
||||
let call = ToolCall {
|
||||
tool_name,
|
||||
call_id: "call-1".to_string(),
|
||||
payload: ToolPayload::Function {
|
||||
arguments: "{}".to_string(),
|
||||
},
|
||||
};
|
||||
let response_task =
|
||||
tokio::spawn(runtime.handle_tool_call(call, cancellation_token.clone()));
|
||||
cancellation_token.cancel();
|
||||
tokio::time::timeout(Duration::from_secs(1), response_task)
|
||||
.await
|
||||
.expect("timed out waiting for cancelled tool response")
|
||||
.expect("cancelled tool response task should join")
|
||||
.expect("cancelled tool call should produce a response");
|
||||
|
||||
let logs = String::from_utf8(
|
||||
buffer
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.clone(),
|
||||
)?;
|
||||
let timing_events = logs
|
||||
.lines()
|
||||
.filter(|line| line.contains("event.name=\"codex.tool_call\""))
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(
|
||||
timing_events.len(),
|
||||
1,
|
||||
"cancelled tool call should emit exactly one timing event; logs:\n{logs}"
|
||||
);
|
||||
let timing_event = timing_events[0];
|
||||
assert!(
|
||||
timing_event.contains("execution_started=false"),
|
||||
"tool cancelled before admission should not report execution started: {timing_event}"
|
||||
);
|
||||
assert!(
|
||||
timing_event.contains("handler_duration_ms=0"),
|
||||
"tool cancelled before admission should report zero handler duration: {timing_event}"
|
||||
);
|
||||
let duration_field = |name: &str| {
|
||||
timing_event.split_whitespace().find_map(|field| {
|
||||
field
|
||||
.strip_prefix(&format!("{name}="))
|
||||
.and_then(|value| value.parse::<u64>().ok())
|
||||
})
|
||||
};
|
||||
let dispatch_duration_ms = duration_field("dispatch_duration_ms")
|
||||
.expect("timing event should include dispatch_duration_ms");
|
||||
let total_duration_ms = duration_field("total_duration_ms")
|
||||
.expect("timing event should include total_duration_ms");
|
||||
assert_eq!(
|
||||
dispatch_duration_ms, total_duration_ms,
|
||||
"tool cancelled before admission should attribute all elapsed time to dispatch: {timing_event}"
|
||||
);
|
||||
release_execution_gate_tx
|
||||
.send(())
|
||||
.expect("execution gate task should remain available");
|
||||
execution_gate_task
|
||||
.await
|
||||
.expect("execution gate task should join");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
struct ImmediateHandler {
|
||||
tool_name: codex_tools::ToolName,
|
||||
|
||||
Reference in New Issue
Block a user