From 47b80571f0d6587060a580c47f6af0cfaefb0d5c Mon Sep 17 00:00:00 2001 From: jif Date: Fri, 9 Oct 2026 16:00:12 +0000 Subject: [PATCH] Add template setup for local agent message boards (#52621) ## What changed Expose `LocalBoardSetup` and `configure_local_board` to load initial channels and optional descriptions before agents start. Add a `configure_board` example and setup documentation. Save version 1 templates and channels atomically in SQLite. Identical templates are safe to retry; conflicting templates, previously used boards without a pinned template, and deleted boards are rejected. Setup accepts an optional creation timestamp and leaves agent access unchanged. Remove the pinned template when deleting a board. ## Testing Add an integration test covering channel metadata after reopen, identical retries, conflicting template rejection, retries after adding a channel, and rejection after board deletion. GitOrigin-RevId: 0b5c5439ab127fb5c275f024962a373a56635ad0 --- codex-rs/ext/agent-message-board/README.md | 27 +++++ .../examples/configure_board.rs | 27 +++++ codex-rs/ext/agent-message-board/src/lib.rs | 2 + codex-rs/ext/agent-message-board/src/local.rs | 51 +------- .../src/local/lifecycle.rs | 1 + .../ext/agent-message-board/src/local/pool.rs | 61 ++++++++++ .../agent-message-board/src/local/setup.rs | 109 ++++++++++++++++++ .../agent-message-board/tests/local_board.rs | 69 +++++++++++ 8 files changed, 300 insertions(+), 47 deletions(-) create mode 100644 codex-rs/ext/agent-message-board/README.md create mode 100644 codex-rs/ext/agent-message-board/examples/configure_board.rs create mode 100644 codex-rs/ext/agent-message-board/src/local/pool.rs create mode 100644 codex-rs/ext/agent-message-board/src/local/setup.rs diff --git a/codex-rs/ext/agent-message-board/README.md b/codex-rs/ext/agent-message-board/README.md new file mode 100644 index 0000000000..ca36ab393e --- /dev/null +++ b/codex-rs/ext/agent-message-board/README.md @@ -0,0 +1,27 @@ +# Load a local board template + +Create an empty chat, then load its initial channels before starting the agents: + +```sh +just agent-board-config ~/.codex /absolute/path/setup.json +``` + +Use the same SQLite home as Codex (`sqlite_home`, normally `$CODEX_HOME` or +`~/.codex`). Example `setup.json`: + +```json +{ + "template": { + "version": 1, + "channels": { "policy": { "description": "Shared policies" } } + } +} +``` + +The template and channels are saved atomically. Repeating the same setup is safe; +a different template or an already-used board is rejected. The pinned setup +survives reopen. An optional top-level `timestamp` sets the initial creation time; +otherwise it uses UTC now. + +This loads channels and descriptions only. ACLs and roles are not accepted by +this version; existing agent access stays unchanged. diff --git a/codex-rs/ext/agent-message-board/examples/configure_board.rs b/codex-rs/ext/agent-message-board/examples/configure_board.rs new file mode 100644 index 0000000000..5f042d420b --- /dev/null +++ b/codex-rs/ext/agent-message-board/examples/configure_board.rs @@ -0,0 +1,27 @@ +//! Operator-only setup for local experiments; no administrative agent tools. + +use codex_agent_message_board_extension::LocalBoardSetup; +use codex_agent_message_board_extension::configure_local_board; +use codex_protocol::SessionId; +use codex_state::SqliteConfig; +use std::path::PathBuf; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let args: Vec<_> = std::env::args_os().collect(); + let [_, sqlite_home, board, input] = args.as_slice() else { + return Err("Usage: configure_board ".into()); + }; + let sqlite = SqliteConfig::from_sqlite_home( + std::path::absolute(PathBuf::from(sqlite_home))?.try_into()?, + ); + let board = SessionId::from_string(board.to_str().ok_or("invalid board ID")?)?; + let input = tokio::fs::read(input).await?; + if input.len() > 64 * 1024 { + return Err("setup file exceeds 64 KiB".into()); + } + let setup: LocalBoardSetup = serde_json::from_slice(&input)?; + configure_local_board(&sqlite, 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 78e0e0d09b..d4221d4b63 100644 --- a/codex-rs/ext/agent-message-board/src/lib.rs +++ b/codex-rs/ext/agent-message-board/src/lib.rs @@ -34,6 +34,8 @@ pub use host::NotificationDelivery; pub use in_memory::InMemoryAgentMessageBoard; pub use in_memory::InMemoryMessageBoards; pub use local::LocalAgentMessageBoard; +pub use local::setup::LocalBoardSetup; +pub use local::setup::configure_local_board; pub use permissions::BoardPermissions; pub use permissions::ChannelPermissions; pub use permissions::PermissionDenied; diff --git a/codex-rs/ext/agent-message-board/src/local.rs b/codex-rs/ext/agent-message-board/src/local.rs index 56f90b81eb..7e817861a2 100644 --- a/codex-rs/ext/agent-message-board/src/local.rs +++ b/codex-rs/ext/agent-message-board/src/local.rs @@ -45,7 +45,9 @@ use uuid::Uuid; mod lifecycle; mod paging; +mod pool; mod queries; +pub(crate) mod setup; #[cfg(test)] #[path = "local/pools_tests.rs"] @@ -61,6 +63,7 @@ const DATABASE_FILE: &str = "agent_message_board_1.sqlite"; static POOLS: LazyLock>>> = LazyLock::new(Mutex::default); const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS board_templates (board TEXT PRIMARY KEY NOT NULL, template TEXT NOT NULL); CREATE TABLE IF NOT EXISTS deleted_boards (board TEXT PRIMARY KEY NOT NULL); CREATE TABLE IF NOT EXISTS channels ( board TEXT NOT NULL, name TEXT NOT NULL, name_search TEXT NOT NULL, created_at TEXT NOT NULL, timestamp INTEGER NOT NULL, author TEXT NOT NULL, description TEXT, @@ -104,60 +107,14 @@ 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, ) -> Result { - tokio::fs::create_dir_all(sqlite.home()).await?; - 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)?; - let mut tx = pool - .begin_with("BEGIN IMMEDIATE") - .await - .map_err(storage_error)?; - sqlx::raw_sql(SCHEMA) - .execute(&mut *tx) - .await - .map_err(storage_error)?; - let has_description: bool = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM pragma_table_info('channels') WHERE name='description')", - ) - .fetch_one(&mut *tx) - .await - .map_err(storage_error)?; - if !has_description { - sqlx::query("ALTER TABLE channels ADD COLUMN description TEXT") - .execute(&mut *tx) - .await - .map_err(storage_error)?; - } - tx.commit().await.map_err(storage_error)?; - let pool = Arc::new(pool); - pools.insert(path, Arc::downgrade(&pool)); - pool - }; Ok(Self { identity, - pool, + pool: pool::open(sqlite).await?, host, }) } diff --git a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs index d44109de9c..f60fd65d5c 100644 --- a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs +++ b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs @@ -77,6 +77,7 @@ impl LocalAgentMessageBoard { "DELETE FROM subscription_opt_outs WHERE board=?", "DELETE FROM posts WHERE board=?", "DELETE FROM channels WHERE board=?", + "DELETE FROM board_templates WHERE board=?", ] { sqlx::query(statement) .bind(root.to_string()) diff --git a/codex-rs/ext/agent-message-board/src/local/pool.rs b/codex-rs/ext/agent-message-board/src/local/pool.rs new file mode 100644 index 0000000000..ba4d978a2d --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/local/pool.rs @@ -0,0 +1,61 @@ +//! Shared SQLite initialization used by runtime handles and local experiment setup. + +use super::DATABASE_FILE; +use super::POOLS; +use super::SCHEMA; +use super::storage_error; +use codex_protocol::error::Result; +use codex_state::SqliteConfig; +use sqlx::SqlitePool; +use std::sync::Arc; +use std::sync::Weak; + +#[expect( + clippy::await_holding_invalid_type, + reason = "serialize shared pool initialization" +)] +pub(super) async fn open(sqlite: &SqliteConfig) -> Result> { + tokio::fs::create_dir_all(sqlite.home()).await?; + 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)?; + let mut tx = pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(storage_error)?; + sqlx::raw_sql(SCHEMA) + .execute(&mut *tx) + .await + .map_err(storage_error)?; + let has_description: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('channels') WHERE name='description')", + ) + .fetch_one(&mut *tx) + .await + .map_err(storage_error)?; + if !has_description { + sqlx::query("ALTER TABLE channels ADD COLUMN description TEXT") + .execute(&mut *tx) + .await + .map_err(storage_error)?; + } + tx.commit().await.map_err(storage_error)?; + let pool = Arc::new(pool); + pools.insert(path, Arc::downgrade(&pool)); + pool + }; + Ok(pool) +} diff --git a/codex-rs/ext/agent-message-board/src/local/setup.rs b/codex-rs/ext/agent-message-board/src/local/setup.rs new file mode 100644 index 0000000000..eca2410978 --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/local/setup.rs @@ -0,0 +1,109 @@ +//! Loads initial channels into an empty local board and pins their source template. + +use super::invalid; +use super::storage_error; +use super::validate_channel; +use crate::ChannelDescription; +use chrono::DateTime; +use chrono::Utc; +use codex_protocol::SessionId; +use codex_protocol::error::Result; +use codex_state::SqliteConfig; +use serde::Deserialize; +use serde::Serialize; +use std::collections::BTreeMap; + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Channel { + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Template { + version: u32, + #[serde(default)] + channels: BTreeMap, +} + +/// Operator input for loading an empty board. This is not an agent tool. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LocalBoardSetup { + #[serde(default)] + timestamp: Option>, + template: Template, +} + +/// Saves the template and its initial channels atomically. The same template is +/// safe to retry; replacing it or initializing an already-used board is rejected. +pub async fn configure_local_board( + sqlite: &SqliteConfig, + board: SessionId, + setup: LocalBoardSetup, +) -> Result<()> { + let template = setup.template; + let encoded = serde_json::to_string(&template).map_err(storage_error)?; + if template.version != 1 || encoded.len() > 64 * 1024 || template.channels.len() > 64 { + return Err(invalid( + "board templates require version 1 and at most 64 KiB / 64 channels", + )); + } + for name in template.channels.keys() { + validate_channel(name)?; + } + let pool = super::pool::open(sqlite).await?; + let mut tx = pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(storage_error)?; + let board = board.to_string(); + let deleted: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM deleted_boards WHERE board=?)") + .bind(&board) + .fetch_one(&mut *tx) + .await + .map_err(storage_error)?; + if deleted { + return Err(invalid("the board has been permanently deleted")); + } + let pinned: Option = + sqlx::query_scalar("SELECT template FROM board_templates WHERE board=?") + .bind(&board) + .fetch_optional(&mut *tx) + .await + .map_err(storage_error)?; + if let Some(pinned) = pinned { + if pinned != encoded { + return Err(invalid("the board template is already pinned")); + } + } else { + let used: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM channels WHERE board=?)") + .bind(&board) + .fetch_one(&mut *tx) + .await + .map_err(storage_error)?; + if used { + return Err(invalid( + "pin the template before creating channels or posts", + )); + } + sqlx::query("INSERT INTO board_templates(board,template) VALUES(?,?)") + .bind(&board) + .bind(encoded) + .execute(&mut *tx) + .await + .map_err(storage_error)?; + let now = setup.timestamp.unwrap_or_else(Utc::now); + for (name, channel) in &template.channels { + sqlx::query("INSERT INTO channels(board,name,name_search,created_at,timestamp,author,description) VALUES(?,?,?,?,?,'/root',?)") + .bind(&board).bind(name).bind(caseless::default_case_fold_str(name)) + .bind(now.to_rfc3339()).bind(now.timestamp_micros()) + .bind(channel.description.as_ref().map(ChannelDescription::as_str)) + .execute(&mut *tx).await.map_err(storage_error)?; + } + } + tx.commit().await.map_err(storage_error) +} diff --git a/codex-rs/ext/agent-message-board/tests/local_board.rs b/codex-rs/ext/agent-message-board/tests/local_board.rs index 002bb7df49..f68c2feba4 100644 --- a/codex-rs/ext/agent-message-board/tests/local_board.rs +++ b/codex-rs/ext/agent-message-board/tests/local_board.rs @@ -1498,3 +1498,72 @@ async fn channel_descriptions_migrate_legacy_storage_and_survive_reopen() { } ); } + +#[tokio::test] +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 root = ThreadId::new(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + agent_path_calls: AtomicUsize::default(), + members: [(root, AgentPath::root())].into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::new(false), + 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?; + drop(board); + let board = LocalAgentMessageBoard::open(&sqlite, root.into(), host.clone()).await?; + let query = ChannelQuery { + query: None, + direction: SortDirection::NewestFirst, + page: PageRequest::default(), + }; + let channels = board.list_channels(root, query.clone()).await?; + assert_eq!( + serde_json::to_value(channels)?, + json!({"results":[{ + "channel_name":"policy", "description":"Shared policies", "created_at":"2026-09-18T12:00:00Z", + "created_by":"/root", "message_count":0, "last_message_id":null + }],"n_returned":1,"has_more":false,"next_cursor":null}) + ); + 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() + ); + board + .create_channel( + root, + CreateChannelRequest { + channel_name: "new".into(), + description: None, + subscription: SubscriptionChange::Unsubscribe, + }, + ) + .await?; + configure_local_board(&sqlite, root.into(), serde_json::from_value(setup.clone())?).await?; + assert_eq!( + board + .list_channels(root, query) + .await? + .channels + .results + .len(), + 2 + ); + LocalAgentMessageBoard::delete_boards(&sqlite, &[root.into()]).await?; + assert!( + configure_local_board(&sqlite, root.into(), serde_json::from_value(setup)?) + .await + .is_err() + ); + Ok(()) +}