Remove response.processed websocket request

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

@@ -239,11 +239,6 @@ pub struct ResponseCreateWsRequest {
     pub client_metadata: Option<HashMap<String, String>>,
 }
 
-#[derive(Debug, Serialize)]
-pub struct ResponseProcessedWsRequest {
-    pub response_id: String,
-}
-
 pub fn response_create_client_metadata(
     client_metadata: Option<HashMap<String, String>>,
     trace: Option<&W3cTraceContext>,
@@ -272,8 +267,6 @@ pub fn response_create_client_metadata(
 pub enum ResponsesWsRequest {
     #[serde(rename = "response.create")]
     ResponseCreate(ResponseCreateWsRequest),
-    #[serde(rename = "response.processed")]
-    ResponseProcessed(ResponseProcessedWsRequest),
 }
 
 pub fn create_text_param_for_request(

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

@@ -1,6 +1,5 @@
 use crate::auth::SharedAuthProvider;
 use crate::common::ResponseEvent;
-use crate::common::ResponseProcessedWsRequest;
 use crate::common::ResponseStream;
 use crate::common::ResponsesWsRequest;
 use crate::error::ApiError;
@@ -206,40 +205,6 @@ impl ResponsesWebsocketConnection {
         self.stream.lock().await.is_none()
     }
 
-    #[instrument(
-        name = "responses_websocket.send_response_processed",
-        level = "info",
-        skip_all,
-        fields(transport = "responses_websocket", api.path = "responses")
-    )]
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "the guard serializes exclusive use of the websocket while sending a request frame"
-    )]
-    pub async fn send_response_processed(&self, response_id: String) -> Result<(), ApiError> {
-        let request =
-            ResponsesWsRequest::ResponseProcessed(ResponseProcessedWsRequest { response_id });
-        let request_body = serde_json::to_value(&request).map_err(|err| {
-            ApiError::Stream(format!("failed to encode websocket request: {err}"))
-        })?;
-
-        let mut guard = self.stream.lock().await;
-        let Some(ws_stream) = guard.as_mut() else {
-            return Err(ApiError::Stream(
-                "websocket connection is closed".to_string(),
-            ));
-        };
-
-        send_websocket_request(
-            ws_stream,
-            request_body,
-            self.idle_timeout,
-            self.telemetry.as_ref(),
-            /*connection_reused*/ true,
-        )
-        .await
-    }
-
     #[instrument(
         name = "responses_websocket.stream_request",
         level = "info",

codex-rs/codex-api/src/lib.rs

@@ -32,7 +32,6 @@ pub use crate::common::RawMemoryMetadata;
 pub use crate::common::Reasoning;
 pub use crate::common::ResponseCreateWsRequest;
 pub use crate::common::ResponseEvent;
-pub use crate::common::ResponseProcessedWsRequest;
 pub use crate::common::ResponseStream;
 pub use crate::common::ResponsesApiRequest;
 pub use crate::common::ResponsesWsRequest;

codex-rs/core/config.schema.json

@@ -556,9 +556,6 @@
             "request_rule": {
               "type": "boolean"
             },
-            "responses_websocket_response_processed": {
-              "type": "boolean"
-            },
             "responses_websockets": {
               "type": "boolean"
             },
@@ -4679,9 +4676,6 @@
         "request_rule": {
           "type": "boolean"
         },
-        "responses_websocket_response_processed": {
-          "type": "boolean"
-        },
         "responses_websockets": {
           "type": "boolean"
         },
@@ -5205,4 +5199,4 @@
   },
   "title": "ConfigToml",
   "type": "object"
-}
+}

codex-rs/core/src/client.rs

@@ -100,7 +100,6 @@ use tokio::sync::oneshot::error::TryRecvError;
 use tokio_tungstenite::tungstenite::Error;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_util::sync::CancellationToken;
-use tracing::debug;
 use tracing::instrument;
 use tracing::trace;
 use tracing::warn;
@@ -967,18 +966,6 @@ impl ModelClientSession {
             .set_connection_reused(/*connection_reused*/ false);
     }
 
-    pub(crate) async fn send_response_processed(&self, response_id: &str) {
-        let Some(connection) = self.websocket_session.connection.as_ref() else {
-            return;
-        };
-        if let Err(err) = connection
-            .send_response_processed(response_id.to_string())
-            .await
-        {
-            debug!("failed to send response.processed websocket request: {err}");
-        }
-    }
-
     #[allow(clippy::too_many_arguments)]
     /// Builds shared Responses API transport options and request-body options.
     ///
