[observability] Add compaction ended observation

Introduce a typed compaction.ended observation and project it through the existing analytics compaction fact so the legacy reducer remains the only analytics emission path.

Add a side-by-side reducer test covering the subagent thread metadata join and verifying the observation path produces the same codex_compaction_event payload as the legacy fact.

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")

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(),

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<TrackEventRequest>,
+    ) {
+        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,

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::*;
 

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<i64>,
+}