[codex] Record exec-server lifecycle metrics (#27467)

## Summary

- Record bounded connection, request, and process lifecycle metrics.
- Report active gauges from callbacks on every collection, including
delta exports.
- Serialize active-count updates so concurrent starts and finishes
cannot publish stale values.
- Serialize process exit, explicit termination, and shutdown through the
process registry so exactly one completion result wins.
- Keep the implementation small with single-owner RAII guards and one
real OTLP/HTTP integration test using the existing `wiremock`
dependency.

## Root cause

Process exit and session shutdown previously used cloned completion
state. That avoided duplicate emission, but it duplicated lifecycle
ownership and made the ordering harder to reason about. The process
registry mutex already defines the lifecycle ordering, so the final
implementation stores the metric guard and termination flag directly on
the process entry. Whichever path claims the entry first owns the
completion result.

Production metric export uses delta temporality. Event-only synchronous
gauge recordings disappear after the next collection when no count
changes, so active counts now use observable callbacks that report
current state on every collection.

The cleanup also removes the constant `result="accepted"` connection
tag, redundant route and response assertions, a custom HTTP collector,
and fallback initialization machinery that did not add behavior.

## Stack

Review and land this stack in order:

1. #27466 — trace exec-server JSON-RPC requests
2. #27467 — record bounded connection, request, and process lifecycle
metrics **(this PR)**
3. #27470 — observe remote registration and Noise rendezvous lifecycle

## Validation

- `just test -p codex-exec-server --lib` (158 passed)
- `just test -p codex-cli --test exec_server` (3 passed)
- `just test -p codex-otel
observable_gauge_is_collected_on_every_delta_snapshot` (1 passed)
- `CARGO_BUILD_JOBS=1 just fix -p codex-otel -p codex-exec-server`
- `just fmt`
- `git diff --check`

codex-rs/cli/src/exec_server_telemetry.rs

@@ -7,7 +7,7 @@ const OTEL_SERVICE_NAME: &str = "codex-exec-server";
 
 pub(crate) fn init(
     config: Option<&codex_core::config::Config>,
-) -> Result<impl Send + Sync, Box<dyn std::error::Error>> {
+) -> (impl Send + Sync, codex_exec_server::ExecServerTelemetry) {
     let fmt_layer = tracing_subscriber::fmt::layer()
         .with_writer(std::io::stderr)
         .with_filter(stderr_env_filter());
@@ -17,21 +17,30 @@ pub(crate) fn init(
             env!("CARGO_PKG_VERSION"),
             Some(OTEL_SERVICE_NAME),
             DEFAULT_ANALYTICS_ENABLED,
-        ),
-        None => Ok(None),
+        )
+        .unwrap_or_else(|error| {
+            eprintln!("Could not create otel exporter: {error}");
+            None
+        }),
+        None => None,
     };
-    let provider = otel.as_ref().ok().and_then(Option::as_ref);
+    let provider = otel.as_ref();
     codex_core::otel_init::record_process_start(provider, OTEL_SERVICE_NAME);
 
     let otel_logger_layer = provider.and_then(|otel| otel.logger_layer());
     let otel_tracing_layer = provider.and_then(|otel| otel.tracing_layer());
+    let telemetry = provider
+        .and_then(|otel| otel.metrics())
+        .cloned()
+        .map(codex_exec_server::ExecServerTelemetry::new)
+        .unwrap_or_default();
     let _ = tracing_subscriber::registry()
         .with(fmt_layer)
         .with(otel_tracing_layer)
         .with(otel_logger_layer)
         .try_init();
     tracing::callsite::rebuild_interest_cache();
