Add trace-safe diagnostics for Noise relay request progress (#58025)

## Why WebSocket heartbeat progress only establishes relay reachability, not executor receipt or completion. A blocked write or full application queue can also leave a Pong unread, so diagnosing stalled requests requires separate transport and application progress signals. ## What changed - Trace request dequeue and completed writes under the request's trace context, and emit an executor receipt span immediately after decoding, before queueing or execution completes. - Record RPC enqueue wait time and queue occupancy. - Add bounded heartbeat events and a transport closure summary with fixed exit categories, queue depths, message counters, and read/write timing. Distinguish write timeouts from missing Pongs and report application delivery backpressure separately. - Keep request methods, IDs, payloads, raw errors, and peer-provided text out of the new trace-safe transport diagnostics. ## Testing Add coverage for blocked data and keepalive writes, missing Pongs, saturated application queues, read errors, and responsive relays without application responses. Verify trace correlation, immediate receipt span export, bounded Pong telemetry, and exclusion of payloads and raw error text. GitOrigin-RevId: b58b2f747c4050dcde8efb408c35ecadd3561b5f
2026-10-29 19:08:43 +00:00 · 2026-10-29 08:51:18 +00:00
parent a29c7b97e8
commit a3697361df
6 changed files with 735 additions and 46 deletions
@@ -50,6 +50,11 @@ use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT;
 use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT_REASON;
 use crate::websocket_pong_watchdog::WebSocketPongWatchdog;

+#[path = "harness_diagnostics.rs"]
+mod diagnostics;
+use diagnostics::TransportExitReason;
+use diagnostics::WebsocketWriteError;
+
 /// Values that bind one harness websocket to the intended executor registration.
 ///
 /// These fields all come from the same registry response. Keeping them together
@@ -111,7 +116,7 @@ where
    let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
    let (disconnected_tx, disconnected_rx) = watch::channel(false);
    let (handshake_ready_tx, handshake_ready_rx) = oneshot::channel();
-    let stream_span = tracing::debug_span!("noise_relay.stream", noise_side = "harness",);
+    let stream_span = tracing::debug_span!("noise_relay.stream", noise_side = "harness", stream_id);
    debug!(
        environment_id,
        executor_registration_id, stream_id, "Noise harness relay details"
@@ -261,6 +266,7 @@ where
                    match initiator_handshake.finish(&response) {
                        Ok(transport) => {
                            info!(
+                                target: "codex_otel.trace_safe",
                                noise_event = "handshake",
                                noise_outcome = "ok",
                                noise_side = "harness",
@@ -333,12 +339,26 @@ where
        let mut pending_outbound = None;
        let mut force_incoming = false;
        let mut frames_drained_after_pong_deadline = 0usize;
-        'relay: loop {
+        let connected_at = Instant::now();
+        let mut last_read_at = connected_at;
+        let mut last_write_at = connected_at;
+        let mut write_started_at = None;
+        let mut write_kind = "none";
+        let mut records_written = 0_u64;
+        let mut frames_read = 0_u64;
+        let mut read_reason = "none";
+        let mut messages_dequeued = 0_u64;
+        let mut messages_written = 0_u64;
+        let mut pending_request: Option<(tracing::Span, Instant)> = None;
+        let mut heartbeat = diagnostics::HeartbeatDiagnostics::default();
+        let exit_reason = 'relay: loop {
            // Consume a due tick before the always-ready record arm below can win
            // another select iteration and postpone the keepalive.
            if pong_watchdog.deadline().is_none()
                && keepalive.tick().now_or_never().is_some()
            {
+                write_started_at = Some(Instant::now());
+                write_kind = "keepalive";
                if let Err(error) = send_keepalive_ping(
                    &mut websocket,
                    &mut pong_watchdog,
@@ -347,8 +367,11 @@ where
                .await
                {
                    warn!("failed to write Noise relay keepalive ping: {error}");
-                    break;
+                    break error.exit_reason();
                }
+                last_write_at = Instant::now();
+                write_started_at = None;
+                heartbeat.record(diagnostics::HeartbeatProgress::PingWritten, &stream_id);
                frames_drained_after_pong_deadline = 0;
                continue;
            }
@@ -368,16 +391,10 @@ where
                {
                    force_incoming = true;
                } else {
-                    warn!(
-                        noise_reason = WEBSOCKET_PONG_TIMEOUT_REASON,
-                        "Noise harness rendezvous websocket disconnected"
-                    );
-                    send_disconnected(
-                        &incoming_tx,
-                        &disconnected_tx,
-                        WEBSOCKET_PONG_TIMEOUT_REASON.to_string(),
-                    );
-                    return;
+                    let _ = incoming_tx.try_send(JsonRpcConnectionEvent::Disconnected {
+                        reason: Some(WEBSOCKET_PONG_TIMEOUT_REASON.to_string()),
+                    });
+                    break TransportExitReason::PongTimeout;
                }
            }

@@ -397,13 +414,36 @@ where
            tokio::select! {
                maybe_message = outgoing_rx.recv(), if pending_outbound.is_none() && !force_incoming && !pong_deadline_expired => {
                    let Some(message) = maybe_message else {
-                        break;
+                        break TransportExitReason::OutgoingClosed;
                    };
+                    messages_dequeued += 1;
+                    if let codex_exec_server_protocol::JSONRPCMessage::Request(request) = &message {
+                        // Link transport progress to the existing RPC trace without copying
+                        // its method, ID, payload or unvalidated trace carrier into fields.
+                        let span = tracing::info_span!("codex.exec_server.noise_request_write", stream_id);
+                        if let Some(trace) = &request.trace {
+                            codex_otel::set_parent_from_w3c_trace_context(&span, trace);
+                        }
+                        // Live JSON correlation reads OtelData before this span closes.
+                        // Setting a remote parent alone leaves its trace ID lazy.
+                        let _ = codex_otel::span_w3c_trace_context(&span);
+                        tracing::event!(
+                            name: "codex.exec_server.request_dequeued",
+                            target: "codex_otel.trace_safe",
+                            parent: &span,
+                            tracing::Level::INFO,
+                            event.name = "codex.exec_server.request_dequeued",
+                            stream_id,
+                            message_sequence = messages_dequeued,
+                            outgoing_queue_depth = outgoing_rx.len(),
+                        );
+                        pending_request = Some((span, Instant::now()));
+                    }
                    let framed = match frame_jsonrpc_message(&message) {
                        Ok(framed) => framed,
                        Err(error) => {
                            warn!("failed to frame JSON-RPC payload for Noise relay: {error}");
-                            break;
+                            break TransportExitReason::Framing;
                        }
                    };
                    let request_trace = match message {
@@ -419,7 +459,7 @@ where
                        Ok(seq) => seq,
                        Err(error) => {
                            warn!("Noise relay sequence exhausted: {error}");
-                            break 'relay;
+                            break 'relay TransportExitReason::SequenceExhausted;
                        }
                    };
                    let (ciphertext, next_offset, message_complete, request_trace) = {
@@ -431,7 +471,7 @@ where
                            Ok(ciphertext) => ciphertext,
                            Err(error) => {
                                warn!("failed to encrypt JSON-RPC payload for Noise relay: {error}");
-                                break 'relay;
+                                break 'relay TransportExitReason::Encryption;
                            }
                        };
                        (
@@ -447,6 +487,8 @@ where
                    // combined sink and stream. A single bounded record can therefore hit the
                    // deadline and disconnect with that Pong queued. Treat that as write
                    // backpressure; this loop yields only between records.
+                    write_started_at = Some(Instant::now());
+                    write_kind = "data";
                    if let Err(error) = send_websocket_message(
                        &mut websocket,
                        Message::Binary(encode_relay_message_frame(&frame).into()),
@@ -455,9 +497,25 @@ where
                    .await
                    {
                        warn!("failed to write Noise relay websocket: {error}");
-                        break 'relay;
+                        break 'relay error.exit_reason();
                    }
+                    last_write_at = Instant::now();
+                    write_started_at = None;
+                    records_written += 1;
                    if message_complete {
+                        messages_written += 1;
+                        if let Some((span, started_at)) = pending_request.take() {
+                            tracing::event!(
+                                name: "codex.exec_server.request_written",
+                                target: "codex_otel.trace_safe",
+                                parent: &span,
+                                tracing::Level::INFO,
+                                event.name = "codex.exec_server.request_written",
+                                stream_id,
+                                message_sequence = messages_dequeued,
+                                write_elapsed_ms = started_at.elapsed().as_millis() as u64,
+                            );
+                        }
                        pending_outbound = None;
                    } else if let Some((_framed, offset, _request_trace)) = pending_outbound.as_mut() {
                        *offset = next_offset;
@@ -467,6 +525,8 @@ where
                    continue;
                }
                _ = keepalive.tick(), if pong_watchdog.deadline().is_none() => {
+                    write_started_at = Some(Instant::now());
+                    write_kind = "keepalive";
                    if let Err(error) = send_keepalive_ping(
                        &mut websocket,
                        &mut pong_watchdog,
@@ -475,15 +535,22 @@ where
                    .await
                    {
                        warn!("failed to write Noise relay keepalive ping: {error}");
-                        break;
+                        break error.exit_reason();
                    }
+                    last_write_at = Instant::now();
+                    write_started_at = None;
+                    heartbeat.record(diagnostics::HeartbeatProgress::PingWritten, &stream_id);
                    frames_drained_after_pong_deadline = 0;
                }
                incoming_message = websocket.next() => {
                    force_incoming = false;
                    let Some(incoming_message) = incoming_message else {
-                        break;
+                        break TransportExitReason::WebsocketEnded;
                    };
+                    if incoming_message.is_ok() {
+                        last_read_at = Instant::now();
+                        frames_read += 1;
+                    }
                    // Count each completed read after expiry. If only the deadline arm
                    // advanced this counter, reads won by a simultaneously ready incoming
                    // arm would not count toward the 32-frame cap.
@@ -499,7 +566,7 @@ where
                                Ok(frame) => frame,
                                Err(error) => {
                                    send_malformed(&incoming_tx, error.to_string());
-                                    break;
+                                    break TransportExitReason::InvalidFrame;
                                }
                            };
                            if frame.stream_id != stream_id {
@@ -511,7 +578,7 @@ where
                                        Ok(data) => data,
                                        Err(error) => {
                                            send_malformed(&incoming_tx, error.to_string());
-                                            break;
+                                            break TransportExitReason::InvalidFrame;
                                        }
                                    };
                                    if let Err(error) = receive_data(
@@ -525,10 +592,10 @@ where
                                    .await
                                    {
                                        if matches!(error, ExecServerError::Closed) {
-                                            break;
+                                            break TransportExitReason::ApplicationDelivery;
                                        }
                                        send_malformed(&incoming_tx, error.to_string());
-                                        break;
+                                        break TransportExitReason::InvalidData;
                                    }
                                }
                                Ok(RelayFrameBodyKind::Reset) => {
@@ -539,7 +606,7 @@ where
                                            ),
                                        },
                                    );
-                                    break;
+                                    break TransportExitReason::RelayReset;
                                }
                                Ok(
                                    RelayFrameBodyKind::Ack
@@ -551,12 +618,13 @@ where
                                        &incoming_tx,
                                        "Noise relay received invalid post-handshake frame".to_string(),
                                    );
-                                    break;
+                                    break TransportExitReason::InvalidFrame;
                                }
                            }
                        }
-                        Ok(Message::Close(_)) => break,
+                        Ok(Message::Close(_)) => break TransportExitReason::WebsocketClose,
                        Ok(Message::Pong(_)) => {
+                            heartbeat.record(diagnostics::HeartbeatProgress::PongReceived, &stream_id);
                            pong_watchdog.received_pong();
                            frames_drained_after_pong_deadline = 0;
                        }
@@ -566,16 +634,46 @@ where
                                &incoming_tx,
                                "Noise relay transport expects binary protobuf frames".to_string(),
                            );
-                            break;
+                            break TransportExitReason::UnexpectedText;
                        }
                        Err(error) => {
                            debug!("Noise relay websocket read failed: {error}");
-                            break;
+                            read_reason = websocket_read_error_reason(&error);
+                            break TransportExitReason::WebsocketRead;
                        }
                    }
                }
            }
