turn interrupt

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

@@ -405,6 +405,7 @@ pub struct TurnStartResponse {
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
 pub struct TurnInterruptParams {
+    pub thread_id: String,
     pub turn_id: String,
 }
 

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

@@ -161,11 +161,17 @@ pub(crate) struct CodexMessageProcessor {
     subscription_to_conversation: HashMap<Uuid, ConversationId>,
     active_login: Arc<Mutex<Option<ActiveLogin>>>,
     // Queue of pending interrupt requests per conversation. We reply when TurnAborted arrives.
-    pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<RequestId>>>>,
+    pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<(RequestId, ApiVersion)>>>>,
     pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
     feedback: CodexFeedback,
 }
 
+#[derive(Clone, Copy, Debug)]
+enum ApiVersion {
+    V1,
+    V2,
+}
+
 impl CodexMessageProcessor {
     pub fn new(
         auth_manager: Arc<AuthManager>,
@@ -219,12 +225,8 @@ impl CodexMessageProcessor {
             ClientRequest::TurnStart { request_id, params } => {
                 self.turn_start(request_id, params).await;
             }
-            ClientRequest::TurnInterrupt {
-                request_id,
-                params: _,
-            } => {
-                self.send_unimplemented_error(request_id, "turn/interrupt")
-                    .await;
+            ClientRequest::TurnInterrupt { request_id, params } => {
+                self.turn_interrupt(request_id, params).await;
             }
             ClientRequest::NewConversation { request_id, params } => {
                 // Do not tokio::spawn() to process new_conversation()
@@ -1939,7 +1941,56 @@ impl CodexMessageProcessor {
         // Record the pending interrupt so we can reply when TurnAborted arrives.
         {
             let mut map = self.pending_interrupts.lock().await;
-            map.entry(conversation_id).or_default().push(request_id);
+            map.entry(conversation_id)
+                .or_default()
+                .push((request_id, ApiVersion::V1));
+        }
+
+        // Submit the interrupt; we'll respond upon TurnAborted.
+        let _ = conversation.submit(Op::Interrupt).await;
+    }
+
+    async fn turn_interrupt(
+        &mut self,
+        request_id: RequestId,
+        params: codex_app_server_protocol::TurnInterruptParams,
+    ) {
+        let codex_app_server_protocol::TurnInterruptParams { thread_id, .. } = params;
+
+        // Resolve conversation id from v2 thread id string.
+        let conversation_id = match ConversationId::from_string(&thread_id) {
+            Ok(id) => id,
+            Err(err) => {
+                let error = JSONRPCErrorError {
+                    code: INVALID_REQUEST_ERROR_CODE,
+                    message: format!("invalid thread id: {err}"),
+                    data: None,
+                };
+                self.outgoing.send_error(request_id, error).await;
+                return;
+            }
+        };
+
+        let Ok(conversation) = self
+            .conversation_manager
+            .get_conversation(conversation_id)
+            .await
+        else {
+            let error = JSONRPCErrorError {
+                code: INVALID_REQUEST_ERROR_CODE,
+                message: format!("conversation not found: {conversation_id}"),
+                data: None,
+            };
+            self.outgoing.send_error(request_id, error).await;
+            return;
+        };
+
+        // Record the pending interrupt so we can reply when TurnAborted arrives.
+        {
+            let mut map = self.pending_interrupts.lock().await;
+            map.entry(conversation_id)
+                .or_default()
+                .push((request_id, ApiVersion::V2));
         }
 
         // Submit the interrupt; we'll respond upon TurnAborted.
@@ -2085,12 +2136,13 @@ impl CodexMessageProcessor {
                 let _ = sender.send(());
                 // Clean reverse mappings.
                 if let Some(conv_id) = self.subscription_to_conversation.remove(&subscription_id)
-                    && let Some(list) = self.conversation_listener_index.get_mut(&conv_id) {
-                        list.retain(|id| id != &subscription_id);
-                        if list.is_empty() {
-                            self.conversation_listener_index.remove(&conv_id);
-                        }
+                    && let Some(list) = self.conversation_listener_index.get_mut(&conv_id)
+                {
+                    list.retain(|id| id != &subscription_id);
+                    if list.is_empty() {
+                        self.conversation_listener_index.remove(&conv_id);
                     }
+                }
                 let response = RemoveConversationSubscriptionResponse {};
                 self.outgoing.send_response(request_id, response).await;
             }
@@ -2243,7 +2295,7 @@ async fn apply_bespoke_event_handling(
     conversation_id: ConversationId,
     conversation: Arc<CodexConversation>,
     outgoing: Arc<OutgoingMessageSender>,
-    pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<RequestId>>>>,
+    pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<(RequestId, ApiVersion)>>>>,
 ) {
     let Event { id: event_id, msg } = event;
     match msg {
@@ -2313,11 +2365,19 @@ async fn apply_bespoke_event_handling(
                 map.remove(&conversation_id).unwrap_or_default()
             };
             if !pending.is_empty() {
-                let response = InterruptConversationResponse {
-                    abort_reason: turn_aborted_event.reason,
-                };
-                for rid in pending {
-                    outgoing.send_response(rid, response.clone()).await;
+                for (rid, ver) in pending {
+                    match ver {
+                        ApiVersion::V1 => {
+                            let response = InterruptConversationResponse {
+                                abort_reason: turn_aborted_event.reason.clone(),
+                            };
+                            outgoing.send_response(rid, response).await;
+                        }
+                        ApiVersion::V2 => {
+                            let response = codex_app_server_protocol::TurnInterruptResponse {};
+                            outgoing.send_response(rid, response).await;
+                        }
+                    }
                 }
             }
         }

codex-rs/app-server/tests/common/mcp_process.rs

@@ -35,6 +35,7 @@ use codex_app_server_protocol::ThreadArchiveParams;
 use codex_app_server_protocol::ThreadListParams;
 use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::TurnInterruptParams as V2TurnInterruptParams;
 use codex_app_server_protocol::TurnStartParams as V2TurnStartParams;
 
 use codex_app_server_protocol::JSONRPCError;
@@ -366,6 +367,15 @@ impl McpProcess {
         self.send_request("turn/start", params).await
     }
 
+    /// Send a `turn/interrupt` JSON-RPC request (v2).
+    pub async fn send_turn_interrupt_request(
+        &mut self,
+        params: V2TurnInterruptParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("turn/interrupt", params).await
+    }
+
     /// Send a `cancelLoginChatGpt` JSON-RPC request.
     pub async fn send_cancel_login_chat_gpt_request(
         &mut self,

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

@@ -2,4 +2,5 @@ mod thread_archive;
 mod thread_list;
 mod thread_resume;
 mod thread_start;
+mod turn_interrupt;
 mod turn_start;

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

@@ -0,0 +1,134 @@
+#![cfg(unix)]
+
+use anyhow::Result;
+use app_test_support::McpProcess;
+use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_shell_sse_response;
+use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnInterruptParams;
+use codex_app_server_protocol::TurnInterruptResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::TurnStartResponse;
+use codex_app_server_protocol::UserInput as V2UserInput;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn turn_interrupt_aborts_running_turn() -> Result<()> {
+    // Use a portable sleep command to keep the turn running.
+    #[cfg(target_os = "windows")]
+    let shell_command = vec![
+        "powershell".to_string(),
+        "-Command".to_string(),
+        "Start-Sleep -Seconds 10".to_string(),
+    ];
+    #[cfg(not(target_os = "windows"))]
+    let shell_command = vec!["sleep".to_string(), "10".to_string()];
+
+    let tmp = TempDir::new()?;
+    let codex_home = tmp.path().join("codex_home");
+    std::fs::create_dir(&codex_home)?;
+    let working_directory = tmp.path().join("workdir");
+    std::fs::create_dir(&working_directory)?;
+
+    // Mock server: long-running shell command then (after abort) nothing else needed.
+    let server = create_mock_chat_completions_server(vec![create_shell_sse_response(
+        shell_command.clone(),
+        Some(&working_directory),
+        Some(10_000),
+        "call_sleep",
+    )?])
+    .await;
+    create_config_toml(&codex_home, &server.uri())?;
+
+    let mut mcp = McpProcess::new(&codex_home).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    // Start a v2 thread and capture its id.
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread } = to_response::<ThreadStartResponse>(thread_resp)?;
+
+    // Start a turn that triggers a long-running command.
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: "run sleep".to_string(),
+            }],
+            cwd: Some(working_directory.clone()),
+            approval_policy: None,
+            sandbox_policy: None,
+            model: None,
+            effort: None,
+            summary: None,
+        })
+        .await?;
+    let turn_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+    let TurnStartResponse { turn } = to_response::<TurnStartResponse>(turn_resp)?;
+
+    // Give the command a brief moment to start.
+    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+    // Interrupt the in-progress turn by id (v2 API).
+    let interrupt_id = mcp
+        .send_turn_interrupt_request(TurnInterruptParams {
+            thread_id: thread.id,
+            turn_id: turn.id,
+        })
+        .await?;
+    let interrupt_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(interrupt_id)),
+    )
+    .await??;
+    let _resp: TurnInterruptResponse = to_response::<TurnInterruptResponse>(interrupt_resp)?;
+
+    // No fields to assert on; successful deserialization confirms proper response shape.
+
+    drop(server);
+    Ok(())
+}
+
+// Helper to create a config.toml pointing at the mock model server.
+fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> {
+    let config_toml = codex_home.join("config.toml");
+    std::fs::write(
+        config_toml,
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "never"
+sandbox_mode = "danger-full-access"
+
+model_provider = "mock_provider"
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{server_uri}/v1"
+wire_api = "chat"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )
+}