@@ -1672,9 +1659,7 @@ fn parse_turn_metadata_header(turn_metadata_header: Option<&str>) -> Option<Head
 /// Meant to be called just before sending the request over the socket, to capture realistic
 /// transport timing.
 fn stamp_ws_stream_request_start_ms(request: &mut ResponsesWsRequest) {
-    let ResponsesWsRequest::ResponseCreate(payload) = request else {
-        return;
-    };
+    let ResponsesWsRequest::ResponseCreate(payload) = request;
     payload
         .client_metadata
         .get_or_insert_with(HashMap::new)

codex-rs/core/src/compact_remote_v2.rs

@@ -26,7 +26,6 @@ use codex_analytics::CompactionImplementation;
 use codex_analytics::CompactionPhase;
 use codex_analytics::CompactionReason;
 use codex_analytics::CompactionTrigger;
-use codex_features::Feature;
 use codex_protocol::error::CodexErr;
 use codex_protocol::error::Result as CodexResult;
 use codex_protocol::items::ContextCompactionItem;
@@ -281,7 +280,6 @@ async fn run_remote_compact_task_inner_impl(
     );
     let RemoteCompactionV2Output {
         compaction_output,
-        response_id,
         token_usage,
     } = compaction_output_result?;
     if let Some(token_usage) = token_usage {
@@ -314,18 +312,11 @@ async fn run_remote_compact_task_inner_impl(
 
     sess.emit_turn_item_completed(turn_context, compaction_item)
         .await;
-    if turn_context
-        .features
-        .enabled(Feature::ResponsesWebsocketResponseProcessed)
-    {
-        client_session.send_response_processed(&response_id).await;
-    }
     Ok(())
 }
 
 struct RemoteCompactionV2Output {
     compaction_output: ResponseItem,
-    response_id: String,
     token_usage: Option<TokenUsage>,
 }
 
@@ -409,7 +400,7 @@ async fn collect_compaction_output(
     let mut output_item_count = 0usize;
     let mut compaction_count = 0usize;
     let mut compaction_output = None;
-    let mut completed_response_id = None;
+    let mut saw_completed = false;
     let mut completed_token_usage = None;
     while let Some(event) = stream.next().await {
         match event? {
@@ -422,25 +413,21 @@ async fn collect_compaction_output(
                     }
                 }
             }
-            ResponseEvent::Completed {
-                response_id,
-                token_usage,
-                ..
-            } => {
-                completed_response_id = Some(response_id);
+            ResponseEvent::Completed { token_usage, .. } => {
+                saw_completed = true;
                 completed_token_usage = token_usage;
                 break;
             }
             _ => {}
         }
     }
 
-    let Some(response_id) = completed_response_id else {
+    if !saw_completed {
         return Err(CodexErr::Stream(
             "remote compaction v2 stream closed before response.completed".to_string(),
             None,
         ));
-    };
+    }
 
     if compaction_count != 1 {
         return Err(CodexErr::Fatal(format!(
@@ -453,7 +440,6 @@ async fn collect_compaction_output(
     };
     Ok(RemoteCompactionV2Output {
         compaction_output,
-        response_id,
         token_usage: completed_token_usage,
     })
 }
@@ -804,7 +790,6 @@ mod tests {
             .expect("compaction should be collected");
 
         assert_eq!(output.compaction_output, compaction);
-        assert_eq!(output.response_id, "resp-compact");
         assert_eq!(
             output.token_usage,
             Some(TokenUsage {

codex-rs/core/src/session/turn.rs

@@ -1827,7 +1827,6 @@ async fn try_run_sampling_request(
         !sess.services.extensions.turn_item_contributors().is_empty();
     let mut active_item_is_streaming_to_client = false;
     let receiving_span = trace_span!("receiving_stream");
-    let mut completed_response_id: Option<String> = None;
     let outcome: CodexResult<SamplingRequestResult> = loop {
         let handle_responses = trace_span!(
             parent: &receiving_span,
@@ -2071,9 +2070,9 @@ async fn try_run_sampling_request(
                 sess.services.models_manager.refresh_if_new_etag(etag).await;
             }
             ResponseEvent::Completed {
-                response_id,
                 token_usage,
                 end_turn,
+                ..
             } => {
                 flush_assistant_text_segments_all(
                     &sess,
@@ -2089,7 +2088,6 @@ async fn try_run_sampling_request(
                 if let Some(false) = end_turn {
                     needs_follow_up = true;
                 }
-                completed_response_id = Some(response_id);
                 break Ok(SamplingRequestResult {
                     needs_follow_up,
                     last_agent_message,
@@ -2213,15 +2211,6 @@ async fn try_run_sampling_request(
     )
     .await;
 
-    if sess
-        .features
-        .enabled(Feature::ResponsesWebsocketResponseProcessed)
-        && outcome.is_ok()
-        && let Some(response_id) = completed_response_id.as_deref()
-    {
-        client_session.send_response_processed(response_id).await;
-    }
-
     drain_in_flight(&mut in_flight, sess.clone(), turn_context.clone()).await?;
 
     if should_emit_token_count {