From 62b7386b07e488b39cc9cbf651cb5acf3acacd04 Mon Sep 17 00:00:00 2001 From: dshakiba-OAI Date: Fri, 7 Aug 2026 19:50:46 +0000 Subject: [PATCH] Limit payload traces in diagnostic logs (#37497) ## Why High-volume request and streamed-response payloads can overwhelm the SQLite log database and diagnostic ring buffer used for reports. ## What changed - Limit HTTP transport, SSE, and WebSocket diagnostics to `DEBUG` in persistent log sinks while leaving unrelated trace diagnostics available. - Log known unhandled response events and delta events at `TRACE`, and surface unexpected event kinds at `DEBUG` without including their payloads. - Report structured parse-error metadata instead of logging an unparseable SSE payload. ## Testing - Cover filtering for transport, SSE, WebSocket, and unrelated trace records in both report and SQLite log sinks. - Cover unknown and unsupported delta response events. GitOrigin-RevId: 6d9121e093ddadf6834da394df544d9c09d8aeb9 --- codex-rs/Cargo.lock | 1 + codex-rs/codex-api/src/sse/responses.rs | 51 ++++++++++++++++++++-- codex-rs/feedback/Cargo.toml | 1 + codex-rs/feedback/src/lib.rs | 52 ++++++++++++++++++++--- codex-rs/state/src/log_db.rs | 4 ++ codex-rs/state/src/log_db_filter_tests.rs | 20 ++++++++- 6 files changed, 117 insertions(+), 12 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index e2f5c6bd58..b353bcb007 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3162,6 +3162,7 @@ dependencies = [ "anyhow", "codex-login", "codex-protocol", + "log", "mime_guess", "pretty_assertions", "sentry", diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index 44ed78737b..7f18d7ca2d 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -467,9 +467,28 @@ pub fn process_responses_event( })); } } - _ => { + "codex.response.metadata" + | "response.content_part.added" + | "response.content_part.done" + | "response.custom_tool_call_input.done" + | "response.function_call_arguments.delta" + | "response.function_call_arguments.done" + | "response.in_progress" + | "response.metadata" + | "response.output_text.done" + | "response.reasoning_summary_part.done" + | "responsesapi.websocket_timing" => { trace!("unhandled responses event: {}", event.kind); } + kind if kind.ends_with(".delta") => { + trace!("unhandled responses event: {kind}"); + } + _ => { + debug!( + "unhandled responses event: {:?}", + event.kind.chars().take(128).collect::() + ); + } } Ok(None) @@ -536,7 +555,13 @@ async fn process_sse_with_treatment( let event: ResponsesStreamEvent = match serde_json::from_str(&sse.data) { Ok(event) => event, Err(e) => { - debug!("Failed to parse SSE event: {e}, data: {}", &sse.data); + debug!( + error_category = ?e.classify(), + error_line = e.line(), + error_column = e.column(), + payload_bytes = sse.data.len(), + "Failed to parse SSE event" + ); continue; } }; @@ -1203,7 +1228,27 @@ mod tests { }, TestCase { name: "unknown", - event: json!({"type": "response.new_tool_event"}), + event: json!({"type": "response.new_tool_event", "sequence_number": 1}), + expect_first: is_completed, + expected_len: 1, + }, + TestCase { + name: "refusal_delta", + event: json!({ + "type": "response.refusal.delta", + "delta": "no", + "sequence_number": 1 + }), + expect_first: is_completed, + expected_len: 1, + }, + TestCase { + name: "mcp_call_arguments_delta", + event: json!({ + "type": "response.mcp_call_arguments.delta", + "delta": "chunk", + "sequence_number": 1 + }), expect_first: is_completed, expected_len: 1, }, diff --git a/codex-rs/feedback/Cargo.toml b/codex-rs/feedback/Cargo.toml index dd89ddb76b..7cfdc9327d 100644 --- a/codex-rs/feedback/Cargo.toml +++ b/codex-rs/feedback/Cargo.toml @@ -17,6 +17,7 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true } [dev-dependencies] +log = { workspace = true } pretty_assertions = { workspace = true } [lib] diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index 79b079893d..e4c0a10042 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -194,7 +194,7 @@ impl CodexFeedback { } } - /// Returns a [`tracing_subscriber`] layer that captures full-fidelity logs into this feedback + /// Returns a [`tracing_subscriber`] layer that captures diagnostic logs into this feedback /// ring buffer. /// /// This is intended for initialization code so call sites don't have to duplicate the exact @@ -208,11 +208,17 @@ impl CodexFeedback { .with_timer(tracing_subscriber::fmt::time::SystemTime) .with_ansi(false) .with_target(false) - // Capture everything, regardless of the caller's `RUST_LOG`, so feedback includes the - // full trace when the user uploads a report. + // Capture diagnostics independently of `RUST_LOG` without filling the feedback ring + // with high-volume request and response payloads. .with_filter( Targets::new() .with_default(Level::TRACE) + .with_target("codex_http_client::transport", LevelFilter::DEBUG) + .with_target("codex_api::sse", LevelFilter::DEBUG) + // `tracing-log` checks legacy log records against their original + // target before re-emitting them as `log`; tungstenite TRACE + // includes full websocket frames and authenticated handshakes. + .with_target("tungstenite", LevelFilter::DEBUG) .with_target("codex_api::responses_websocket_timing", LevelFilter::OFF) .with_target("codex_core::post_sampling_token_estimate", LevelFilter::OFF), ) @@ -721,18 +727,50 @@ mod tests { } #[test] - fn logger_layer_excludes_responses_websocket_timing_payloads() { + fn logger_layer_filters_noisy_trace_payloads() { let fb = CodexFeedback::new(); let _guard = tracing_subscriber::registry() + // Keep another TRACE subscriber interested so bridged records are + // emitted; feedback must still reject them with its own filter. + .with(tracing_subscriber::fmt::layer().with_writer(std::io::sink)) .with(fb.logger_layer()) .set_default(); tracing::trace!(target: "codex_api::responses_websocket_timing", payload = "secret"); - tracing::trace!(target: "codex_feedback_test", "retained"); + tracing::trace!(target: "codex_http_client::transport", "transport-trace"); + tracing::trace!(target: "codex_api::sse", "sse-trace"); + tracing::trace!(target: "codex_api::sse::responses", "nested-sse-trace"); + tracing::debug!(target: "codex_http_client::transport", "transport-debug"); + tracing::debug!(target: "codex_api::sse::responses", "sse-debug"); + tracing::trace!(target: "codex_feedback_test", "unrelated-trace"); + log::trace!(target: "codex_feedback_test", "unrelated-log-trace"); + log::trace!( + target: "tungstenite::handshake::client", + "websocket-handshake-payload" + ); + log::trace!(target: "tungstenite::protocol", "websocket-frame-payload"); + log::debug!(target: "tungstenite::protocol", "websocket-debug"); let logs = String::from_utf8(fb.snapshot(/*session_id*/ None).bytes).unwrap(); - assert!(!logs.contains("secret")); - assert!(logs.contains("retained")); + for excluded in [ + "secret", + "transport-trace", + "sse-trace", + "nested-sse-trace", + "websocket-handshake-payload", + "websocket-frame-payload", + ] { + assert!(!logs.contains(excluded)); + } + for retained in [ + "transport-debug", + "sse-debug", + "unrelated-trace", + "unrelated-log-trace", + "websocket-debug", + ] { + assert!(logs.contains(retained)); + } } #[test] diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index 7179d3cd8e..e500354ecc 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -61,6 +61,10 @@ pub fn default_filter() -> Targets { .with_target("rmcp", LevelFilter::INFO) .with_target("codex_api::responses_websocket_timing", LevelFilter::OFF) .with_target("codex_core::post_sampling_token_estimate", LevelFilter::OFF) + // Full model request bodies and streamed response payloads overwhelm the + // SQLite log database, but remain available to explicit TRACE subscribers. + .with_target("codex_http_client::transport", LevelFilter::DEBUG) + .with_target("codex_api::sse", LevelFilter::DEBUG) } #[derive(Clone, Copy, Debug, Eq, PartialEq)] diff --git a/codex-rs/state/src/log_db_filter_tests.rs b/codex-rs/state/src/log_db_filter_tests.rs index a30f3de225..8396c89112 100644 --- a/codex-rs/state/src/log_db_filter_tests.rs +++ b/codex-rs/state/src/log_db_filter_tests.rs @@ -10,6 +10,9 @@ use super::*; async fn sqlite_sink_drops_low_level_opentelemetry_sdk_logs() { let codex_home = std::env::temp_dir().join(format!("codex-state-log-db-filter-{}", Uuid::new_v4())); + let _cleanup = scopeguard::guard(codex_home.clone(), |codex_home| { + let _ = std::fs::remove_dir_all(codex_home); + }); let runtime = StateRuntime::init( crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()), "test-provider".to_string(), @@ -35,6 +38,11 @@ async fn sqlite_sink_drops_low_level_opentelemetry_sdk_logs() { target: "codex_rmcp_client::oauth", "retained-codex-rmcp-client-info" ); + tracing::trace!(target: "codex_http_client::transport", "dropped-request-body"); + tracing::debug!(target: "codex_http_client::transport", "retained-request-diagnostic"); + tracing::trace!(target: "codex_api::sse", "dropped-sse-parent"); + tracing::trace!(target: "codex_api::sse::responses", "dropped-sse-payload"); + tracing::debug!(target: "codex_api::sse::responses", "retained-sse-diagnostic"); tracing::trace!(target: "codex_state", "retained-trace"); tracing::trace!( target: "codex_api::responses_websocket_timing", @@ -65,9 +73,17 @@ async fn sqlite_sink_drops_low_level_opentelemetry_sdk_logs() { "codex_rmcp_client::oauth", Some("retained-codex-rmcp-client-info") ), + ( + "DEBUG", + "codex_http_client::transport", + Some("retained-request-diagnostic") + ), + ( + "DEBUG", + "codex_api::sse::responses", + Some("retained-sse-diagnostic") + ), ("TRACE", "codex_state", Some("retained-trace")), ] ); - - let _ = tokio::fs::remove_dir_all(codex_home).await; }