app-server: bound deferred cleanup state

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<()> {

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<ThreadId, JSONRPCError
     ThreadId::from_string(thread_id)
         .map_err(|err| invalid_request(format!("invalid thread id: {err}")))
 }
+
+#[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<OutgoingEnvelope>) -> 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(())
+    }
+}

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<OutgoingEnvelope>) -> 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(())
+    }
+}