Add opt-in OTLP logging for final agent responses (#47649)

## What changed

Add `otel.log_agent_responses`, disabled by default and requiring an OTLP HTTP or gRPC log exporter. Emit `codex.agent_response` for completed `final_answer` messages from main agents and spawned task agents, with session, turn, item, and available parent/root attribution.

Strip memory citations from exported text, preserve the stored answer, and cap responses at 64 KiB on a UTF-8 character boundary. Include the original byte length and a truncation flag, and suppress duplicate item completions within each turn. Exclude commentary and non-task agents.

## Testing

Extend configuration tests for default-off and explicit opt-in behavior, the agent execution test for main/subagent response attribution, and the log-buffer test for response-event exclusion.

GitOrigin-RevId: eacda1150de2e6bac2c667d71b57214b0a086f1a
This commit is contained in:
xli-oai
2026-09-23 20:07:03 +00:00
committed by copyberry
parent ad26b2520a
commit e6da6fd7c7
12 changed files with 245 additions and 3 deletions
+1
View File
@@ -4107,6 +4107,7 @@ dependencies = [
"codex-http-client",
"codex-protocol",
"codex-utils-absolute-path",
"codex-utils-stream-parser",
"codex-utils-string",
"eventsource-stream",
"gethostname",
+16
View File
@@ -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<bool>,
/// 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<bool>,
/// Mark traces with environment (dev, staging, prod, test). Defaults to dev.
pub environment: Option<String>,
@@ -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 {
+4
View File
@@ -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"
+3
View File
@@ -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 { .. }
+1
View File
@@ -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,
+16
View File
@@ -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(
+49 -3
View File
@@ -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(())
}
+7
View File
@@ -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",
+1
View File
@@ -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 }
+33
View File
@@ -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:
+111
View File
@@ -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<String>,
pub root_turn_id: Option<String>,
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<HashSet<ResponseItemId>>,
}
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::<String>();
// 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(),
);
}
}
+3
View File
@@ -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;