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
This commit is contained in:
jif
2026-10-09 16:11:43 +00:00
committed by copyberry
parent c63929af50
commit 47b80571f0
8 changed files with 300 additions and 47 deletions
@@ -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 <root-session-id> /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.
@@ -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<dyn std::error::Error>> {
let args: Vec<_> = std::env::args_os().collect();
let [_, sqlite_home, board, input] = args.as_slice() else {
return Err("Usage: configure_board <sqlite-home> <board-session-id> <setup.json>".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(())
}
@@ -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;
+4 -47
View File
@@ -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<Mutex<HashMap<PathBuf, Weak<SqlitePool>>>> = 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<dyn MessageBoardHost>,
) -> Result<Self> {
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,
})
}
@@ -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())
@@ -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<Arc<SqlitePool>> {
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)
}
@@ -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<ChannelDescription>,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Template {
version: u32,
#[serde(default)]
channels: BTreeMap<String, Channel>,
}
/// 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<DateTime<Utc>>,
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<String> =
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)
}
@@ -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(())
}