Add correlated, credential-free rendezvous connection diagnostics (#51483)

## Why

Rendezvous connection warnings report only a generic WebSocket error, while debug logs include raw error text. Structured diagnostics make failures easier to investigate without logging potentially sensitive error payloads.

## What changed

- Generate a request ID for each connection attempt, send it as `x-request-id`, and include it in the connection span and connection, disconnection, and failure logs alongside the attempt count.
- Replace raw error logging with typed error categories, I/O error kinds, HTTP status codes, and allowlisted values from `x-codex-rendezvous-rejection-reason`.
- Ignore unknown rejection reasons and response bodies.

## Testing

Add tests for rejection-header allowlisting, typed I/O failures, and a rejected WebSocket handshake that verifies the request ID header and preserves the HTTP status and known rejection reason.

GitOrigin-RevId: a8274fec965bcf5b8453b047c0bb79396cdf6ba4

codex-rs/exec-server/src/remote.rs

@@ -58,6 +58,8 @@ use direct::run_direct_environment;
 const ERROR_BODY_PREVIEW_BYTES: usize = 4096;
 const NOISE_RELAY_SECURITY_PROFILE: &str = "noise_hybrid_ik_v1";
 
+#[path = "remote/connection_diagnostics.rs"]
+mod connection_diagnostics;
 mod reconnect_backoff;
 mod registration_retry;
 
@@ -712,9 +714,13 @@ async fn run_remote_environment_connections<H: NoiseStreamHandler>(
         .register_environment_with_retry(&config.environment_id, &identity.public_key())
         .await?;
 
+    let mut connection_attempt = 0_u64;
     loop {
+        connection_attempt += 1;
+        let request_id = format!("req_{}", uuid::Uuid::new_v4().simple());
         match connect_rendezvous(
             &response.url,
+            &request_id,
             &config.telemetry,
             &config.http_client_factory,
         )
@@ -726,6 +732,8 @@ async fn run_remote_environment_connections<H: NoiseStreamHandler>(
                 info!(
                     noise_event = "rendezvous_connection",
                     noise_outcome = "ok",
+                    request_id,
+                    connection_attempt,
                     "Noise executor connected to rendezvous"
                 );
                 let disconnect_reason = run_multiplexed_environment(
@@ -745,6 +753,8 @@ async fn run_remote_environment_connections<H: NoiseStreamHandler>(
                     noise_event = "rendezvous_connection",
                     noise_outcome = "disconnected",
                     noise_reason = disconnect_reason.as_str(),
+                    request_id,
+                    connection_attempt,
                     "Noise executor disconnected from rendezvous"
                 );
                 config
@@ -757,13 +767,19 @@ async fn run_remote_environment_connections<H: NoiseStreamHandler>(
                     tokio_tungstenite::tungstenite::Error::Http(response)
                         if response.status().is_client_error()
                 );
+                let failure = connection_diagnostics::ConnectionFailure::from(&error);
                 warn!(
                     noise_event = "rendezvous_connection",
                     noise_outcome = "error",
                     noise_reason = "websocket_error",
+                    request_id,
+                    connection_attempt,
+                    error_kind = failure.error_kind,
+                    io_error_kind = failure.io_error_kind,
+                    http_status = failure.http_status,
+                    rejection_reason = failure.rejection_reason,
                     "Noise executor failed to connect to rendezvous"
                 );
-                debug!(error = %error, "Noise executor rendezvous connection error");
                 if registration_rejected {
                     config.telemetry.remote_reconnect("registration_rejected");
                     response = client
@@ -793,11 +809,13 @@ async fn run_remote_environment_connections<H: NoiseStreamHandler>(
     fields(
         otel.kind = "client",
         otel.name = "codex.exec_server.remote.rendezvous.connect",
+        request_id = request_id,
         result = tracing::field::Empty,
     )
 )]
 async fn connect_rendezvous(
     url: &str,
+    request_id: &str,
     telemetry: &ExecServerTelemetry,
     http_client_factory: &HttpClientFactory,
 ) -> Result<WebSocketConnection, tokio_tungstenite::tungstenite::Error> {
@@ -805,6 +823,9 @@ async fn connect_rendezvous(
     let result = async {
         let mut request = url.into_client_request()?;
         request.headers_mut().extend(current_rendezvous_headers());
+        request
+            .headers_mut()
+            .insert("x-request-id", HeaderValue::from_str(request_id)?);
         let connector = WebSocketConnector::new_with_tls_mode(
             http_client_factory,
             WebSocketTlsMode::TungsteniteDefault,

codex-rs/exec-server/src/remote/connection_diagnostics.rs

@@ -0,0 +1,96 @@
+//! Credential-free diagnostics for failed rendezvous connections.
+//! Wire headers are untrusted: accept only known rejection codes and never inspect response bodies.
+
+use std::io::ErrorKind;
+
+use tokio_tungstenite::tungstenite::Error;
+
+const REJECTION_REASON_HEADER: &str = "x-codex-rendezvous-rejection-reason";
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) struct ConnectionFailure {
+    pub error_kind: &'static str,
+    pub io_error_kind: Option<&'static str>,
+    pub http_status: Option<u16>,
+    pub rejection_reason: Option<&'static str>,
+}
+
+impl From<&Error> for ConnectionFailure {
+    fn from(error: &Error) -> Self {
+        let (error_kind, io_error_kind) = match error {
+            Error::Http(response) => {
+                return Self {
+                    error_kind: "http",
+                    io_error_kind: None,
+                    http_status: Some(response.status().as_u16()),
+                    rejection_reason: response
+                        .headers()
+                        .get(REJECTION_REASON_HEADER)
+                        .and_then(|value| rejection_reason(value.as_bytes())),
+                };
+            }
+            Error::Io(error) => (
+                "io",
+                Some(match error.kind() {
+                    ErrorKind::TimedOut => "timed_out",
+                    ErrorKind::ConnectionRefused => "connection_refused",
+                    ErrorKind::ConnectionReset => "connection_reset",
+                    ErrorKind::ConnectionAborted => "connection_aborted",
+                    ErrorKind::NotConnected => "not_connected",
+                    ErrorKind::NetworkDown => "network_down",
+                    ErrorKind::NetworkUnreachable => "network_unreachable",
+                    ErrorKind::HostUnreachable => "host_unreachable",
+                    ErrorKind::AddrNotAvailable => "addr_not_available",
+                    ErrorKind::PermissionDenied => "permission_denied",
+                    ErrorKind::BrokenPipe => "broken_pipe",
+                    ErrorKind::UnexpectedEof => "unexpected_eof",
+                    ErrorKind::Interrupted => "interrupted",
+                    ErrorKind::WouldBlock => "would_block",
+                    _ => "other",
+                }),
+            ),
+            Error::Tls(_) => ("tls", None),
+            Error::Protocol(_) => ("protocol", None),
+            Error::ConnectionClosed | Error::AlreadyClosed => ("connection_closed", None),
+            Error::Capacity(_)
+            | Error::WriteBufferFull(_)
+            | Error::Utf8(_)
+            | Error::AttackAttempt
+            | Error::Url(_)
+            | Error::HttpFormat(_) => ("other", None),
+        };
+        Self {
+            error_kind,
+            io_error_kind,
+            http_status: None,
+            rejection_reason: None,
+        }
+    }
+}
+
+fn rejection_reason(header: &[u8]) -> Option<&'static str> {
+    match header {
+        b"expired" => Some("expired"),
+        b"missing_role" => Some("missing_role"),
+        b"missing_expiry" => Some("missing_expiry"),
+        b"missing_signature" => Some("missing_signature"),
+        b"missing_version" => Some("missing_version"),
+        b"missing_executor_registration_id" => Some("missing_executor_registration_id"),
+        b"unexpected_executor_registration_id" => Some("unexpected_executor_registration_id"),
+        b"malformed_query" => Some("malformed_query"),
+        b"invalid_role" => Some("invalid_role"),
+        b"invalid_expiry" => Some("invalid_expiry"),
+        b"invalid_signature" => Some("invalid_signature"),
+        b"invalid_secret" => Some("invalid_secret"),
+        b"extra_query_param" => Some("extra_query_param"),
+        b"pod_draining" => Some("pod_draining"),
+        b"unknown_route" => Some("unknown_route"),
+        b"wrong_rendezvous_cluster" => Some("wrong_rendezvous_cluster"),
+        b"route_unavailable" => Some("route_unavailable"),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+#[path = "connection_diagnostics_tests.rs"]
+mod tests;

codex-rs/exec-server/src/remote/connection_diagnostics_tests.rs

@@ -0,0 +1,116 @@
+//! Verify that connection failures remain correlated without exposing wire payloads.
+
+use pretty_assertions::assert_eq;
+use tokio_tungstenite::tungstenite::Error;
+
+use super::ConnectionFailure;
+use super::REJECTION_REASON_HEADER;
+
+#[test]
+fn only_known_http_rejection_headers_are_reported() {
+    for (header, expected) in [
+        (Some("expired"), Some("expired")),
+        (Some("invalid_signature"), Some("invalid_signature")),
+        (Some("pod_draining"), Some("pod_draining")),
+        (Some("expired secret=credential"), None),
+        (Some("private-proxy-hostname"), None),
+        (None, None),
+    ] {
+        let mut response = http::Response::builder().status(401);
+        if let Some(header) = header {
+            response = response.header(REJECTION_REASON_HEADER, header);
+        }
+        // A known legacy body must not override a header or stand in for a missing one.
+        let error = Error::Http(Box::new(
+            response.body(Some(b"pod_draining".to_vec())).unwrap(),
+        ));
+        assert_eq!(
+            ConnectionFailure::from(&error),
+            ConnectionFailure {
+                error_kind: "http",
+                io_error_kind: None,
+                http_status: Some(401),
+                rejection_reason: expected,
+            }
+        );
+    }
+}
+
+#[test]
+fn io_failure_reports_typed_kind_without_error_text() {
+    use std::io::ErrorKind;
+
+    for (kind, expected) in [
+        (ErrorKind::TimedOut, "timed_out"),
+        (ErrorKind::ConnectionRefused, "connection_refused"),
+        (ErrorKind::ConnectionReset, "connection_reset"),
+        (ErrorKind::ConnectionAborted, "connection_aborted"),
+        (ErrorKind::NotConnected, "not_connected"),
+        (ErrorKind::NetworkDown, "network_down"),
+        (ErrorKind::NetworkUnreachable, "network_unreachable"),
+        (ErrorKind::HostUnreachable, "host_unreachable"),
+        (ErrorKind::AddrNotAvailable, "addr_not_available"),
+        (ErrorKind::PermissionDenied, "permission_denied"),
+        (ErrorKind::BrokenPipe, "broken_pipe"),
+        (ErrorKind::UnexpectedEof, "unexpected_eof"),
+        (ErrorKind::Interrupted, "interrupted"),
+        (ErrorKind::WouldBlock, "would_block"),
+        (ErrorKind::Other, "other"),
+        (ErrorKind::InvalidData, "other"),
+    ] {
+        let error = Error::Io(std::io::Error::new(kind, "private host and credentials"));
+        assert_eq!(
+            ConnectionFailure::from(&error),
+            ConnectionFailure {
+                error_kind: "io",
+                io_error_kind: Some(expected),
+                http_status: None,
+                rejection_reason: None,
+            }
+        );
+    }
+}
+
+#[tokio::test]
+async fn websocket_rejection_preserves_status_and_sends_correlation_id() {
+    use wiremock::Mock;
+    use wiremock::MockServer;
+    use wiremock::ResponseTemplate;
+    use wiremock::matchers::header;
+    use wiremock::matchers::method;
+
+    let server = MockServer::start().await;
+    let request_id = "req_0123456789abcdef0123456789abcdef";
+    Mock::given(method("GET"))
+        .and(header("x-request-id", request_id))
+        .respond_with(
+            ResponseTemplate::new(401)
+                .insert_header(REJECTION_REASON_HEADER, "expired")
+                .set_body_string("private proxy response that must not be logged"),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+    let result = super::super::connect_rendezvous(
+        &format!("ws://{}/", server.address()),
+        request_id,
+        &crate::ExecServerTelemetry::default(),
+        &codex_http_client::HttpClientFactory::new(
+            codex_http_client::OutboundProxyPolicy::ReqwestDefault,
+        ),
+    )
+    .await;
+    let error = match result {
+        Err(error) => error,
+        Ok(_) => panic!("expected the websocket upgrade to be rejected"),
+    };
+    assert_eq!(
+        ConnectionFailure::from(&error),
+        ConnectionFailure {
+            error_kind: "http",
+            io_error_kind: None,
+            http_status: Some(401),
+            rejection_reason: Some("expired"),
+        }
+    );
+}