diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 087d9165cb..410620324b 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -1283,6 +1283,7 @@ async fn handle_turn_plan_update( }); } } + flush_pending_terminal_plan_cleanup(conversation_id, thread_state, outgoing).await; emit_turn_plan_updated(conversation_id, event_turn_id, plan_update_event, outgoing).await; } @@ -3640,6 +3641,88 @@ mod tests { Ok(()) } + #[tokio::test] + async fn new_live_plan_flushes_older_deferred_cleanup() -> Result<()> { + let conversation_id = ThreadId::new(); + let active_turn_id = "active-plan-turn".to_string(); + let thread_state = new_thread_state(); + { + let mut state = thread_state.lock().await; + state.track_current_turn_event( + &active_turn_id, + &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { + turn_id: active_turn_id.clone(), + started_at: Some(42), + model_context_window: None, + collaboration_mode_kind: Default::default(), + }), + ); + state.pending_terminal_plan_cleanups = vec![PendingTerminalPlanCleanup { + turn_id: "older-terminal-turn".to_string(), + plan_update: UpdatePlanArgs { + explanation: Some("still working".to_string()), + plan: vec![PlanItemArg { + step: "older".to_string(), + status: StepStatus::InProgress, + }], + }, + }]; + } + let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )); + let outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + vec![ConnectionId(1)], + conversation_id, + ); + + handle_turn_plan_update( + conversation_id, + &active_turn_id, + UpdatePlanArgs { + explanation: Some("next turn".to_string()), + plan: vec![PlanItemArg { + step: "new".to_string(), + status: StepStatus::InProgress, + }], + }, + &outgoing, + &thread_state, + ) + .await; + + let first = recv_broadcast_message(&mut rx).await?; + match first { + OutgoingMessage::AppServerNotification(ServerNotification::TurnPlanUpdated(n)) => { + assert_eq!(n.turn_id, "older-terminal-turn"); + assert_eq!(n.plan[0].status, TurnPlanStepStatus::Pending); + } + other => bail!("unexpected message: {other:?}"), + } + + let second = recv_broadcast_message(&mut rx).await?; + match second { + OutgoingMessage::AppServerNotification(ServerNotification::TurnPlanUpdated(n)) => { + assert_eq!(n.turn_id, active_turn_id); + assert_eq!(n.plan[0].status, TurnPlanStepStatus::InProgress); + } + other => bail!("unexpected message: {other:?}"), + } + + let pending_terminal_plan_cleanups = + &thread_state.lock().await.pending_terminal_plan_cleanups; + assert_eq!(pending_terminal_plan_cleanups.len(), 1); + assert_eq!( + pending_terminal_plan_cleanups[0].turn_id, + "active-plan-turn" + ); + assert!(rx.try_recv().is_err(), "no extra messages expected"); + Ok(()) + } + #[tokio::test] async fn test_handle_turn_complete_downgrades_in_progress_plan_without_active_goal() -> Result<()> { diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index e7f901c79a..121fd05741 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -521,3 +521,92 @@ fn parse_thread_id_for_request(thread_id: &str) -> Result) -> OutgoingMessage { + match rx.recv().await.expect("expected outgoing message") { + OutgoingEnvelope::Broadcast { message } + | OutgoingEnvelope::ToConnection { message, .. } => message, + } + } + + #[tokio::test] + async fn goal_clear_fallback_flushes_pending_terminal_plan_cleanup() -> anyhow::Result<()> { + let codex_home = TempDir::new()?; + let config = load_default_config_for_test(&codex_home).await; + let thread_manager = Arc::new( + codex_core::test_support::thread_manager_with_models_provider_and_home( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + ), + ); + let thread_id = ThreadId::new(); + let thread_state_manager = ThreadStateManager::new(); + let connection_id = ConnectionId(1); + thread_state_manager + .connection_initialized(connection_id, ConnectionCapabilities::default()) + .await; + thread_state_manager + .try_ensure_connection_subscribed( + thread_id, + connection_id, + /*experimental_raw_events*/ false, + ) + .await + .expect("connection should subscribe"); + let thread_state = thread_state_manager.thread_state(thread_id).await; + thread_state.lock().await.pending_terminal_plan_cleanups = + vec![PendingTerminalPlanCleanup { + turn_id: "terminal-turn".to_string(), + plan_update: UpdatePlanArgs { + explanation: Some("still working".to_string()), + plan: vec![PlanItemArg { + step: "first".to_string(), + status: StepStatus::InProgress, + }], + }, + }]; + let (tx, mut rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )); + let processor = ThreadGoalRequestProcessor::new( + thread_manager, + outgoing, + Arc::new(config), + thread_state_manager, + /*state_db*/ None, + ); + + processor + .emit_thread_goal_cleared_ordered(thread_id, /*listener_command_tx*/ None) + .await; + + assert!(matches!( + recv_message(&mut rx).await, + OutgoingMessage::AppServerNotification(ServerNotification::ThreadGoalCleared(_)) + )); + assert!(matches!( + recv_message(&mut rx).await, + OutgoingMessage::AppServerNotification(ServerNotification::TurnPlanUpdated(_)) + )); + Ok(()) + } +} diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index c76820bb97..f0c3dc49bc 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -794,3 +794,98 @@ pub(super) fn set_thread_status_and_interrupt_stale_turns( } thread.status = status; } + +#[cfg(test)] +mod tests { + use super::*; + use crate::outgoing_message::ConnectionId; + use crate::outgoing_message::OutgoingEnvelope; + use crate::outgoing_message::OutgoingMessage; + use crate::thread_state::ConnectionCapabilities; + use crate::thread_state::PendingTerminalPlanCleanup; + use codex_protocol::plan_tool::PlanItemArg; + use codex_protocol::plan_tool::StepStatus; + use codex_protocol::plan_tool::UpdatePlanArgs; + use core_test_support::load_default_config_for_test; + use tempfile::TempDir; + use tokio::sync::mpsc; + + async fn recv_message(rx: &mut mpsc::Receiver) -> OutgoingMessage { + match rx.recv().await.expect("expected outgoing message") { + OutgoingEnvelope::Broadcast { message } + | OutgoingEnvelope::ToConnection { message, .. } => message, + } + } + + #[tokio::test] + async fn listener_goal_clear_flushes_pending_terminal_plan_cleanup() -> anyhow::Result<()> { + let codex_home = TempDir::new()?; + let config = load_default_config_for_test(&codex_home).await; + let thread_manager = Arc::new( + codex_core::test_support::thread_manager_with_models_provider_and_home( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + ), + ); + let codex_core::NewThread { + thread_id, + thread: conversation, + .. + } = thread_manager.start_thread(config).await?; + let thread_state_manager = ThreadStateManager::new(); + let connection_id = ConnectionId(1); + thread_state_manager + .connection_initialized(connection_id, ConnectionCapabilities::default()) + .await; + thread_state_manager + .try_ensure_connection_subscribed( + thread_id, + connection_id, + /*experimental_raw_events*/ false, + ) + .await + .expect("connection should subscribe"); + let thread_state = thread_state_manager.thread_state(thread_id).await; + thread_state.lock().await.pending_terminal_plan_cleanups = + vec![PendingTerminalPlanCleanup { + turn_id: "terminal-turn".to_string(), + plan_update: UpdatePlanArgs { + explanation: Some("still working".to_string()), + plan: vec![PlanItemArg { + step: "first".to_string(), + status: StepStatus::InProgress, + }], + }, + }]; + let (tx, mut rx) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )); + + handle_thread_listener_command( + thread_id, + &conversation, + codex_home.path(), + &thread_state_manager, + &thread_state, + &ThreadWatchManager::new(), + &outgoing, + &Arc::new(Mutex::new(HashSet::new())), + ThreadListenerCommand::EmitThreadGoalCleared, + ) + .await; + + assert!(matches!( + recv_message(&mut rx).await, + OutgoingMessage::AppServerNotification(ServerNotification::ThreadGoalCleared(_)) + )); + assert!(matches!( + recv_message(&mut rx).await, + OutgoingMessage::AppServerNotification(ServerNotification::TurnPlanUpdated(_)) + )); + Ok(()) + } +}