diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9fcb43afd2..f761c1f8d7 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4107,6 +4107,7 @@ dependencies = [ "codex-http-client", "codex-protocol", "codex-utils-absolute-path", + "codex-utils-stream-parser", "codex-utils-string", "eventsource-stream", "gethostname", diff --git a/codex-rs/config/src/types.rs b/codex-rs/config/src/types.rs index a59f1252c3..0072df443c 100644 --- a/codex-rs/config/src/types.rs +++ b/codex-rs/config/src/types.rs @@ -601,6 +601,9 @@ pub struct OtelConfigToml { pub tool_result: codex_protocol::config_types::ToolResultLogConfig, /// Log user prompt in traces pub log_user_prompt: Option, + /// Opt in to logging final main-agent and spawned-subagent responses to an OTLP log exporter. + /// Defaults to false. Response text can be sensitive and is capped at 64 KiB. + pub log_agent_responses: Option, /// Mark traces with environment (dev, staging, prod, test). Defaults to dev. pub environment: Option, @@ -626,6 +629,7 @@ pub struct OtelConfigToml { pub struct OtelConfig { pub tool_result: codex_protocol::config_types::ToolResultLogConfig, pub log_user_prompt: bool, + pub log_agent_responses: bool, pub environment: String, pub exporter: OtelExporterKind, pub trace_exporter: OtelExporterKind, @@ -639,6 +643,7 @@ impl Default for OtelConfig { OtelConfig { tool_result: Default::default(), log_user_prompt: false, + log_agent_responses: false, environment: DEFAULT_OTEL_ENVIRONMENT.to_owned(), exporter: OtelExporterKind::None, trace_exporter: OtelExporterKind::None, @@ -649,6 +654,17 @@ impl Default for OtelConfig { } } +impl OtelConfig { + /// Response text requires a separate opt-in and an explicit OTLP log destination. + pub fn agent_response_logging_enabled(&self) -> bool { + self.log_agent_responses + && matches!( + self.exporter, + OtelExporterKind::OtlpHttp { .. } | OtelExporterKind::OtlpGrpc { .. } + ) + } +} + #[derive(Serialize, Debug, Clone, PartialEq, Eq, Deserialize, JsonSchema)] #[serde(untagged)] pub enum Notifications { diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 68b12c3bd2..082b7b5162 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -3187,6 +3187,10 @@ ], "description": "Optional log exporter" }, + "log_agent_responses": { + "description": "Opt in to logging final main-agent and spawned-subagent responses to an OTLP log exporter. Defaults to false. Response text can be sensitive and is capped at 64 KiB.", + "type": "boolean" + }, "log_user_prompt": { "description": "Log user prompt in traces", "type": "boolean" diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 08d2d80d6c..c619534677 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -9900,6 +9900,7 @@ async fn metrics_exporter_defaults_to_statsig_when_missing() -> std::io::Result< .await?; assert_eq!(config.otel.metrics_exporter, OtelExporterKind::Statsig); + assert!(!config.otel.agent_response_logging_enabled()); Ok(()) } @@ -9909,6 +9910,7 @@ async fn trace_exporter_defaults_to_none_when_log_exporter_is_set() -> std::io:: let mut cfg = fixture.cfg.clone(); cfg.otel = Some(OtelConfigToml { tool_result: toml::from_str("max_bytes = 8192").expect("tool-result logging config"), + log_agent_responses: Some(true), exporter: Some(OtelExporterKind::OtlpHttp { endpoint: "http://localhost:14318/v1/logs".to_string(), headers: HashMap::new(), @@ -9930,6 +9932,7 @@ async fn trace_exporter_defaults_to_none_when_log_exporter_is_set() -> std::io:: .await?; assert_eq!(config.otel.tool_result.max_bytes, 8192); + assert!(config.otel.agent_response_logging_enabled()); assert!(matches!( config.otel.exporter, OtelExporterKind::OtlpHttp { .. } diff --git a/codex-rs/core/src/config/otel.rs b/codex-rs/core/src/config/otel.rs index 56f3b79cf4..e6f792a309 100644 --- a/codex-rs/core/src/config/otel.rs +++ b/codex-rs/core/src/config/otel.rs @@ -28,6 +28,7 @@ pub(crate) fn resolve_config( OtelConfig { tool_result: config.tool_result, log_user_prompt, + log_agent_responses: config.log_agent_responses.unwrap_or(false), environment, exporter, trace_exporter, diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 3e3456e9ba..767eba32e8 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -104,6 +104,22 @@ pub(crate) async fn record_completed_response_item_with_finalized_facts( std::slice::from_ref(item), ) .await; + if turn_context.config.otel.agent_response_logging_enabled() { + turn_context + .extension_data + .get_or_init(codex_otel::AgentResponseLogger::default) + .record( + &step_context.session_telemetry, + item, + codex_otel::AgentResponseContext { + turn_id: &turn_context.sub_id, + session_source: &turn_context.session_source, + parent_turn_id: turn_context.turn_metadata_state.parent_turn_id(), + root_turn_id: turn_context.turn_metadata_state.root_turn_id(), + initiating_agent_path: turn_context.turn_metadata_state.initiating_agent_path(), + }, + ); + } let defers_mailbox_delivery = finalized_facts.map_or_else( || { completed_item_defers_mailbox_delivery_to_next_turn( diff --git a/codex-rs/core/tests/suite/agent_execution.rs b/codex-rs/core/tests/suite/agent_execution.rs index 0eddad8b5b..303aa510b6 100644 --- a/codex-rs/core/tests/suite/agent_execution.rs +++ b/codex-rs/core/tests/suite/agent_execution.rs @@ -75,12 +75,14 @@ async fn mount_root_collaboration_call( .await; let completion_id = format!("resp-{call_id}-complete"); + let mut answer = ev_assistant_message(&format!("msg-{call_id}"), "collaboration completed"); + answer["item"]["phase"] = json!("final_answer"); mount_sse_once_match( server, move |request: &wiremock::Request| has_function_call_output(request, call_id), sse(vec![ ev_response_created(&completion_id), - ev_assistant_message(&format!("msg-{call_id}"), "collaboration completed"), + answer, ev_completed(&completion_id), ]), ) @@ -93,6 +95,9 @@ async fn mount_completed_worker( parent_call_id: &'static str, ) -> ResponseMock { let response_id = format!("resp-worker-{parent_call_id}"); + let mut answer = + ev_assistant_message(&format!("msg-worker-{parent_call_id}"), "worker completed"); + answer["item"]["phase"] = json!("final_answer"); mount_sse_once_match( server, move |request: &wiremock::Request| { @@ -100,7 +105,7 @@ async fn mount_completed_worker( }, sse(vec![ ev_response_created(&response_id), - ev_assistant_message(&format!("msg-worker-{parent_call_id}"), "worker completed"), + answer, ev_completed(&response_id), ]), ) @@ -205,7 +210,8 @@ async fn v2_nested_spawn_checks_shared_active_execution_capacity() -> Result<()> Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] +#[tracing_test::traced_test] async fn child_turn_start_preserves_root_attribution() -> Result<()> { let server = start_mock_server().await; mount_root_collaboration_call( @@ -222,6 +228,12 @@ async fn child_turn_start_preserves_root_attribution() -> Result<()> { let test = test_codex() .with_model("gpt-5.6-sol") .with_config(|config| { + config.otel.log_agent_responses = true; + config.otel.exporter = codex_config::types::OtelExporterKind::OtlpGrpc { + endpoint: "http://127.0.0.1:1".into(), + headers: Default::default(), + tls: None, + }; config.features.enable(Feature::Collab).unwrap(); config.features.enable(Feature::MultiAgentV2).unwrap(); }) @@ -258,6 +270,40 @@ async fn child_turn_start_preserves_root_attribution() -> Result<()> { .iter() .all(|(_, event)| { event.root_turn_id.as_ref() == Some(root_turn_id) }) ); + logs_assert(|lines: &[&str]| { + let logs: Vec<_> = lines + .iter() + .filter(|line| line.contains("codex.agent_response")) + .collect(); + assert_eq!(logs.len(), 2); + for (id, turn) in &starts { + let line = logs + .iter() + .find(|line| line.contains(&format!(" conversation.id={id}"))) + .expect("response log"); + let (agent, item, text) = if *id == test.session_configured.thread_id { + ("main", "msg-first-call", "collaboration completed") + } else { + assert!(line.contains(&format!( + " parent.conversation.id=\"{}\"", + test.session_configured.thread_id + ))); + assert!(line.contains(&format!(" parent.turn.id={root_turn_id:?}"))); + assert!(line.contains(" initiating.agent.path=\"/root\"")); + ("subagent", "msg-worker-first-call", "worker completed") + }; + for field in [ + format!(" agent.type={agent:?}"), + format!(" turn.id={:?}", turn.turn_id), + format!(" root.turn.id={root_turn_id:?}"), + format!(" item.id={item:?}"), + format!(" response={text:?}"), + ] { + assert!(line.contains(&field), "missing {field}: {line}"); + } + } + Ok(()) + }); Ok(()) } diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index f75e32bbb8..60a9af40db 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -238,6 +238,8 @@ impl CodexFeedback { .with_filter( Targets::new() .with_default(Level::TRACE) + // Opted-in content belongs to the configured OTLP destination, not feedback. + .with_target("codex_otel.log_only", LevelFilter::OFF) .with_target("codex_http_client::transport", LevelFilter::DEBUG) .with_target("codex_api::sse", LevelFilter::DEBUG) // `tracing-log` checks legacy log records against their original @@ -942,6 +944,10 @@ mod tests { .set_default(); tracing::trace!(target: "codex_api::responses_websocket_timing", payload = "secret"); + tracing::event!( + target: "codex_otel.log_only", tracing::Level::INFO, + event.name = "codex.agent_response", response = "private-agent-response" + ); 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"); @@ -959,6 +965,7 @@ mod tests { let logs = String::from_utf8(fb.snapshot(/*session_id*/ None).bytes).unwrap(); for excluded in [ "secret", + "private-agent-response", "transport-trace", "sse-trace", "nested-sse-trace", diff --git a/codex-rs/otel/Cargo.toml b/codex-rs/otel/Cargo.toml index 362a788972..3197eaa846 100644 --- a/codex-rs/otel/Cargo.toml +++ b/codex-rs/otel/Cargo.toml @@ -18,6 +18,7 @@ codex-http-client = { workspace = true } opentelemetry-http = "0.31.0" chrono = { workspace = true } codex-utils-absolute-path = { workspace = true } +codex-utils-stream-parser = { workspace = true } codex-utils-string = { workspace = true } codex-api = { workspace = true } codex-protocol = { workspace = true } diff --git a/codex-rs/otel/README.md b/codex-rs/otel/README.md index 68d05d0b61..9342ed887c 100644 --- a/codex-rs/otel/README.md +++ b/codex-rs/otel/README.md @@ -96,6 +96,39 @@ let manager = SessionTelemetry::new( manager.user_prompt(&prompt_items); ``` +### Agent response logs + +Set `otel.log_agent_responses = true` with an OTLP HTTP or gRPC log exporter to +emit `codex.agent_response` for completed assistant messages explicitly marked +`final_answer`. The default is false, independent of `otel.log_user_prompt`; +disabled response logging emits no event. Main agents and spawned task agents +are included. Internal/review/compaction/memory agents, commentary, untagged +messages, streaming deltas, tool-generated async messages, and history replay +are excluded. Plans are preserved and memory-citation markup is removed from +the exported copy without changing the stored answer. + +Opting in exports potentially sensitive response text, including subagent work +absent from the main answer, to your log destination. Text is sent only to logs, +capped at 65,536 UTF-8 bytes at a character boundary, without a truncation notice. +`response_length` is the original byte count after citation removal, and +`response_truncated` indicates whether the cap removed text. +The log-only target is excluded from trace export, local state logs, and the +feedback log buffer. + +Events include standard session attributes, `agent.type` (`main` or `subagent`), +`turn.id`, and `item.id`. `conversation.id` identifies the emitting thread. +Available lineage includes `parent.conversation.id`, `parent.turn.id`, +`root.turn.id`, and `initiating.agent.path`. The parent conversation is the +structural owner; the parent turn is the trigger and may belong to another +agent. They are not a guaranteed matching pair or a supported Compliance API join. +Completion is logged even if the turn later fails. Duplicate item completions +are suppressed within a turn; delivery retains the existing best-effort exporter +behavior, without an exactly-once guarantee. + +`AgentResponseLogger` owns filtering, text preparation, and per-turn deduplication. +Core supplies completed raw items and step attribution, retaining the logger in +the existing turn extension data; it does not own response-logging policy. + ## Metrics (OTLP or in-memory) Modes: diff --git a/codex-rs/otel/src/agent_response.rs b/codex-rs/otel/src/agent_response.rs new file mode 100644 index 0000000000..c7164e1f20 --- /dev/null +++ b/codex-rs/otel/src/agent_response.rs @@ -0,0 +1,111 @@ +//! Exports opted-in live final responses, with turn-local deduplication and bounded text. +//! Hosts supply completed raw messages and attribution; stored history is never changed. + +use crate::SessionTelemetry; +use crate::events::shared::log_event; +use codex_protocol::AgentPath; +use codex_protocol::ResponseItemId; +use codex_protocol::models::ContentItem; +use codex_protocol::models::MessagePhase; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_utils_stream_parser::strip_citations; +use codex_utils_string::take_bytes_at_char_boundary; +use std::collections::HashSet; +use std::sync::Mutex; + +const MAX_RESPONSE_BYTES: usize = 65_536; + +/// Host-owned attribution for one live completed response item. +pub struct AgentResponseContext<'a> { + pub turn_id: &'a str, + pub session_source: &'a SessionSource, + pub parent_turn_id: Option, + pub root_turn_id: Option, + pub initiating_agent_path: Option<&'a AgentPath>, +} + +/// One turn's response-log state. Create a fresh instance for each turn; never replay history. +/// Only construct when response logging is opted in and an OTLP log exporter is configured. +#[derive(Default)] +pub struct AgentResponseLogger { + logged: Mutex>, +} + +impl AgentResponseLogger { + /// Record a live final message once, using the telemetry of its actual sampling step. + pub fn record( + &self, + telemetry: &SessionTelemetry, + item: &ResponseItem, + context: AgentResponseContext<'_>, + ) { + let ResponseItem::Message { + id: Some(item_id), + role, + phase: Some(MessagePhase::FinalAnswer), + content, + .. + } = item + else { + return; + }; + if role != "assistant" { + return; + } + let (agent_type, parent_conversation_id) = match context.session_source { + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, .. + }) => ("subagent", Some(parent_thread_id.to_string())), + SessionSource::Cli + | SessionSource::VSCode + | SessionSource::Exec + | SessionSource::Mcp + | SessionSource::Custom(_) + | SessionSource::Unknown => ("main", None), + SessionSource::Internal(_) + | SessionSource::SubAgent( + SubAgentSource::Review + | SubAgentSource::Compact + | SubAgentSource::MemoryConsolidation + | SubAgentSource::Other(_), + ) => return, + }; + let text = content + .iter() + .filter_map(|content| match content { + ContentItem::OutputText { text } => Some(text.as_str()), + ContentItem::InputText { .. } + | ContentItem::InputImage { .. } + | ContentItem::InputAudio { .. } => None, + }) + .collect::(); + // Keep plan content, stripping memory citations only from the exported copy. + let (text, _) = strip_citations(&text); + if text.trim().is_empty() + || !self + .logged + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(item_id.clone()) + { + return; + } + let response = take_bytes_at_char_boundary(&text, MAX_RESPONSE_BYTES); + log_event!( + telemetry, + event.name = "codex.agent_response", + agent.type = agent_type, + turn.id = context.turn_id, + item.id = item_id.as_str(), + parent.conversation.id = parent_conversation_id.as_deref(), + parent.turn.id = context.parent_turn_id.as_deref(), + root.turn.id = context.root_turn_id.as_deref(), + initiating.agent.path = context.initiating_agent_path.map(AgentPath::as_str), + response = response, + response_length = text.len() as u64, + response_truncated = response.len() < text.len(), + ); + } +} diff --git a/codex-rs/otel/src/lib.rs b/codex-rs/otel/src/lib.rs index ca4eeab2c7..55f18e1690 100644 --- a/codex-rs/otel/src/lib.rs +++ b/codex-rs/otel/src/lib.rs @@ -4,6 +4,7 @@ pub(crate) mod metrics; pub(crate) mod provider; pub(crate) mod trace_context; +mod agent_response; mod network_policy; mod otlp; mod targets; @@ -14,6 +15,8 @@ use codex_protocol::auth::AuthMode; use serde::Serialize; use strum_macros::Display; +pub use crate::agent_response::AgentResponseContext; +pub use crate::agent_response::AgentResponseLogger; pub use crate::config::OtelExporter; pub use crate::config::OtelHttpProtocol; pub use crate::config::OtelSettings;