-    otel
+    (otel, telemetry)
 }
 
 fn stderr_env_filter() -> EnvFilter {

codex-rs/cli/src/main.rs

@@ -1693,9 +1693,7 @@ async fn run_exec_server_command(
             .environment_id
             .ok_or_else(|| anyhow::anyhow!("--environment-id is required when --remote is set"))?;
         let config = load_exec_server_config(root_config_overrides, strict_config).await?;
-        let _otel = exec_server_telemetry::init(Some(&config))
-            .inspect_err(|err| eprintln!("Could not create otel exporter: {err}"))
-            .ok();
+        let (_otel, telemetry) = exec_server_telemetry::init(Some(&config));
         let auth_provider =
             load_exec_server_remote_auth_provider(&config, &base_url, cmd.use_agent_identity_auth)
                 .await?;
@@ -1707,6 +1705,7 @@ async fn run_exec_server_command(
         if let Some(name) = cmd.name {
             remote_config.name = name;
         }
+        let remote_config = remote_config.with_telemetry(telemetry);
         codex_exec_server::run_remote_environment(remote_config, runtime_paths).await?;
         Ok(())
     } else {
@@ -1716,14 +1715,12 @@ async fn run_exec_server_command(
         } else {
             config_result.ok()
         };
-        let _otel = exec_server_telemetry::init(config.as_ref())
-            .inspect_err(|err| eprintln!("Could not create otel exporter: {err}"))
-            .ok();
+        let (_otel, telemetry) = exec_server_telemetry::init(config.as_ref());
         let listen_url = cmd
             .listen
             .as_deref()
             .unwrap_or(codex_exec_server::DEFAULT_LISTEN_URL);
-        codex_exec_server::run_main(listen_url, runtime_paths)
+        codex_exec_server::run_main_with_telemetry(listen_url, runtime_paths, telemetry)
             .await
             .map_err(anyhow::Error::from_boxed)
     }

codex-rs/cli/tests/exec_server.rs

@@ -1,9 +1,20 @@
 use std::path::Path;
+use std::process::Stdio;
+use std::time::Duration;
 
 use anyhow::Result;
 use predicates::prelude::PredicateBooleanExt;
 use predicates::str::contains;
 use tempfile::TempDir;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
 
 fn codex_command(codex_home: &Path) -> Result<assert_cmd::Command> {
     let mut cmd = assert_cmd::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?);
@@ -48,3 +59,225 @@ fn local_exec_server_ignores_invalid_config_without_strict_config() -> Result<()
 
     Ok(())
 }
+
+#[tokio::test]
+async fn local_exec_server_flushes_telemetry_on_stdio_disconnect() -> Result<()> {
+    let collector = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/metrics"))
+        .respond_with(ResponseTemplate::new(202))
+        .mount(&collector)
+        .await;
+    let codex_home = TempDir::new()?;
+    let base_url = collector.uri();
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"
+[analytics]
+enabled = true
+
+[otel]
+environment = "test"
+metrics_exporter = {{ otlp-http = {{ endpoint = "{base_url}/v1/metrics", protocol = "json" }} }}
+"#
+        ),
+    )?;
+
+    let cwd = url::Url::from_directory_path(std::env::current_dir()?)
+        .map_err(|()| anyhow::anyhow!("could not convert cwd to file URL"))?;
+    #[cfg(windows)]
+    let argv = vec!["ping.exe", "-n", "61", "127.0.0.1"];
+    #[cfg(not(windows))]
+    let argv = vec!["/bin/sleep", "60"];
+    let codex_bin = codex_utils_cargo_bin::cargo_bin("codex")?;
+    let codex_home = codex_home.path().to_path_buf();
+    let subprocess = async move {
+        let mut command = tokio::process::Command::new(codex_bin);
+        command
+            .env("CODEX_HOME", codex_home)
+            .env("NO_PROXY", "127.0.0.1,localhost")
+            .env("no_proxy", "127.0.0.1,localhost")
+            .args(["exec-server", "--listen", "stdio"])
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .kill_on_drop(true);
+        let mut child = command.spawn()?;
+        let mut stdin = child
+            .stdin
+            .take()
+            .ok_or_else(|| anyhow::anyhow!("exec-server stdin was not piped"))?;
+        let stdout = child
+            .stdout
+            .take()
+            .ok_or_else(|| anyhow::anyhow!("exec-server stdout was not piped"))?;
+        let mut stdout = BufReader::new(stdout);
+        send_json_line(
+            &mut stdin,
+            &serde_json::json!({
+                "id": 1,
+                "method": "initialize",
+                "params": {"clientName": "otel-test", "resumeSessionId": null}
+            }),
+        )
+        .await?;
+        wait_for_response(&mut stdout, /*expected_id*/ 1).await?;
+        send_json_line(
+            &mut stdin,
+            &serde_json::json!({"method": "initialized", "params": {}}),
+        )
+        .await?;
+        send_json_line(
+            &mut stdin,
+            &serde_json::json!({
+                "id": 2,
+                "method": "process/start",
+                "params": {
+                    "processId": "otel-process",
+                    "argv": argv,
+                    "cwd": cwd,
+                    "env": {},
+                    "tty": false,
+                    "pipeStdin": false,
+                    "arg0": null
+                }
+            }),
+        )
+        .await?;
+        wait_for_response(&mut stdout, /*expected_id*/ 2).await?;
+        drop(stdin);
+        let mut remaining_stdout = String::new();
+        stdout.read_to_string(&mut remaining_stdout).await?;
+        let status = child.wait().await?;
+        anyhow::ensure!(
+            status.success(),
+            "exec-server exited with {status}; remaining stdout: {remaining_stdout}"
+        );
+        Ok::<(), anyhow::Error>(())
+    };
+    let subprocess_result = tokio::time::timeout(Duration::from_secs(30), subprocess)
+        .await
+        .map_err(|_| anyhow::anyhow!("exec-server subprocess timed out"))?;
+    subprocess_result?;
+
+    let requests = collector
+        .received_requests()
+        .await
+        .ok_or_else(|| anyhow::anyhow!("failed to read OTLP collector requests"))?;
+    let metrics = requests
+        .iter()
+        .filter(|request| request.url.path() == "/v1/metrics")
+        .map(|request| serde_json::from_slice::<serde_json::Value>(&request.body))
+        .collect::<serde_json::Result<Vec<_>>>()?;
+    assert_metric_point(
+        &metrics,
+        "exec_server_connections_active",
+        &[("transport", "stdio")],
+        Some(0),
+    );
+    assert_metric_point(
+        &metrics,
+        "exec_server_connections_total",
+        &[("transport", "stdio")],
+        Some(1),
+    );
+    assert_metric_point(
+        &metrics,
+        "exec_server_requests_total",
+        &[("method", "process/start"), ("result", "success")],
+        Some(1),
+    );
+    assert_metric_point(&metrics, "exec_server_processes_active", &[], Some(0));
+    assert_metric_point(
+        &metrics,
+        "exec_server_processes_finished_total",
+        &[("result", "terminated")],
+        Some(1),
+    );
+    assert_metric_point(
+        &metrics,
+        "exec_server_request_duration_seconds",
+        &[("method", "process/start"), ("result", "success")],
+        /*value*/ None,
+    );
+    assert_metric_point(
+        &metrics,
+        "exec_server_process_duration_seconds",
+        &[("result", "terminated")],
+        /*value*/ None,
+    );
+    Ok(())
+}
+
+async fn send_json_line(
+    stdin: &mut (impl tokio::io::AsyncWrite + Unpin),
+    message: &serde_json::Value,
+) -> Result<()> {
+    let mut encoded = serde_json::to_vec(message)?;
+    encoded.push(b'\n');
+    stdin.write_all(&encoded).await?;
+    stdin.flush().await?;
+    Ok(())
+}
+
+async fn wait_for_response(
+    stdout: &mut (impl tokio::io::AsyncBufRead + Unpin),
+    expected_id: i64,
+) -> Result<()> {
+    loop {
+        let mut line = String::new();
+        if stdout.read_line(&mut line).await? == 0 {
+            anyhow::bail!("exec-server stdout closed before response {expected_id}");
+        }
+        let message: serde_json::Value = serde_json::from_str(&line)?;
+        if message["id"].as_i64() == Some(expected_id) {
+            anyhow::ensure!(
+                message.get("error").is_none(),
+                "exec-server request {expected_id} failed: {message}"
+            );
+            return Ok(());
+        }
+    }
+}
+
+fn assert_metric_point(
+    payloads: &[serde_json::Value],
+    name: &str,
+    attributes: &[(&str, &str)],
+    value: Option<i64>,
+) {
+    let found = payloads
+        .iter()
+        .flat_map(|payload| payload["resourceMetrics"].as_array().into_iter().flatten())
+        .flat_map(|resource| resource["scopeMetrics"].as_array().into_iter().flatten())
+        .flat_map(|scope| scope["metrics"].as_array().into_iter().flatten())
+        .filter(|metric| metric["name"].as_str() == Some(name))
+        .flat_map(|metric| {
+            ["gauge", "sum", "histogram"]
+                .into_iter()
+                .find_map(|kind| metric[kind]["dataPoints"].as_array())
+                .into_iter()
+                .flatten()
+        })
+        .any(|point| {
+            let actual_attributes = point["attributes"]
+                .as_array()
+                .map(Vec::as_slice)
+                .unwrap_or_default();
+            let attributes_match = actual_attributes.len() == attributes.len()
+                && attributes.iter().all(|(expected_key, expected_value)| {
+                    actual_attributes.iter().any(|actual| {
+                        actual["key"].as_str() == Some(*expected_key)
+                            && actual["value"]["stringValue"].as_str() == Some(*expected_value)
+                    })
+                });
+            let actual_value = point["asInt"]
+                .as_i64()
+                .or_else(|| point["asInt"].as_str()?.parse().ok());
+            attributes_match && value.is_none_or(|expected| actual_value == Some(expected))
+        });
+    assert!(
+        found,
+        "metric {name} with attributes {attributes:?} and value {value:?} missing"
+    );
+}