diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index 2ea8b3b9c0..a3a7ae6614 100644 --- a/codex-rs/core/src/agent_message_board.rs +++ b/codex-rs/core/src/agent_message_board.rs @@ -13,6 +13,7 @@ use chrono::DateTime; use chrono::Utc; use codex_agent_message_board_client::AccessToken; use codex_agent_message_board_client::RemoteAgentMessageBoard; +use codex_agent_message_board_extension::local_board_options; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ThreadStartInput; use codex_features::Feature; @@ -110,7 +111,13 @@ pub fn install_agent_message_board( Arc::new(in_memory_boards.open(tree, host).await) } else { Arc::new( - LocalAgentMessageBoard::open(config.sqlite_config(), tree, host).await?, + LocalAgentMessageBoard::open( + config.sqlite_config(), + &local_board_options(), + tree, + host, + ) + .await?, ) }; Ok(Some(board)) diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index ef30f44137..b2d22649a6 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -39,6 +39,7 @@ use crate::tasks::interrupted_turn_history_marker; use crate::thread_startup_metadata::ThreadStartupMetadata; use codex_agent_graph_store::AgentGraphStore; use codex_agent_graph_store::LocalAgentGraphStore; +use codex_agent_message_board_extension::local_board_options; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ThreadHistoryBuilder; use codex_app_server_protocol::TurnStatus; @@ -501,13 +502,19 @@ pub fn thread_store_from_config( let sqlite = sqlite.clone(); Box::pin(async move { let boards = thread_ids.into_iter().map(Into::into).collect::>(); - LocalAgentMessageBoard::delete_boards(&sqlite, &boards) - .await - .map_err(|err| ThreadStoreError::Internal { + LocalAgentMessageBoard::delete_boards( + &sqlite, + &local_board_options(), + &boards, + ) + .await + .map_err(|err| { + ThreadStoreError::Internal { message: format!( "failed to delete agent message boards: {err}" ), - }) + } + }) }) }), ); diff --git a/codex-rs/ext/agent-message-board/examples/configure_board.rs b/codex-rs/ext/agent-message-board/examples/configure_board.rs index f3f5156bcc..978c611b83 100644 --- a/codex-rs/ext/agent-message-board/examples/configure_board.rs +++ b/codex-rs/ext/agent-message-board/examples/configure_board.rs @@ -1,5 +1,6 @@ //! Operator-only setup for local experiments; no administrative agent tools. +use codex_agent_message_board_extension::local_board_options; use codex_message_board::LocalBoardSetup; use codex_message_board::configure_local_board; use codex_protocol::SessionId; @@ -21,7 +22,7 @@ async fn main() -> Result<(), Box> { return Err("setup file exceeds 64 KiB".into()); } let setup: LocalBoardSetup = serde_json::from_slice(&input)?; - configure_local_board(&sqlite, board, setup).await?; + configure_local_board(&sqlite, &local_board_options(), board, setup).await?; println!("Configured local board {board}"); Ok(()) } diff --git a/codex-rs/ext/agent-message-board/src/lib.rs b/codex-rs/ext/agent-message-board/src/lib.rs index 601ac0f247..10aea44a8b 100644 --- a/codex-rs/ext/agent-message-board/src/lib.rs +++ b/codex-rs/ext/agent-message-board/src/lib.rs @@ -4,3 +4,10 @@ mod extension; pub use extension::install; + +/// Storage selected by the agent consumer for opening, setup and permanent deletion. +pub fn local_board_options() -> codex_message_board::LocalBoardOptions { + codex_message_board::LocalBoardOptions { + storage_namespace: "agent_message_board".into(), + } +} diff --git a/codex-rs/message-board/src/lib.rs b/codex-rs/message-board/src/lib.rs index eb560a66e5..a2c426e326 100644 --- a/codex-rs/message-board/src/lib.rs +++ b/codex-rs/message-board/src/lib.rs @@ -34,6 +34,7 @@ pub use in_memory::InMemoryMessageBoards; pub use local::LocalAgentMessageBoard; pub use local::connection::LocalBoard; pub use local::connection::LocalBoardConnection; +pub use local::options::LocalBoardOptions; pub use local::setup::LocalBoardSetup; pub use local::setup::configure_local_board; pub use permissions::BoardPermissions; diff --git a/codex-rs/message-board/src/local.rs b/codex-rs/message-board/src/local.rs index e58ee55c23..c8112d385c 100644 --- a/codex-rs/message-board/src/local.rs +++ b/codex-rs/message-board/src/local.rs @@ -8,6 +8,7 @@ //! Explicit connections can restrict writes; default handles retain legacy access. use self::connection::Access; +use self::options::LocalBoardOptions; use crate::ChannelDescription; use crate::ChannelPermissions; use crate::ChannelSummary; @@ -43,6 +44,7 @@ use uuid::Uuid; pub(crate) mod connection; mod lifecycle; +pub(crate) mod options; mod paging; mod pool; mod queries; @@ -73,14 +75,16 @@ 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. + /// The caller selects the storage namespace. pub async fn open( sqlite: &SqliteConfig, + options: &LocalBoardOptions, identity: SessionId, host: Arc, ) -> Result { Ok(Self { identity, - pool: pool::open(sqlite).await?, + pool: pool::open(sqlite, options).await?, host, access: None, }) @@ -252,11 +256,11 @@ impl LocalAgentMessageBoard { thread_id: root, created_at: now, }, - text: request.text.clone(), + text: request.text, }; sqlx::query("INSERT INTO posts(board,id,channel,root,author,timestamp,body_search,payload,request_id,request) VALUES(?,?,?,?,?,?,?,?,?,?)") .bind(self.identity.to_string()).bind(id.to_string()).bind(channel).bind(root.to_string()) - .bind(author.to_string()).bind(now.timestamp_micros()).bind(default_case_fold_str(&request.text)) + .bind(author.to_string()).bind(now.timestamp_micros()).bind(default_case_fold_str(&post.text)) .bind(serde_json::to_string(&post).map_err(storage_error)?).bind(request_id).bind(request_json) .execute(&mut *tx).await.map_err(storage_error)?; // Participation subscribes by default, without overriding an explicit opt-out. diff --git a/codex-rs/message-board/src/local/connection.rs b/codex-rs/message-board/src/local/connection.rs index 405b74723d..c79240be10 100644 --- a/codex-rs/message-board/src/local/connection.rs +++ b/codex-rs/message-board/src/local/connection.rs @@ -2,6 +2,7 @@ //! independent of the board's storage or the agent's tree identity. use super::LocalAgentMessageBoard; +use super::options::LocalBoardOptions; use crate::MessageBoardHost; use crate::PermissionDenied; use codex_protocol::AgentPath; @@ -19,11 +20,21 @@ use std::sync::OnceLock; pub struct LocalBoard { sqlite: SqliteConfig, identity: SessionId, + options: LocalBoardOptions, } impl LocalBoard { pub fn new(sqlite: SqliteConfig, identity: SessionId) -> Self { - Self { sqlite, identity } + Self { + sqlite, + identity, + options: LocalBoardOptions::default(), + } + } + + pub fn with_options(mut self, options: LocalBoardOptions) -> Self { + self.options = options; + self } pub fn read_only(&self) -> LocalBoardConnection { @@ -77,8 +88,13 @@ impl LocalBoardConnection { } pub async fn open(&self, host: Arc) -> Result { - let mut board = - LocalAgentMessageBoard::open(&self.board.sqlite, self.identity(), host).await?; + let mut board = LocalAgentMessageBoard::open( + &self.board.sqlite, + &self.board.options, + self.identity(), + host, + ) + .await?; // Pool initialization canonicalizes the directory before opening SQLite. let options = board.pool.connect_options(); let database = options.get_filename(); diff --git a/codex-rs/message-board/src/local/lifecycle.rs b/codex-rs/message-board/src/local/lifecycle.rs index 12e30dda61..c7fb4efbee 100644 --- a/codex-rs/message-board/src/local/lifecycle.rs +++ b/codex-rs/message-board/src/local/lifecycle.rs @@ -6,7 +6,7 @@ use super::LocalAgentMessageBoard; use super::invalid; -use super::pool::DATABASE_FILE; +use super::options::LocalBoardOptions; use super::pool::POOLS; use super::pool::SCHEMA; use super::storage_error; @@ -27,14 +27,17 @@ impl LocalAgentMessageBoard { 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); + pub async fn delete_boards( + sqlite: &SqliteConfig, + options: &LocalBoardOptions, + roots: &[SessionId], + ) -> Result<()> { + let filename = options.database_file()?; + let path = sqlite.home().join(&filename); if roots.is_empty() || !tokio::fs::try_exists(&path).await? { return Ok(()); } - let path = tokio::fs::canonicalize(sqlite.home()) - .await? - .join(DATABASE_FILE); + let path = tokio::fs::canonicalize(sqlite.home()).await?.join(filename); let mut pools = POOLS.lock().await; let pool = match sqlite.open_read_write_pool(&path).await { Ok(pool) => pool, diff --git a/codex-rs/message-board/src/local/options.rs b/codex-rs/message-board/src/local/options.rs new file mode 100644 index 0000000000..3959dc9665 --- /dev/null +++ b/codex-rs/message-board/src/local/options.rs @@ -0,0 +1,43 @@ +//! Storage namespace selected by the board's owner. +//! Owners supply the same namespace when reopening storage. + +use super::invalid; +use codex_protocol::error::Result; + +pub(super) const DEFAULT_STORAGE_NAMESPACE: &str = "agent_message_board"; + +/// Host-selected storage, independent of tool selection and connection permissions. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LocalBoardOptions { + /// A namespace isolates boards with the same session identity in separate databases. + /// Use 1–64 lowercase ASCII letters, digits, underscores or hyphens. + pub storage_namespace: String, +} + +impl Default for LocalBoardOptions { + fn default() -> Self { + Self { + storage_namespace: DEFAULT_STORAGE_NAMESPACE.into(), + } + } +} + +impl LocalBoardOptions { + pub(super) fn database_file(&self) -> Result { + if self.storage_namespace.is_empty() + || self.storage_namespace.len() > 64 + || !self.storage_namespace.bytes().all(|byte| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-') + }) + { + return Err(invalid( + "board storage namespaces require 1–64 lowercase ASCII letters, digits, underscores or hyphens", + )); + } + if self.storage_namespace == DEFAULT_STORAGE_NAMESPACE { + return Ok("agent_message_board_1.sqlite".into()); + } + // Custom boards must not share filenames with other runtime databases. + Ok(format!("message_board_{}_1.sqlite", self.storage_namespace)) + } +} diff --git a/codex-rs/message-board/src/local/pool.rs b/codex-rs/message-board/src/local/pool.rs index c39b48d5ea..9b2ea43703 100644 --- a/codex-rs/message-board/src/local/pool.rs +++ b/codex-rs/message-board/src/local/pool.rs @@ -1,6 +1,7 @@ //! Owns the local database schema and pool registry shared by runtime handles and setup. //! Initialization and deletion recovery share the registry lock to exclude stale pools. +use super::options::LocalBoardOptions; use super::storage_error; use codex_protocol::error::Result; use codex_state::SqliteConfig; @@ -13,8 +14,6 @@ use std::sync::LazyLock; use std::sync::Weak; use tokio::sync::Mutex; -pub(super) 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. pub(super) static POOLS: LazyLock>>> = @@ -53,11 +52,13 @@ CREATE TABLE IF NOT EXISTS subscription_opt_outs ( clippy::await_holding_invalid_type, reason = "serialize shared pool initialization" )] -pub(super) async fn open(sqlite: &SqliteConfig) -> Result> { +pub(super) async fn open( + sqlite: &SqliteConfig, + options: &LocalBoardOptions, +) -> Result> { + let filename = options.database_file()?; tokio::fs::create_dir_all(sqlite.home()).await?; - let path = tokio::fs::canonicalize(sqlite.home()) - .await? - .join(DATABASE_FILE); + let path = tokio::fs::canonicalize(sqlite.home()).await?.join(filename); let mut pools = POOLS.lock().await; pools.retain(|_, pool| pool.strong_count() > 0); let pool = if let Some(pool) = pools diff --git a/codex-rs/message-board/src/local/pools_tests.rs b/codex-rs/message-board/src/local/pools_tests.rs index 364f7def89..1d9fb92e24 100644 --- a/codex-rs/message-board/src/local/pools_tests.rs +++ b/codex-rs/message-board/src/local/pools_tests.rs @@ -34,9 +34,10 @@ impl MessageBoardHost for UnusedHost { 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 options = LocalBoardOptions::default(); let (first, second) = tokio::join!( - LocalAgentMessageBoard::open(&sqlite, SessionId::new(), Arc::new(UnusedHost)), - LocalAgentMessageBoard::open(&sqlite, SessionId::new(), Arc::new(UnusedHost)), + LocalAgentMessageBoard::open(&sqlite, &options, SessionId::new(), Arc::new(UnusedHost)), + LocalAgentMessageBoard::open(&sqlite, &options, SessionId::new(), Arc::new(UnusedHost)), ); let first = first.unwrap(); let second = second.unwrap(); @@ -47,9 +48,10 @@ async fn concurrent_handles_share_and_release_the_pool() { drop(second); assert!(original.upgrade().is_none()); - let reopened = LocalAgentMessageBoard::open(&sqlite, SessionId::new(), Arc::new(UnusedHost)) - .await - .unwrap(); + let reopened = + LocalAgentMessageBoard::open(&sqlite, &options, SessionId::new(), Arc::new(UnusedHost)) + .await + .unwrap(); assert_ne!(original.as_ptr(), Arc::as_ptr(&reopened.pool)); reopened.pool.close().await; } diff --git a/codex-rs/message-board/src/local/setup.rs b/codex-rs/message-board/src/local/setup.rs index b767a4f38d..02ab4900a1 100644 --- a/codex-rs/message-board/src/local/setup.rs +++ b/codex-rs/message-board/src/local/setup.rs @@ -2,6 +2,7 @@ use super::insert_channel; use super::invalid; +use super::options::LocalBoardOptions; use super::storage_error; use super::validate_channel; use crate::ChannelDescription; @@ -43,6 +44,7 @@ pub struct LocalBoardSetup { /// safe to retry; replacing it or initializing an already-used board is rejected. pub async fn configure_local_board( sqlite: &SqliteConfig, + options: &LocalBoardOptions, board: SessionId, setup: LocalBoardSetup, ) -> Result<()> { @@ -56,7 +58,7 @@ pub async fn configure_local_board( for name in template.channels.keys() { validate_channel(name)?; } - let pool = super::pool::open(sqlite).await?; + let pool = super::pool::open(sqlite, options).await?; let mut tx = pool .begin_with("BEGIN IMMEDIATE") .await diff --git a/codex-rs/message-board/tests/local_board.rs b/codex-rs/message-board/tests/local_board.rs index b57ccbd169..34679340a1 100644 --- a/codex-rs/message-board/tests/local_board.rs +++ b/codex-rs/message-board/tests/local_board.rs @@ -91,6 +91,7 @@ impl MessageBoardHost for Host { async fn opening_board_preserves_posts_when_corruption_has_no_startup_recovery() { let dir = tempfile::tempdir().unwrap(); let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let options = LocalBoardOptions::default(); let root = ThreadId::new(); let host = Arc::new(Host { clock: AtomicI64::default(), @@ -131,7 +132,7 @@ async fn opening_board_preserves_posts_when_corruption_has_no_startup_recovery() pool.close().await; let tree = SessionId::from(root); - let board = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + let board = LocalAgentMessageBoard::open(&sqlite, &options, tree, host.clone()) .await .unwrap(); let request = PostRequest { @@ -142,7 +143,7 @@ async fn opening_board_preserves_posts_when_corruption_has_no_startup_recovery() }; let metadata = board.post(root, request.clone()).await.unwrap(); drop(board); - let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host) + let resumed = LocalAgentMessageBoard::open(&sqlite, &options, tree, host) .await .unwrap(); assert_eq!(resumed.post(root, request).await.unwrap(), metadata); @@ -152,6 +153,7 @@ async fn opening_board_preserves_posts_when_corruption_has_no_startup_recovery() async fn shared_handles_resume_posts_and_preserve_subscription_rules() { let dir = tempfile::tempdir().unwrap(); let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let options = LocalBoardOptions::default(); let root = ThreadId::new(); let child = ThreadId::new(); let child_path = AgentPath::root().join("worker").unwrap(); @@ -164,7 +166,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { notifications: Mutex::default(), }); let tree = SessionId::from(root); - let first = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + let first = LocalAgentMessageBoard::open(&sqlite, &options, tree, host.clone()) .await .unwrap(); first @@ -186,7 +188,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { // The creator receives this through its channel subscription. agents_to_notify: vec![AgentPath::root()], }; - let second = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + let second = LocalAgentMessageBoard::open(&sqlite, &options, tree, host.clone()) .await .unwrap(); let (metadata, duplicate) = tokio::join!( @@ -213,7 +215,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { .unwrap(); legacy_pool.close().await; - let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + let resumed = LocalAgentMessageBoard::open(&sqlite, &options, tree, host.clone()) .await .unwrap(); assert_eq!(resumed.post(root, request).await.unwrap(), metadata); @@ -337,7 +339,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { .await .unwrap(); drop(resumed); - let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + let resumed = LocalAgentMessageBoard::open(&sqlite, &options, tree, host.clone()) .await .unwrap(); resumed @@ -418,7 +420,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { .unwrap(); assert_eq!(*host.notifications.lock().unwrap(), vec![(root, reply)]); - let other = LocalAgentMessageBoard::open(&sqlite, SessionId::new(), host.clone()) + let other = LocalAgentMessageBoard::open(&sqlite, &options, SessionId::new(), host.clone()) .await .unwrap(); assert!( @@ -434,7 +436,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { .await .is_err() ); - LocalAgentMessageBoard::delete_boards(&sqlite, &[tree]) + LocalAgentMessageBoard::delete_boards(&sqlite, &options, &[tree]) .await .unwrap(); let pool = sqlite @@ -519,7 +521,7 @@ impl Backend { ) -> Arc { match self { Self::Local => Arc::new( - LocalAgentMessageBoard::open(sqlite, identity, host) + LocalAgentMessageBoard::open(sqlite, &LocalBoardOptions::default(), identity, host) .await .unwrap(), ), @@ -1427,6 +1429,7 @@ fn channel_descriptions_validate_construction_and_serialization() { async fn channel_descriptions_migrate_legacy_storage_and_survive_reopen() { let dir = tempfile::tempdir().unwrap(); let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let options = LocalBoardOptions::default(); let root = ThreadId::new(); let host = Arc::new(Host { clock: AtomicI64::default(), @@ -1449,7 +1452,7 @@ async fn channel_descriptions_migrate_legacy_storage_and_survive_reopen() { .await .unwrap(); pool.close().await; - let board = LocalAgentMessageBoard::open(&sqlite, root.into(), host.clone()) + let board = LocalAgentMessageBoard::open(&sqlite, &options, root.into(), host.clone()) .await .unwrap(); let legacy = ChannelSummary { @@ -1478,7 +1481,7 @@ async fn channel_descriptions_migrate_legacy_storage_and_survive_reopen() { .await .unwrap(); drop(board); - let reopened = LocalAgentMessageBoard::open(&sqlite, root.into(), host) + let reopened = LocalAgentMessageBoard::open(&sqlite, &options, root.into(), host) .await .unwrap(); let listed = reopened @@ -1506,6 +1509,9 @@ async fn local_board_setup_survives_reopen() -> Result<()> { use serde_json::json; let dir = tempfile::tempdir()?; let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let options = LocalBoardOptions { + storage_namespace: "setup_test".into(), + }; let root = ThreadId::new(); let host = Arc::new(Host { clock: AtomicI64::default(), @@ -1516,11 +1522,23 @@ async fn local_board_setup_survives_reopen() -> Result<()> { notifications: Mutex::default(), }); let setup = json!({"timestamp":"2026-09-18T12:00:00Z","template":{"version":1,"channels":{"policy":{"description":"Shared policies"}}}}); - configure_local_board(&sqlite, root.into(), serde_json::from_value(setup.clone())?).await?; - configure_local_board(&sqlite, root.into(), serde_json::from_value(setup.clone())?).await?; - let board = LocalAgentMessageBoard::open(&sqlite, root.into(), host.clone()).await?; + configure_local_board( + &sqlite, + &options, + root.into(), + serde_json::from_value(setup.clone())?, + ) + .await?; + configure_local_board( + &sqlite, + &options, + root.into(), + serde_json::from_value(setup.clone())?, + ) + .await?; + let board = LocalAgentMessageBoard::open(&sqlite, &options, root.into(), host.clone()).await?; drop(board); - let board = LocalAgentMessageBoard::open(&sqlite, root.into(), host.clone()).await?; + let board = LocalAgentMessageBoard::open(&sqlite, &options, root.into(), host.clone()).await?; let query = ChannelQuery { query: None, direction: SortDirection::NewestFirst, @@ -1537,9 +1555,14 @@ async fn local_board_setup_survives_reopen() -> Result<()> { let mut changed = setup.clone(); changed["template"]["channels"]["policy"]["description"] = json!("Changed"); assert!( - configure_local_board(&sqlite, root.into(), serde_json::from_value(changed)?) - .await - .is_err() + configure_local_board( + &sqlite, + &options, + root.into(), + serde_json::from_value(changed)? + ) + .await + .is_err() ); board .create_channel( @@ -1551,7 +1574,13 @@ async fn local_board_setup_survives_reopen() -> Result<()> { }, ) .await?; - configure_local_board(&sqlite, root.into(), serde_json::from_value(setup.clone())?).await?; + configure_local_board( + &sqlite, + &options, + root.into(), + serde_json::from_value(setup.clone())?, + ) + .await?; assert_eq!( board .list_channels(root, query) @@ -1561,11 +1590,111 @@ async fn local_board_setup_survives_reopen() -> Result<()> { .len(), 2 ); - LocalAgentMessageBoard::delete_boards(&sqlite, &[root.into()]).await?; + LocalAgentMessageBoard::delete_boards(&sqlite, &options, &[root.into()]).await?; assert!( - configure_local_board(&sqlite, root.into(), serde_json::from_value(setup)?) - .await - .is_err() + configure_local_board( + &sqlite, + &options, + root.into(), + serde_json::from_value(setup)? + ) + .await + .is_err() ); Ok(()) } + +#[tokio::test] +async fn storage_namespaces_preserve_isolation_and_reopen() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let root = ThreadId::new(); + let child = ThreadId::new(); + let child_path = AgentPath::root().join("worker").unwrap(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + agent_path_calls: AtomicUsize::default(), + members: [(root, AgentPath::root()), (child, child_path)].into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::new(false), + notifications: Mutex::default(), + }); + for storage_namespace in ["queue", "goals"] { + // Opening a board must not modify another runtime database. + let runtime_database = sqlite + .open_read_write_pool(&dir.path().join(format!("{storage_namespace}_1.sqlite"))) + .await + .unwrap(); + sqlx::query("CREATE TABLE runtime_data (value TEXT NOT NULL)") + .execute(&runtime_database) + .await + .unwrap(); + let identity = SessionId::new(); + let options = LocalBoardOptions { + storage_namespace: storage_namespace.into(), + }; + let descriptor = LocalBoard::new(sqlite.clone(), identity).with_options(options); + let writer = descriptor.read_write().open(host.clone()).await?; + let seed = PostRequest { + request_id: "seed".into(), + destination: PostDestination::NewChannel("records".into()), + text: "Saved evidence".into(), + agents_to_notify: Vec::new(), + }; + let metadata = writer.post(root, seed.clone()).await?; + let read = ReadPostRequest { + message_id: metadata.message_id, + offset_chars: 0, + limit_chars: NonZeroU32::MAX, + }; + let legacy = LocalAgentMessageBoard::open( + &sqlite, + &LocalBoardOptions::default(), + identity, + host.clone(), + ) + .await?; + assert!(legacy.read_post(root, read.clone()).await.is_err()); + let legacy_metadata = legacy.post(root, seed.clone()).await?; + // Release the writer before reopening its saved posts. + drop(writer); + let reopened = descriptor.read_write().open(host.clone()).await?; + assert_eq!( + reopened.read_post(child, read).await?, + PostContent { + metadata, + text: seed.text.clone(), + n_chars: seed.text.chars().count(), + next_offset_chars: seed.text.chars().count(), + } + ); + assert_eq!( + legacy + .read_post( + root, + ReadPostRequest { + message_id: legacy_metadata.message_id, + offset_chars: 0, + limit_chars: NonZeroU32::MAX, + }, + ) + .await?, + PostContent { + metadata: legacy_metadata, + text: seed.text.clone(), + n_chars: seed.text.chars().count(), + next_offset_chars: seed.text.chars().count(), + } + ); + assert_eq!( + sqlx::query_scalar::<_, String>( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name", + ) + .fetch_all(&runtime_database) + .await + .unwrap(), + vec!["runtime_data".to_owned()] + ); + } + Ok(()) +}