From fa5fa861ef698d8d15fb5cacb9e0cf5831f9267e Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sun, 2 Nov 2025 16:36:14 -0800 Subject: [PATCH] turn interrupt --- .../app-server-protocol/src/protocol/v2.rs | 1 + .../app-server/src/codex_message_processor.rs | 98 ++++++++++--- .../app-server/tests/common/mcp_process.rs | 10 ++ codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../tests/suite/v2/turn_interrupt.rs | 134 ++++++++++++++++++ 5 files changed, 225 insertions(+), 19 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/turn_interrupt.rs diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 209c184c8c..9ba53b041d 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/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, } diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index a76545b10d..c8fc2b852e 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -161,11 +161,17 @@ pub(crate) struct CodexMessageProcessor { subscription_to_conversation: HashMap, active_login: Arc>>, // Queue of pending interrupt requests per conversation. We reply when TurnAborted arrives. - pending_interrupts: Arc>>>, + pending_interrupts: Arc>>>, pending_fuzzy_searches: Arc>>>, feedback: CodexFeedback, } +#[derive(Clone, Copy, Debug)] +enum ApiVersion { + V1, + V2, +} + impl CodexMessageProcessor { pub fn new( auth_manager: Arc, @@ -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, outgoing: Arc, - pending_interrupts: Arc>>>, + pending_interrupts: Arc>>>, ) { 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; + } + } } } } diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index aa78e8286f..d4b376d823 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/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 { + 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, diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 87af7aaf3b..2be6824d26 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/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; diff --git a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs new file mode 100644 index 0000000000..971d5a4a6e --- /dev/null +++ b/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::(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::(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::(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 +"# + ), + ) +}