-        }
+        };
+        // Fixed reasons and local counters only. Raw errors, frames and peer-provided
+        // close/reset text must remain outside the trace-safe target.
+        tracing::event!(
+            name: "codex.exec_server.noise_transport_closed",
+            target: "codex_otel.trace_safe",
+            tracing::Level::INFO,
+            event.name = "codex.exec_server.noise_transport_closed",
+            noise_side = "harness",
+            noise_reason = exit_reason.as_str(),
+            read_reason,
+            stream_id,
+            messages_dequeued,
+            messages_written,
+            pings_written = heartbeat.pings_written,
+            pongs_received = heartbeat.pongs_received,
+            connection_age_ms = connected_at.elapsed().as_millis() as u64,
+            last_read_age_ms = last_read_at.elapsed().as_millis() as u64,
+            last_write_age_ms = last_write_at.elapsed().as_millis() as u64,
+            write_pending = write_started_at.is_some(),
+            write_kind = if write_started_at.is_some() { write_kind } else { "none" },
+            write_wait_ms = write_started_at.map(|at| at.elapsed().as_millis() as u64).unwrap_or(0),
+            pong_pending = pong_watchdog.deadline().is_some(),
+            pong_deadline_overdue_ms = pong_watchdog.deadline()
+                .map(|at| Instant::now().saturating_duration_since(at).as_millis() as u64).unwrap_or(0),
+            frames_read,
+            records_written,
+            outgoing_queue_depth = outgoing_rx.len(),
+            incoming_queue_depth = incoming_tx.max_capacity() - incoming_tx.capacity(),
+        );
        let _ = disconnected_tx.send(true);
    }
    .instrument(stream_span));
