openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex] cover emited events by otel with tests

Browse Source
This commit is contained in:
Anton Panasenko
2025-09-19 14:43:26 -07:00
parent a51cf9266d
commit d958b66179
10 changed files with 1471 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+22
View File
@@ -762,6 +762,7 @@ dependencies = [
"toml",
"toml_edit",
"tracing",
"tracing-test",
"tree-sitter",
"tree-sitter-bash",
"uuid",
@@ -5311,6 +5312,27 @@ dependencies = [
"tracing-log",
]
[[package]]
name = "tracing-test"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "557b891436fe0d5e0e363427fc7f217abf9ccd510d5136549847bdcbcd011d68"
dependencies = [
"tracing-core",
"tracing-subscriber",
"tracing-test-macro",
]
[[package]]
name = "tracing-test-macro"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04659ddb06c87d233c566112c1c9c5b9e98256d9af50ec3bc9c8327f873a7568"
dependencies = [
"quote",
"syn 2.0.104",
]
[[package]]
name = "tree-sitter"
version = "0.25.9"
codex-rs/core/Cargo.toml
+1
View File
@@ -88,6 +88,7 @@ tempfile = "3"
tokio-test = "0.4"
walkdir = "2.5.0"
wiremock = "0.6"
tracing-test = { version = "0.2.5", features = ["no-env-filter"] }
[package.metadata.cargo-shear]
ignored = ["openssl-sys"]
codex-rs/core/src/apply_patch.rs
+7 -6
View File
@@ -52,12 +52,13 @@ pub(crate) async fn apply_patch(
&turn_context.sandbox_policy,
&turn_context.cwd,
) {
SafetyCheck::AutoApprove { user_explicitly_approved, .. } => {
InternalApplyPatchInvocation::DelegateToExec(ApplyPatchExec {
action,
user_explicitly_approved_this_action: user_explicitly_approved,
})
}
SafetyCheck::AutoApprove {
user_explicitly_approved,
..
} => InternalApplyPatchInvocation::DelegateToExec(ApplyPatchExec {
action,
user_explicitly_approved_this_action: user_explicitly_approved,
}),
SafetyCheck::AskUser => {
// Compute a readable summary of path changes to include in the
// approval request so the user can make an informed decision.
codex-rs/core/src/chat_completions.rs
+2 -2
View File
@@ -421,7 +421,7 @@ async fn process_chat_sse<S>(
// OpenAI Chat streaming sends a literal string "[DONE]" when finished.
if sse.data.trim() == "[DONE]" {
otel_event_manager.sse_event(sse.event, start.elapsed());
otel_event_manager.sse_event(&sse.event, start.elapsed());
// Emit any finalized items before closing so downstream consumers receive
// terminal events for both assistant content and raw reasoning.
if !assistant_text.is_empty() {
@@ -466,7 +466,7 @@ async fn process_chat_sse<S>(
}
};
trace!("chat_completions received SSE chunk: {chunk:?}");
otel_event_manager.sse_event(sse.event, start.elapsed());
otel_event_manager.sse_event(&sse.event, start.elapsed());
let choice_opt = chunk.get("choices").and_then(|c| c.get(0));
codex-rs/core/src/client.rs
+406 -108
View File
@@ -600,6 +600,8 @@ async fn process_sse<S>(
}
};
otel_event_manager.sse_event(event.kind.as_str(), start.elapsed());
match event.kind.as_str() {
// Individual output item finalised. Forward immediately so the
// rest of the agent can stream assistant text/functions *live*
@@ -621,7 +623,6 @@ async fn process_sse<S>(
// drop the duplicated list inside `response.completed`.
"response.output_item.done" => {
let Some(item_val) = event.item else {
otel_event_manager.sse_event(event.kind, start.elapsed());
continue
};
let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) else {
@@ -631,15 +632,12 @@ async fn process_sse<S>(
continue;
};
otel_event_manager.sse_event(event.kind, start.elapsed());
let event = ResponseEvent::OutputItemDone(item);
if tx_event.send(Ok(event)).await.is_err() {
return;
}
}
"response.output_text.delta" => {
otel_event_manager.sse_event(event.kind, start.elapsed());
if let Some(delta) = event.delta {
let event = ResponseEvent::OutputTextDelta(delta);
if tx_event.send(Ok(event)).await.is_err() {
@@ -648,7 +646,6 @@ async fn process_sse<S>(
}
}
"response.reasoning_summary_text.delta" => {
otel_event_manager.sse_event(event.kind, start.elapsed());
if let Some(delta) = event.delta {
let event = ResponseEvent::ReasoningSummaryDelta(delta);
if tx_event.send(Ok(event)).await.is_err() {
@@ -657,7 +654,6 @@ async fn process_sse<S>(
}
}
"response.reasoning_text.delta" => {
otel_event_manager.sse_event(event.kind, start.elapsed());
if let Some(delta) = event.delta {
let event = ResponseEvent::ReasoningContentDelta(delta);
if tx_event.send(Ok(event)).await.is_err() {
@@ -666,7 +662,6 @@ async fn process_sse<S>(
}
}
"response.created" => {
otel_event_manager.sse_event(event.kind, start.elapsed());
if event.response.is_some() {
let _ = tx_event.send(Ok(ResponseEvent::Created {})).await;
}
@@ -723,7 +718,6 @@ async fn process_sse<S>(
| "response.in_progress"
| "response.output_text.done" => {}
"response.output_item.added" => {
otel_event_manager.sse_event(event.kind, start.elapsed());
if let Some(item) = event.item.as_ref() {
// Detect web_search_call begin and forward a synthetic event upstream.
if let Some(ty) = item.get("type").and_then(|v| v.as_str())
@@ -743,17 +737,14 @@ async fn process_sse<S>(
}
"response.reasoning_summary_part.added" => {
// Boundary between reasoning summary sections (e.g., titles).
otel_event_manager.sse_event(event.kind, start.elapsed());
let event = ResponseEvent::ReasoningSummaryPartAdded;
if tx_event.send(Ok(event)).await.is_err() {
return;
}
}
"response.reasoning_summary_text.done" => {
otel_event_manager.sse_event(event.kind, start.elapsed());
}
_ => {
otel_event_manager.sse_event(event.kind, start.elapsed());
}
}
}
@@ -828,7 +819,7 @@ mod tests {
use tokio::sync::mpsc;
use tokio_test::io::Builder as IoBuilder;
use tokio_util::io::ReaderStream;
use tracing_test::traced_test;
// ────────────────────────────
// Helpers
// ────────────────────────────
@@ -862,6 +853,29 @@ mod tests {
events
}
async fn run_sse_allow_errors(
events: Vec<serde_json::Value>,
provider: ModelProviderInfo,
otel_event_manager: OtelEventManager,
) {
let chunks: Vec<Vec<u8>> = events
.into_iter()
.map(|event| {
let kind = event
.get("type")
.and_then(|v| v.as_str())
.expect("fixture event missing type");
if event.as_object().map(|o| o.len() == 1).unwrap_or(false) {
format!("event: {kind}\n\n").into_bytes()
} else {
format!("event: {kind}\ndata: {event}\n\n").into_bytes()
}
})
.collect();
let chunk_refs: Vec<&[u8]> = chunks.iter().map(Vec::as_slice).collect();
let _ = collect_events(&chunk_refs, provider, otel_event_manager).await;
}
/// Builds an in-memory SSE stream from JSON fixtures and returns only the
/// successfully parsed events (panics on internal channel errors).
async fn run_sse(
@@ -898,6 +912,378 @@ mod tests {
out
}
fn otel_event_manager() -> OtelEventManager {
OtelEventManager::new(
ConversationId::new(),
"test-model",
"test-slug",
None,
Some(AuthMode::ChatGPT),
true,
"test-terminal".to_string(),
)
}
fn model_provider() -> ModelProviderInfo {
ModelProviderInfo {
name: "test".to_string(),
base_url: Some("https://example.com".to_string()),
env_key: Some("TEST_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1_000),
requires_openai_auth: false,
}
}
#[tokio::test]
#[traced_test]
async fn process_sse_emits_tracing_for_output_item() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse(
vec![
json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"content": [{ "type": "output_text", "text": "hi" }]
}
}),
json!({
"type": "response.completed",
"response": { "id": "resp1" }
}),
],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.output_item.done")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing response.output_item.done event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_emits_failed_event_on_parse_error() {
let chunk = "data: not-json\n\n";
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = collect_events(&[chunk.as_bytes()], provider, otel_event_manager).await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("error.message")
&& line.contains("Failed to parse SSE event")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_stream_error_records_failed_event() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(1);
let stream = stream::iter(vec![Err::<Bytes, CodexErr>(
std::io::Error::new(std::io::ErrorKind::Other, "boom").into(),
)]);
let handle = tokio::spawn(process_sse(
stream,
tx,
provider.stream_idle_timeout(),
otel_event_manager,
));
while rx.recv().await.is_some() {}
handle.await.expect("process_sse task");
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("error.message")
&& line.contains("boom")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_idle_timeout_records_failed_event() {
let otel_event_manager = otel_event_manager();
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(1);
let stream = stream::pending::<std::result::Result<Bytes, CodexErr>>();
let idle_timeout = std::time::Duration::from_millis(1);
let handle = tokio::spawn(process_sse(stream, tx, idle_timeout, otel_event_manager));
let _ = rx.recv().await;
handle.await.expect("process_sse task");
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("error.message")
&& line.contains("idle timeout")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_records_failed_event_when_stream_closes_without_completed() {
let item = json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "hi"}]
}
})
.to_string();
let sse = format!("event: response.output_item.done\ndata: {item}\n\n");
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = collect_events(&[sse.as_bytes()], provider, otel_event_manager).await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("error.message")
&& line.contains("stream closed before response.completed")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_output_item_parse_failure_logs_event() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse_allow_errors(
vec![json!({
"type": "response.output_item.done",
"item": "unexpected"
})],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.output_item.done")
&& line.contains("error.message")
&& line.contains("failed to parse ResponseItem")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_failed_event_records_response_error_message() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse_allow_errors(
vec![json!({
"type": "response.failed",
"response": {
"error": {
"message": "boom",
"code": "bad"
}
}
})],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.failed")
&& line.contains("error.message")
&& line.contains("boom")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_failed_event_logs_parse_error() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse_allow_errors(
vec![json!({
"type": "response.failed",
"response": {
"error": "not-an-object"
}
})],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.failed")
&& line.contains("error.message")
&& line.contains("failed to parse ErrorResponse")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_failed_event_logs_missing_error() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse_allow_errors(
vec![json!({
"type": "response.failed",
"response": {}
})],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.failed")
&& line.contains("error.message")
&& line.contains("response.failed event received")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_failed_event_logs_response_completed_parse_error() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse_allow_errors(
vec![json!({
"type": "response.completed",
"response": {}
})],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.completed")
&& line.contains("error.message")
&& line.contains("failed to parse ResponseCompleted")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing codex.sse_event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn process_sse_emits_completed_telemetry() {
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let _ = run_sse_allow_errors(
vec![json!({
"type": "response.completed",
"response": {
"id": "resp1",
"usage": {
"input_tokens": 3,
"input_tokens_details": { "cached_tokens": 1 },
"output_tokens": 5,
"output_tokens_details": { "reasoning_tokens": 2 },
"total_tokens": 9
}
}
})],
provider,
otel_event_manager,
)
.await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("event.kind=response.completed")
&& line.contains("input_token_count=3")
&& line.contains("output_token_count=5")
&& line.contains("cached_token_count=1")
&& line.contains("reasoning_token_count=2")
&& line.contains("tool_token_count=9")
})
.map(|_| Ok(()))
.unwrap_or(Err("missing response.completed telemetry".to_string()))
});
}
// ────────────────────────────
// Tests from `implement-test-for-responses-api-sse-parser`
// ────────────────────────────
@@ -934,30 +1320,8 @@ mod tests {
let sse2 = format!("event: response.output_item.done\ndata: {item2}\n\n");
let sse3 = format!("event: response.completed\ndata: {completed}\n\n");
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
};
let otel_event_manager = OtelEventManager::new(
ConversationId::new(),
"test",
"test",
None,
Some(AuthMode::ChatGPT),
false,
"test".to_string(),
);
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let events = collect_events(
&[sse1.as_bytes(), sse2.as_bytes(), sse3.as_bytes()],
@@ -1005,30 +1369,8 @@ mod tests {
.to_string();
let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n");
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
};
let otel_event_manager = OtelEventManager::new(
ConversationId::new(),
"test",
"test",
None,
Some(AuthMode::ChatGPT),
false,
"test".to_string(),
);
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await;
@@ -1049,30 +1391,8 @@ mod tests {
let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_689bcf18d7f08194bf3440ba62fe05d803fee0cdac429894","object":"response","created_at":1755041560,"status":"failed","background":false,"error":{"code":"rate_limit_exceeded","message":"Rate limit reached for gpt-5 in organization org-AAA on tokens per min (TPM): Limit 30000, Used 22999, Requested 12528. Please try again in 11.054s. Visit https://platform.openai.com/account/rate-limits to learn more."}, "usage":null,"user":null,"metadata":{}}}"#;
let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
};
let otel_event_manager = OtelEventManager::new(
ConversationId::new(),
"test",
"test",
None,
Some(AuthMode::ChatGPT),
false,
"test".to_string(),
);
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await;
@@ -1164,30 +1484,8 @@ mod tests {
let mut evs = vec![case.event];
evs.push(completed.clone());
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: Some("https://test.com".to_string()),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
requires_openai_auth: false,
};
let otel_event_manager = OtelEventManager::new(
ConversationId::new(),
"test",
"test",
None,
Some(AuthMode::ChatGPT),
false,
"test".to_string(),
);
let provider = model_provider();
let otel_event_manager = otel_event_manager();
let out = run_sse(evs, provider, otel_event_manager).await;
assert_eq!(out.len(), case.expected_len, "case {}", case.name);
codex-rs/core/src/codex.rs
+843 -3
View File
@@ -1,8 +1,8 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::Path;
use std::fmt::Debug;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
@@ -467,7 +467,7 @@ impl Session {
config.approval_policy,
config.sandbox_policy.clone(),
config.mcp_servers.keys().map(|k| k.as_str()).collect(),
config.active_profile.clone()
config.active_profile.clone(),
);
// Now that the conversation id is final (may have been updated by resume),
@@ -3540,19 +3540,30 @@ mod tests {
use super::*;
use crate::config::ConfigOverrides;
use crate::config::ConfigToml;
use crate::exec::ExecParams;
use crate::protocol::CompactedItem;
use crate::protocol::InitialHistory;
use crate::protocol::ResumedHistory;
use crate::turn_diff_tracker::TurnDiffTracker;
use codex_protocol::mcp_protocol::AuthMode;
use codex_protocol::models::ContentItem;
use codex_protocol::models::LocalShellAction;
use codex_protocol::models::LocalShellExecAction;
use codex_protocol::models::LocalShellStatus;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::ReviewDecision;
use mcp_types::ContentBlock;
use mcp_types::TextContent;
use pretty_assertions::assert_eq;
use serde::Deserialize;
use serde_json::json;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration as StdDuration;
use tempfile::tempdir;
use tracing_test::traced_test;
#[test]
fn reconstruct_history_matches_live_compactions() {
@@ -3595,6 +3606,52 @@ mod tests {
assert_eq!(expected, actual);
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn session_new_emits_conversation_start_event() {
let (tx_event, _rx_event) = async_channel::unbounded();
let codex_home = tempfile::tempdir().expect("create temp dir");
let config = Config::load_from_base_config_with_overrides(
ConfigToml::default(),
ConfigOverrides::default(),
codex_home.path().to_path_buf(),
)
.expect("load default test config");
let codex_home_path = config.codex_home.clone();
let config = Arc::new(config);
let auth_manager = AuthManager::shared(codex_home_path);
let configure_session = ConfigureSession {
provider: config.model_provider.clone(),
model: config.model.clone(),
model_reasoning_effort: config.model_reasoning_effort,
model_reasoning_summary: config.model_reasoning_summary,
user_instructions: config.user_instructions.clone(),
base_instructions: config.base_instructions.clone(),
approval_policy: config.approval_policy,
sandbox_policy: config.sandbox_policy.clone(),
notify: config.notify.clone(),
cwd: config.cwd.clone(),
};
let _ = Session::new(
configure_session,
Arc::clone(&config),
auth_manager,
tx_event,
InitialHistory::New,
)
.await
.expect("session setup succeeds");
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| line.contains("codex.conversation_starts"))
.map(|_| Ok(()))
.unwrap_or_else(|| Err("expected codex.conversation_starts event".to_string()))
});
}
#[test]
fn prefers_structured_content_when_present() {
let ctr = CallToolResult {
@@ -3990,6 +4047,7 @@ mod tests {
let mut turn_diff_tracker = TurnDiffTracker::new();
let tool_name = "shell";
let sub_id = "test-sub".to_string();
let call_id = "test-call".to_string();
@@ -3998,6 +4056,7 @@ mod tests {
&session,
&turn_context,
&mut turn_diff_tracker,
tool_name,
sub_id,
call_id,
)
@@ -4012,7 +4071,7 @@ mod tests {
policy = turn_context.approval_policy
);
pretty_assertions::assert_eq!(output.content, expected);
assert_eq!(output.content, expected);
// Now retry the same command WITHOUT escalated permissions; should succeed.
// Force DangerFullAccess to avoid platform sandbox dependencies in tests.
@@ -4023,6 +4082,7 @@ mod tests {
&session,
&turn_context,
&mut turn_diff_tracker,
tool_name,
"test-sub".to_string(),
"test-call-2".to_string(),
)
@@ -4050,4 +4110,784 @@ mod tests {
assert!(exec_output.output.contains("hi"));
pretty_assertions::assert_eq!(output.success, Some(true));
}
fn exec_params(command: Vec<String>, cwd: &Path) -> ExecParams {
ExecParams {
command,
cwd: cwd.to_path_buf(),
timeout_ms: None,
env: std::collections::HashMap::new(),
with_escalated_permissions: None,
justification: None,
}
}
async fn respond_to_pending_approval(
session: Arc<Session>,
sub_id: String,
decision: ReviewDecision,
) {
for _ in 0..100 {
let requested = session
.state
.lock()
.await
.pending_approvals
.contains_key(&sub_id);
if requested {
session.notify_approval(&sub_id, decision).await;
return;
}
tokio::time::sleep(StdDuration::from_millis(10)).await;
}
panic!("approval {sub_id} was never requested");
}
async fn run_handle_container_exec(
session: Arc<Session>,
turn_context: &TurnContext,
params: ExecParams,
tool_name: &str,
sub_id: &str,
call_id: &str,
approval: Option<ReviewDecision>,
) -> ResponseInputItem {
let approval_task = approval.map(|decision| {
tokio::spawn(respond_to_pending_approval(
Arc::clone(&session),
sub_id.to_owned(),
decision,
))
});
let mut tracker = TurnDiffTracker::new();
let response = handle_container_exec_with_params(
params,
session.as_ref(),
turn_context,
&mut tracker,
&tool_name,
sub_id.to_owned(),
call_id.to_owned(),
)
.await;
if let Some(task) = approval_task {
task.await.expect("approval task completed");
}
response
}
async fn run_handle_sandbox(
session: Arc<Session>,
turn_context: &TurnContext,
params: ExecParams,
exec_command_context: ExecCommandContext,
sandbox_type: SandboxType,
approval: Option<ReviewDecision>,
) -> ResponseInputItem {
let sub_id = exec_command_context.sub_id.clone();
let approval_task = approval.map(|decision| {
tokio::spawn(respond_to_pending_approval(
Arc::clone(&session),
sub_id.clone(),
decision,
))
});
let mut tracker = TurnDiffTracker::new();
let otel_event_manager = turn_context.client.get_otel_event_manager();
let response = handle_sandbox_error(
"shell",
&mut tracker,
params,
exec_command_context,
SandboxErr::Signal(9),
sandbox_type,
session.as_ref(),
turn_context,
&otel_event_manager,
)
.await;
if let Some(task) = approval_task {
task.await.expect("approval task completed");
}
response
}
#[tokio::test]
#[traced_test]
async fn handle_response_item_records_tool_result_for_custom_tool_call() {
let mut tracker = TurnDiffTracker::new();
let (session, turn_context) = make_session_and_context();
let call_id = "custom-tool-call".to_string();
let tool_name = "unsupported_tool".to_string();
let tool_input = "{\"key\":\"value\"}".to_string();
let item = ResponseItem::CustomToolCall {
id: Some("item".to_string()),
status: None,
call_id: call_id.clone(),
name: tool_name.clone(),
input: tool_input.clone(),
};
let result = handle_response_item(&session, &turn_context, &mut tracker, "subtask", item)
.await
.expect("handle_response_item succeeds");
match result {
Some(ResponseInputItem::CustomToolCallOutput { output, .. }) => {
assert_eq!(format!("unsupported custom tool call: {tool_name}"), output);
}
other => panic!("unexpected response: {other:?}"),
}
logs_assert(|lines: &[&str]| {
let line = lines
.iter()
.find(|line| {
line.contains("codex.tool_result") && line.contains("call_id=custom-tool-call")
})
.ok_or_else(|| "missing codex.tool_result event".to_string())?;
if !line.contains("tool_name=unsupported_tool") {
return Err("missing tool_name field".to_string());
}
if !line.contains("arguments={\"key\":\"value\"}") {
return Err("missing arguments field".to_string());
}
if !line.contains("output=unsupported custom tool call: unsupported_tool") {
return Err("missing output field".to_string());
}
if !line.contains("success=true") {
return Err("missing success field".to_string());
}
Ok(())
});
}
#[tokio::test]
#[traced_test]
async fn handle_response_item_records_tool_result_for_function_call() {
let mut tracker = TurnDiffTracker::new();
let (session, turn_context) = make_session_and_context();
let call_id = "function-call".to_string();
let tool_name = "nonexistent".to_string();
let arguments = "{\"value\":1}".to_string();
let sub_id = "function-sub";
let item = ResponseItem::FunctionCall {
id: Some("item".to_string()),
name: tool_name.clone(),
arguments: arguments.clone(),
call_id: call_id.clone(),
};
let result = handle_response_item(&session, &turn_context, &mut tracker, sub_id, item)
.await
.expect("handle_response_item succeeds");
match result {
Some(ResponseInputItem::FunctionCallOutput { output, .. }) => {
assert_eq!(format!("unsupported call: {tool_name}"), output.content);
}
other => panic!("unexpected response: {other:?}"),
}
logs_assert(|lines: &[&str]| {
let line = lines
.iter()
.find(|line| {
line.contains("codex.tool_result") && line.contains("call_id=function-call")
})
.ok_or_else(|| "missing codex.tool_result event".to_string())?;
if !line.contains("tool_name=nonexistent") {
return Err("missing tool_name field".to_string());
}
if !line.contains("arguments={\"value\":1}") {
return Err("missing arguments field".to_string());
}
if !line.contains("output=unsupported call: nonexistent") {
return Err("missing output field".to_string());
}
if !line.contains("success=false") {
return Err("missing success field".to_string());
}
Ok(())
});
}
#[tokio::test]
#[traced_test]
async fn handle_response_item_records_tool_result_for_local_shell_missing_ids() {
let mut tracker = TurnDiffTracker::new();
let (session, turn_context) = make_session_and_context();
let sub_id = "shell-sub";
let item = ResponseItem::LocalShellCall {
id: None,
call_id: None,
status: LocalShellStatus::Completed,
action: LocalShellAction::Exec(LocalShellExecAction {
command: vec!["/bin/echo".to_string(), "hello".to_string()],
timeout_ms: None,
working_directory: None,
env: None,
user: None,
}),
};
let result = handle_response_item(&session, &turn_context, &mut tracker, sub_id, item)
.await
.expect("handle_response_item succeeds");
match result {
Some(ResponseInputItem::FunctionCallOutput { output, call_id }) => {
assert!(call_id.is_empty());
assert_eq!("LocalShellCall without call_id or id", output.content);
}
other => panic!("unexpected response: {other:?}"),
}
logs_assert(|lines: &[&str]| {
let line = lines
.iter()
.find(|line| {
line.contains("codex.tool_result")
&& line.contains("tool_name=local_shell")
&& line.contains("output=LocalShellCall without call_id or id")
})
.ok_or_else(|| "missing codex.tool_result event".to_string())?;
if !line.contains("call_id=") {
return Err("missing call_id field".to_string());
}
if !line.contains("arguments=") {
return Err("missing arguments field".to_string());
}
if !line.contains("success=false") {
return Err("missing success field".to_string());
}
Ok(())
});
}
#[tokio::test]
#[traced_test]
async fn handle_response_item_records_tool_result_for_local_shell_call() {
let mut tracker = TurnDiffTracker::new();
let (session, turn_context) = make_session_and_context();
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let call_id = "shell-call".to_string();
let sub_id = "shell-sub";
let item = ResponseItem::LocalShellCall {
id: Some("item".to_string()),
call_id: Some(call_id.clone()),
status: LocalShellStatus::Completed,
action: LocalShellAction::Exec(LocalShellExecAction {
command: vec!["/bin/echo".to_string(), "shell".to_string()],
timeout_ms: None,
working_directory: None,
env: None,
user: None,
}),
};
let result = handle_response_item(&session, &turn_context, &mut tracker, sub_id, item)
.await
.expect("handle_response_item succeeds");
match result {
Some(ResponseInputItem::FunctionCallOutput { output, .. }) => {
assert_eq!(Some(true), output.success);
}
other => panic!("unexpected response: {other:?}"),
}
logs_assert(|lines: &[&str]| {
let line = lines
.iter()
.find(|line| {
line.contains("codex.tool_result") && line.contains("call_id=shell-call")
})
.ok_or_else(|| "missing codex.tool_result event".to_string())?;
if !line.contains("tool_name=local_shell") {
return Err("missing tool_name field".to_string());
}
if !line.contains("arguments=/bin/echo shell") {
return Err("missing arguments field".to_string());
}
let output_idx = line
.find("output=")
.ok_or_else(|| "missing output field".to_string())?;
if line[output_idx + "output=".len()..].is_empty() {
return Err("empty output field".to_string());
}
if !line.contains("success=true") {
return Err("missing success field".to_string());
}
Ok(())
});
}
fn tool_decision_assertion<'a>(
call_id: &'a str,
expected_decision: &'a str,
expected_source: &'a str,
) -> impl Fn(&[&str]) -> Result<(), String> + 'a {
let call_id = call_id.to_string();
let expected_decision = expected_decision.to_string();
let expected_source = expected_source.to_string();
move |lines: &[&str]| {
let line = lines
.iter()
.find(|line| {
line.contains("codex.tool_decision")
&& line.contains(&format!("call_id={call_id}"))
})
.ok_or_else(|| format!("missing codex.tool_decision event for {call_id}"))?;
let lower = line.to_lowercase();
if !lower.contains("tool_name=shell") {
return Err(format!("missing tool_name for {call_id}"));
}
if !lower.contains(&format!("decision={}", expected_decision)) {
return Err(format!("unexpected decision for {call_id}"));
}
if !lower.contains(&format!("source={}", expected_source)) {
return Err(format!("unexpected source for {call_id}"));
}
Ok(())
}
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_autoapprove_from_config_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::OnRequest;
turn_context.sandbox_policy = SandboxPolicy::DangerFullAccess;
let command = vec!["/bin/echo".to_string(), "hello".to_string()];
let params = exec_params(command, &turn_context.cwd);
let _response = run_handle_container_exec(
Arc::new(session),
&turn_context,
params,
"shell",
"auto_config_sub",
"auto_config_call",
None,
)
.await;
logs_assert(tool_decision_assertion(
"auto_config_call",
"approved",
"config",
));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_apply_patch_user_autoapprove_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let temp_dir = tempdir().expect("create apply_patch temp dir");
turn_context.cwd = temp_dir.path().to_path_buf();
let patch =
"*** Begin Patch\n*** Add File: new_file.txt\n+hello\n*** End Patch\n".to_string();
let params = ExecParams {
command: vec!["apply_patch".to_string(), patch],
cwd: turn_context.cwd.clone(),
timeout_ms: None,
env: std::collections::HashMap::new(),
with_escalated_permissions: None,
justification: None,
};
let _response = run_handle_container_exec(
Arc::new(session),
&turn_context,
params,
"shell",
"auto_user_sub",
"auto_user_call",
Some(ReviewDecision::Approved),
)
.await;
logs_assert(tool_decision_assertion(
"auto_user_call",
"approved",
"user",
));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_user_approved_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let params = exec_params(
vec!["/bin/echo".to_string(), "approved".to_string()],
&turn_context.cwd,
);
let _response = run_handle_container_exec(
Arc::new(session),
&turn_context,
params,
"shell",
"user_approved_sub",
"user_approved_call",
Some(ReviewDecision::Approved),
)
.await;
logs_assert(tool_decision_assertion(
"user_approved_call",
"approved",
"user",
));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_user_approved_for_session_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let command = vec!["/bin/echo".to_string(), "persist".to_string()];
let params = exec_params(command.clone(), &turn_context.cwd);
let session = Arc::new(session);
let _response = run_handle_container_exec(
session.clone(),
&turn_context,
params,
"shell",
"user_approved_session_sub",
"user_approved_session_call",
Some(ReviewDecision::ApprovedForSession),
)
.await;
logs_assert(tool_decision_assertion(
"user_approved_session_call",
"approvedforsession",
"user",
));
let state = session.state.lock().await;
assert!(state.approved_commands.contains(&command));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_user_denies_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let params = exec_params(
vec!["/bin/echo".to_string(), "deny".to_string()],
&turn_context.cwd,
);
let response = run_handle_container_exec(
Arc::new(session),
&turn_context,
params,
"shell",
"user_denied_sub",
"user_denied_call",
Some(ReviewDecision::Denied),
)
.await;
logs_assert(tool_decision_assertion(
"user_denied_call",
"denied",
"user",
));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert_eq!("exec command rejected by user", output.content);
assert_eq!(None, output.success);
}
other => panic!("unexpected response for denied branch: {other:?}"),
}
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_user_aborts_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let params = exec_params(
vec!["/bin/echo".to_string(), "abort".to_string()],
&turn_context.cwd,
);
let response = run_handle_container_exec(
Arc::new(session),
&turn_context,
params,
"shell",
"user_abort_sub",
"user_abort_call",
Some(ReviewDecision::Abort),
)
.await;
logs_assert(tool_decision_assertion("user_abort_call", "abort", "user"));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert_eq!("exec command aborted by user", output.content);
assert_eq!(None, output.success);
}
other => panic!("unexpected response for abort branch: {other:?}"),
}
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_sandbox_error_user_approves_retry_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let command = vec!["/bin/echo".to_string(), "retry".to_string()];
let params = exec_params(command.clone(), &turn_context.cwd);
let exec_command_context = ExecCommandContext {
sub_id: "sandbox_retry_sub".to_string(),
call_id: "sandbox_retry_call".to_string(),
command_for_display: command.clone(),
cwd: turn_context.cwd.clone(),
apply_patch: None,
};
let session = Arc::new(session);
let response = run_handle_sandbox(
Arc::clone(&session),
&turn_context,
params,
exec_command_context,
SandboxType::MacosSeatbelt,
Some(ReviewDecision::Approved),
)
.await;
logs_assert(tool_decision_assertion(
"sandbox_retry_call",
"approved",
"user",
));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert_eq!(Some(true), output.success);
}
other => panic!("unexpected response for sandbox retry: {other:?}"),
}
let state = session.state.lock().await;
assert!(state.approved_commands.contains(&command));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_sandbox_error_user_approves_for_session_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let command = vec!["/bin/echo".to_string(), "persist".to_string()];
let params = exec_params(command.clone(), &turn_context.cwd);
let exec_command_context = ExecCommandContext {
sub_id: "sandbox_session_sub".to_string(),
call_id: "sandbox_session_call".to_string(),
command_for_display: command.clone(),
cwd: turn_context.cwd.clone(),
apply_patch: None,
};
let session = Arc::new(session);
let response = run_handle_sandbox(
Arc::clone(&session),
&turn_context,
params,
exec_command_context,
SandboxType::MacosSeatbelt,
Some(ReviewDecision::ApprovedForSession),
)
.await;
logs_assert(tool_decision_assertion(
"sandbox_session_call",
"approvedforsession",
"user",
));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert_eq!(Some(true), output.success);
}
other => {
panic!("unexpected response for sandbox approved session retry: {other:?}")
}
}
let state = session.state.lock().await;
assert!(state.approved_commands.contains(&command));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_sandbox_error_user_denies_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let command = vec!["/bin/echo".to_string(), "deny".to_string()];
let params = exec_params(command.clone(), &turn_context.cwd);
let exec_command_context = ExecCommandContext {
sub_id: "sandbox_deny_sub".to_string(),
call_id: "sandbox_deny_call".to_string(),
command_for_display: command.clone(),
cwd: turn_context.cwd.clone(),
apply_patch: None,
};
let session = Arc::new(session);
let response = run_handle_sandbox(
Arc::clone(&session),
&turn_context,
params,
exec_command_context,
SandboxType::MacosSeatbelt,
Some(ReviewDecision::Denied),
)
.await;
logs_assert(tool_decision_assertion(
"sandbox_deny_call",
"denied",
"user",
));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert_eq!("exec command rejected by user", output.content);
assert_eq!(None, output.success);
}
other => panic!("unexpected response for sandbox deny: {other:?}"),
}
let state = session.state.lock().await;
assert!(!state.approved_commands.contains(&command));
}
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_sandbox_error_user_aborts_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::UnlessTrusted;
let command = vec!["/bin/echo".to_string(), "abort".to_string()];
let params = exec_params(command.clone(), &turn_context.cwd);
let exec_command_context = ExecCommandContext {
sub_id: "sandbox_abort_sub".to_string(),
call_id: "sandbox_abort_call".to_string(),
command_for_display: command,
cwd: turn_context.cwd.clone(),
apply_patch: None,
};
let session = Arc::new(session);
let response = run_handle_sandbox(
session,
&turn_context,
params,
exec_command_context,
SandboxType::MacosSeatbelt,
Some(ReviewDecision::Abort),
)
.await;
logs_assert(tool_decision_assertion(
"sandbox_abort_call",
"abort",
"user",
));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert_eq!("exec command rejected by user", output.content);
assert_eq!(None, output.success);
}
other => panic!("unexpected response for sandbox abort: {other:?}"),
}
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
#[tokio::test(flavor = "current_thread")]
#[traced_test]
async fn handle_container_exec_config_reject_records_tool_decision() {
let (session, mut turn_context) = make_session_and_context();
turn_context.approval_policy = AskForApproval::Never;
turn_context.sandbox_policy = SandboxPolicy::ReadOnly;
let params = exec_params(
vec!["/bin/echo".to_string(), "reject".to_string()],
&turn_context.cwd,
);
let response = run_handle_container_exec(
Arc::new(session),
&turn_context,
params,
"shell",
"config_reject_sub",
"config_reject_call",
None,
)
.await;
logs_assert(tool_decision_assertion(
"config_reject_call",
"denied",
"config",
));
match response {
ResponseInputItem::FunctionCallOutput { output, .. } => {
assert!(output.content.starts_with("exec command rejected"));
assert_eq!(None, output.success);
}
other => panic!("unexpected response for reject branch: {other:?}"),
}
}
}
codex-rs/core/src/otel_init.rs
+1 -1
View File
@@ -1,12 +1,12 @@
use crate::config::Config;
use crate::config_types::OtelExporterKind as Kind;
use crate::config_types::OtelHttpProtocol as Protocol;
use crate::default_client::ORIGINATOR;
use codex_otel::config::OtelExporter;
use codex_otel::config::OtelHttpProtocol;
use codex_otel::config::OtelSettings;
use codex_otel::otel_provider::OtelProvider;
use std::error::Error;
use crate::default_client::ORIGINATOR;
/// Build an OpenTelemetry provider from the app Config.
///
codex-rs/core/tests/chat_completions_sse.rs
+93 -2
View File
@@ -1,4 +1,5 @@
use std::sync::Arc;
use tracing_test::traced_test;
use codex_core::ContentItem;
use codex_core::ModelClient;
@@ -25,11 +26,15 @@ fn network_disabled() -> bool {
}
async fn run_stream(sse_body: &str) -> Vec<ResponseEvent> {
run_stream_with_bytes(sse_body.as_bytes()).await
}
async fn run_stream_with_bytes(sse_body: &[u8]) -> Vec<ResponseEvent> {
let server = MockServer::start().await;
let template = ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(sse_body.to_string(), "text/event-stream");
.set_body_bytes(sse_body.to_vec());
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
@@ -104,7 +109,8 @@ async fn run_stream(sse_body: &str) -> Vec<ResponseEvent> {
while let Some(event) = stream.next().await {
match event {
Ok(ev) => events.push(ev),
Err(e) => panic!("stream event error: {e}"),
// We still collect the error to exercise telemetry and complete the task.
Err(_e) => break,
}
}
events
@@ -333,3 +339,88 @@ async fn streams_reasoning_before_tool_call() {
assert!(matches!(events[3], ResponseEvent::Completed { .. }));
}
#[tokio::test]
#[traced_test]
async fn chat_sse_emits_failed_on_parse_error() {
if network_disabled() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
let sse_body = concat!("data: not-json\n\n", "data: [DONE]\n\n");
let _ = run_stream(sse_body).await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.api_request") && line.contains("http.response.status_code=200")
})
.map(|_| Ok(()))
.unwrap_or(Err("cannot find codex.api_request event".to_string()))
});
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("error.message")
&& line.contains("Failed to parse SSE event")
})
.map(|_| Ok(()))
.unwrap_or(Err("cannot find SSE event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn chat_sse_done_chunk_emits_event() {
if network_disabled() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
let sse_body = concat!("data: [DONE]\n\n",);
let _ = run_stream(sse_body).await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| line.contains("codex.sse_event") && line.contains("event.kind=message"))
.map(|_| Ok(()))
.unwrap_or(Err("cannot find SSE event".to_string()))
});
}
#[tokio::test]
#[traced_test]
async fn chat_sse_emits_error_on_invalid_utf8() {
if network_disabled() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
let _ = run_stream_with_bytes(b"data: \x80\x80\n\n").await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("codex.sse_event")
&& line.contains("error.message")
&& line.contains("UTF8 error: invalid utf-8 sequence of 1 bytes from index 0")
})
.map(|_| Ok(()))
.unwrap_or(Err("cannot find SSE event".to_string()))
});
}
codex-rs/core/tests/suite/client.rs
+55
View File
@@ -30,6 +30,7 @@ use serde_json::json;
use std::io::Write;
use std::sync::Arc;
use tempfile::TempDir;
use tracing_test::traced_test;
use uuid::Uuid;
use wiremock::Mock;
use wiremock::MockServer;
@@ -1089,3 +1090,57 @@ async fn history_dedupes_streamed_and_final_messages_across_turns() {
"request 3 tail mismatch",
);
}
#[tokio::test]
#[traced_test]
async fn responses_api_emits_api_request_event() {
let server = MockServer::start().await;
let first = ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(sse_completed("resp1"), "text/event-stream");
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(first)
.expect(1)
.mount(&server)
.await;
let model_provider = ModelProviderInfo {
base_url: Some(format!("{}/v1", server.uri())),
..built_in_model_providers()["openai"].clone()
};
let codex_home = TempDir::new().unwrap();
let mut config = load_default_config_for_test(&codex_home);
config.model_provider = model_provider;
config.user_instructions = Some("be nice".to_string());
let conversation_manager =
ConversationManager::with_auth(CodexAuth::from_api_key("Test API Key"));
let codex = conversation_manager
.new_conversation(config)
.await
.expect("create new conversation")
.conversation;
codex
.submit(Op::UserInput {
items: vec![InputItem::Text {
text: "hello".into(),
}],
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| line.contains("codex.api_request"))
.map(|_| Ok(()))
.unwrap_or_else(|| Err("expected codex.api_request event".to_string()))
});
}
codex-rs/otel/src/otel_event_manager.rs
+41 -21
View File
@@ -1,15 +1,18 @@
use chrono::SecondsFormat;
use chrono::Utc;
use codex_protocol::config_types::ReasoningEffort;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::mcp_protocol::AuthMode;
use codex_protocol::mcp_protocol::ConversationId;
use codex_protocol::protocol::{AskForApproval, InputItem, SandboxPolicy};
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::InputItem;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::SandboxPolicy;
use reqwest::Error;
use reqwest::Response;
use serde::Serialize;
use std::time::Duration;
use strum_macros::Display;
use codex_protocol::config_types::{ReasoningEffort, ReasoningSummary};
#[derive(Debug, Clone, Serialize, Display)]
#[serde(rename_all = "snake_case")]
@@ -66,6 +69,7 @@ impl OtelEventManager {
manager
}
#[allow(clippy::too_many_arguments)]
pub fn conversation_starts(
&self,
provider_name: &str,
@@ -134,7 +138,7 @@ impl OtelEventManager {
);
}
pub fn sse_event(&self, kind: String, duration: Duration) {
pub fn sse_event(&self, kind: &str, duration: Duration) {
tracing::event!(
tracing::Level::INFO,
event.name = "codex.sse_event",
@@ -152,21 +156,37 @@ impl OtelEventManager {
}
pub fn sse_event_failed(&self, kind: Option<String>, duration: Duration, error: &str) {
tracing::event!(
tracing::Level::INFO,
event.name = "codex.sse_event",
event.timestamp = %timestamp(),
event.kind = kind,
conversation.id = %self.metadata.conversation_id,
app.version = %self.metadata.app_version,
auth_mode = self.metadata.auth_mode,
user.account_id = self.metadata.account_id,
terminal.type = %self.metadata.terminal_type,
model = %self.metadata.model,
slug = %self.metadata.slug,
duration_ms = %duration.as_millis(),
error.message = %error,
);
match kind {
Some(kind) => tracing::event!(
tracing::Level::INFO,
event.name = "codex.sse_event",
event.timestamp = %timestamp(),
event.kind = %kind,
conversation.id = %self.metadata.conversation_id,
app.version = %self.metadata.app_version,
auth_mode = self.metadata.auth_mode,
user.account_id = self.metadata.account_id,
terminal.type = %self.metadata.terminal_type,
model = %self.metadata.model,
slug = %self.metadata.slug,
duration_ms = %duration.as_millis(),
error.message = %error,
),
None => tracing::event!(
tracing::Level::INFO,
event.name = "codex.sse_event",
event.timestamp = %timestamp(),
conversation.id = %self.metadata.conversation_id,
app.version = %self.metadata.app_version,
auth_mode = self.metadata.auth_mode,
user.account_id = self.metadata.account_id,
terminal.type = %self.metadata.terminal_type,
model = %self.metadata.model,
slug = %self.metadata.slug,
duration_ms = %duration.as_millis(),
error.message = %error,
),
}
}
pub fn sse_event_completed(
@@ -182,7 +202,7 @@ impl OtelEventManager {
tracing::Level::INFO,
event.name = "codex.sse_event",
event.timestamp = %timestamp(),
event.kind = "response.completed",
event.kind = %"response.completed",
conversation.id = %self.metadata.conversation_id,
app.version = %self.metadata.app_version,
auth_mode = self.metadata.auth_mode,
@@ -250,8 +270,8 @@ impl OtelEventManager {
slug = %self.metadata.slug,
tool_name = %tool_name,
call_id = %call_id,
decision = decision.to_string().to_lowercase(),
source = source.to_string(),
decision = %decision.to_string().to_lowercase(),
source = %source.to_string(),
);
}