From f7167ac895c869c60ae00a2d7f518a9977cd1205 Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 3 Nov 2026 14:02:48 +0000 Subject: [PATCH] Move agent message board cleanup into the app server (#59167) ## What changed Move permanent agent message board deletion from `codex-core` into the app server's thread-store cleanup callback. Keep board cleanup active when the board feature is disabled, and run it before optional Guardian evidence-board cleanup. Remove the message board extension dependency from `codex-core` and retain `codex-message-board` only as a development dependency. Update the core test harness to install the app server's cleanup callback. GitOrigin-RevId: 61813bbdcd8b57073b8c249e1f771e3fb0ea061e --- codex-rs/Cargo.lock | 1 - .../app-server/src/agent_message_board.rs | 24 ++++++++++++++ codex-rs/app-server/src/lib.rs | 1 + codex-rs/app-server/src/message_processor.rs | 31 +++++++++-------- codex-rs/core/Cargo.toml | 3 +- codex-rs/core/src/thread_store_builder.rs | 33 +++---------------- codex-rs/core/tests/common/test_codex.rs | 15 ++++++--- 7 files changed, 60 insertions(+), 48 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 0384f36a25..6ff6fbc902 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2953,7 +2953,6 @@ dependencies = [ "bm25", "chrono", "codex-agent-graph-store", - "codex-agent-message-board-extension", "codex-agent-roles", "codex-analytics", "codex-api", diff --git a/codex-rs/app-server/src/agent_message_board.rs b/codex-rs/app-server/src/agent_message_board.rs index 6b9054e6fb..3399a5adb1 100644 --- a/codex-rs/app-server/src/agent_message_board.rs +++ b/codex-rs/app-server/src/agent_message_board.rs @@ -3,11 +3,13 @@ //! The extension owns backend construction, storage and tool registration. This adapter //! selects configuration and supplies host capabilities. It never starts or restores //! recipients during lookup. Notifications use ordinary agent-message scheduling. +//! Permanent board deletion uses the thread store's existing host cleanup callback. use chrono::DateTime; use chrono::Utc; use codex_agent_message_board_client::RemoteAgentMessageBoard; use codex_agent_message_board_extension::BoardBackend; +use codex_agent_message_board_extension::local_board_options; use codex_core::CodexThread; use codex_core::MULTI_AGENT_V2_NAMESPACE_DESCRIPTION; use codex_core::ThreadManager; @@ -15,6 +17,7 @@ use codex_core::config::Config; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ThreadStartInput; use codex_features::Feature; +use codex_message_board::LocalAgentMessageBoard; use codex_message_board::LocalBoardConnection; use codex_message_board::MessageBoardHost; use codex_message_board::NotificationDelivery; @@ -27,6 +30,9 @@ use codex_protocol::error::CodexErrorDetails; use codex_protocol::error::Result; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::Op; +use codex_state::SqliteConfig; +use codex_thread_store::ThreadStoreError; +use codex_thread_store::ThreadStoreResult; use futures::future::BoxFuture; use std::sync::Arc; use std::sync::Weak; @@ -98,6 +104,24 @@ pub fn install_agent_message_board( ); } +/// Removes persisted agent-board data through the thread store's deletion callback. +/// Cleanup also applies when the board feature is currently disabled. +pub async fn delete_agent_message_boards( + sqlite: &SqliteConfig, + thread_ids: &[ThreadId], +) -> ThreadStoreResult<()> { + let boards = thread_ids + .iter() + .copied() + .map(Into::into) + .collect::>(); + LocalAgentMessageBoard::delete_boards(sqlite, &local_board_options(), &boards) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to delete agent message boards: {err}"), + }) +} + #[derive(Clone)] struct LocalBoardHost { manager: Weak, diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 4eee045679..a03f5888b4 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -96,6 +96,7 @@ fn is_unsupported_untrusted_approval_policy_error(err: &std::io::Error) -> bool } mod agent_message_board; +pub use agent_message_board::delete_agent_message_boards; pub use agent_message_board::install_agent_message_board; mod analytics_utils; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 52e788e3ce..6755a8483a 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -319,22 +319,27 @@ impl MessageProcessor { // The thread store is intentionally process-scoped. Config reloads can // affect per-thread behavior, but they must not move newly started, // resumed, or forked threads to a different persistence backend/root. - let mut thread_store = ThreadStoreBuilder::from_config(config.as_ref(), state_db.clone()); - if config.features.enabled(Feature::GuardianV3) { - let sqlite = config.sqlite_config().clone(); - thread_store = thread_store.with_additional_cleanup(move |thread_ids| { + let sqlite = config.sqlite_config().clone(); + let guardian_v3_enabled = config.features.enabled(Feature::GuardianV3); + let thread_store = ThreadStoreBuilder::from_config(config.as_ref(), state_db.clone()) + .with_additional_cleanup(move |thread_ids| { let sqlite = sqlite.clone(); Box::pin(async move { - let boards = thread_ids.into_iter().map(Into::into).collect::>(); - codex_guardian_v3::delete_local_boards(&sqlite, &boards) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to delete Guardian evidence boards: {err}"), - }) + crate::delete_agent_message_boards(&sqlite, &thread_ids).await?; + if guardian_v3_enabled { + let boards = thread_ids.into_iter().map(Into::into).collect::>(); + codex_guardian_v3::delete_local_boards(&sqlite, &boards) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to delete Guardian evidence boards: {err}" + ), + })?; + } + Ok(()) }) - }); - } - let thread_store = thread_store.build(); + }) + .build(); // Queue persistence requires SQLite, so in-memory thread stores and // app servers without a state database do not have a queue backend. let queue_store: Option> = match &config.experimental_thread_store { diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index f456872236..30d92564c2 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -70,8 +70,6 @@ codex-sandboxing = { workspace = true } codex-skills = { workspace = true } codex-skills-extension = { workspace = true } codex-state = { workspace = true } -codex-agent-message-board-extension = { workspace = true } -codex-message-board = { workspace = true } codex-terminal-detection = { workspace = true } codex-thread-store = { workspace = true } codex-tools = { workspace = true } @@ -145,6 +143,7 @@ openssl-sys = { workspace = true, features = ["vendored"] } openssl-sys = { workspace = true, features = ["vendored"] } [dev-dependencies] +codex-message-board = { workspace = true } codex-app-server = { workspace = true } assert_cmd = { workspace = true } assert_matches = { workspace = true } diff --git a/codex-rs/core/src/thread_store_builder.rs b/codex-rs/core/src/thread_store_builder.rs index 52d6acbade..fb10b8bb1b 100644 --- a/codex-rs/core/src/thread_store_builder.rs +++ b/codex-rs/core/src/thread_store_builder.rs @@ -1,26 +1,23 @@ -//! Configured thread-store construction and permanent-deletion cleanup. -//! Host cleanup runs after agent-board cleanup under the store's existing deletion locks. +//! Configured thread-store construction and host-owned permanent-deletion cleanup. +//! Host cleanup runs under the store's existing deletion locks. use crate::config::Config; use crate::config::ThreadStoreConfig; -use codex_agent_message_board_extension::local_board_options; use codex_features::Feature; -use codex_message_board::LocalAgentMessageBoard; use codex_protocol::ThreadId; use codex_rollout::state_db::StateDbHandle; use codex_thread_store::InMemoryThreadStore; use codex_thread_store::LocalThreadStore; use codex_thread_store::LocalThreadStoreConfig; use codex_thread_store::ThreadStore; -use codex_thread_store::ThreadStoreError; use codex_thread_store::ThreadStoreFuture; use std::sync::Arc; use tracing::warn; type AdditionalCleanup = dyn Fn(Vec) -> ThreadStoreFuture<'static, ()> + Send + Sync; -/// Builds the configured thread store with its existing agent-board cleanup. -/// Hosts can supply additional cleanup before the store is shared with callers. +/// Builds the configured thread store with optional host-owned data cleanup. +/// Hosts install cleanup for their services before sharing the store with callers. pub struct ThreadStoreBuilder<'config> { config: &'config Config, state_db: Option, @@ -37,7 +34,7 @@ impl<'config> ThreadStoreBuilder<'config> { } /// Sets host-owned cleanup for permanent deletion in local stores. - /// Runs after successful agent-board cleanup and before deleting rollouts, under the same locks. + /// Runs before deleting rollouts, under the store's existing deletion locks. /// Cleanup must be idempotent because later deletion steps can fail and be retried. /// /// Cleanup must not call, or await work that calls, thread-store operations that reacquire @@ -69,31 +66,11 @@ impl<'config> ThreadStoreBuilder<'config> { .features .enabled(Feature::BackgroundPaginatedRolloutMigration); let has_state_db = state_db.is_some(); - let sqlite = config.sqlite_config().clone(); let store = Arc::new( LocalThreadStore::new(LocalThreadStoreConfig::from_config(config), state_db) .with_thread_data_cleanup(move |thread_ids| { - let sqlite = sqlite.clone(); let additional_cleanup = additional_cleanup.clone(); Box::pin(async move { - let boards = thread_ids - .iter() - .copied() - .map(Into::into) - .collect::>(); - LocalAgentMessageBoard::delete_boards( - &sqlite, - &local_board_options(), - &boards, - ) - .await - .map_err(|err| { - ThreadStoreError::Internal { - message: format!( - "failed to delete agent message boards: {err}" - ), - } - })?; if let Some(cleanup) = additional_cleanup { cleanup(thread_ids).await?; } diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index 722d2cf514..25a09467ec 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -844,10 +844,17 @@ impl TestCodexBuilder { environment_manager: Arc, ) -> anyhow::Result { let state_db = codex_core::init_state_db(&config).await; - let thread_store = self - .thread_store - .clone() - .unwrap_or_else(|| ThreadStoreBuilder::from_config(&config, state_db.clone()).build()); + let thread_store = self.thread_store.clone().unwrap_or_else(|| { + let sqlite = config.sqlite_config().clone(); + ThreadStoreBuilder::from_config(&config, state_db.clone()) + .with_additional_cleanup(move |thread_ids| { + let sqlite = sqlite.clone(); + Box::pin(async move { + codex_app_server::delete_agent_message_boards(&sqlite, &thread_ids).await + }) + }) + .build() + }); let installation_id = resolve_installation_id(&config.codex_home).await?; let user_instructions_provider = self.user_instructions_provider.clone().unwrap_or_else(|| {