@@ -618,15 +716,15 @@ async fn send_websocket_message<T, E>(
    websocket: &mut T,
    message: Message,
    deadline: tokio::time::Instant,
-) -> Result<(), String>
+) -> Result<(), WebsocketWriteError>
 where
    T: Sink<Message, Error = E> + Unpin,
    E: std::fmt::Display,
 {
    match tokio::time::timeout_at(deadline, websocket.send(message)).await {
        Ok(Ok(())) => Ok(()),
-        Ok(Err(error)) => Err(error.to_string()),
-        Err(_) => Err("websocket write timed out".to_string()),
+        Ok(Err(error)) => Err(WebsocketWriteError::Transport(error.to_string())),
+        Err(_) => Err(WebsocketWriteError::Timeout),
    }
 }

@@ -634,7 +732,7 @@ async fn send_keepalive_ping<T, E>(
    websocket: &mut T,
    pong_watchdog: &mut WebSocketPongWatchdog,
    pong_deadline: std::pin::Pin<&mut tokio::time::Sleep>,
-) -> Result<(), String>
+) -> Result<(), WebsocketWriteError>
 where
    T: Sink<Message, Error = E> + Unpin,
    E: std::fmt::Display,
@@ -692,13 +790,21 @@ async fn send_incoming_event(
    event: JsonRpcConnectionEvent,
    deadline: tokio::time::Instant,
 ) -> Result<(), ExecServerError> {
+    let delivery_started = Instant::now();
    match tokio::time::timeout_at(deadline, incoming_tx.send(event)).await {
        Ok(Ok(())) => Ok(()),
        Ok(Err(_)) => Err(ExecServerError::Closed),
        Err(_) => {
-            warn!(
+            tracing::event!(
+                name: "codex.exec_server.noise_application_backpressure",
+                target: "codex_otel.trace_safe",
+                tracing::Level::WARN,
+                event.name = "codex.exec_server.noise_application_backpressure",
+                noise_side = "harness",
                noise_reason = "application_backpressure",
-                "Noise harness application event delivery timed out"
+                delivery_wait_ms = delivery_started.elapsed().as_millis() as u64,
+                incoming_queue_depth = incoming_tx.max_capacity() - incoming_tx.capacity(),
+                incoming_queue_capacity = incoming_tx.max_capacity(),
            );
            Err(ExecServerError::Closed)
        }
@@ -0,0 +1,109 @@
+//! Fixed transport exit categories and bounded heartbeat progress telemetry.
+//!
+//! Keep wire values and error text out of these categories: CCA retains this
+//! telemetry even when ordinary exec-server diagnostics are filtered out.
+
+pub(super) enum TransportExitReason {
+    PongTimeout,
+    OutgoingClosed,
+    Framing,
+    SequenceExhausted,
+    Encryption,
+    WebsocketWriteTimeout,
+    WebsocketWrite,
+    WebsocketEnded,
+    InvalidFrame,
+    ApplicationDelivery,
+    InvalidData,
+    RelayReset,
+    WebsocketClose,
+    UnexpectedText,
+    WebsocketRead,
+}
+
+impl TransportExitReason {
+    pub(super) fn as_str(&self) -> &'static str {
+        match self {
+            Self::PongTimeout => "pong_timeout",
+            Self::OutgoingClosed => "outgoing_closed",
+            Self::Framing => "framing",
+            Self::SequenceExhausted => "sequence_exhausted",
+            Self::Encryption => "encryption",
+            Self::WebsocketWriteTimeout => "websocket_write_timeout",
+            Self::WebsocketWrite => "websocket_write",
+            Self::WebsocketEnded => "websocket_ended",
+            Self::InvalidFrame => "invalid_frame",
+            Self::ApplicationDelivery => "application_delivery",
+            Self::InvalidData => "invalid_data",
+            Self::RelayReset => "relay_reset",
+            Self::WebsocketClose => "websocket_close",
+            Self::UnexpectedText => "unexpected_text",
+            Self::WebsocketRead => "websocket_read",
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(super) enum WebsocketWriteError {
+    #[error("websocket write timed out")]
+    Timeout,
+    #[error("{0}")]
+    Transport(String),
+}
+
+impl WebsocketWriteError {
+    pub(super) fn exit_reason(&self) -> TransportExitReason {
+        match self {
+            Self::Timeout => TransportExitReason::WebsocketWriteTimeout,
+            Self::Transport(_) => TransportExitReason::WebsocketWrite,
+        }
+    }
+}
+
+/// WebSocket heartbeat progress only proves reachability of the relay peer, not
+/// delivery to the executor behind it. Emit at most once per local Ping or Pong.
+pub(super) enum HeartbeatProgress {
+    PingWritten,
+    PongReceived,
+}
+
+#[derive(Default)]
+pub(super) struct HeartbeatDiagnostics {
+    pub(super) pings_written: u64,
+    pub(super) pongs_received: u64,
+    last_ping_at: Option<tokio::time::Instant>,
+}
+
+impl HeartbeatDiagnostics {
+    pub(super) fn record(&mut self, progress: HeartbeatProgress, stream_id: &str) {
+        let phase = match progress {
+            HeartbeatProgress::PingWritten => {
+                self.pings_written += 1;
+                self.last_ping_at = Some(tokio::time::Instant::now());
+                "ping_written"
+            }
+            HeartbeatProgress::PongReceived => {
+                // Unsolicited Pongs must not create unbounded telemetry volume.
+                if self.last_ping_at.is_none() {
+                    return;
+                }
+                self.pongs_received += 1;
+                "pong_received"
+            }
+        };
+        tracing::event!(
+            name: "codex.exec_server.noise_heartbeat",
+            target: "codex_otel.trace_safe",
+            tracing::Level::INFO,
+            event.name = "codex.exec_server.noise_heartbeat",
+            stream_id,
+            phase,
+            pings_written = self.pings_written,
+            pongs_received = self.pongs_received,
+            ping_elapsed_ms = self.last_ping_at.map(|at| at.elapsed().as_millis() as u64).unwrap_or(0),
+        );
+        if matches!(progress, HeartbeatProgress::PongReceived) {
+            self.last_ping_at = None;
+        }
+    }
+}
@@ -66,6 +66,28 @@ fn websocket_read_diagnostics_use_typed_errors_only() {

 #[tokio::test(start_paused = true)]
 async fn first_encrypted_request_frame_exposes_only_its_trace_context() -> Result<()> {
+    use opentelemetry::trace::TracerProvider as _;
+    use opentelemetry_sdk::trace::InMemorySpanExporter;
+    use opentelemetry_sdk::trace::SdkTracerProvider;
+    use tracing_subscriber::prelude::*;
+    let exporter = InMemorySpanExporter::default();
+    let provider = SdkTracerProvider::builder()
+        .with_simple_exporter(exporter.clone())
+        .build();
+    let events = SafeEvents::default();
+    let _subscriber = tracing::subscriber::set_default(
+        tracing_subscriber::registry()
+            .with(
+                events
+                    .clone()
+                    .with_filter(tracing_subscriber::filter::LevelFilter::INFO),
+            )
+            .with(
+                tracing_opentelemetry::layer()
+                    .with_tracer(provider.tracer("harness-test"))
+                    .with_filter(tracing_subscriber::filter::LevelFilter::INFO),
+            ),
+    );
    let (connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
    let traceparent = "00-00000000000000000000000000000001-0000000000000002-01";
    let tracestate = "dd=s:1";
@@ -86,12 +108,34 @@ async fn first_encrypted_request_frame_exposes_only_its_trace_context() -> Resul
        .await?;

    control.wait_for_blocked_write(/*expected*/ 1).await?;
+    {
+        let captured = events.0.lock().unwrap();
+        let dequeue = captured
+            .iter()
+            .find(|event| {
+                event
+                    .get("event.name")
+                    .is_some_and(|name| name.contains("request_dequeued"))
+            })
+            .context("live dequeue")?;
+        assert_eq!(
+            dequeue.get("trace_id").map(String::as_str),
+            Some("00000000000000000000000000000001")
+        );
+        assert!(
+            exporter
+                .get_finished_spans()?
+                .iter()
+                .all(|span| span.name != "codex.exec_server.noise_request_write")
+        );
+    }
    control.grant_writes(/*count*/ 1);
    let first = read_outbound_frame(&mut outbound_rx).await?;
    assert_eq!(
        (first.traceparent.as_deref(), first.tracestate.as_deref()),
        (Some(traceparent), Some(tracestate))
    );
+    let stream_id = first.stream_id.clone();
    let first_payload = first.into_data()?.payload;
    assert!(
        !first_payload
@@ -103,6 +147,43 @@ async fn first_encrypted_request_frame_exposes_only_its_trace_context() -> Resul
    control.grant_writes(/*count*/ 1);
    let second = read_outbound_frame(&mut outbound_rx).await?;
    assert_eq!((second.traceparent, second.tracestate), (None, None));
+    assert!(!format!("{:?}", events.0.lock().unwrap()).contains("request_written"));
+    control.wait_for_blocked_write(/*expected*/ 3).await?;
+    control.grant_writes(/*count*/ 1);
+    read_outbound_frame(&mut outbound_rx).await?;
+    let spans = exporter.get_finished_spans()?;
+    let write_span = spans
+        .iter()
+        .find(|span| span.name == "codex.exec_server.noise_request_write")
+        .context("expected finished request write span")?;
+    assert_eq!(
+        write_span.span_context.trace_id().to_string(),
+        "00000000000000000000000000000001"
+    );
+    assert_eq!(write_span.parent_span_id.to_string(), "0000000000000002");
+    let captured = events.0.lock().unwrap();
+    let progress: Vec<_> = captured
+        .iter()
+        .filter(|e| {
+            e.get("event.name").is_some_and(|name| {
+                name.contains("request_dequeued") || name.contains("request_written")
+            })
+        })
+        .collect();
+    assert_eq!(progress.len(), 2);
+    for event in &progress {
+        assert_eq!(
+            event.get("trace_id").map(String::as_str),
+            Some("00000000000000000000000000000001")
+        );
+    }
+    assert_eq!(progress[0]["stream_id"], format!("{stream_id:?}"));
+    assert_eq!(progress[0]["stream_id"], progress[1]["stream_id"]);
+    assert_eq!(
+        progress[0]["message_sequence"],
+        progress[1]["message_sequence"]
+    );
+    drop(captured);

    for task in &connection.task_handles {
        task.abort();
@@ -294,8 +375,13 @@ async fn pong_keeps_harness_alive_until_peer_stops_responding() -> Result<()> {
    Ok(())
 }

-#[tokio::test]
+/// A full consumer queue times out with occupancy and wait duration, without logging its data.
+#[tokio::test(start_paused = true)]
 async fn application_event_delivery_is_bounded() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
    let (incoming_tx, _incoming_rx) = mpsc::channel(1);
    incoming_tx
        .send(JsonRpcConnectionEvent::MalformedMessage {
@@ -313,6 +399,14 @@ async fn application_event_delivery_is_bounded() -> Result<()> {
    .await;

    assert!(matches!(result, Err(ExecServerError::Closed)));
+    let events = events.0.lock().unwrap();
+    assert_eq!(events.len(), 1);
+    let event = &events[0];
+    assert_eq!(event["noise_reason"], "\"application_backpressure\"");
+    assert_eq!(event["incoming_queue_depth"], "1");
+    assert_eq!(event["incoming_queue_capacity"], "1");
+    assert!(event["delivery_wait_ms"].parse::<u64>()? >= 10);
+    assert!(!format!("{events:?}").contains("blocked event"));
    Ok(())
 }

@@ -343,6 +437,18 @@ async fn connected_controlled_harness() -> Result<(
    JsonRpcConnection,
    ControlledWebSocketHandle,
    futures_mpsc::UnboundedReceiver<Message>,
+)> {
+    let (connection, control, outbound_rx, _transport, _stream_id) =
+        connected_controlled_harness_with_transport().await?;
+    Ok((connection, control, outbound_rx))
+}
+
+async fn connected_controlled_harness_with_transport() -> Result<(
+    JsonRpcConnection,
+    ControlledWebSocketHandle,
+    futures_mpsc::UnboundedReceiver<Message>,
+    NoiseTransport,
+    String,
 )> {
    let (websocket, control, mut outbound_rx) = ControlledWebSocket::new(/*write_permits*/ 2);
    let executor_identity = NoiseChannelIdentity::generate()?;
@@ -381,11 +487,12 @@ async fn connected_controlled_harness() -> Result<(
        &prologue,
        &handshake.into_handshake_payload()?,
    )?;
-    let (_transport, response) = pending.complete()?;
+    let (transport, response) = pending.complete()?;
    control.send_inbound(Message::Binary(
-        encode_relay_message_frame(&RelayMessageFrame::handshake(stream_id, response)).into(),
+        encode_relay_message_frame(&RelayMessageFrame::handshake(stream_id.clone(), response))
+            .into(),
    ))?;
-    Ok((connection, control, outbound_rx))
+    Ok((connection, control, outbound_rx, transport, stream_id))
 }

 struct ControlledWebSocket {
@@ -528,3 +635,324 @@ impl futures::Stream for ControlledWebSocket {
        result
    }
 }
+
+#[derive(Clone, Default)]
+struct SafeEvents(Arc<std::sync::Mutex<Vec<std::collections::BTreeMap<String, String>>>>);
+
+impl<S> tracing_subscriber::Layer<S> for SafeEvents
+where
+    S: tracing::Subscriber + for<'lookup> tracing_subscriber::registry::LookupSpan<'lookup>,
+{
+    fn on_event(
+        &self,
+        event: &tracing::Event<'_>,
+        context: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        if event.metadata().target() != "codex_otel.trace_safe" {
+            return;
+        }
+        #[derive(Default)]
+        struct Fields(std::collections::BTreeMap<String, String>);
+        impl tracing::field::Visit for Fields {
+            fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+                self.0.insert(field.name().to_owned(), format!("{value:?}"));
+            }
+        }
+        let mut fields = Fields::default();
+        event.record(&mut fields);
+        // Match CCA's live JSON lookup: inspecting OtelData must not materialize
+        // the context and accidentally hide a missing producer initialization.
+        if let Some(trace_id) = context.event_scope(event).and_then(|mut scope| {
+            scope.find_map(|span| {
+                span.extensions()
+                    .get::<tracing_opentelemetry::OtelData>()?
+                    .trace_id()
+            })
+        }) {
+            fields.0.insert("trace_id".to_owned(), trace_id.to_string());
+        }
+        self.0.lock().unwrap().push(fields.0);
+    }
+}
+
+/// A blocked write can time out with a Pong queued; its diagnostic must distinguish this
+/// local write stall from the read loop's missing-Pong watchdog.
+#[tokio::test(start_paused = true)]
+async fn blocked_write_reports_timeout_with_pong_still_queued() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+    let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
+    control.wait_for_blocked_write(/*expected*/ 1).await?;
+    control.grant_writes(/*count*/ 1);
+    let Message::Ping(payload) = outbound_rx.next().await.context("expected Ping")? else {
+        anyhow::bail!("expected Ping before blocked write");
+    };
+    connection
+        .outgoing_tx
+        .send(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(1),
+            method: "SYNTHETIC_PRIVATE_METHOD".to_owned(),
+            params: Some(serde_json::json!({"secret": "SYNTHETIC_PRIVATE_PAYLOAD"})),
+            trace: None,
+        }))
+        .await?;
+    control.wait_for_blocked_write(/*expected*/ 2).await?;
+    connection
+        .outgoing_tx
+        .send(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(2),
+            method: "SYNTHETIC_PRIVATE_QUEUED".to_owned(),
+            params: None,
+            trace: None,
+        }))
+        .await?;
+    let reads = control.inbound_reads();
+    control.send_inbound(Message::Pong(payload))?;
+    tokio::time::advance(WEBSOCKET_PONG_TIMEOUT + Duration::from_millis(1)).await;
+    for task in connection.task_handles.drain(..) {
+        task.await?;
+    }
+    assert_eq!(control.inbound_reads(), reads);
+    assert!(*connection.disconnected_rx.borrow());
+    let events = events.0.lock().unwrap();
+    let terminal: Vec<_> = events
+        .iter()
+        .filter(|e| {
+            e.get("event.name")
+                .is_some_and(|name| name == "\"codex.exec_server.noise_transport_closed\"")
+        })
+        .collect();
+    assert_eq!(terminal.len(), 1);
+    let event = terminal[0];
+    assert_eq!(event["noise_reason"], "\"websocket_write_timeout\"");
+    assert_eq!(event["write_pending"], "true");
+    assert_eq!(event["write_kind"], "\"data\"");
+    assert_eq!(event["pong_pending"], "true");
+    assert_eq!(event["records_written"], "0");
+    assert_eq!(event["messages_dequeued"], "1");
+    assert_eq!(event["messages_written"], "0");
+    assert_eq!(event["outgoing_queue_depth"], "1");
+    assert_eq!(event["pings_written"], "1");
+    assert_eq!(event["pongs_received"], "0");
+    assert!(!format!("{events:?}").contains("request_written"));
+    assert!(event["write_wait_ms"].parse::<u64>()? >= WEBSOCKET_PONG_TIMEOUT.as_millis() as u64);
+    assert!(!format!("{events:?}").contains("SYNTHETIC_PRIVATE"));
+    Ok(())
+}
+
+/// Peer-controlled read errors must produce a fixed terminal category without retaining
+/// the raw error, even though ordinary debug logs still carry it.
+#[tokio::test(start_paused = true)]
+async fn read_error_reports_safe_terminal_category() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+    let (mut connection, control, _outbound_rx) = connected_controlled_harness().await?;
+    control
+        .inbound_tx
+        .unbounded_send(Err(WebSocketError::Io(std::io::Error::other(
+            "SYNTHETIC_PRIVATE_ERROR",
+        ))))?;
+    for task in connection.task_handles.drain(..) {
+        task.await?;
+    }
+    let events = events.0.lock().unwrap();
+    let event = events.last().context("expected terminal event")?;
+    assert_eq!(
+        event["event.name"],
+        "\"codex.exec_server.noise_transport_closed\""
+    );
+    assert_eq!(event["noise_reason"], "\"websocket_read\"");
+    assert_eq!(event["read_reason"], "\"websocket_read_io_other\"");
+    assert!(!format!("{events:?}").contains("SYNTHETIC_PRIVATE_ERROR"));
+    Ok(())
+}
+
+/// A keepalive write stall has no outstanding Pong yet; retain that distinction
+/// when a disconnect aligns with the periodic keepalive schedule.
+#[tokio::test(start_paused = true)]
+async fn blocked_keepalive_reports_write_phase_without_pong_deadline() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+    let (mut connection, mut control, _outbound_rx) = connected_controlled_harness().await?;
+    control.wait_for_blocked_write(/*expected*/ 1).await?;
+    tokio::time::advance(WEBSOCKET_PONG_TIMEOUT + Duration::from_millis(1)).await;
+    for task in connection.task_handles.drain(..) {
+        task.await?;
+    }
+    let events = events.0.lock().unwrap();
+    let event = events.last().context("expected terminal event")?;
+    assert_eq!(event["noise_reason"], "\"websocket_write_timeout\"");
+    assert_eq!(event["write_kind"], "\"keepalive\"");
+    assert_eq!(event["pong_pending"], "false");
+    assert_eq!(event["write_pending"], "true");
+    Ok(())
+}
+
+/// Successful local writes with no inbound traffic reproduce a blackholed path,
+/// which must report a Pong timeout rather than a blocked write.
+#[tokio::test(start_paused = true)]
+async fn blackholed_peer_reports_pong_timeout_after_successful_writes() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+    let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
+    control.wait_for_blocked_write(/*expected*/ 1).await?;
+    control.grant_writes(/*count*/ 1);
+    assert!(matches!(outbound_rx.next().await, Some(Message::Ping(_))));
+    connection
+        .outgoing_tx
+        .send(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(1),
+            method: "test".to_owned(),
+            params: None,
+            trace: None,
+        }))
+        .await?;
+    control.wait_for_blocked_write(/*expected*/ 2).await?;
+    control.grant_writes(/*count*/ 1);
+    read_outbound_data(&mut outbound_rx).await?;
+    tokio::time::advance(WEBSOCKET_PONG_TIMEOUT / 2).await;
+    assert!(!*connection.disconnected_rx.borrow());
+    tokio::time::advance(WEBSOCKET_PONG_TIMEOUT / 2 + Duration::from_millis(1)).await;
+    for task in connection.task_handles.drain(..) {
+        task.await?;
+    }
+    let events = events.0.lock().unwrap();
+    let event = events.last().context("expected terminal event")?;
+    assert_eq!(event["noise_reason"], "\"pong_timeout\"");
+    assert_eq!(event["write_pending"], "false");
+    assert_eq!(event["records_written"], "1");
+    assert_eq!(event["messages_dequeued"], "1");
+    assert_eq!(event["messages_written"], "1");
+    assert_eq!(event["pings_written"], "1");
+    assert_eq!(event["pongs_received"], "0");
+    assert!(format!("{events:?}").contains("request_written"));
+    assert_eq!(event["pong_pending"], "true");
+    Ok(())
+}
+
+/// A stalled application consumer can prevent the shared loop from reading an
+/// already queued Pong; retain queue saturation separately from network timeout.
+#[tokio::test(start_paused = true)]
+async fn full_application_queue_reports_backpressure_with_pong_queued() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+    let (mut connection, mut control, mut outbound_rx, mut transport, stream_id) =
+        connected_controlled_harness_with_transport().await?;
+    control.wait_for_blocked_write(/*expected*/ 1).await?;
+    control.grant_writes(/*count*/ 1);
+    let Some(Message::Ping(payload)) = outbound_rx.next().await else {
+        anyhow::bail!("expected Ping");
+    };
+    let framed = frame_jsonrpc_message(&JSONRPCMessage::Response(
+        codex_exec_server_protocol::JSONRPCResponse {
+            id: RequestId::Integer(1),
+            result: serde_json::json!({"private": "SYNTHETIC_PRIVATE_RESPONSE"}),
+        },
+    ))?;
+    let reads = control.inbound_reads();
+    for seq in 0..=CHANNEL_CAPACITY {
+        let frame = RelayMessageFrame::data(
+            stream_id.clone(),
+            seq as u32,
+            transport.encrypt(&framed)?,
+            /*trace*/ None,
+        );
+        control.send_inbound(Message::Binary(encode_relay_message_frame(&frame).into()))?;
+    }
+    control.send_inbound(Message::Pong(payload))?;
+    // Awaiting the task lets the paused clock advance only once frame processing
+    // has filled the application queue and is waiting on its delivery deadline.
+    for task in connection.task_handles.drain(..) {
+        task.await?;
+    }
+    assert_eq!(control.inbound_reads() - reads, CHANNEL_CAPACITY + 1);
+    assert_eq!(connection.incoming_rx.len(), CHANNEL_CAPACITY);
+    let events = events.0.lock().unwrap();
+    let backpressure = events
+        .iter()
+        .find(|event| {
+            event
+                .get("noise_reason")
+                .is_some_and(|value| value == "\"application_backpressure\"")
+        })
+        .context("expected application backpressure")?;
+    assert_eq!(
+        backpressure["incoming_queue_depth"],
+        CHANNEL_CAPACITY.to_string()
+    );
+    let terminal = events.last().context("expected terminal event")?;
+    assert_eq!(terminal["noise_reason"], "\"application_delivery\"");
+    assert_eq!(terminal["pong_pending"], "true");
+    assert!(!format!("{events:?}").contains("SYNTHETIC_PRIVATE_RESPONSE"));
+    Ok(())
+}
+
+/// A responsive relay can coexist with a missing executor response. Ping/Pong
+/// progress must remain distinct from successful application delivery.
+#[tokio::test(start_paused = true)]
+async fn responsive_relay_without_application_response_reports_progress() -> Result<()> {
+    use tracing_subscriber::prelude::*;
+    let events = SafeEvents::default();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+    let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
+    connection
+        .outgoing_tx
+        .send(JSONRPCMessage::Request(JSONRPCRequest {
+            id: RequestId::Integer(1),
+            method: "SYNTHETIC_PRIVATE_METHOD".to_owned(),
+            params: None,
+            trace: None,
+        }))
+        .await?;
+    control.wait_for_blocked_write(/*expected*/ 1).await?;
+    control.grant_writes(/*count*/ 1);
+    read_outbound_data(&mut outbound_rx).await?;
+    for expected in 2..=3 {
+        control.wait_for_blocked_write(expected).await?;
+        control.grant_writes(/*count*/ 1);
+        let Some(Message::Ping(payload)) = outbound_rx.next().await else {
+            anyhow::bail!("expected Ping")
+        };
+        for _ in 0..3 {
+            control.send_inbound(Message::Pong(payload.clone()))?;
+        }
+    }
+    // A subsequent Close is ordered behind the last Pong and completes the summary.
+    control.send_inbound(Message::Close(None))?;
+    for task in connection.task_handles.drain(..) {
+        task.await?;
+    }
+    assert!(connection.incoming_rx.is_empty());
+    let captured = events.0.lock().unwrap();
+    let terminal = captured.last().context("expected terminal")?;
+    assert_eq!(terminal["noise_reason"], "\"websocket_close\"");
+    assert_eq!(terminal["messages_written"], "1");
+    assert_eq!(terminal["pings_written"], "2");
+    assert_eq!(terminal["pongs_received"], "2");
+    let heartbeats: Vec<_> = captured
+        .iter()
+        .filter_map(|event| event.get("phase"))
+        .collect();
+    assert_eq!(
+        heartbeats,
+        vec![
+            "\"ping_written\"",
+            "\"pong_received\"",
+            "\"ping_written\"",
+            "\"pong_received\""
+        ]
+    );
+    assert!(!format!("{captured:?}").contains("SYNTHETIC_PRIVATE"));
+    Ok(())
+}
@@ -49,9 +49,18 @@ impl NoiseStreamHandler for ConnectionProcessor {
    type Outgoing = JSONRPCMessage;

    fn decode(payload: Bytes) -> Result<Self::Incoming, ExecServerError> {
-        Ok(JsonRpcConnectionEvent::message(serde_json::from_slice(
-            &payload,
-        )?))
+        let event = JsonRpcConnectionEvent::message(serde_json::from_slice(&payload)?);
+        if let JsonRpcConnectionEvent::QueuedRequest { request_span, .. } = &event {
+            // The request span can remain open while queued or executing. Finish a
+            // payload-free receipt marker now so a stalled executor is distinguishable
+            // from a request that never arrived, without waiting for that span to close.
+            drop(tracing::info_span!(
+                target: "codex_otel.trace_safe",
+                parent: request_span,
+                "codex.exec_server.noise_request_received",
+            ));
+        }
+        Ok(event)
    }

    fn encode(message: Self::Outgoing) -> Result<NoiseOutboundMessage, ExecServerError> {
@@ -12,14 +12,28 @@ use crate::server::ConnectionProcessor;

 #[test]
 fn local_decode_starts_request_span_before_queueing() -> Result<()> {
-    let _subscriber = tracing::subscriber::set_default(tracing_subscriber::registry());
+    use opentelemetry::trace::TracerProvider as _;
+    use opentelemetry_sdk::trace::InMemorySpanExporter;
+    use opentelemetry_sdk::trace::SdkTracerProvider;
+    use tracing_subscriber::prelude::*;
+    let exporter = InMemorySpanExporter::default();
+    let provider = SdkTracerProvider::builder()
+        .with_simple_exporter(exporter.clone())
+        .build();
+    let _subscriber =
+        tracing::subscriber::set_default(tracing_subscriber::registry().with(
+            tracing_opentelemetry::layer().with_tracer(provider.tracer("noise-receipt-test")),
+        ));
    tracing::callsite::rebuild_interest_cache();

    let request = JSONRPCRequest {
        id: RequestId::Integer(1),
        method: "test/queued".to_string(),
-        params: None,
-        trace: None,
+        params: Some(serde_json::json!({"secret": "SYNTHETIC_PRIVATE_PAYLOAD"})),
+        trace: Some(codex_protocol::protocol::W3cTraceContext {
+            traceparent: Some("00-00000000000000000000000000000001-0000000000000002-01".to_owned()),
+            tracestate: None,
+        }),
    };
    let before_decode = Instant::now();
    let event = <ConnectionProcessor as NoiseStreamHandler>::decode(
@@ -37,5 +51,23 @@ fn local_decode_starts_request_span_before_queueing() -> Result<()> {
    assert_eq!(actual_request, request);
    assert!(!request_span.is_disabled());
    assert!((before_decode..=after_decode).contains(&queued_at));
+    let spans = exporter.get_finished_spans()?;
+    assert_eq!(
+        spans.len(),
+        1,
+        "receipt must export while the request is still queued"
+    );
+    assert_eq!(spans[0].name, "codex.exec_server.noise_request_received");
+    assert_eq!(
+        spans[0].span_context.trace_id().to_string(),
+        "00000000000000000000000000000001"
+    );
+    assert!(!format!("{spans:?}").contains("SYNTHETIC_PRIVATE_PAYLOAD"));
+    assert!(!format!("{spans:?}").contains("test/queued"));
+    let receipt_parent = spans[0].parent_span_id;
+    drop(request_span);
+    let spans = exporter.get_finished_spans()?;
+    assert_eq!(spans.len(), 2);
+    assert_eq!(receipt_parent, spans[1].span_context.span_id());
    Ok(())
 }
@@ -28,6 +28,7 @@ use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::sync::watch;
 use tokio::task::JoinHandle;
+use tokio::time::Instant;
 use tokio::time::timeout;

 use crate::client_telemetry::record_client_request;
@@ -740,6 +741,7 @@ impl RpcClient {
        response_rx: oneshot::Receiver<RpcCompletion>,
        call_timeout: RpcCallTimeout,
    ) -> Result<Value, RpcCallError> {
+        let enqueue_started = Instant::now();
        if self
            .write_tx
            .send(JSONRPCMessage::Request(JSONRPCRequest {
@@ -759,6 +761,9 @@ impl RpcClient {
            tracing::Level::INFO,
            event.name = "codex.exec_server.request_enqueued",
            rpc.method = method,
+            enqueue_wait_ms = enqueue_started.elapsed().as_millis() as u64,
+            outgoing_queue_depth = self.write_tx.max_capacity() - self.write_tx.capacity(),
+            outgoing_queue_capacity = self.write_tx.max_capacity(),
        );
        // Do not race in-flight requests directly against the transport-close
        // watch value. The connection reader receives JSON-RPC messages and