Allow upstream interruption when canceling Responses Lite turns (#57864)

## What changed

- Add the disabled-by-default `interrupt_on_cancel` feature to interrupt upstream inference when a Responses Lite turn is canceled, with a one-second cleanup deadline.
- Detach canceled WebSocket responses so replacement requests can open a new connection without waiting for cleanup.
- Add an optional `interrupt_on_cancel` override for thread samples, inheriting the parent setting when omitted.

## Testing

Add coverage for cancellation during inference and steering, replacement turns proceeding before cleanup finishes, and prompt cancellation of model response interceptors.

GitOrigin-RevId: 501368746fc04fd3fdbd45c1f34c75ce57f4e955

codex-rs/app-server-protocol/src/protocol/v2/thread_sample.rs

@@ -31,6 +31,9 @@ pub struct ThreadSampleParams {
     /// Disable MCP servers when the inherited window uses incremental tools on Responses Lite.
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub disable_mcp: bool,
+    /// Override upstream interruption on cancellation for this sample; otherwise inherit the parent.
+    #[ts(optional = nullable)]
+    pub interrupt_on_cancel: Option<bool>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]

codex-rs/app-server/src/thread_sample.rs

@@ -195,6 +195,13 @@ impl SampleTask {
             .features
             .enable(Feature::ReasoningEffortOverride)
             .map_err(|error| invalid_request(error.to_string()))?;
+        if let Some(interrupt_on_cancel) = params.interrupt_on_cancel {
+            options
+                .config
+                .features
+                .set_enabled(Feature::InterruptOnCancel, interrupt_on_cancel)
+                .map_err(|error| invalid_request(error.to_string()))?;
+        }
         options
             .thread_extension_init
             .insert(codex_core::compact::AutoCompactionDisabled);

codex-rs/app-server/tests/suite/v2/thread_sample.rs

@@ -305,6 +305,7 @@ async fn sample_preserves_context_and_keeps_source_unchanged(
             expected_turn_id: completed.turn.id.clone(),
             prompt: "Continue identically".to_string(),
             effort: Some(ReasoningEffort::Medium),
+            interrupt_on_cancel: Some(true),
             thread_source: Some(ThreadSource::Feature("composer_predictions".to_string())),
             turn_trigger: Some("composer_predictions".to_string()),
             responsesapi_client_metadata: Some(HashMap::from([

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

@@ -338,54 +338,70 @@ impl ResponsesWebsocketConnection {
                 if tx_event.is_closed() {
                     return;
                 }
+                let Some(mut ws_stream) = guard.take() else {
+                    let _ = tx_event
+                        .send(Err(ApiError::Stream(
+                            "websocket connection is closed".to_string(),
+                        )))
+                        .await;
+                    return;
+                };
                 let mut completion = None;
                 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;
-                    };
+                    let response = run_websocket_response_stream(
+                        &mut ws_stream,
+                        tx_event.clone(),
+                        request_text,
+                        idle_timeout,
+                        telemetry,
+                        turn_state.as_deref(),
+                        &timing_log_context,
+                        interrupt.clone(),
+                        &mut completion,
+                    );
+                    tokio::pin!(response);
 
                     tokio::select! {
                         biased;
-                        result = run_websocket_response_stream(
-                            ws_stream,
-                            tx_event.clone(),
-                            request_text,
-                            idle_timeout,
-                            telemetry,
-                            turn_state.as_deref(),
-                            &timing_log_context,
-                            interrupt.clone(),
-                            &mut completion,
-                        ) => result,
+                        result = &mut response => result,
                         () = &mut interrupt_timeout => Err(ApiError::Stream(
                             "timed out draining interrupted response".into(),
                         )),
-                        _ = tx_event.closed() => Err(ApiError::Stream(
-                            "response event consumer dropped".to_string(),
-                        )),
+                        () = tx_event.closed() => {
+                            // Release the lock so later requests can open a new connection
+                            // while this task finishes cleanup on the detached socket.
+                            drop(guard);
+                            if matches!(
+                                interrupt.clone().now_or_never(),
+                                Some(ResponseInterrupt::Cancel { .. })
+                            ) {
+                                tokio::select! {
+                                    _ = &mut response => {}
+                                    () = &mut interrupt_timeout => {}
+                                }
+                            }
+                            return;
+                        }
                     }
                 };
 
                 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);
+                    drop(ws_stream);
                     // Failed cancellation retires the socket while preserving any received
                     // completion and its usage for the abandoned classifier's cleanup.
                     let event = completion.map_or_else(|| Err(err), Ok);
                     let _ = tx_event.send(event).await;
-                } else if let Some(completed) = completion {
-                    // The handshake is finished. Consumer backpressure must not let the
-                    // interruption deadline discard completion or its usage accounting.
-                    drop(guard);
-                    let _ = tx_event.send(Ok(completed)).await;
+                } else {
+                    *guard = Some(ws_stream);
+                    if let Some(completed) = completion {
+                        // The handshake is finished. Consumer backpressure must not let the
+                        // interruption deadline discard completion or its usage accounting.
+                        drop(guard);
+                        let _ = tx_event.send(Ok(completed)).await;
+                    }
                 }
             }
             .instrument(current_span),
@@ -925,11 +941,7 @@ async fn run_websocket_response_stream(
                     last_server_model = Some(model);
                 }
                 for metadata_event in metadata_events {
-                    if tx_event.send(Ok(metadata_event)).await.is_err() {
-                        return Err(ApiError::Stream(
-                            "response event consumer dropped".to_string(),
-                        ));
-                    }
+                    let _ = tx_event.send(Ok(metadata_event)).await;
                 }
                 match event {
                     Ok(Some(event)) => {

codex-rs/core/config.schema.json

@@ -1128,6 +1128,9 @@
             "instant_interrupt": {
               "type": "boolean"
             },
+            "interrupt_on_cancel": {
+              "type": "boolean"
+            },
             "item_ids": {
               "type": "boolean"
             },
@@ -7780,6 +7783,9 @@
         "instant_interrupt": {
           "type": "boolean"
         },
+        "interrupt_on_cancel": {
+          "type": "boolean"
+        },
         "item_ids": {
           "type": "boolean"
         },

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

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
+use std::time::Duration;
 use std::time::Instant;
 
 use crate::client::ModelClientSession;
@@ -2790,7 +2791,13 @@ async fn try_run_sampling_request(
             return Err(error);
         }
     };
-    if cancellation_token.is_cancelled() {
+    let can_interrupt = turn_context
+        .config
+        .features
+        .enabled(Feature::InterruptOnCancel)
+        && step_context.settings.model_info.use_responses_lite
+        && stream.interrupt.is_some();
+    if cancellation_token.is_cancelled() && !can_interrupt {
         return Err(CodexErr::TurnAborted);
     }
     let mut in_flight: FuturesOrdered<InFlightFuture<'static>> = FuturesOrdered::new();
@@ -2870,7 +2877,6 @@ async fn try_run_sampling_request(
                     continue;
                 }
                 // TODO: Reconcile any response item already being presented to the client.
-                drop(stream);
                 break Ok(SamplingRequestResult {
                     needs_follow_up: true,
                     last_agent_message,
@@ -3431,6 +3437,13 @@ async fn try_run_sampling_request(
             }
         }
     };
+    if cancellation_token.is_cancelled() && !response_completed && can_interrupt {
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
+        if let Some(interrupt) = stream.interrupt.take() {
+            let _ = interrupt.send(codex_api::ResponseInterrupt::Cancel { deadline });
+        }
+    }
+    drop(stream);
     drop(sampling_span);
     drop(sampling_timing_guard);
     if let Some(finish_post_request) = finish_post_request {

codex-rs/core/tests/common/responses.rs

@@ -1283,6 +1283,7 @@ pub async fn start_websocket_server_with_headers(
     let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
 
     let task = tokio::spawn(async move {
+        let mut handlers = tokio::task::JoinSet::new();
         loop {
             let accept_res = tokio::select! {
                 _ = &mut shutdown_rx => return,
@@ -1355,102 +1356,105 @@ pub async fn start_websocket_server_with_headers(
                 log.push(Vec::new());
                 log.len() - 1
             };
-            let close_after_requests = connection.close_after_requests;
-            let mut response_complete = false;
-            let mut batches = connection.requests.into_iter().peekable();
-            while let Some(request_events) = batches.peek() {
-                let message = tokio::select! {
-                    _ = &mut shutdown_rx => return,
-                    message = ws_stream.next() => message,
-                };
-                let Some(Ok(message)) = message else {
-                    break;
-                };
-                if let Some(body) = parse_ws_request_body(message) {
-                    let late_interrupt = response_complete && body["type"] == "response.interrupt";
-                    let mut log = requests.lock().unwrap();
-                    if let Some(connection_log) = log.get_mut(connection_index) {
-                        connection_log.push(WebSocketRequest { body });
-                        let request_index = connection_log.len() - 1;
-                        let request = &connection_log[request_index];
-                        let request_body = request.body_json();
-                        eprintln!(
-                            "[ws test server +{}ms] connection={} received request={} type={:?} role={:?} text={:?} data={:?}",
-                            start.elapsed().as_millis(),
-                            connection_index,
-                            request_index,
-                            request_body.get("type").and_then(Value::as_str),
-                            request_body
-                                .get("item")
-                                .and_then(|item| item.get("role"))
-                                .and_then(Value::as_str),
-                            request_body
-                                .get("item")
-                                .and_then(|item| item.get("content"))
-                                .and_then(Value::as_array)
-                                .and_then(|content| content.first())
-                                .and_then(|content| content.get("text"))
-                                .and_then(Value::as_str),
-                            request_body
-                                .get("item")
-                                .and_then(|item| item.get("content"))
-                                .and_then(Value::as_array)
-                                .and_then(|content| content.first())
-                                .and_then(|content| content.get("data"))
-                                .and_then(Value::as_str),
-                        );
-                    }
-                    request_log.notify_waiters();
-                    // A late interrupt cannot consume the next response's script.
-                    if late_interrupt {
-                        continue;
-                    }
-                }
-
-                eprintln!(
-                    "[ws test server +{}ms] connection={} sending batch_size={} event_types={:?} audio_data={:?}",
-                    start.elapsed().as_millis(),
-                    connection_index,
-                    request_events.len(),
-                    request_events
-                        .iter()
-                        .map(|event| event.get("type").and_then(Value::as_str))
-                        .collect::<Vec<_>>(),
-                    request_events
-                        .iter()
-                        .find_map(|event| event.get("delta").and_then(Value::as_str)),
-                );
-                for event in request_events {
-                    let Ok(payload) = serde_json::to_string(event) else {
-                        continue;
+            let requests = Arc::clone(&requests);
+            let request_log = Arc::clone(&request_log);
+            let gates = Arc::clone(&gates);
+            let closed_tx = closed_tx.clone();
+            handlers.spawn(async move {
+                let close_after_requests = connection.close_after_requests;
+                let mut response_complete = false;
+                let mut batches = connection.requests.into_iter().peekable();
+                while let Some(request_events) = batches.peek() {
+                    let message = ws_stream.next().await;
+                    let Some(Ok(message)) = message else {
+                        break;
                     };
-                    let gate = gates.lock().unwrap().remove(&payload);
-                    if let Some(gate) = gate {
-                        tokio::select! {
-                            _ = &mut shutdown_rx => return,
-                            _ = gate => {},
+                    if let Some(body) = parse_ws_request_body(message) {
+                        let late_interrupt = response_complete && body["type"] == "response.interrupt";
+                        let mut log = requests.lock().unwrap();
+                        if let Some(connection_log) = log.get_mut(connection_index) {
+                            connection_log.push(WebSocketRequest { body });
+                            let request_index = connection_log.len() - 1;
+                            let request = &connection_log[request_index];
+                            let request_body = request.body_json();
+                            eprintln!(
+                                "[ws test server +{}ms] connection={} received request={} type={:?} role={:?} text={:?} data={:?}",
+                                start.elapsed().as_millis(),
+                                connection_index,
+                                request_index,
+                                request_body.get("type").and_then(Value::as_str),
+                                request_body
+                                    .get("item")
+                                    .and_then(|item| item.get("role"))
+                                    .and_then(Value::as_str),
+                                request_body
+                                    .get("item")
+                                    .and_then(|item| item.get("content"))
+                                    .and_then(Value::as_array)
+                                    .and_then(|content| content.first())
+                                    .and_then(|content| content.get("text"))
+                                    .and_then(Value::as_str),
+                                request_body
+                                    .get("item")
+                                    .and_then(|item| item.get("content"))
+                                    .and_then(Value::as_array)
+                                    .and_then(|content| content.first())
+                                    .and_then(|content| content.get("data"))
+                                    .and_then(Value::as_str),
+                            );
+                        }
+                        request_log.notify_waiters();
+                        // A late interrupt cannot consume the next response's script.
+                        if late_interrupt {
+                            continue;
                         }
                     }
-                    if ws_stream.send(Message::Text(payload.into())).await.is_err() {
-                        break;
-                    }
-                    response_complete = matches!(
-                        event["type"].as_str(),
-                        Some("response.completed" | "response.incomplete" | "response.failed")
-                    );
-                }
-                batches.next();
-            }
 
-            if close_after_requests {
-                let _ = ws_stream.close(None).await;
-                closed_tx.send_modify(|count| *count += 1);
-            } else {
-                let _ = shutdown_rx.await;
-                return;
-            }
+                    eprintln!(
+                        "[ws test server +{}ms] connection={} sending batch_size={} event_types={:?} audio_data={:?}",
+                        start.elapsed().as_millis(),
+                        connection_index,
+                        request_events.len(),
+                        request_events
+                            .iter()
+                            .map(|event| event.get("type").and_then(Value::as_str))
+                            .collect::<Vec<_>>(),
+                        request_events
+                            .iter()
+                            .find_map(|event| event.get("delta").and_then(Value::as_str)),
+                    );
+                    for event in request_events {
+                        let Ok(payload) = serde_json::to_string(event) else {
+                            continue;
+                        };
+                        let gate = gates.lock().unwrap().remove(&payload);
+                        if let Some(gate) = gate {
+                            let _ = gate.await;
+                        }
+                        if ws_stream.send(Message::Text(payload.into())).await.is_err() {
+                            break;
+                        }
+                        response_complete = matches!(
+                            event["type"].as_str(),
+                            Some("response.completed" | "response.incomplete" | "response.failed")
+                        );
+                    }
+                    batches.next();
+                }
+
+                if close_after_requests {
+                    let _ = ws_stream.close(None).await;
+                    closed_tx.send_modify(|count| *count += 1);
+                } else {
+                    std::future::pending::<()>().await;
+                }
+            });
 
             if connections.lock().unwrap().is_empty() {
+                tokio::select! {
+                    _ = &mut shutdown_rx => {},
+                    _ = handlers.join_all() => {},
+                }
                 return;
             }
         }

codex-rs/core/tests/suite/model_request.rs

@@ -9,6 +9,7 @@ use codex_extension_api::ModelRequestKind;
 use codex_extension_api::ModelResponseInterceptor;
 use codex_extension_api::ModelResponseStream;
 use codex_extension_api::ResponseEvent;
+use codex_features::Feature;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::Op;
 use codex_protocol::user_input::UserInput;
@@ -30,6 +31,7 @@ use tokio::time::timeout;
 #[derive(Debug)]
 struct Gates {
     delivered: Semaphore,
+    output_received: Semaphore,
     output_waiting: Semaphore,
     cancelled: Semaphore,
     output_items: std::sync::atomic::AtomicUsize,
@@ -87,6 +89,7 @@ impl ModelResponseInterceptor for Permit {
                     }
                     if matches!(&event, Ok(ResponseEvent::OutputItemDone(_))) {
                         output_items += 1;
+                        self.gates.output_received.add_permits(1);
                     }
                     let completed = matches!(&event, Ok(ResponseEvent::Completed { .. }));
                     events.push(event);
@@ -135,6 +138,7 @@ async fn model_request_gates_output_and_cancellation() -> Result<()> {
     let model = responses::mount_sse_sequence(&server, vec![response.clone(), response]).await;
     let gates = Arc::new(Gates {
         delivered: Semaphore::new(0),
+        output_received: Semaphore::new(0),
         output_waiting: Semaphore::new(0),
         cancelled: Semaphore::new(0),
         output_items: std::sync::atomic::AtomicUsize::new(0),
@@ -213,8 +217,9 @@ async fn model_request_gates_output_and_cancellation() -> Result<()> {
     Ok(())
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[tokio::test]
 async fn model_request_gates_websocket_output_but_not_warmup() -> Result<()> {
+    let completed = ev_completed("response");
     let server = responses::start_websocket_server(vec![vec![
         vec![ev_response_created("warm"), ev_completed("warm")],
         vec![
@@ -240,12 +245,18 @@ async fn model_request_gates_websocket_output_but_not_warmup() -> Result<()> {
                 }
             }),
             ev_assistant_message("message", "done"),
-            ev_completed("response"),
         ],
+        vec![
+            serde_json::json!({"type": "response.interrupt.accepted", "response_id": "response"}),
+            completed.clone(),
+        ],
+        vec![],
     ]])
     .await;
+    let release_completion = server.gate_response_event(&completed);
     let gates = Arc::new(Gates {
         delivered: Semaphore::new(0),
+        output_received: Semaphore::new(0),
         output_waiting: Semaphore::new(0),
         cancelled: Semaphore::new(0),
         output_items: std::sync::atomic::AtomicUsize::new(0),
@@ -256,6 +267,15 @@ async fn model_request_gates_websocket_output_but_not_warmup() -> Result<()> {
     registry.model_request_contributor(Arc::new(Contributor(gates.clone())));
     let mut builder = test_codex()
         .with_model("gpt-5.2")
+        .with_model_info_override("gpt-5.2", |model_info| {
+            model_info.use_responses_lite = true;
+        })
+        .with_config(|config| {
+            config
+                .features
+                .enable(Feature::InterruptOnCancel)
+                .expect("enable InterruptOnCancel feature");
+        })
         .with_extensions(Arc::new(registry.build()));
     let test = builder.build_with_websocket_server(&server).await?;
     test.codex
@@ -264,7 +284,7 @@ async fn model_request_gates_websocket_output_but_not_warmup() -> Result<()> {
             text_elements: vec![],
         }]))
         .await?;
-    timeout(Duration::from_secs(10), gates.output_waiting.acquire())
+    timeout(Duration::from_secs(10), gates.output_received.acquire())
         .await??
         .forget();
     let requests = server.single_connection();
@@ -304,10 +324,42 @@ async fn model_request_gates_websocket_output_but_not_warmup() -> Result<()> {
         requests[1].body_json()["client_metadata"]["test-interceptor"],
         "present"
     );
-    test.codex.submit(Op::Interrupt).await?;
-    timeout(Duration::from_secs(10), gates.cancelled.acquire())
-        .await??
-        .forget();
+    // Neither a terminal response nor the cleanup deadline may release this interceptor.
+    tokio::time::pause();
+    {
+        let cancelled = async {
+            test.codex.submit(Op::Interrupt).await?;
+            timeout(Duration::from_secs(10), gates.cancelled.acquire())
+                .await??
+                .forget();
+            server
+                .wait_for_request(/*connection_index*/ 0, /*request_index*/ 2)
+                .await;
+            anyhow::Ok(())
+        };
+        tokio::pin!(cancelled);
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            anyhow::ensure!(
+                std::time::Instant::now() < deadline,
+                "interceptor remained alive during cancelled response cleanup"
+            );
+            tokio::select! {
+                result = &mut cancelled => { result?; break; }
+                // Keep the runtime runnable so paused time cannot auto-advance during I/O.
+                () = tokio::task::yield_now() => {}
+            }
+        }
+    }
+    tokio::time::resume();
+    release_completion
+        .send(())
+        .expect("release interrupted completion");
+    assert!(
+        server
+            .wait_for_closed_connections(/*expected*/ 1, Duration::from_secs(5))
+            .await
+    );
     server.shutdown().await;
     Ok(())
 }

codex-rs/core/tests/suite/pending_input.rs

@@ -1403,6 +1403,131 @@ async fn steer_interrupts_and_drains_websocket(
     Ok(())
 }
 
+#[test_case(false, false; "feature disabled")]
+#[test_case(true, false; "interrupt during inference")]
+#[test_case(true, true; "interrupt while steering drains")]
+#[tokio::test]
+async fn turn_interrupt_stops_websocket_inference(
+    interrupt_on_cancel: bool,
+    steer_first: bool,
+) -> anyhow::Result<()> {
+    core_test_support::skip_if_no_network!(Ok(()));
+
+    let completed = ev_completed("resp-interrupted");
+    let server = responses::start_websocket_server(vec![
+        vec![
+            vec![ev_response_created("warm-1"), ev_completed("warm-1")],
+            vec![
+                ev_response_created("resp-interrupted"),
+                ev_reasoning_item_added("reason-1", &["thinking"]),
+            ],
+            vec![
+                json!({"type": "response.interrupt.accepted", "response_id": "resp-interrupted"}),
+                completed.clone(),
+            ],
+            vec![],
+        ],
+        vec![vec![
+            ev_response_created("replacement"),
+            ev_completed("replacement"),
+        ]],
+    ])
+    .await;
+    let release_completion = server.gate_response_event(&completed);
+    let test = test_codex()
+        .with_model("gpt-5.4")
+        .with_model_info_override("gpt-5.4", |model_info| {
+            model_info.use_responses_lite = true;
+        })
+        .with_config(move |config| {
+            if interrupt_on_cancel {
+                config
+                    .features
+                    .enable(Feature::InterruptOnCancel)
+                    .expect("enable InterruptOnCancel feature");
+            }
+            config
+                .features
+                .enable(Feature::InstantInterrupt)
+                .expect("enable InstantInterrupt feature");
+        })
+        .build_with_websocket_server(&server)
+        .await?;
+    let codex = &test.codex;
+    submit_user_input(codex, "first prompt").await;
+    wait_for_reasoning_item_started(codex).await;
+
+    if steer_first {
+        steer_user_input(codex, "second prompt").await;
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2),
+        )
+        .await?;
+    }
+    // Freeze the cleanup deadline so waiting for it cannot make replacement succeed.
+    tokio::time::pause();
+    {
+        let replacement = async {
+            codex.submit(Op::Interrupt).await?;
+            wait_for_event(codex, |event| matches!(event, EventMsg::TurnAborted(_))).await;
+            if interrupt_on_cancel {
+                let interrupt = tokio::time::timeout(
+                    std::time::Duration::from_secs(/*secs*/ 10),
+                    server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2),
+                )
+                .await?
+                .body_json();
+                assert_eq!(
+                    interrupt,
+                    json!({
+                        "type": "response.interrupt", "response_id": "resp-interrupted",
+                        "mode": "discard_partial_items"
+                    })
+                );
+            }
+            submit_user_input(codex, "replacement prompt").await;
+            wait_for_successful_turn(codex).await;
+            anyhow::Ok(())
+        };
+        tokio::pin!(replacement);
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+        loop {
+            anyhow::ensure!(
+                std::time::Instant::now() < deadline,
+                "replacement waited for cancelled response cleanup"
+            );
+            tokio::select! {
+                result = &mut replacement => { result?; break; }
+                // Keep the runtime runnable so paused time cannot auto-advance during I/O.
+                () = tokio::task::yield_now() => {}
+            }
+        }
+    }
+    tokio::time::resume();
+    release_completion
+        .send(())
+        .expect("release interrupted completion");
+    let connections = server.connections();
+    assert_eq!(connections.len(), 2);
+    assert_eq!(
+        connections[1][0].body_json()["previous_response_id"],
+        Value::Null
+    );
+    codex.shutdown_and_wait().await?;
+    assert!(
+        server
+            .wait_for_closed_connections(/*expected*/ 2, std::time::Duration::from_secs(5))
+            .await
+    );
+    assert_eq!(
+        connections[0].len(),
+        if interrupt_on_cancel { 3 } else { 2 }
+    );
+    server.shutdown().await;
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn steer_during_stream_retry_skips_backoff() {
     let server = responses::start_mock_server().await;

codex-rs/features/src/lib.rs

@@ -253,6 +253,8 @@ pub enum Feature {
     WakeOnAgentMessages,
     /// Preempt responses and yield foreground code-mode observations on new user input.
     InstantInterrupt,
+    /// Interrupt upstream inference when a turn is cancelled.
+    InterruptOnCancel,
     /// Enable shared discussion tools for an agent tree.
     AgentMessageBoard,
     /// Removed compatibility flag retained as a no-op.
@@ -1293,6 +1295,12 @@ pub const FEATURES: &[FeatureSpec] = &[
         stage: Stage::UnderDevelopment,
         default_enabled: true,
     },
+    FeatureSpec {
+        id: Feature::InterruptOnCancel,
+        key: "interrupt_on_cancel",
+        stage: Stage::UnderDevelopment,
+        default_enabled: false,
+    },
     FeatureSpec {
         id: Feature::CodeModeOnly,
         key: "code_mode_only",