Share SQLite pools across local agent message-board handles (#47049)

## Why

Opening each local message-board handle created a separate SQLite pool for the same database.

## What changed

- Cache one shared pool per canonical database path within the process, using weak references so the last handle can release it.
- Serialize pool initialization with deletion and corruption recovery. Close and evict the cached pool before replacing a corrupt database, and block new handles until deletion tombstones are saved.

## Testing

Enable library unit tests and add coverage for concurrent handles sharing a pool, releasing it after the last handle drops, and creating a new pool on reopen.

GitOrigin-RevId: 4d4a5c9ea9f063733df85a2ac120d51099736520

codex-rs/ext/agent-message-board/Cargo.toml

@@ -8,7 +8,6 @@ version.workspace = true
 name = "codex_agent_message_board_extension"
 path = "src/lib.rs"
 doctest = false
-test = false
 
 [lints]
 workspace = true

codex-rs/ext/agent-message-board/src/local.rs

@@ -1,6 +1,7 @@
 //! SQLite-backed boards. The tree ID scopes every read and write.
 //!
 //! Immediate transactions serialize mutations across independently opened handles.
+//! Live handles in this process share one connection pool per database path.
 //! Accepted posts survive runtime unload and process restart, but cannot recreate
 //! a board after its root has been permanently deleted.
 
@@ -31,19 +32,32 @@ use serde::Serialize;
 use sqlx::Row;
 use sqlx::SqliteConnection;
 use sqlx::SqlitePool;
+use std::collections::HashMap;
 use std::collections::HashSet;
+use std::path::PathBuf;
 use std::sync::Arc;
+use std::sync::LazyLock;
+use std::sync::Weak;
+use tokio::sync::Mutex;
 use uuid::Uuid;
 
 mod lifecycle;
 mod paging;
 mod queries;
 
+#[cfg(test)]
+#[path = "local/pools_tests.rs"]
+mod pools_tests;
+
 const MAX_POST_BYTES: usize = 64 * 1024;
 const MAX_CHANNEL_BYTES: usize = 128;
 const MAX_READ_CHARS: usize = 20_000;
 const DATABASE_FILE: &str = "agent_message_board_1.sqlite";
 
+// Weak entries let the last board handle release its pool. Initialization and
+// recovery share one lock so concurrent starts cannot open duplicate or stale pools.
+static POOLS: LazyLock<Mutex<HashMap<PathBuf, Weak<SqlitePool>>>> = LazyLock::new(Mutex::default);
+
 const SCHEMA: &str = "
 CREATE TABLE IF NOT EXISTS deleted_boards (board TEXT PRIMARY KEY NOT NULL);
 CREATE TABLE IF NOT EXISTS channels (
@@ -70,7 +84,7 @@ CREATE TABLE IF NOT EXISTS subscriptions (
 #[derive(Clone)]
 pub struct LocalAgentMessageBoard {
     identity: SessionId,
-    pool: SqlitePool,
+    pool: Arc<SqlitePool>,
     host: Arc<dyn MessageBoardHost>,
 }
 
@@ -83,20 +97,40 @@ struct StoredPost {
 impl LocalAgentMessageBoard {
     /// Reopens the same board for a root, child or resumed runtime. The shared
     /// SQLite configuration preserves the host's connection and journal policy.
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "pool creation and schema initialization stay serialized to avoid duplicate pools"
+    )]
     pub async fn open(
         sqlite: &SqliteConfig,
         identity: SessionId,
         host: Arc<dyn MessageBoardHost>,
     ) -> Result<Self> {
         tokio::fs::create_dir_all(sqlite.home()).await?;
-        let pool = sqlite
-            .open_read_write_pool(&sqlite.home().join(DATABASE_FILE))
-            .await
-            .map_err(storage_error)?;
-        sqlx::raw_sql(SCHEMA)
-            .execute(&pool)
-            .await
-            .map_err(storage_error)?;
+        let path = tokio::fs::canonicalize(sqlite.home())
+            .await?
+            .join(DATABASE_FILE);
+        let mut pools = POOLS.lock().await;
+        pools.retain(|_, pool| pool.strong_count() > 0);
+        let pool = if let Some(pool) = pools
+            .get(&path)
+            .and_then(Weak::upgrade)
+            .filter(|pool| !pool.is_closed())
+        {
+            pool
+        } else {
+            let pool = sqlite
+                .open_read_write_pool(&path)
+                .await
+                .map_err(storage_error)?;
+            sqlx::raw_sql(SCHEMA)
+                .execute(&pool)
+                .await
+                .map_err(storage_error)?;
+            let pool = Arc::new(pool);
+            pools.insert(path, Arc::downgrade(&pool));
+            pool
+        };
         Ok(Self {
             identity,
             pool,

codex-rs/ext/agent-message-board/src/local/lifecycle.rs

@@ -1,10 +1,12 @@
 //! Permanent board deletion, serialized with writes across all open handles.
 //!
 //! Keep only a root ID tombstone so delayed writes cannot resurrect deleted data.
-//! If corruption prevents opening the database, back it up before recreating it.
+//! Recovery closes the cached pool before backing up and recreating the database,
+//! and excludes new handles until the replacement's deletion tombstones are saved.
 
 use super::DATABASE_FILE;
 use super::LocalAgentMessageBoard;
+use super::POOLS;
 use super::SCHEMA;
 use super::invalid;
 use super::storage_error;
@@ -13,17 +15,26 @@ use codex_protocol::error::Result;
 use codex_state::SqliteConfig;
 use sqlx::Sqlite;
 use sqlx::Transaction;
+use std::sync::Weak;
 
 impl LocalAgentMessageBoard {
     /// Permanently removes boards owned by these roots, including their posts and subscriptions.
     /// A child's ID does not match its parent's board. Unload and archive must not call this.
     /// Safe to retry and independent of whether the feature is currently enabled.
     /// Recovering an unopenable corrupt database resets all boards, retaining a backup.
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "recovery excludes new handles until replacement deletion tombstones are saved"
+    )]
     pub async fn delete_boards(sqlite: &SqliteConfig, roots: &[SessionId]) -> Result<()> {
         let path = sqlite.home().join(DATABASE_FILE);
         if roots.is_empty() || !tokio::fs::try_exists(&path).await? {
             return Ok(());
         }
+        let path = tokio::fs::canonicalize(sqlite.home())
+            .await?
+            .join(DATABASE_FILE);
+        let mut pools = POOLS.lock().await;
         let pool = match sqlite.open_read_write_pool(&path).await {
             Ok(pool) => pool,
             Err(error) => {
@@ -31,6 +42,10 @@ impl LocalAgentMessageBoard {
                 if !codex_state::is_sqlite_corruption_error(&error) {
                     return Err(storage_error(error));
                 }
+                if let Some(pool) = pools.get(&path).and_then(Weak::upgrade) {
+                    pool.close().await;
+                }
+                pools.remove(&path);
                 let backups = codex_state::backup_runtime_db_for_fresh_start(&path).await?;
                 tracing::warn!(
                     database = %path.display(),

codex-rs/ext/agent-message-board/src/local/pools_tests.rs

@@ -0,0 +1,54 @@
+//! Verifies pool sharing and release without exposing storage internals.
+
+use super::*;
+use crate::NotificationDelivery;
+use futures::future::BoxFuture;
+use pretty_assertions::assert_ne;
+
+struct UnusedHost;
+
+impl MessageBoardHost for UnusedHost {
+    fn agent_path(&self, _caller: ThreadId) -> BoxFuture<'_, Result<AgentPath>> {
+        unreachable!("pool tests do not invoke host capabilities")
+    }
+
+    fn resolve_agent(&self, _path: AgentPath) -> BoxFuture<'_, Result<ThreadId>> {
+        unreachable!("pool tests do not invoke host capabilities")
+    }
+
+    fn current_time(&self, _caller: ThreadId) -> BoxFuture<'_, Result<DateTime<Utc>>> {
+        unreachable!("pool tests do not invoke host capabilities")
+    }
+
+    fn notify(
+        &self,
+        _recipient: ThreadId,
+        _post: PostMetadata,
+    ) -> BoxFuture<'_, Result<NotificationDelivery>> {
+        unreachable!("pool tests do not invoke host capabilities")
+    }
+}
+
+#[tokio::test]
+async fn concurrent_handles_share_and_release_the_pool() {
+    let dir = tempfile::tempdir().unwrap();
+    let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap());
+    let (first, second) = tokio::join!(
+        LocalAgentMessageBoard::open(&sqlite, SessionId::new(), Arc::new(UnusedHost)),
+        LocalAgentMessageBoard::open(&sqlite, SessionId::new(), Arc::new(UnusedHost)),
+    );
+    let first = first.unwrap();
+    let second = second.unwrap();
+    assert!(Arc::ptr_eq(&first.pool, &second.pool));
+    let original = Arc::downgrade(&first.pool);
+    drop(first);
+    assert!(Arc::ptr_eq(&original.upgrade().unwrap(), &second.pool));
+    drop(second);
+    assert!(original.upgrade().is_none());
+
+    let reopened = LocalAgentMessageBoard::open(&sqlite, SessionId::new(), Arc::new(UnusedHost))
+        .await
+        .unwrap();
+    assert_ne!(original.as_ptr(), Arc::as_ptr(&reopened.pool));
+    reopened.pool.close().await;
+}