From d958b661798395f634b6e57fa92bcf18aa3ee939 Mon Sep 17 00:00:00 2001 From: Anton Panasenko Date: Thu, 18 Sep 2025 13:51:16 -0700 Subject: [PATCH] [codex] cover emited events by otel with tests --- codex-rs/Cargo.lock | 22 + codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/apply_patch.rs | 13 +- codex-rs/core/src/chat_completions.rs | 4 +- codex-rs/core/src/client.rs | 514 +++++++++--- codex-rs/core/src/codex.rs | 846 +++++++++++++++++++- codex-rs/core/src/otel_init.rs | 2 +- codex-rs/core/tests/chat_completions_sse.rs | 95 ++- codex-rs/core/tests/suite/client.rs | 55 ++ codex-rs/otel/src/otel_event_manager.rs | 62 +- 10 files changed, 1471 insertions(+), 143 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 3495e18b9a..f8441f42ce 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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" diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 4c92d133f5..8a8816b208 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -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"] diff --git a/codex-rs/core/src/apply_patch.rs b/codex-rs/core/src/apply_patch.rs index 213b3d3ccd..8a58cc2d5a 100644 --- a/codex-rs/core/src/apply_patch.rs +++ b/codex-rs/core/src/apply_patch.rs @@ -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. diff --git a/codex-rs/core/src/chat_completions.rs b/codex-rs/core/src/chat_completions.rs index 71359ccc0a..53085f8b2b 100644 --- a/codex-rs/core/src/chat_completions.rs +++ b/codex-rs/core/src/chat_completions.rs @@ -421,7 +421,7 @@ async fn process_chat_sse( // 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( } }; 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)); diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index dc22c90275..9e130153be 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -600,6 +600,8 @@ async fn process_sse( } }; + 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( // 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::(item_val) else { @@ -631,15 +632,12 @@ async fn process_sse( 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( } } "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( } } "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( } } "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( | "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( } "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, + provider: ModelProviderInfo, + otel_event_manager: OtelEventManager, + ) { + let chunks: Vec> = 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::>(1); + + let stream = stream::iter(vec![Err::( + 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::>(1); + let stream = stream::pending::>(); + 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); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index d6e97222a9..a978063130 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -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, 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, + 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, + turn_context: &TurnContext, + params: ExecParams, + tool_name: &str, + sub_id: &str, + call_id: &str, + approval: Option, + ) -> 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, + turn_context: &TurnContext, + params: ExecParams, + exec_command_context: ExecCommandContext, + sandbox_type: SandboxType, + approval: Option, + ) -> 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:?}"), + } + } } diff --git a/codex-rs/core/src/otel_init.rs b/codex-rs/core/src/otel_init.rs index 365f795b81..0d6002c225 100644 --- a/codex-rs/core/src/otel_init.rs +++ b/codex-rs/core/src/otel_init.rs @@ -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. /// diff --git a/codex-rs/core/tests/chat_completions_sse.rs b/codex-rs/core/tests/chat_completions_sse.rs index f59df3d810..7991ecdd19 100644 --- a/codex-rs/core/tests/chat_completions_sse.rs +++ b/codex-rs/core/tests/chat_completions_sse.rs @@ -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 { + run_stream_with_bytes(sse_body.as_bytes()).await +} + +async fn run_stream_with_bytes(sse_body: &[u8]) -> Vec { 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 { 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())) + }); +} diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 847174a77c..53699438b0 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -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())) + }); +} diff --git a/codex-rs/otel/src/otel_event_manager.rs b/codex-rs/otel/src/otel_event_manager.rs index f7a0764bab..b4300f51d4 100644 --- a/codex-rs/otel/src/otel_event_manager.rs +++ b/codex-rs/otel/src/otel_event_manager.rs @@ -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, 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(), ); }