diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index e986d2e574..c88258fa56 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -881,7 +881,7 @@ pub(crate) async fn apply_bespoke_event_handling( Some(turn_id) => Some(turn_id), None => { let state = thread_state.lock().await; - state.active_turn_snapshot().map(|turn| turn.id) + state.active_turn_id().map(str::to_owned) } }; let server_name = request.server_name.clone(); diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 37aca5dfe9..df3b049d81 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1607,11 +1607,10 @@ impl TurnRequestProcessor { let is_running = matches!(thread.agent_status().await, AgentStatus::Running); { let mut thread_state = thread_state.lock().await; - if let Some(active_turn) = thread_state.active_turn_snapshot() { - if active_turn.id != turn_id { + if let Some(active_turn_id) = thread_state.active_turn_id() { + if active_turn_id != turn_id { return Err(invalid_request(format!( - "expected active turn id {turn_id} but found {}", - active_turn.id + "expected active turn id {turn_id} but found {active_turn_id}" ))); } } else if thread_state.last_terminal_turn_id.as_deref() == Some(turn_id.as_str()) diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 14012f3f66..d80ce76982 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -165,6 +165,11 @@ impl ThreadState { self.current_turn_history.active_turn_snapshot() } + /// Returns the same turn ID as `active_turn_snapshot` without cloning its items. + pub(crate) fn active_turn_id(&self) -> Option<&str> { + self.current_turn_history.active_turn_id() + } + pub(crate) fn register_shutdown_drain_waiter(&mut self) -> oneshot::Receiver<()> { let (completion_tx, completion_rx) = oneshot::channel(); self.shutdown_drain_waiter = Some(completion_tx); @@ -459,7 +464,7 @@ impl ThreadStateManager { thread_id = %thread_id, listener_generation = thread_state.listener_generation, had_listener = thread_state.cancel_tx.is_some(), - had_active_turn = thread_state.active_turn_snapshot().is_some(), + had_active_turn = thread_state.active_turn_id().is_some(), "clearing thread listener during thread-state teardown" ); thread_state.clear_listener(); @@ -483,7 +488,7 @@ impl ThreadStateManager { thread_id = %thread_id, listener_generation = thread_state.listener_generation, had_listener = thread_state.cancel_tx.is_some(), - had_active_turn = thread_state.active_turn_snapshot().is_some(), + had_active_turn = thread_state.active_turn_id().is_some(), "clearing thread listener during app-server shutdown" ); thread_state.clear_listener(); diff --git a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs index 02e2762292..bdc1740778 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs @@ -9,6 +9,7 @@ use app_test_support::create_mock_responses_server_sequence; use app_test_support::create_mock_responses_server_sequence_unchecked; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::JSONRPCError; +use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequestResolvedNotification; @@ -19,9 +20,11 @@ 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::TurnStartedNotification; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput as V2UserInput; use core_test_support::skip_if_remote; +use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::time::timeout; @@ -98,10 +101,32 @@ async fn turn_interrupt_aborts_running_turn() -> Result<()> { .await?; let turn_id = turn.id.clone(); - // Give the command a brief moment to start. - tokio::time::sleep(std::time::Duration::from_secs(1)).await; + let started: TurnStartedNotification = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("turn/started")).await??; + assert_eq!(started.thread_id, thread.id); + assert_eq!(started.turn.id, turn_id); let thread_id = thread.id.clone(); + let interrupt_id = mcp + .send_turn_interrupt_request(TurnInterruptParams { + thread_id: thread_id.clone(), + turn_id: "wrong-turn".to_string(), + }) + .await?; + let interrupt_err = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(interrupt_id)), + ) + .await??; + assert_eq!( + interrupt_err.error, + JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("expected active turn id wrong-turn but found {turn_id}"), + data: None, + } + ); + // Interrupt the in-progress turn by id (v2 API). let _: TurnInterruptResponse = mcp .request(|request_id| ClientRequest::TurnInterrupt {