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
This commit is contained in:
jif
2026-11-03 14:15:56 +00:00
committed by copyberry
parent bfdbf01bfa
commit f7167ac895
7 changed files with 60 additions and 48 deletions
-1
View File
@@ -2953,7 +2953,6 @@ dependencies = [
"bm25",
"chrono",
"codex-agent-graph-store",
"codex-agent-message-board-extension",
"codex-agent-roles",
"codex-analytics",
"codex-api",
@@ -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::<Vec<_>>();
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<ThreadManager>,
+1
View File
@@ -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;
+18 -13
View File
@@ -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::<Vec<_>>();
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::<Vec<_>>();
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<Arc<dyn QueueStore>> = match &config.experimental_thread_store {
+1 -2
View File
@@ -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 }
+5 -28
View File
@@ -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<ThreadId>) -> 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<StateDbHandle>,
@@ -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::<Vec<_>>();
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?;
}
+11 -4
View File
@@ -844,10 +844,17 @@ impl TestCodexBuilder {
environment_manager: Arc<codex_exec_server::EnvironmentManager>,
) -> anyhow::Result<TestCodex> {
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(|| {