diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b322b05b55..6087fcc24a 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -2312,6 +2312,122 @@ async fn turn_steer_observations_match_legacy_sources() { assert_eq!(observation_payload, legacy_payload); } +#[tokio::test] +async fn compaction_ended_observation_matches_legacy_fact() { + use codex_observability::events as observation_events; + + let mut legacy_reducer = AnalyticsReducer::default(); + let mut observation_reducer = AnalyticsObservationReducer::default(); + let mut legacy_events = Vec::new(); + let mut observation_events = Vec::new(); + let parent_thread_id = + codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222") + .expect("valid parent thread id"); + + ingest_initialize(&mut legacy_reducer, &mut legacy_events).await; + legacy_reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_resume_response_with_source( + "thread-1", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + )), + }, + &mut legacy_events, + ) + .await; + legacy_events.clear(); + + observation_reducer + .ingest_existing_fact_for_test(sample_initialize_fact(), &mut observation_events) + .await; + let parent_thread_id = + codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222") + .expect("valid parent thread id"); + observation_reducer + .ingest_existing_fact_for_test( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_resume_response_with_source( + "thread-1", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + )), + }, + &mut observation_events, + ) + .await; + observation_events.clear(); + + legacy_reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( + CodexCompactionEvent { + thread_id: "thread-1".to_string(), + turn_id: "turn-compact".to_string(), + trigger: CompactionTrigger::Manual, + reason: CompactionReason::UserRequested, + implementation: CompactionImplementation::Responses, + phase: CompactionPhase::StandaloneTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Failed, + error: Some("context limit exceeded".to_string()), + active_context_tokens_before: 131_000, + active_context_tokens_after: 131_000, + started_at: 100, + completed_at: 101, + duration_ms: Some(1200), + }, + ))), + &mut legacy_events, + ) + .await; + + observation_reducer + .ingest_compaction_ended( + observation_events::CompactionEnded { + thread_id: "thread-1", + turn_id: "turn-compact", + trigger: observation_events::CompactionTrigger::Manual, + reason: observation_events::CompactionReason::UserRequested, + implementation: observation_events::CompactionImplementation::Responses, + phase: observation_events::CompactionPhase::StandaloneTurn, + strategy: observation_events::CompactionStrategy::Memento, + status: observation_events::CompactionStatus::Failed { + error: Some("context limit exceeded"), + }, + active_context_tokens_before: 131_000, + active_context_tokens_after: 131_000, + started_at: 100, + ended_at: 101, + duration_ms: Some(1200), + }, + &mut observation_events, + ) + .await; + + let legacy_payload = serde_json::to_value(&legacy_events).expect("serialize legacy events"); + let observation_payload = + serde_json::to_value(&observation_events).expect("serialize observation events"); + assert_eq!(observation_payload, legacy_payload); +} + fn legacy_event_created_at(event: &TrackEventRequest) -> i64 { serde_json::to_value(event) .expect("serialize event") diff --git a/codex-rs/analytics/src/observation_projection.rs b/codex-rs/analytics/src/observation_projection.rs index 77063bbfc2..18c9bd284d 100644 --- a/codex-rs/analytics/src/observation_projection.rs +++ b/codex-rs/analytics/src/observation_projection.rs @@ -291,6 +291,60 @@ pub(crate) fn turn_ended_notification(observation: events::TurnEnded<'_>) -> Ser }) } +pub(crate) fn compaction_ended_event( + observation: events::CompactionEnded<'_>, +) -> facts::CodexCompactionEvent { + let (status, error) = match observation.status { + events::CompactionStatus::Completed => (facts::CompactionStatus::Completed, None), + events::CompactionStatus::Failed { error } => { + (facts::CompactionStatus::Failed, error.map(str::to_string)) + } + events::CompactionStatus::Interrupted => (facts::CompactionStatus::Interrupted, None), + }; + + facts::CodexCompactionEvent { + thread_id: observation.thread_id.to_string(), + turn_id: observation.turn_id.to_string(), + trigger: match observation.trigger { + events::CompactionTrigger::Manual => facts::CompactionTrigger::Manual, + events::CompactionTrigger::Auto => facts::CompactionTrigger::Auto, + }, + reason: match observation.reason { + events::CompactionReason::UserRequested => facts::CompactionReason::UserRequested, + events::CompactionReason::ContextLimit => facts::CompactionReason::ContextLimit, + events::CompactionReason::ModelDownshift => facts::CompactionReason::ModelDownshift, + }, + implementation: match observation.implementation { + events::CompactionImplementation::Responses => { + facts::CompactionImplementation::Responses + } + events::CompactionImplementation::ResponsesCompact => { + facts::CompactionImplementation::ResponsesCompact + } + }, + phase: match observation.phase { + events::CompactionPhase::StandaloneTurn => facts::CompactionPhase::StandaloneTurn, + events::CompactionPhase::PreTurn => facts::CompactionPhase::PreTurn, + events::CompactionPhase::MidTurn => facts::CompactionPhase::MidTurn, + }, + strategy: match observation.strategy { + events::CompactionStrategy::Memento => facts::CompactionStrategy::Memento, + events::CompactionStrategy::PrefixCompaction => { + facts::CompactionStrategy::PrefixCompaction + } + }, + status, + error, + active_context_tokens_before: observation.active_context_tokens_before, + active_context_tokens_after: observation.active_context_tokens_after, + started_at: u64::try_from(observation.started_at).unwrap_or_default(), + completed_at: u64::try_from(observation.ended_at).unwrap_or_default(), + duration_ms: observation + .duration_ms + .and_then(|duration_ms| u64::try_from(duration_ms).ok()), + } +} + fn tracking_from_fields(model_slug: &str, thread_id: &str, turn_id: &str) -> TrackEventsContext { TrackEventsContext { model_slug: model_slug.to_string(), diff --git a/codex-rs/analytics/src/observation_reducer.rs b/codex-rs/analytics/src/observation_reducer.rs index e8a1b75cfa..093bed6be3 100644 --- a/codex-rs/analytics/src/observation_reducer.rs +++ b/codex-rs/analytics/src/observation_reducer.rs @@ -140,6 +140,22 @@ impl AnalyticsObservationReducer { .ingest_observed_turn_steer(connection_id, observation, out); } + /// Ingests a compaction.ended observation and emits the current analytics event. + pub(crate) async fn ingest_compaction_ended( + &mut self, + observation: events::CompactionEnded<'_>, + out: &mut Vec, + ) { + self.legacy + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( + observation_projection::compaction_ended_event(observation), + ))), + out, + ) + .await; + } + /// Ingests a hook.run_completed observation and emits the current analytics event. pub(crate) fn ingest_hook_run_completed( &mut self, diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index e8f467d5a0..a3aef0cd8b 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -3,9 +3,11 @@ use crate::Observation; use serde::Serialize; +mod compaction; mod thread; mod turn; +pub use compaction::*; pub use thread::*; pub use turn::*; diff --git a/codex-rs/observability/src/events/compaction.rs b/codex-rs/observability/src/events/compaction.rs new file mode 100644 index 0000000000..8f4a97e3a5 --- /dev/null +++ b/codex-rs/observability/src/events/compaction.rs @@ -0,0 +1,105 @@ +//! Compaction observation event definitions. + +use crate::Observation; +use serde::Serialize; + +/// What initiated the compaction work. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionTrigger { + Manual, + Auto, +} + +/// Why the runtime chose to compact context. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionReason { + UserRequested, + ContextLimit, + ModelDownshift, +} + +/// Runtime implementation used for the compaction. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionImplementation { + Responses, + ResponsesCompact, +} + +/// Point in the turn lifecycle where compaction ran. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionPhase { + StandaloneTurn, + PreTurn, + MidTurn, +} + +/// Strategy used to build the compacted context. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionStrategy { + Memento, + PrefixCompaction, +} + +/// Terminal status of a compaction run. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionStatus<'a> { + Completed, + Failed { + /// Bounded failure detail intended for remote export. + error: Option<&'a str>, + }, + Interrupted, +} + +/// Observation emitted when a compaction run reaches a terminal state. +#[derive(Observation)] +#[observation(name = "compaction.ended", crate = "crate", uses = ["analytics"])] +pub struct CompactionEnded<'a> { + #[obs(level = "basic", class = "identifier")] + pub thread_id: &'a str, + + #[obs(level = "basic", class = "identifier")] + pub turn_id: &'a str, + + #[obs(level = "basic", class = "operational")] + pub trigger: CompactionTrigger, + + #[obs(level = "basic", class = "operational")] + pub reason: CompactionReason, + + #[obs(level = "basic", class = "operational")] + pub implementation: CompactionImplementation, + + #[obs(level = "basic", class = "operational")] + pub phase: CompactionPhase, + + #[obs(level = "basic", class = "operational")] + pub strategy: CompactionStrategy, + + #[obs(level = "basic", class = "operational")] + pub status: CompactionStatus<'a>, + + #[obs(level = "basic", class = "operational")] + pub active_context_tokens_before: i64, + + #[obs(level = "basic", class = "operational")] + pub active_context_tokens_after: i64, + + /// Unix timestamp in seconds when compaction work began. + #[obs(level = "basic", class = "operational")] + pub started_at: i64, + + /// Unix timestamp in seconds when compaction reached its terminal state. + #[obs(level = "basic", class = "operational")] + pub ended_at: i64, + + /// Absent when the callsite cannot measure elapsed wall time. + #[obs(level = "basic", class = "operational")] + pub duration_ms: Option, +}