Merge branch 'main' into dev/pbakkum/codex-realtimeapi-port

codex-rs/codex-api/src/common.rs

@@ -21,6 +21,12 @@ pub struct CompactionInput<'a> {
     pub model: &'a str,
     pub input: &'a [ResponseItem],
     pub instructions: &'a str,
+    pub tools: Vec<Value>,
+    pub parallel_tool_calls: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reasoning: Option<Reasoning>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub text: Option<TextControls>,
 }
 
 /// Canonical input payload for the memory summarize endpoint.

codex-rs/codex-api/src/endpoint/responses.rs

@@ -75,6 +75,9 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
         }
 
         let mut headers = extra_headers;
+        if let Some(ref conv_id) = conversation_id {
+            insert_header(&mut headers, "x-client-request-id", conv_id);
+        }
         headers.extend(build_conversation_headers(conversation_id));
         if let Some(subagent) = subagent_header(&session_source) {
             insert_header(&mut headers, "x-openai-subagent", &subagent);

codex-rs/codex-api/src/endpoint/responses_websocket.rs

@@ -53,9 +53,6 @@ enum WsCommand {
         message: Message,
         tx_result: oneshot::Sender<Result<(), WsError>>,
     },
-    Close {
-        tx_result: oneshot::Sender<Result<(), WsError>>,
-    },
 }
 
 impl WsStream {
@@ -80,11 +77,6 @@ impl WsStream {
                                     break;
                                 }
                             }
-                            WsCommand::Close { tx_result } => {
-                                let result = inner.close(None).await;
-                                let _ = tx_result.send(result);
-                                break;
-                            }
                         }
                     }
                     message = inner.next() => {
@@ -144,11 +136,6 @@ impl WsStream {
             .await
     }
 
-    async fn close(&self) -> Result<(), WsError> {
-        self.request(|tx_result| WsCommand::Close { tx_result })
-            .await
-    }
-
     async fn next(&mut self) -> Option<Result<Message, WsError>> {
         self.rx_message.recv().await
     }
@@ -242,26 +229,32 @@ impl ResponsesWebsocketConnection {
                     .await;
             }
             let mut guard = stream.lock().await;
-            let Some(ws_stream) = guard.as_mut() else {
-                let _ = tx_event
-                    .send(Err(ApiError::Stream(
-                        "websocket connection is closed".to_string(),
-                    )))
-                    .await;
-                return;
+            let result = {
+                let Some(ws_stream) = guard.as_mut() else {
+                    let _ = tx_event
+                        .send(Err(ApiError::Stream(
+                            "websocket connection is closed".to_string(),
+                        )))
+                        .await;
+                    return;
+                };
+
+                run_websocket_response_stream(
+                    ws_stream,
+                    tx_event.clone(),
+                    request_body,
+                    idle_timeout,
+                    telemetry,
+                )
+                .await
             };
 
-            if let Err(err) = run_websocket_response_stream(
-                ws_stream,
-                tx_event.clone(),
-                request_body,
-                idle_timeout,
-                telemetry,
-            )
-            .await
-            {
-                let _ = ws_stream.close().await;
-                *guard = None;
+            if let Err(err) = result {
+                // A terminal stream error should reach the caller immediately. Waiting for a
+                // graceful close handshake here can stall indefinitely and mask the error.
+                let failed_stream = guard.take();
+                drop(guard);
+                drop(failed_stream);
                 let _ = tx_event.send(Err(err)).await;
             }
         });