feat: phase 1 and phase 2 e2e latencies (#12124)

codex-rs/core/src/memories/mod.rs

@@ -67,12 +67,16 @@ mod phase_two {
 mod metrics {
     /// Number of phase-1 startup jobs grouped by status.
     pub(super) const MEMORY_PHASE_ONE_JOBS: &str = "codex.memory.phase1";
+    /// End-to-end latency for a single phase-1 startup run.
+    pub(super) const MEMORY_PHASE_ONE_E2E_MS: &str = "codex.memory.phase1.e2e_ms";
     /// Number of raw memories produced by phase-1 startup extraction.
     pub(super) const MEMORY_PHASE_ONE_OUTPUT: &str = "codex.memory.phase1.output";
     /// Histogram for aggregate token usage across one phase-1 startup run.
     pub(super) const MEMORY_PHASE_ONE_TOKEN_USAGE: &str = "codex.memory.phase1.token_usage";
     /// Number of phase-2 startup jobs grouped by status.
     pub(super) const MEMORY_PHASE_TWO_JOBS: &str = "codex.memory.phase2";
+    /// End-to-end latency for a single phase-2 consolidation run.
+    pub(super) const MEMORY_PHASE_TWO_E2E_MS: &str = "codex.memory.phase2.e2e_ms";
     /// Number of stage-1 memories included in each phase-2 consolidation step.
     pub(super) const MEMORY_PHASE_TWO_INPUT: &str = "codex.memory.phase2.input";
 }

codex-rs/core/src/memories/phase1.rs

@@ -80,6 +80,12 @@ struct StageOneOutput {
 /// 3) run stage-1 extraction jobs in parallel
 /// 4) emit metrics and logs
 pub(in crate::memories) async fn run(session: &Arc<Session>, config: &Config) {
+    let _phase_one_e2e_timer = session
+        .services
+        .otel_manager
+        .start_timer(metrics::MEMORY_PHASE_ONE_E2E_MS, &[])
+        .ok();
+
     // 1. Claim startup job.
     let Some(claimed_candidates) = claim_startup_jobs(session, &config.memories).await else {
         return;

codex-rs/core/src/memories/phase2.rs

@@ -36,6 +36,12 @@ struct Counters {
 /// Runs memory phase 2 (aka consolidation) in strict order. The method represents the linear
 /// flow of the consolidation phase.
 pub(super) async fn run(session: &Arc<Session>, config: Arc<Config>) {
+    let phase_two_e2e_timer = session
+        .services
+        .otel_manager
+        .start_timer(metrics::MEMORY_PHASE_TWO_E2E_MS, &[])
+        .ok();
+
     let Some(db) = session.services.state_db.as_deref() else {
         // This should not happen.
         return;
@@ -117,7 +123,13 @@ pub(super) async fn run(session: &Arc<Session>, config: Arc<Config>) {
     };
 
     // 6. Spawn the agent handler.
-    agent::handle(session, claim, new_watermark, thread_id);
+    agent::handle(
+        session,
+        claim,
+        new_watermark,
+        thread_id,
+        phase_two_e2e_timer,
+    );
 
     // 7. Metrics and logs.
     let counters = Counters {
@@ -264,6 +276,7 @@ mod agent {
         claim: Claim,
         new_watermark: i64,
         thread_id: ThreadId,
+        phase_two_e2e_timer: Option<codex_otel::Timer>,
     ) {
         let Some(db) = session.services.state_db.clone() else {
             return;
@@ -271,6 +284,7 @@ mod agent {
         let session = session.clone();
 
         tokio::spawn(async move {
+            let _phase_two_e2e_timer = phase_two_e2e_timer;
             let agent_control = session.services.agent_control.clone();
 
             // TODO(jif) we might have a very small race here.