Add attributed exec process lifecycle telemetry (#42373)

## Why Exec-server process and network-policy events need stable attribution to the launching tool call and executor without recording process payloads. ## What changed - Add optional `ExecMetadata` to `ExecParams` and propagate the thread and tool call IDs from unified exec. - Emit bounded OpenTelemetry events for process start, spawn failure, sandbox denial, and exit. Correlate them with the launch trace and registry-issued executor identity while excluding arguments, paths, environment values, output, and error text. - Preserve launch attribution across long-running processes and reconnects, and attach the same metadata to network-policy audit events. - Keep the protocol backward compatible when metadata is omitted, and prevent invalid trace headers from inheriting an unrelated active span. ## Testing Add coverage for metadata serialization and propagation, lifecycle event fields, trace relationships, reconnect behavior, spawn failures, sandbox denials, and network-policy attribution. GitOrigin-RevId: 7aa480a7289c73cb95e2c124c35500bb6f0d5084
2026-11-12 21:20:34 +00:00 · 2026-09-02 21:58:58 +00:00
parent 5037919777
commit f53c91be2c
47 changed files with 1461 additions and 178 deletions
@@ -6,6 +6,7 @@ use std::sync::atomic::Ordering;
 use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_exec_server_protocol::RequestId;
 use codex_http_client::HttpClientFactory;
+use opentelemetry::trace::SpanContext;
 use serde_json::to_value;
 use std::collections::HashSet;
 use tokio::sync::Mutex;
@@ -69,8 +70,10 @@ use crate::rpc::invalid_request;
 use crate::server::file_system_handler::FileSystemHandler;
 use crate::server::session_registry::SessionHandle;
 use crate::server::session_registry::SessionRegistry;
+use crate::telemetry::ExecutorRegistration;

 pub(crate) struct ExecServerHandler {
+    pub(super) executor_registration: Option<Arc<ExecutorRegistration>>,
    session_registry: Arc<SessionRegistry>,
    notifications: RpcNotificationSender,
    session: StdMutex<Option<SessionHandle>>,
@@ -92,6 +95,7 @@ impl ExecServerHandler {
        http_client_factory: HttpClientFactory,
    ) -> Self {
        Self {
+            executor_registration: None,
            session_registry,
            notifications,
            session: StdMutex::new(None),
@@ -172,9 +176,23 @@ impl ExecServerHandler {
        Ok(())
    }

-    pub(crate) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, JSONRPCErrorError> {
+    pub(crate) async fn exec(
+        &self,
+        params: ExecParams,
+        launch_context: Option<SpanContext>,
+    ) -> Result<ExecResponse, JSONRPCErrorError> {
        let session = self.require_initialized_for("exec")?;
-        session.process().exec(params).await
+        session
+            .process()
+            .exec(
+                params,
+                crate::process_telemetry::ProcessTelemetry {
+                    launch_context,
+                    executor_registration: self.executor_registration.clone(),
+                    ..Default::default()
+                },
+            )
+            .await
    }

    pub(crate) fn environment_info(&self) -> Result<EnvironmentInfo, JSONRPCErrorError> {
@@ -27,6 +27,7 @@ fn exec_params(process_id: &str) -> ExecParams {

 fn exec_params_with_argv(process_id: &str, argv: Vec<String>) -> ExecParams {
    ExecParams {
+        metadata: Default::default(),
        process_id: ProcessId::from(process_id),
        argv,
        cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd"))
@@ -116,8 +117,8 @@ async fn duplicate_process_ids_allow_only_one_successful_start() {
    let second_handler = Arc::clone(&handler);

    let (first, second) = tokio::join!(
-        first_handler.exec(exec_params("proc-1")),
-        second_handler.exec(exec_params("proc-1")),
+        first_handler.exec(exec_params("proc-1"), /*launch_context*/ None),
+        second_handler.exec(exec_params("proc-1"), /*launch_context*/ None),
    );

    let (successes, failures): (Vec<_>, Vec<_>) =
@@ -141,7 +142,7 @@ async fn duplicate_process_ids_allow_only_one_successful_start() {
 async fn terminate_reports_false_after_process_exit() {
    let handler = initialized_handler().await;
    handler
-        .exec(exec_params("proc-1"))
+        .exec(exec_params("proc-1"), /*launch_context*/ None)
        .await
        .expect("start process");

@@ -188,10 +189,13 @@ async fn long_poll_read_fails_after_session_resume() {
    // Keep the process quiet and alive so the pending read can only complete
    // after session resume, not because the process produced output or exited.
    first_handler
-        .exec(exec_params_with_argv(
-            "proc-long-poll",
-            shell_argv("sleep 5", "ping -n 6 127.0.0.1 >NUL"),
-        ))
+        .exec(
+            exec_params_with_argv(
+                "proc-long-poll",
+                shell_argv("sleep 5", "ping -n 6 127.0.0.1 >NUL"),
+            ),
+            /*launch_context*/ None,
+        )
        .await
        .expect("start process");

@@ -306,13 +310,16 @@ async fn output_and_exit_are_retained_after_notification_receiver_closes() {

    let process_id = ProcessId::from("proc-notification-fail");
    handler
-        .exec(exec_params_with_argv(
-            process_id.as_str(),
-            shell_argv(
-                "sleep 0.05; printf 'first\\n'; sleep 0.05; printf 'second\\n'",
-                "echo first&& ping -n 2 127.0.0.1 >NUL&& echo second",
+        .exec(
+            exec_params_with_argv(
+                process_id.as_str(),
+                shell_argv(
+                    "sleep 0.05; printf 'first\\n'; sleep 0.05; printf 'second\\n'",
+                    "echo first&& ping -n 2 127.0.0.1 >NUL&& echo second",
+                ),
            ),
-        ))
+            /*launch_context*/ None,
+        )
        .await
        .expect("start process");

@@ -324,7 +331,10 @@ async fn output_and_exit_are_retained_after_notification_receiver_closes() {

    tokio::time::sleep(Duration::from_millis(100)).await;
    handler
-        .exec(exec_params(process_id.as_str()))
+        .exec(
+            exec_params(process_id.as_str()),
+            /*launch_context*/ None,
+        )
        .await
        .expect("process id should be reusable after exit retention");

@@ -1,3 +1,4 @@
+use crate::process_telemetry::ProcessTelemetry;
 use codex_exec_server_protocol::JSONRPCErrorError;

 use crate::ExecServerRuntimePaths;
@@ -39,8 +40,12 @@ impl ProcessHandler {
        self.process.set_notification_sender(notifications);
    }

-    pub(crate) async fn exec(&self, params: ExecParams) -> Result<ExecResponse, JSONRPCErrorError> {
-        self.process.exec(params).await
+    pub(crate) async fn exec(
+        &self,
+        params: ExecParams,
+        telemetry: ProcessTelemetry,
+    ) -> Result<ExecResponse, JSONRPCErrorError> {
+        self.process.exec(params, telemetry).await
    }

    pub(crate) async fn exec_read(
@@ -0,0 +1,752 @@
+//! Verifies exported lifecycle and network logs retain launch identity without private payloads.
+
+#![allow(clippy::expect_used)]
+
+use std::collections::BTreeMap;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCRequest;
+use codex_exec_server_protocol::RequestId;
+use codex_http_client::HttpClientFactory;
+use codex_http_client::OutboundProxyPolicy;
+use codex_network_proxy::NetworkProxyConfig;
+use codex_network_proxy::RemoteNetworkProxyConfig;
+use codex_network_proxy::RemoteNetworkProxyLaunchConfig;
+use codex_otel::OtelExporter;
+use codex_otel::OtelHttpProtocol;
+use codex_otel::OtelProvider;
+use codex_otel::OtelSettings;
+use codex_protocol::protocol::W3cTraceContext;
+use codex_utils_path_uri::PathUri;
+use pretty_assertions::assert_eq;
+use serde_json::Value;
+use serde_json::json;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::sync::mpsc;
+use tracing::Instrument;
+use tracing_subscriber::prelude::*;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+
+use super::ExecServerHandler;
+use super::registry::build_router;
+use super::session_registry::SessionRegistry;
+use crate::ExecServerRuntimePaths;
+use crate::ExecServerTelemetry;
+use crate::connection::JsonRpcConnectionEvent;
+use crate::protocol::EXEC_METHOD;
+use crate::protocol::EXEC_READ_METHOD;
+use crate::protocol::EXEC_TERMINATE_METHOD;
+use crate::protocol::InitializeParams;
+use crate::protocol::ReadResponse;
+use crate::rpc::RpcNotificationSender;
+use crate::rpc::RpcRouter;
+use crate::rpc::RpcServerOutboundMessage;
+use crate::telemetry::ExecutorRegistration;
+
+const PRIVATE_PAYLOAD: &str = "private-process-payload";
+const TRACE_ID: &str = "11111111111111111111111111111111";
+const LAUNCH_SPANS: [&str; 2] = ["2222222222222222", "3333333333333333"];
+const THREADS: [&str; 2] = [
+    "11111111-1111-4111-8111-111111111111",
+    "22222222-2222-4222-8222-222222222222",
+];
+const CALLS: [&str; 2] = ["call-first", "call-second"];
+const LATER_TRACE: &str = "00-44444444444444444444444444444444-5555555555555555-01";
+
+#[test]
+fn exported_process_and_network_logs_keep_the_validated_launch_reference() {
+    // Keep the subscriber on the runtime's only thread, including proxy background tasks.
+    for (export_traces, registered) in [(false, true), (true, true), (false, false)] {
+        let records = exported_logs(export_traces, |runtime, _, _| {
+            runtime.block_on(async {
+                let server = TestServer::new();
+                let mut handlers = Vec::new();
+                let mut session_ids = Vec::new();
+                for client_name in ["first-orchestrator", "second-orchestrator"] {
+                    let (handler, session_id) = server.new_orchestrator(
+                        client_name, registered.then_some("original"),
+                    ).await;
+                    handlers.push(handler);
+                    session_ids.push(session_id);
+                }
+                let proxy = RemoteNetworkProxyLaunchConfig::new(
+                    RemoteNetworkProxyConfig::from_effective_config(&NetworkProxyConfig {
+                        enabled: true,
+                        ..NetworkProxyConfig::default()
+                    }).expect("proxy config"),
+                );
+                // Unsampled incoming context must still correlate logs-only export.
+                let flags = if export_traces { "01" } else { "00" };
+                let launches = LAUNCH_SPANS.map(|span| format!("00-{TRACE_ID}-{span}-{flags}"));
+                for (batch, traces) in [
+                    [Some(launches[0].as_str()), Some(launches[1].as_str())],
+                    [None, Some(PRIVATE_PAYLOAD)],
+                ].into_iter().enumerate() {
+                    // Process IDs are session-scoped and deliberately reused across clients.
+                    let process_id = format!("{PRIVATE_PAYLOAD}-{batch}");
+                    for (index, (handler, trace)) in handlers.iter().zip(traces).enumerate() {
+                        let mut params = process_start_params(
+                            &process_id,
+                            json!(["/bin/sh", "-c", "printf '%s\\n' \"$HTTP_PROXY\"; read ignored", PRIVATE_PAYLOAD]),
+                            PathUri::from_host_native_path(std::env::current_dir().expect("cwd")).expect("cwd URI"),
+                        );
+                        params["metadata"] = if batch == 0 {
+                            process_metadata(index)
+                        } else {
+                            json!({"toolCallId": if index == 0 { None } else { Some("private-process-payload\n") }})
+                        };
+                        params["metadata"]["executorRegistrationId"] = json!(PRIVATE_PAYLOAD);
+                        params["metadata"]["environmentId"] = json!(PRIVATE_PAYLOAD);
+                        params["executorRegistrationId"] = json!(PRIVATE_PAYLOAD);
+                        params["environmentId"] = json!(PRIVATE_PAYLOAD);
+                        params["pipeStdin"] = json!(true);
+                        params["networkProxy"] = json!(proxy);
+                        request(&server.router, handler, EXEC_METHOD, params, trace).await;
+                    }
+                    if batch == 0 {
+                        handlers[0].shutdown().await;
+                        let resumed = new_handler(
+                            &server.sessions, server.outgoing.clone(), registered.then_some("resumed"),
+                        );
+                        resumed.initialize(InitializeParams {
+                            client_name: "resumed-orchestrator".to_string(),
+                            resume_session_id: Some(session_ids[0].clone()),
+                        }).await.expect("resume session");
+                        resumed.initialized().expect("initialized resumed orchestrator");
+                        handlers[0] = resumed;
+                    }
+                    for handler in &handlers {
+                        let output: ReadResponse = serde_json::from_value(request(
+                            &server.router, handler, EXEC_READ_METHOD,
+                            json!({"processId": process_id, "waitMs": 1000}), Some(LATER_TRACE),
+                        ).await).expect("read proxy address");
+                        let proxy_address = String::from_utf8(output.chunks.into_iter()
+                            .flat_map(|chunk| chunk.chunk.into_inner()).collect()).expect("UTF-8 output");
+                        let mut connection = tokio::net::TcpStream::connect(
+                            proxy_address.trim().strip_prefix("http://").expect("HTTP proxy address"),
+                        ).await.expect("connect to process proxy");
+                        connection.write_all(b"CONNECT 8.8.8.8:443 HTTP/1.1\r\nHost: 8.8.8.8:443\r\n\r\n")
+                            .await.expect("request denied destination");
+                        let mut response = [0_u8; 256];
+                        let length = tokio::time::timeout(Duration::from_secs(5), connection.read(&mut response))
+                            .await.expect("proxy response timeout").expect("proxy response");
+                        assert!(String::from_utf8_lossy(&response[..length]).starts_with("HTTP/1.1 403"));
+                        request(&server.router, handler, EXEC_TERMINATE_METHOD,
+                            json!({"processId": process_id}), Some(LATER_TRACE)).await;
+                        read_until_closed(&server.router, handler, &process_id).await;
+                    }
+                }
+                server.sessions.shutdown().await;
+            });
+        });
+        let mut events = Vec::new();
+        for record in &records {
+            let name = attribute(record, "event.name").expect("event name");
+            let trace = attribute(record, "launch.trace_id");
+            let span = attribute(record, "launch.span_id");
+            if export_traces
+                && name.starts_with("codex.exec_server.process_")
+                && let Some(span) = span
+            {
+                assert_eq!(record["traceId"].as_str(), trace);
+                assert_ne!(record["spanId"].as_str().expect("native child span"), span);
+            }
+            events.push((
+                name.to_string(),
+                trace.map(str::to_string),
+                span.map(str::to_string),
+                attribute(record, "conversation.id").map(str::to_string),
+                attribute(record, "tool.call_id").map(str::to_string),
+                attribute(record, "executor.environment_id").map(str::to_string),
+                attribute(record, "executor.registration_id").map(str::to_string),
+            ));
+        }
+        let mut expected = Vec::new();
+        for (index, span) in [Some(LAUNCH_SPANS[0]), Some(LAUNCH_SPANS[1]), None, None]
+            .into_iter()
+            .enumerate()
+        {
+            for name in [
+                "codex.exec_server.process_start",
+                "codex.exec_server.process_exit",
+                "codex.network_proxy.policy_decision",
+            ] {
+                expected.push((
+                    name.to_string(),
+                    span.map(|_| TRACE_ID.to_string()),
+                    span.map(str::to_string),
+                    (index < 2).then(|| THREADS[index].to_string()),
+                    (index < 2).then(|| CALLS[index].to_string()),
+                    registered.then(|| "environment".to_string()),
+                    registered.then(|| if index == 2 { "resumed" } else { "original" }.to_string()),
+                ));
+            }
+        }
+        events.sort();
+        expected.sort();
+        assert_eq!(events, expected);
+    }
+}
+
+#[test]
+fn launch_rpc_span_finishes_before_its_process_exits() {
+    let mut observed = None;
+    let records = exported_logs(/*export_traces*/ true, |runtime, otel, collector| {
+        runtime.block_on(async {
+            let server = TestServer::new();
+            let (handler, _) = server
+                .new_orchestrator("process-span-orchestrator", Some("original"))
+                .await;
+            let launch = format!("00-{TRACE_ID}-{}-01", LAUNCH_SPANS[0]);
+            let mut params = process_start_params(
+                PRIVATE_PAYLOAD,
+                json!(["/bin/sh", "-c", "read ignored", PRIVATE_PAYLOAD]),
+                PathUri::from_host_native_path(std::env::current_dir().expect("cwd"))
+                    .expect("cwd URI"),
+            );
+            params["metadata"] = process_metadata(/*index*/ 0);
+            params["pipeStdin"] = json!(true);
+            request(&server.router, &handler, EXEC_METHOD, params, Some(&launch)).await;
+
+            // Flush completed spans while the process is still waiting on its open stdin.
+            // The old task's in_current_span() retains the RPC span, so this snapshot lacks it.
+            let spans_before_exit = flushed_spans(otel, collector).await;
+            let running: ReadResponse = serde_json::from_value(
+                request(
+                    &server.router,
+                    &handler,
+                    EXEC_READ_METHOD,
+                    json!({"processId": PRIVATE_PAYLOAD, "waitMs": 0}),
+                    Some(LATER_TRACE),
+                )
+                .await,
+            )
+            .expect("read running process");
+            request(
+                &server.router,
+                &handler,
+                EXEC_TERMINATE_METHOD,
+                json!({"processId": PRIVATE_PAYLOAD}),
+                Some(LATER_TRACE),
+            )
+            .await;
+            let exited = read_until_closed(&server.router, &handler, PRIVATE_PAYLOAD).await;
+            server.sessions.shutdown().await;
+            let spans_after_exit = flushed_spans(otel, collector).await;
+            // Finish cleanup before asserting the regression so a failure leaves no child behind.
+            observed = Some((running, exited, spans_before_exit, spans_after_exit));
+        });
+    });
+    let (running, exited, spans_before_exit, spans_after_exit) =
+        observed.expect("process observations");
+    assert_eq!(
+        (running.exited, running.closed, running.exit_code),
+        (false, false, None)
+    );
+    let launch_spans: Vec<_> = spans_before_exit
+        .iter()
+        .filter(|span| span["name"] == EXEC_METHOD)
+        .collect();
+    assert_eq!(
+        launch_spans.len(),
+        1,
+        "process/start span must export before child exit"
+    );
+    let launch_span = launch_spans[0];
+    assert_eq!(
+        (
+            launch_span["traceId"].as_str(),
+            launch_span["parentSpanId"].as_str()
+        ),
+        (Some(TRACE_ID), Some(LAUNCH_SPANS[0]))
+    );
+    let exit_record = records
+        .iter()
+        .find(|record| attribute(record, "event.name") == Some("codex.exec_server.process_exit"))
+        .expect("process exit log");
+    let exit_span_id = exit_record["spanId"]
+        .as_str()
+        .expect("process exit span ID");
+    // ProcessMetricGuard has the same span name; select the span that emitted the exit log.
+    let process_spans: Vec<_> = spans_after_exit
+        .iter()
+        .filter(|span| span["spanId"].as_str() == Some(exit_span_id))
+        .collect();
+    assert_eq!(
+        process_spans.len(),
+        1,
+        "process completion has its own span"
+    );
+    let process_span = process_spans[0];
+    assert_eq!(process_span["name"], "codex.exec_server.process");
+    assert_eq!(
+        (
+            process_span["traceId"].as_str(),
+            process_span["parentSpanId"].as_str()
+        ),
+        (Some(TRACE_ID), Some(LAUNCH_SPANS[0]))
+    );
+    assert_ne!(process_span["spanId"], launch_span["spanId"]);
+    let mut expected_exit =
+        expected_process_attributes("codex.exec_server.process_exit", /*index*/ 0, "None");
+    expected_exit["process.exit_code"] =
+        json!({"intValue": exited.exit_code.expect("exit code").to_string()});
+    expected_exit["process.termination_requested"] = json!({"boolValue": true});
+    assert_exported_attributes(
+        &records,
+        vec![
+            expected_process_attributes(
+                "codex.exec_server.process_start",
+                /*index*/ 0,
+                "None",
+            ),
+            expected_exit,
+        ],
+    );
+    for record in &records {
+        assert_eq!(record["traceId"].as_str(), Some(TRACE_ID));
+        let span = if attribute(record, "event.name") == Some("codex.exec_server.process_start") {
+            launch_span
+        } else {
+            process_span
+        };
+        assert_eq!(record["spanId"], span["spanId"]);
+    }
+}
+
+async fn flushed_spans(otel: &OtelProvider, collector: &MockServer) -> Vec<Value> {
+    let provider = otel
+        .tracer_provider
+        .as_ref()
+        .expect("trace exporter")
+        .clone();
+    // The collector uses this runtime, so the synchronous export flush must not block it.
+    tokio::time::timeout(
+        Duration::from_secs(5),
+        tokio::task::spawn_blocking(move || provider.force_flush()),
+    )
+    .await
+    .expect("trace flush timeout")
+    .expect("trace flush task")
+    .expect("trace flush");
+    let mut spans = Vec::new();
+    for request in collector
+        .received_requests()
+        .await
+        .expect("exported requests")
+    {
+        let body: Value = serde_json::from_slice(&request.body).expect("OTLP JSON");
+        let Some(resources) = body["resourceSpans"].as_array() else {
+            continue;
+        };
+        for resource in resources {
+            for scope in resource["scopeSpans"].as_array().expect("scope spans") {
+                spans.extend(scope["spans"].as_array().expect("spans").iter().cloned());
+            }
+        }
+    }
+    spans
+}
+
+#[test]
+fn exported_spawn_failure_keeps_launch_identity_without_outcome_or_error_text() {
+    let records = exported_logs(/*export_traces*/ false, |runtime, _, _| {
+        runtime.block_on(async {
+            let server = TestServer::new();
+            let (handler, _) = server
+                .new_orchestrator("spawn-failure-orchestrator", Some("original"))
+                .await;
+            let directory = tempfile::tempdir().expect("process directory");
+            let missing_executable = directory.path().join(PRIVATE_PAYLOAD);
+            let launch = format!("00-{TRACE_ID}-{}-00", LAUNCH_SPANS[0]);
+            let mut params = process_start_params(
+                PRIVATE_PAYLOAD,
+                json!([missing_executable, PRIVATE_PAYLOAD]),
+                PathUri::from_host_native_path(directory.path()).expect("cwd URI"),
+            );
+            params["metadata"] = process_metadata(/*index*/ 0);
+            let response =
+                request_result(&server.router, &handler, EXEC_METHOD, params, Some(&launch)).await;
+            let Some(RpcServerOutboundMessage::Error { error, .. }) = response else {
+                panic!("missing executable must fail to spawn: {response:?}");
+            };
+            assert_eq!(error.code, -32603);
+            assert!(
+                !error.message.is_empty(),
+                "caller still receives the spawn error"
+            );
+            server.sessions.shutdown().await;
+        });
+    });
+    assert_exported_attributes(
+        &records,
+        vec![expected_process_attributes(
+            "codex.exec_server.process_spawn_failed",
+            /*index*/ 0,
+            "None",
+        )],
+    );
+}
+
+#[cfg(target_os = "macos")]
+#[test]
+fn exported_sandbox_denial_keeps_launch_identity_and_separate_exit_outcome() {
+    let records = exported_logs(/*export_traces*/ false, |runtime, _, _| {
+        runtime.block_on(async {
+            let server = TestServer::new();
+            let (handler, _) = server.new_orchestrator("sandbox-denial-orchestrator", Some("original")).await;
+            let directory = tempfile::tempdir().expect("process directory");
+            let private_file = directory.path().join(PRIVATE_PAYLOAD);
+            std::fs::write(&private_file, PRIVATE_PAYLOAD).expect("write test file");
+            let cwd = PathUri::from_host_native_path(directory.path()).expect("cwd URI");
+            let sandbox = crate::FileSystemSandboxContext::from_legacy_sandbox_policy(
+                codex_protocol::protocol::SandboxPolicy::new_read_only_policy(), cwd.clone(),
+            ).expect("read-only sandbox");
+            for (index, span) in LAUNCH_SPANS.into_iter().enumerate() {
+                let process_id = format!("{PRIVATE_PAYLOAD}-{index}");
+                let launch = format!("00-{TRACE_ID}-{span}-00");
+                let argv = if index == 0 {
+                    json!(["/bin/cat", private_file])
+                } else {
+                    // Emit private output, then attempt a real denied write. Exit 23 is produced
+                    // by the child shell only after that write fails, not by sandbox startup.
+                    json!(["/bin/sh", "-c",
+                        "printf '%s' \"$PRIVATE_TEST_VALUE\"; if printf changed > \"$1\"; then exit 0; else exit 23; fi",
+                        PRIVATE_PAYLOAD, private_file])
+                };
+                let mut params = process_start_params(&process_id, argv, cwd.clone());
+                params["metadata"] = process_metadata(index);
+                params["env"]["LC_ALL"] = json!("C");
+                params["sandbox"] = json!(sandbox);
+                let started: crate::protocol::ExecResponse = serde_json::from_value(request(
+                    &server.router, &handler, EXEC_METHOD, params, Some(&launch),
+                ).await).expect("start sandboxed process");
+                assert_eq!(started.sandbox_type, Some(crate::protocol::ProcessSandboxType::MacosSeatbelt));
+                let output = read_until_closed(&server.router, &handler, &process_id).await;
+                let stdout: Vec<u8> = output.chunks.iter()
+                    .filter(|chunk| chunk.stream == crate::protocol::ExecOutputStream::Stdout)
+                    .flat_map(|chunk| chunk.chunk.0.iter().copied()).collect();
+                assert_eq!(stdout, PRIVATE_PAYLOAD.as_bytes());
+                assert_eq!((output.exit_code, output.sandbox_denied), if index == 0 {
+                    (Some(0), false)
+                } else {
+                    (Some(23), true)
+                });
+                if index == 1 {
+                    let stderr: Vec<u8> = output.chunks.iter()
+                        .filter(|chunk| chunk.stream == crate::protocol::ExecOutputStream::Stderr)
+                        .flat_map(|chunk| chunk.chunk.0.iter().copied()).collect();
+                    assert!(String::from_utf8_lossy(&stderr).contains("Operation not permitted"));
+                }
+            }
+            assert_eq!(std::fs::read(&private_file).expect("read test file"), PRIVATE_PAYLOAD.as_bytes());
+            server.sessions.shutdown().await;
+        });
+    });
+    let mut expected = Vec::new();
+    for index in 0..2 {
+        expected.push(expected_process_attributes(
+            "codex.exec_server.process_start",
+            index,
+            "MacosSeatbelt",
+        ));
+        let mut exited =
+            expected_process_attributes("codex.exec_server.process_exit", index, "MacosSeatbelt");
+        exited["process.exit_code"] = json!({"intValue": if index == 0 { "0" } else { "23" }});
+        exited["process.termination_requested"] = json!({"boolValue": false});
+        expected.push(exited);
+    }
+    let mut denied = expected_process_attributes(
+        "codex.exec_server.sandbox_denied",
+        /*index*/ 1,
+        "MacosSeatbelt",
+    );
+    denied["reason"] = json!({"stringValue": "inferred_denial"});
+    expected.push(denied);
+    assert_exported_attributes(&records, expected);
+}
+
+async fn read_until_closed(
+    router: &RpcRouter<ExecServerHandler>,
+    handler: &Arc<ExecServerHandler>,
+    process_id: &str,
+) -> ReadResponse {
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let output: ReadResponse = serde_json::from_value(
+                request(
+                    router,
+                    handler,
+                    EXEC_READ_METHOD,
+                    json!({"processId": process_id, "waitMs": 100}),
+                    Some(LATER_TRACE),
+                )
+                .await,
+            )
+            .expect("read process exit");
+            if output.closed {
+                return output;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("process exits")
+}
+
+fn expected_process_attributes(name: &str, index: usize, sandbox: &str) -> Value {
+    json!({
+        "event.name": {"stringValue": name},
+        "launch.trace_id": {"stringValue": TRACE_ID},
+        "launch.span_id": {"stringValue": LAUNCH_SPANS[index]},
+        "conversation.id": {"stringValue": THREADS[index]},
+        "tool.call_id": {"stringValue": CALLS[index]},
+        "executor.environment_id": {"stringValue": "environment"},
+        "executor.registration_id": {"stringValue": "original"},
+        "sandbox.type": {"stringValue": sandbox},
+    })
+}
+
+fn assert_exported_attributes(records: &[Value], mut expected: Vec<Value>) {
+    let mut actual: Vec<Value> = records
+        .iter()
+        .map(|record| {
+            Value::Object(
+                record["attributes"]
+                    .as_array()
+                    .expect("log attributes")
+                    .iter()
+                    .map(|attribute| {
+                        (
+                            attribute["key"]
+                                .as_str()
+                                .expect("attribute key")
+                                .to_string(),
+                            attribute["value"].clone(),
+                        )
+                    })
+                    .collect(),
+            )
+        })
+        .collect();
+    actual.sort_by_key(Value::to_string);
+    expected.sort_by_key(Value::to_string);
+    assert_eq!(actual, expected);
+}
+
+async fn request(
+    router: &RpcRouter<ExecServerHandler>,
+    handler: &Arc<ExecServerHandler>,
+    method: &str,
+    params: Value,
+    traceparent: Option<&str>,
+) -> Value {
+    let response = request_result(router, handler, method, params, traceparent).await;
+    let Some(RpcServerOutboundMessage::Response { result, .. }) = response else {
+        panic!("request failed: {response:?}");
+    };
+    result
+}
+
+async fn request_result(
+    router: &RpcRouter<ExecServerHandler>,
+    handler: &Arc<ExecServerHandler>,
+    method: &str,
+    params: Value,
+    traceparent: Option<&str>,
+) -> Option<RpcServerOutboundMessage> {
+    let request = JSONRPCRequest {
+        id: RequestId::Integer(1),
+        method: method.to_string(),
+        params: Some(params),
+        trace: traceparent.map(|traceparent| W3cTraceContext {
+            traceparent: Some(traceparent.to_string()),
+            tracestate: None,
+        }),
+    };
+    let JsonRpcConnectionEvent::QueuedRequest {
+        request,
+        request_span,
+        ..
+    } = JsonRpcConnectionEvent::message(JSONRPCMessage::Request(request))
+    else {
+        panic!("request span")
+    };
+    let (_, route) = router
+        .request_route(method)
+        .expect("registered request route");
+    request_span.record("otel.name", method);
+    route(Arc::clone(handler), request)
+        .instrument(request_span)
+        .await
+}
+
+fn attribute<'a>(record: &'a Value, key: &str) -> Option<&'a str> {
+    record["attributes"]
+        .as_array()
+        .expect("log attributes")
+        .iter()
+        .find(|attribute| attribute["key"] == key)
+        .map(|attribute| {
+            attribute["value"]["stringValue"]
+                .as_str()
+                .expect("string attribute")
+        })
+}
+
+fn exported_logs(
+    export_traces: bool,
+    run: impl FnOnce(&tokio::runtime::Runtime, &OtelProvider, &MockServer),
+) -> Vec<Value> {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("test runtime");
+    let collector = runtime.block_on(MockServer::start());
+    runtime.block_on(
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&collector),
+    );
+    let exporter = OtelExporter::OtlpHttp {
+        endpoint: format!("{}/v1/logs", collector.uri()),
+        headers: HashMap::new(),
+        protocol: OtelHttpProtocol::Json,
+        tls: None,
+    };
+    let otel = OtelProvider::try_new(&OtelSettings {
+        environment: "test".to_string(),
+        service_name: "codex-exec-server".to_string(),
+        service_version: env!("CARGO_PKG_VERSION").to_string(),
+        codex_home: std::env::current_dir().expect("cwd"),
+        exporter,
+        trace_exporter: if export_traces {
+            OtelExporter::OtlpHttp {
+                endpoint: format!("{}/v1/traces", collector.uri()),
+                headers: HashMap::new(),
+                protocol: OtelHttpProtocol::Json,
+                tls: None,
+            }
+        } else {
+            OtelExporter::None
+        },
+        metrics_exporter: OtelExporter::None,
+        runtime_metrics: false,
+        span_attributes: BTreeMap::new(),
+        tracestate: BTreeMap::new(),
+    })
+    .expect("OTEL settings")
+    .expect("OTEL provider");
+    let subscriber = tracing_subscriber::registry()
+        .with(otel.tracing_layer())
+        .with(otel.logger_layer());
+    tracing::subscriber::with_default(subscriber, || {
+        tracing::callsite::rebuild_interest_cache();
+        run(&runtime, &otel, &collector);
+    });
+    runtime
+        .block_on(otel.shutdown_with_timeout(Duration::from_secs(5)))
+        .expect("flush OTEL");
+
+    let mut records = Vec::new();
+    for request in runtime
+        .block_on(collector.received_requests())
+        .expect("exported requests")
+    {
+        let body: Value = serde_json::from_slice(&request.body).expect("OTLP JSON");
+        let Some(resources) = body["resourceLogs"].as_array() else {
+            continue;
+        };
+        assert!(!String::from_utf8_lossy(&request.body).contains(PRIVATE_PAYLOAD));
+        for resource in resources {
+            for scope in resource["scopeLogs"].as_array().expect("scope logs") {
+                for record in scope["logRecords"].as_array().expect("log records") {
+                    assert!(record["body"].is_null(), "lifecycle logs have no raw body");
+                    records.push(record.clone());
+                }
+            }
+        }
+    }
+    records
+}
+
+struct TestServer {
+    sessions: Arc<SessionRegistry>,
+    router: RpcRouter<ExecServerHandler>,
+    outgoing: mpsc::Sender<RpcServerOutboundMessage>,
+    _notifications: mpsc::Receiver<RpcServerOutboundMessage>,
+}
+
+impl TestServer {
+    fn new() -> Self {
+        let (outgoing, notifications) = mpsc::channel(/*buffer*/ 128);
+        Self {
+            sessions: SessionRegistry::new(ExecServerTelemetry::default()),
+            router: build_router(),
+            outgoing,
+            _notifications: notifications,
+        }
+    }
+
+    async fn new_orchestrator(
+        &self,
+        client_name: &str,
+        registration: Option<&str>,
+    ) -> (Arc<ExecServerHandler>, String) {
+        let handler = new_handler(&self.sessions, self.outgoing.clone(), registration);
+        let initialized = handler
+            .initialize(InitializeParams {
+                client_name: client_name.to_string(),
+                resume_session_id: None,
+            })
+            .await
+            .expect("initialize orchestrator");
+        handler.initialized().expect("initialized orchestrator");
+        (handler, initialized.session_id)
+    }
+}
+
+fn process_start_params(process_id: &str, argv: Value, cwd: PathUri) -> Value {
+    json!({
+        "processId": process_id,
+        "argv": argv,
+        "cwd": cwd,
+        "env": {"PRIVATE_TEST_VALUE": PRIVATE_PAYLOAD},
+        "tty": false,
+        "arg0": null,
+    })
+}
+
+fn process_metadata(index: usize) -> Value {
+    json!({"threadId": THREADS[index], "toolCallId": CALLS[index]})
+}
+
+fn new_handler(
+    sessions: &Arc<SessionRegistry>,
+    outgoing: mpsc::Sender<RpcServerOutboundMessage>,
+    registration: Option<&str>,
+) -> Arc<ExecServerHandler> {
+    let mut handler = ExecServerHandler::new(
+        Arc::clone(sessions),
+        RpcNotificationSender::new(outgoing),
+        ExecServerRuntimePaths::new(
+            std::env::current_exe().expect("test executable"),
+            /*codex_linux_sandbox_exe*/ None,
+        )
+        .expect("runtime paths"),
+        HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
+    );
+    handler.executor_registration = registration
+        .and_then(|registration| {
+            ExecutorRegistration::new("environment".to_string(), registration.to_string())
+        })
+        .map(Arc::new);
+    Arc::new(handler)
+}
@@ -24,6 +24,7 @@ use crate::server::request_dispatcher::RequestTaskResult;
 use crate::server::session_registry::SessionRegistry;
 use crate::telemetry::ConnectionTransport;
 use crate::telemetry::ExecServerTelemetry;
+use crate::telemetry::ExecutorRegistration;
 use codex_http_client::HttpClientFactory;

 #[derive(Clone)]
@@ -72,12 +73,23 @@ impl ConnectionProcessor {
    ) {
        run_connection(
            connection,
-            Arc::clone(&self.session_registry),
-            self.runtime_paths.clone(),
-            self.telemetry.clone(),
-            self.http_client_factory.clone(),
+            self.clone(),
            transport,
-            self.request_dispatch_mode,
+            /*executor_registration*/ None,
+        )
+        .await;
+    }
+
+    pub(crate) async fn run_registered_connection(
+        &self,
+        connection: JsonRpcConnection,
+        executor_registration: Option<Arc<ExecutorRegistration>>,
+    ) {
+        run_connection(
+            connection,
+            self.clone(),
+            ConnectionTransport::Relay,
+            executor_registration,
        )
        .await;
    }
@@ -89,13 +101,17 @@ impl ConnectionProcessor {

 async fn run_connection(
    connection: JsonRpcConnection,
-    session_registry: Arc<SessionRegistry>,
-    runtime_paths: ExecServerRuntimePaths,
-    telemetry: ExecServerTelemetry,
-    http_client_factory: HttpClientFactory,
+    processor: ConnectionProcessor,
    transport: ConnectionTransport,
-    request_dispatch_mode: RequestDispatchMode,
+    executor_registration: Option<Arc<ExecutorRegistration>>,
 ) {
+    let ConnectionProcessor {
+        session_registry,
+        runtime_paths,
+        telemetry,
+        http_client_factory,
+        request_dispatch_mode,
+    } = processor;
    let _connection_metrics = telemetry.connection_started(transport);
    let JsonRpcConnection {
        outgoing_tx: json_outgoing_tx,
@@ -108,12 +124,14 @@ async fn run_connection(
        mpsc::channel::<RpcServerOutboundMessage>(CHANNEL_CAPACITY);
    let notifications = RpcNotificationSender::new(outgoing_tx.clone());
    let requests = notifications.request_sender();
-    let handler = Arc::new(ExecServerHandler::new(
+    let mut handler = ExecServerHandler::new(
        session_registry,
        notifications,
        runtime_paths,
        http_client_factory,
-    ));
+    );
+    handler.executor_registration = executor_registration;
+    let handler = Arc::new(handler);

    let outbound_task = tokio::spawn(async move {
        while let Some(message) = outgoing_rx.recv().await {
@@ -297,7 +315,6 @@ mod tests {
    use crate::protocol::TerminateResponse;
    use crate::rpc::RpcServerOutboundMessage;
    use crate::rpc_server_requests::RpcServerRequestSender;
-    use crate::server::RequestDispatchMode;
    use crate::server::session_registry::SessionRegistry;

    #[tokio::test]
@@ -494,14 +511,12 @@ mod tests {
            JsonRpcConnection::from_stdio(server_reader, server_writer, label.to_string());
        let task = tokio::spawn(run_connection(
            connection,
-            registry,
-            test_runtime_paths(),
-            crate::ExecServerTelemetry::default(),
-            codex_http_client::HttpClientFactory::new(
-                codex_http_client::OutboundProxyPolicy::ReqwestDefault,
-            ),
+            super::ConnectionProcessor {
+                session_registry: registry,
+                ..super::ConnectionProcessor::new(test_runtime_paths())
+            },
            crate::telemetry::ConnectionTransport::Stdio,
-            RequestDispatchMode::Inline,
+            /*executor_registration*/ None,
        ));
        (client_writer, BufReader::new(client_reader).lines(), task)
    }
@@ -574,6 +589,7 @@ mod tests {
            env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
        }
        ExecParams {
+            metadata: Default::default(),
            process_id,
            argv: sleep_then_print_argv(),
            cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd"))
@@ -1,3 +1,4 @@
+use opentelemetry::trace::TraceContextExt;
 use std::sync::Arc;

 use crate::protocol::CAPABILITY_ROOTS_DISCOVER_METHOD;
@@ -68,9 +69,15 @@ pub(crate) fn build_router() -> RpcRouter<ExecServerHandler> {
            handler.http_request(request_id, params).await
        },
    );
-    router.request(
+    router.request_with_trace(
        EXEC_METHOD,
-        |handler: Arc<ExecServerHandler>, params: ExecParams| async move { handler.exec(params).await },
+        |handler: Arc<ExecServerHandler>, params: ExecParams, trace| async move {
+            let launch_context = trace
+                .as_ref()
+                .and_then(codex_otel::context_from_w3c_trace_context)
+                .map(|context| context.span().span_context().clone());
+            handler.exec(params, launch_context).await
+        },
    );
    router.request(
        ENVIRONMENT_INFO_METHOD,