diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 4c7ea92a77..5a2d0584e8 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread_sample.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread_sample.rs index e02dcab80f..adb65b8a2c 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread_sample.rs +++ b/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, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] diff --git a/codex-rs/app-server/src/thread_sample.rs b/codex-rs/app-server/src/thread_sample.rs index 57c4cc1b0d..fe96b92cf3 100644 --- a/codex-rs/app-server/src/thread_sample.rs +++ b/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); diff --git a/codex-rs/app-server/tests/suite/v2/thread_sample.rs b/codex-rs/app-server/tests/suite/v2/thread_sample.rs index b01dd5d5f0..0d72d343cf 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_sample.rs +++ b/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([ diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index f43e285d00..ce32a374c1 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/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)) => { diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 1da9d7ba0c..62b87d2e08 100644 --- a/codex-rs/core/config.schema.json +++ b/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" }, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index e4a092ed65..dbc78f4046 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/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> = 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 { diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 672d960660..13b446e291 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/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::>(), - 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::>(), + 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; } } diff --git a/codex-rs/core/tests/suite/model_request.rs b/codex-rs/core/tests/suite/model_request.rs index 3163db9886..c72f7f884c 100644 --- a/codex-rs/core/tests/suite/model_request.rs +++ b/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(()) } diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index 2aa2c35ec3..30fbb97704 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/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; diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 9015a20c35..f76b7118fa 100644 --- a/codex-rs/features/src/lib.rs +++ b/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",