diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 4ac1ee62f5..a005541894 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -1217,9 +1217,9 @@ pub(crate) async fn apply_bespoke_event_handling( )) .await; if thread_goal_event.goal.status != codex_protocol::protocol::ThreadGoalStatus::Active { - emit_terminal_plan_cleanup( + flush_pending_terminal_plan_cleanup( thread_goal_event.thread_id, - take_pending_terminal_plan_cleanup(&thread_state).await, + &thread_state, &outgoing, ) .await; @@ -1324,11 +1324,12 @@ async fn emit_turn_completed_with_status( .await; } -pub(crate) async fn emit_terminal_plan_cleanup( +pub(crate) async fn flush_pending_terminal_plan_cleanup( conversation_id: ThreadId, - pending_terminal_plan_cleanups: Vec, + thread_state: &Arc>, outgoing: &ThreadScopedOutgoingMessageSender, ) { + let pending_terminal_plan_cleanups = take_pending_terminal_plan_cleanup(thread_state).await; for (turn_id, latest_plan_update) in terminal_plan_cleanup_updates(pending_terminal_plan_cleanups) { @@ -1368,20 +1369,8 @@ async fn emit_turn_plan_updated( plan_update_event: UpdatePlanArgs, outgoing: &ThreadScopedOutgoingMessageSender, ) { - let notification = - turn_plan_updated_notification(conversation_id, event_turn_id, plan_update_event); - outgoing - .send_server_notification(ServerNotification::TurnPlanUpdated(notification)) - .await; -} - -fn turn_plan_updated_notification( - conversation_id: ThreadId, - event_turn_id: &str, - plan_update_event: UpdatePlanArgs, -) -> TurnPlanUpdatedNotification { // `update_plan` is a todo/checklist tool; it is not related to plan-mode updates. - TurnPlanUpdatedNotification { + let notification = TurnPlanUpdatedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.to_string(), explanation: plan_update_event.explanation, @@ -1390,7 +1379,10 @@ fn turn_plan_updated_notification( .into_iter() .map(TurnPlanStep::from) .collect(), - } + }; + outgoing + .send_server_notification(ServerNotification::TurnPlanUpdated(notification)) + .await; } async fn terminal_plan_cleanup_may_be_needed(thread_state: &Arc>) -> bool { @@ -1401,7 +1393,7 @@ async fn terminal_plan_cleanup_may_be_needed(thread_state: &Arc>, ) -> Vec { std::mem::take(&mut thread_state.lock().await.pending_terminal_plan_cleanups) @@ -1592,12 +1584,7 @@ async fn handle_turn_complete( ) { let turn_summary = find_and_remove_turn_summary(thread_state).await; if !preserve_terminal_plan_progress { - emit_terminal_plan_cleanup( - conversation_id, - take_pending_terminal_plan_cleanup(thread_state).await, - outgoing, - ) - .await; + flush_pending_terminal_plan_cleanup(conversation_id, thread_state, outgoing).await; } let (status, error) = match turn_summary.last_error { @@ -1630,12 +1617,7 @@ async fn handle_turn_interrupted( ) { let turn_summary = find_and_remove_turn_summary(thread_state).await; if !preserve_terminal_plan_progress { - emit_terminal_plan_cleanup( - conversation_id, - take_pending_terminal_plan_cleanup(thread_state).await, - outgoing, - ) - .await; + flush_pending_terminal_plan_cleanup(conversation_id, thread_state, outgoing).await; } emit_turn_completed_with_status( 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 2cef82bc94..e7f901c79a 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 @@ -464,9 +464,9 @@ impl ThreadGoalRequestProcessor { subscribed_connection_ids, thread_id, ); - crate::bespoke_event_handling::emit_terminal_plan_cleanup( + crate::bespoke_event_handling::flush_pending_terminal_plan_cleanup( thread_id, - crate::bespoke_event_handling::take_pending_terminal_plan_cleanup(&thread_state).await, + &thread_state, &thread_outgoing, ) .await; 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 0f74d579c1..c76820bb97 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -484,10 +484,9 @@ pub(super) async fn handle_thread_listener_command( subscribed_connection_ids, conversation_id, ); - crate::bespoke_event_handling::emit_terminal_plan_cleanup( + crate::bespoke_event_handling::flush_pending_terminal_plan_cleanup( conversation_id, - crate::bespoke_event_handling::take_pending_terminal_plan_cleanup(thread_state) - .await, + thread_state, &thread_outgoing, ) .await; @@ -509,10 +508,9 @@ pub(super) async fn handle_thread_listener_command( subscribed_connection_ids, conversation_id, ); - crate::bespoke_event_handling::emit_terminal_plan_cleanup( + crate::bespoke_event_handling::flush_pending_terminal_plan_cleanup( conversation_id, - crate::bespoke_event_handling::take_pending_terminal_plan_cleanup(thread_state) - .await, + thread_state, &thread_outgoing, ) .await;