mirror of
https://github.com/openai/codex.git
synced 2026-10-29 19:08:43 +00:00
Add an option to disable successful SSE delta logs (#58034)
## What changed Add `otel.log_sse_deltas`, enabled by default. Setting it to `false` suppresses successful SSE event logs whose kind ends in `.delta` while preserving errors, completion events, and metrics. Apply the setting to session and memory telemetry, support managed requirements, and expose it in the app-server configuration requirements API and generated schemas and bindings. ## Testing Add coverage for delta log suppression with errors, completion logs, event counts, and duration metrics preserved. Extend configuration and app-server tests to cover defaults, overrides, and managed requirements. GitOrigin-RevId: 751c7f357c41f51848b025c838246a733eb92e2f
This commit is contained in:
+6
@@ -16737,6 +16737,12 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"log_sse_deltas": {
|
||||
"type": [
|
||||
"boolean",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"log_user_prompt": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
+6
@@ -12729,6 +12729,12 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"log_sse_deltas": {
|
||||
"type": [
|
||||
"boolean",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"log_user_prompt": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
+6
@@ -1278,6 +1278,12 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"log_sse_deltas": {
|
||||
"type": [
|
||||
"boolean",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"log_user_prompt": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
@@ -4,4 +4,4 @@
|
||||
import type { OtelExporter } from "./OtelExporter";
|
||||
import type { OtelToolResultLogConfig } from "./OtelToolResultLogConfig";
|
||||
|
||||
export type OtelRequirements = { tool_result: OtelToolResultLogConfig | null, log_user_prompt: boolean | null, log_agent_responses: boolean | null, log_guardian_assessments: boolean | null, environment: string | null, exporter: OtelExporter | null, trace_exporter: OtelExporter | null, metrics_exporter: OtelExporter | null, span_attributes: { [key in string]?: string } | null, tracestate: { [key in string]?: { [key in string]?: string } } | null, };
|
||||
export type OtelRequirements = { tool_result: OtelToolResultLogConfig | null, log_sse_deltas: boolean | null, log_user_prompt: boolean | null, log_agent_responses: boolean | null, log_guardian_assessments: boolean | null, environment: string | null, exporter: OtelExporter | null, trace_exporter: OtelExporter | null, metrics_exporter: OtelExporter | null, span_attributes: { [key in string]?: string } | null, tracestate: { [key in string]?: { [key in string]?: string } } | null, };
|
||||
|
||||
@@ -506,6 +506,7 @@ pub struct FeedbackRequirements {
|
||||
#[ts(export_to = "v2/")]
|
||||
pub struct OtelRequirements {
|
||||
pub tool_result: Option<OtelToolResultLogConfig>,
|
||||
pub log_sse_deltas: Option<bool>,
|
||||
pub log_user_prompt: Option<bool>,
|
||||
pub log_agent_responses: Option<bool>,
|
||||
pub log_guardian_assessments: Option<bool>,
|
||||
|
||||
@@ -1870,6 +1870,7 @@ tracestate = { managed = { tenant = "enterprise" }, empty = {} }"#,
|
||||
otel,
|
||||
&serde_json::json!({
|
||||
"tool_result": { "max_bytes": 1234 },
|
||||
"log_sse_deltas": null,
|
||||
"log_user_prompt": false,
|
||||
"log_agent_responses": null,
|
||||
"log_guardian_assessments": null,
|
||||
|
||||
@@ -18,6 +18,7 @@ pub(crate) fn map_otel_requirements_to_api(otel: OtelRequirementsToml) -> OtelRe
|
||||
tool_result: otel.tool_result.map(|value| OtelToolResultLogConfig {
|
||||
max_bytes: value.max_bytes,
|
||||
}),
|
||||
log_sse_deltas: otel.log_sse_deltas,
|
||||
log_user_prompt: otel.log_user_prompt,
|
||||
log_agent_responses: otel.log_agent_responses,
|
||||
log_guardian_assessments: otel.log_guardian_assessments,
|
||||
|
||||
@@ -717,6 +717,7 @@ async fn config_requirements_read_redacts_otel_exporter_headers(
|
||||
r#"[otel]
|
||||
tool_result = { max_bytes = 1024 }
|
||||
log_user_prompt = false
|
||||
log_sse_deltas = false
|
||||
log_agent_responses = true
|
||||
log_guardian_assessments = false
|
||||
environment = "managed"
|
||||
@@ -736,6 +737,7 @@ tracestate = { managed = { tenant = "enterprise" } }"#
|
||||
json!({
|
||||
"tool_result": { "max_bytes": 1024 },
|
||||
"log_user_prompt": false,
|
||||
"log_sse_deltas": false,
|
||||
"log_agent_responses": true,
|
||||
"log_guardian_assessments": false,
|
||||
"environment": "managed",
|
||||
|
||||
@@ -1115,6 +1115,7 @@ pub struct ConfigRequirementsToml {
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct OtelRequirementsToml {
|
||||
pub tool_result: Option<ToolResultLogConfig>,
|
||||
pub log_sse_deltas: Option<bool>,
|
||||
pub log_user_prompt: Option<bool>,
|
||||
pub log_agent_responses: Option<bool>,
|
||||
pub log_guardian_assessments: Option<bool>,
|
||||
@@ -1129,6 +1130,7 @@ pub struct OtelRequirementsToml {
|
||||
impl OtelRequirementsToml {
|
||||
pub(crate) fn is_empty(&self) -> bool {
|
||||
self.tool_result.is_none()
|
||||
&& self.log_sse_deltas.is_none()
|
||||
&& self.log_user_prompt.is_none()
|
||||
&& self.log_agent_responses.is_none()
|
||||
&& self.log_guardian_assessments.is_none()
|
||||
@@ -1772,7 +1774,7 @@ impl ConfigRequirementsToml {
|
||||
return Some("model_providers");
|
||||
}
|
||||
let otel = self.otel.as_ref();
|
||||
let managed_fields: [(bool, &[&str], &'static str); 17] = [
|
||||
let managed_fields: [(bool, &[&str], &'static str); 18] = [
|
||||
(
|
||||
self.model_provider.is_some(),
|
||||
&["model_provider"],
|
||||
@@ -1823,6 +1825,11 @@ impl ConfigRequirementsToml {
|
||||
&["otel", "log_user_prompt"],
|
||||
"otel.log_user_prompt",
|
||||
),
|
||||
(
|
||||
otel.is_some_and(|otel| otel.log_sse_deltas.is_some()),
|
||||
&["otel", "log_sse_deltas"],
|
||||
"otel.log_sse_deltas",
|
||||
),
|
||||
(
|
||||
otel.is_some_and(|otel| otel.log_agent_responses.is_some()),
|
||||
&["otel", "log_agent_responses"],
|
||||
@@ -1914,6 +1921,7 @@ macro_rules! otel_merge_functions {
|
||||
|
||||
otel_merge_functions!(
|
||||
tool_result,
|
||||
log_sse_deltas,
|
||||
log_user_prompt,
|
||||
log_agent_responses,
|
||||
log_guardian_assessments,
|
||||
@@ -2563,6 +2571,7 @@ mod tests {
|
||||
#[test_case::test_case(&["otel", "tool_result"], Some("otel.tool_result"); "tool_result")]
|
||||
#[test_case::test_case(&["otel", "tool_result", "max_bytes"], Some("otel.tool_result"); "tool_result_max_bytes")]
|
||||
#[test_case::test_case(&["otel", "log_user_prompt"], Some("otel.log_user_prompt"); "log_user_prompt")]
|
||||
#[test_case::test_case(&["otel", "log_sse_deltas"], Some("otel.log_sse_deltas"); "log_sse_deltas")]
|
||||
#[test_case::test_case(&["otel", "environment"], Some("otel.environment"); "environment")]
|
||||
#[test_case::test_case(&["otel", "exporter", "otlp-http", "endpoint"], Some("otel.exporter"); "exporter_endpoint")]
|
||||
#[test_case::test_case(&["otel", "trace_exporter"], Some("otel.trace_exporter"); "trace_exporter")]
|
||||
@@ -2586,6 +2595,7 @@ mod tests {
|
||||
tool_result = { max_bytes = 1024 }
|
||||
log_user_prompt = false
|
||||
environment = "managed"
|
||||
log_sse_deltas = false
|
||||
exporter = "none"
|
||||
trace_exporter = "none"
|
||||
metrics_exporter = "none"
|
||||
@@ -2599,6 +2609,7 @@ tracestate = { managed = { tenant = "enterprise" }, empty = {} }
|
||||
tool_result = { max_bytes = 4096 }
|
||||
log_user_prompt = true
|
||||
environment = "user"
|
||||
log_sse_deltas = true
|
||||
exporter = "statsig"
|
||||
trace_exporter = { otlp-grpc = { endpoint = "https://user.example" } }
|
||||
metrics_exporter = "statsig"
|
||||
@@ -2616,6 +2627,7 @@ tracestate = { managed = { tenant = "user", region = "us" }, other = { keep = "y
|
||||
tool_result = { max_bytes = 1024 }
|
||||
log_user_prompt = false
|
||||
environment = "managed"
|
||||
log_sse_deltas = false
|
||||
exporter = "none"
|
||||
trace_exporter = "none"
|
||||
metrics_exporter = "none"
|
||||
|
||||
@@ -639,6 +639,9 @@ pub enum OtelExporterKind {
|
||||
pub struct OtelConfigToml {
|
||||
/// Byte limit for tool-result log output; independent of model-visible output.
|
||||
pub tool_result: Option<codex_protocol::config_types::ToolResultLogConfig>,
|
||||
/// Log successful SSE events whose kind ends in `.delta`. Defaults to true.
|
||||
/// Disabling this preserves errors, completion events, and metrics.
|
||||
pub log_sse_deltas: Option<bool>,
|
||||
/// 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.
|
||||
@@ -672,6 +675,7 @@ pub struct OtelConfigToml {
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct OtelConfig {
|
||||
pub tool_result: codex_protocol::config_types::ToolResultLogConfig,
|
||||
pub log_sse_deltas: bool,
|
||||
pub log_user_prompt: bool,
|
||||
pub log_agent_responses: bool,
|
||||
pub log_guardian_assessments: bool,
|
||||
@@ -687,6 +691,7 @@ impl Default for OtelConfig {
|
||||
fn default() -> Self {
|
||||
OtelConfig {
|
||||
tool_result: Default::default(),
|
||||
log_sse_deltas: true,
|
||||
log_user_prompt: false,
|
||||
log_agent_responses: false,
|
||||
log_guardian_assessments: false,
|
||||
|
||||
@@ -3780,6 +3780,10 @@
|
||||
"description": "Opt in to logging completed Guardian assessments to an OTLP log exporter. Defaults to false. Rationales can be sensitive and are capped at 64 KiB.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"log_sse_deltas": {
|
||||
"description": "Log successful SSE events whose kind ends in `.delta`. Defaults to true. Disabling this preserves errors, completion events, and metrics.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"log_user_prompt": {
|
||||
"description": "Log user prompt in traces",
|
||||
"type": "boolean"
|
||||
|
||||
@@ -10327,6 +10327,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.log_sse_deltas);
|
||||
assert!(!config.otel.agent_response_logging_enabled());
|
||||
assert!(!config.otel.guardian_assessment_logging_enabled());
|
||||
Ok(())
|
||||
@@ -10338,6 +10339,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: Some(toml::from_str("max_bytes = 8192").expect("tool-result logging config")),
|
||||
log_sse_deltas: Some(false),
|
||||
log_agent_responses: Some(true),
|
||||
log_guardian_assessments: Some(true),
|
||||
exporter: Some(OtelExporterKind::OtlpHttp {
|
||||
@@ -10361,6 +10363,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.log_sse_deltas);
|
||||
assert!(config.otel.agent_response_logging_enabled());
|
||||
assert!(config.otel.guardian_assessment_logging_enabled());
|
||||
assert!(matches!(
|
||||
@@ -13975,6 +13978,10 @@ tracestate = { vendor = { tenant = "managed", region = "us" }, other = { keep =
|
||||
"[otel]\nlog_agent_responses = true\nlog_guardian_assessments = false",
|
||||
"[otel]\nlog_agent_responses = true\nlog_guardian_assessments = false", true; "managed_logging_flags"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
"[otel]\nlog_sse_deltas = true", "[otel]\nlog_sse_deltas = false",
|
||||
"[otel]\nlog_sse_deltas = false", true; "managed_sse_delta_logs"
|
||||
)]
|
||||
#[tokio::test]
|
||||
async fn otel_requirements_resolve_config(
|
||||
user_config: &str,
|
||||
|
||||
@@ -28,6 +28,7 @@ pub(crate) fn resolve_config(
|
||||
|
||||
OtelConfig {
|
||||
tool_result: config.tool_result.unwrap_or_default(),
|
||||
log_sse_deltas: config.log_sse_deltas.unwrap_or(true),
|
||||
log_user_prompt,
|
||||
log_agent_responses: config.log_agent_responses.unwrap_or(false),
|
||||
log_guardian_assessments: config.log_guardian_assessments.unwrap_or(false),
|
||||
|
||||
@@ -1381,6 +1381,9 @@ impl Session {
|
||||
.with_auth_env(auth_env_telemetry.to_otel_metadata())
|
||||
.with_user_id(telemetry_auth.and_then(CodexAuth::get_chatgpt_user_id))
|
||||
.with_tool_result_log_config(config.otel.tool_result);
|
||||
if !config.otel.log_sse_deltas {
|
||||
session_telemetry = session_telemetry.without_sse_delta_logs();
|
||||
}
|
||||
if let Some(metrics) = thread_extension_data.get::<codex_otel::MetricsClient>() {
|
||||
session_telemetry = session_telemetry.with_metrics(metrics.as_ref().clone());
|
||||
}
|
||||
|
||||
@@ -518,13 +518,14 @@ async fn process_sse_failed_event_logs_response_completed_parse_error() {
|
||||
|
||||
#[tokio::test]
|
||||
#[traced_test]
|
||||
async fn process_sse_emits_completed_telemetry() {
|
||||
async fn process_sse_emits_completed_telemetry_with_delta_logs_disabled() {
|
||||
let server = start_mock_server().await;
|
||||
|
||||
mount_sse_once(
|
||||
let mock = mount_sse_once(
|
||||
&server,
|
||||
sse(vec![
|
||||
ev_reasoning_item_added("reasoning-1", &[]),
|
||||
ev_reasoning_summary_text_delta("Thinking"),
|
||||
serde_json::json!({
|
||||
"type": "response.completed",
|
||||
"response": {
|
||||
@@ -545,7 +546,13 @@ async fn process_sse_emits_completed_telemetry() {
|
||||
)
|
||||
.await;
|
||||
|
||||
let TestCodex { codex, .. } = test_codex().build(&server).await.unwrap();
|
||||
let TestCodex { codex, .. } = test_codex()
|
||||
.with_config(|config| {
|
||||
config.otel.log_sse_deltas = false;
|
||||
})
|
||||
.build_with_auto_env(&server)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
codex
|
||||
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
|
||||
@@ -557,6 +564,19 @@ async fn process_sse_emits_completed_telemetry() {
|
||||
|
||||
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
|
||||
|
||||
let _request = mock.single_request();
|
||||
logs_assert(|lines: &[&str]| {
|
||||
let delta_logged = lines.iter().any(|line| {
|
||||
line.contains("codex.sse_event")
|
||||
&& line.contains("event.kind=response.reasoning_summary_text.delta")
|
||||
});
|
||||
if delta_logged {
|
||||
Err("unexpected successful SSE delta log".to_string())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
});
|
||||
|
||||
logs_assert(|lines: &[&str]| {
|
||||
lines
|
||||
.iter()
|
||||
|
||||
@@ -119,7 +119,7 @@ fn build_session_telemetry(
|
||||
&config.model_provider,
|
||||
auth_manager.codex_api_key_env_enabled(),
|
||||
);
|
||||
SessionTelemetry::new(
|
||||
let mut session_telemetry = SessionTelemetry::new(
|
||||
thread_id,
|
||||
model,
|
||||
model,
|
||||
@@ -131,7 +131,11 @@ fn build_session_telemetry(
|
||||
user_agent(),
|
||||
source,
|
||||
)
|
||||
.with_auth_env(auth_env_telemetry.to_otel_metadata())
|
||||
.with_auth_env(auth_env_telemetry.to_otel_metadata());
|
||||
if !config.otel.log_sse_deltas {
|
||||
session_telemetry = session_telemetry.without_sse_delta_logs();
|
||||
}
|
||||
session_telemetry
|
||||
}
|
||||
|
||||
impl MemoryStartupContext {
|
||||
|
||||
@@ -120,6 +120,7 @@ pub struct SessionTelemetryMetadata {
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SessionTelemetry {
|
||||
tool_result_log_config: ToolResultLogConfig,
|
||||
log_sse_deltas: bool,
|
||||
pub(crate) metadata: SessionTelemetryMetadata,
|
||||
pub(crate) metrics: Option<MetricsClient>,
|
||||
pub(crate) metrics_use_metadata_tags: bool,
|
||||
@@ -142,6 +143,12 @@ pub fn bounded_product_sku(product_sku: Option<&str>) -> Option<&'static str> {
|
||||
}
|
||||
|
||||
impl SessionTelemetry {
|
||||
/// Suppress successful SSE delta logs while retaining errors and all metrics.
|
||||
pub fn without_sse_delta_logs(mut self) -> Self {
|
||||
self.log_sse_deltas = false;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_tool_result_log_config(mut self, config: ToolResultLogConfig) -> Self {
|
||||
self.tool_result_log_config = config;
|
||||
self
|
||||
@@ -609,6 +616,7 @@ impl SessionTelemetry {
|
||||
});
|
||||
Self {
|
||||
tool_result_log_config: ToolResultLogConfig::default(),
|
||||
log_sse_deltas: true,
|
||||
metadata: SessionTelemetryMetadata {
|
||||
conversation_id,
|
||||
agent_name,
|
||||
@@ -1065,6 +1073,9 @@ impl SessionTelemetry {
|
||||
duration,
|
||||
&[("kind", kind), ("success", "true")],
|
||||
);
|
||||
if !self.log_sse_deltas && kind.ends_with(".delta") {
|
||||
return;
|
||||
}
|
||||
log_event!(
|
||||
self,
|
||||
event.name = "codex.sse_event",
|
||||
|
||||
@@ -1,6 +1,12 @@
|
||||
use crate::harness::attributes_to_map;
|
||||
use crate::harness::build_metrics_with_defaults;
|
||||
use crate::harness::find_metric;
|
||||
use crate::harness::latest_metrics;
|
||||
use codex_api::AgentIdentityTelemetry;
|
||||
use codex_otel::AuthEnvTelemetryMetadata;
|
||||
use codex_otel::OtelProvider;
|
||||
use codex_otel::SSE_EVENT_COUNT_METRIC;
|
||||
use codex_otel::SSE_EVENT_DURATION_METRIC;
|
||||
use codex_otel::SessionTelemetry;
|
||||
use codex_otel::TelemetryAuthMode;
|
||||
use opentelemetry::KeyValue;
|
||||
@@ -9,12 +15,15 @@ use opentelemetry::trace::TracerProvider as _;
|
||||
use opentelemetry_sdk::logs::InMemoryLogExporter;
|
||||
use opentelemetry_sdk::logs::SdkLogRecord;
|
||||
use opentelemetry_sdk::logs::SdkLoggerProvider;
|
||||
use opentelemetry_sdk::metrics::data::AggregatedMetrics;
|
||||
use opentelemetry_sdk::metrics::data::MetricData;
|
||||
use opentelemetry_sdk::trace::InMemorySpanExporter;
|
||||
use opentelemetry_sdk::trace::SdkTracerProvider;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::borrow::Cow;
|
||||
use std::collections::BTreeMap;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
use tracing_subscriber::Layer;
|
||||
use tracing_subscriber::filter::filter_fn;
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
@@ -32,7 +41,9 @@ use codex_protocol::protocol::GuardianAssessmentStatus;
|
||||
use codex_protocol::protocol::SandboxPolicy;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use codex_protocol::protocol::TokenUsage;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use eventsource_stream::Event as StreamEvent;
|
||||
|
||||
fn log_attributes(record: &SdkLogRecord) -> BTreeMap<String, String> {
|
||||
record
|
||||
@@ -100,6 +111,154 @@ fn auth_env_metadata() -> AuthEnvTelemetryMetadata {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn suppressing_sse_delta_logs_preserves_errors_completions_and_metrics() -> codex_otel::Result<()> {
|
||||
for log_sse_deltas in [true, false] {
|
||||
let log_exporter = InMemoryLogExporter::default();
|
||||
let logger_provider = SdkLoggerProvider::builder()
|
||||
.with_simple_exporter(log_exporter.clone())
|
||||
.build();
|
||||
let subscriber = tracing_subscriber::registry().with(
|
||||
opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge::new(
|
||||
&logger_provider,
|
||||
)
|
||||
.with_filter(filter_fn(OtelProvider::log_export_filter)),
|
||||
);
|
||||
let (metrics, metric_exporter) = build_metrics_with_defaults(&[])?;
|
||||
let mut manager = SessionTelemetry::new(
|
||||
ThreadId::new(),
|
||||
"gpt-5.1",
|
||||
"gpt-5.1",
|
||||
/*account_id*/ None,
|
||||
/*account_email*/ None,
|
||||
/*auth_mode*/ None,
|
||||
"test_originator".to_string(),
|
||||
/*log_user_prompts*/ false,
|
||||
"tty".to_string(),
|
||||
SessionSource::Cli,
|
||||
)
|
||||
.with_metrics_without_metadata_tags(metrics);
|
||||
if !log_sse_deltas {
|
||||
manager = manager.without_sse_delta_logs();
|
||||
}
|
||||
let events = [
|
||||
("response.output_text.delta", "{}", true),
|
||||
("response.reasoning_summary_text.delta", "{}", true),
|
||||
("response.custom_tool_call_input.delta", "{}", true),
|
||||
("response.output_text.done", "{}", true),
|
||||
("response.completed", "{}", true),
|
||||
("response.output_text.delta", "{", false),
|
||||
("response.failed", r#"{"error":"failed"}"#, false),
|
||||
];
|
||||
tracing::subscriber::with_default(subscriber, || {
|
||||
tracing::callsite::rebuild_interest_cache();
|
||||
for (kind, data, _) in events {
|
||||
manager.log_sse_event::<&str>(
|
||||
&Ok(Some(Ok(StreamEvent {
|
||||
event: kind.to_string(),
|
||||
data: data.to_string(),
|
||||
id: String::new(),
|
||||
retry: None,
|
||||
}))),
|
||||
Duration::from_millis(10),
|
||||
);
|
||||
}
|
||||
manager.sse_event_completed(
|
||||
&TokenUsage {
|
||||
input_tokens: 7,
|
||||
..Default::default()
|
||||
},
|
||||
/*ttft_ms*/ Some(10),
|
||||
);
|
||||
});
|
||||
logger_provider.force_flush().expect("flush logs");
|
||||
let logs = log_exporter.get_emitted_logs().expect("exported logs");
|
||||
let exported_events = logs
|
||||
.iter()
|
||||
.map(|log| {
|
||||
let attributes = log_attributes(&log.record);
|
||||
(
|
||||
attributes["event.kind"].clone(),
|
||||
attributes.contains_key("error.message"),
|
||||
attributes.get("input_token_count").cloned(),
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let mut expected_events = Vec::new();
|
||||
if log_sse_deltas {
|
||||
expected_events.extend([
|
||||
("response.output_text.delta".to_string(), false, None),
|
||||
(
|
||||
"response.reasoning_summary_text.delta".to_string(),
|
||||
false,
|
||||
None,
|
||||
),
|
||||
(
|
||||
"response.custom_tool_call_input.delta".to_string(),
|
||||
false,
|
||||
None,
|
||||
),
|
||||
]);
|
||||
}
|
||||
expected_events.extend([
|
||||
("response.output_text.done".to_string(), false, None),
|
||||
("response.completed".to_string(), false, None),
|
||||
("response.output_text.delta".to_string(), true, None),
|
||||
("response.failed".to_string(), true, None),
|
||||
(
|
||||
"response.completed".to_string(),
|
||||
false,
|
||||
Some("7".to_string()),
|
||||
),
|
||||
]);
|
||||
assert_eq!(exported_events, expected_events);
|
||||
|
||||
manager.shutdown_metrics()?;
|
||||
let resource_metrics = latest_metrics(&metric_exporter);
|
||||
let expected_counts = events
|
||||
.iter()
|
||||
.map(|(kind, _, success)| {
|
||||
(
|
||||
BTreeMap::from([
|
||||
("kind".to_string(), kind.to_string()),
|
||||
("success".to_string(), success.to_string()),
|
||||
]),
|
||||
1,
|
||||
)
|
||||
})
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
let counter = find_metric(&resource_metrics, SSE_EVENT_COUNT_METRIC).expect("SSE counter");
|
||||
let AggregatedMetrics::U64(MetricData::Sum(sum)) = counter.data() else {
|
||||
panic!("expected SSE counter");
|
||||
};
|
||||
assert_eq!(
|
||||
sum.data_points()
|
||||
.map(|point| (attributes_to_map(point.attributes()), point.value()))
|
||||
.collect::<BTreeMap<_, _>>(),
|
||||
expected_counts,
|
||||
);
|
||||
let duration =
|
||||
find_metric(&resource_metrics, SSE_EVENT_DURATION_METRIC).expect("SSE duration");
|
||||
let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = duration.data() else {
|
||||
panic!("expected SSE duration histogram");
|
||||
};
|
||||
assert_eq!(
|
||||
histogram
|
||||
.data_points()
|
||||
.map(|point| (
|
||||
attributes_to_map(point.attributes()),
|
||||
(point.count(), point.sum())
|
||||
))
|
||||
.collect::<BTreeMap<_, _>>(),
|
||||
expected_counts
|
||||
.into_keys()
|
||||
.map(|tags| (tags, (1, 10.0)))
|
||||
.collect(),
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn otel_export_routing_policy_routes_prompt_and_guardian_payloads() {
|
||||
let log_exporter = InMemoryLogExporter::default();
|
||||
|
||||
@@ -9257,6 +9257,7 @@ class OtelRequirements(BaseModel):
|
||||
exporter: OtelExporter | None = None
|
||||
log_agent_responses: bool | None = None
|
||||
log_guardian_assessments: bool | None = None
|
||||
log_sse_deltas: bool | None = None
|
||||
log_user_prompt: bool | None = None
|
||||
metrics_exporter: OtelExporter | None = None
|
||||
span_attributes: dict[str, Any] | None = None
|
||||
|
||||
Reference in New Issue
Block a user