diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index bd2e8f7316..a5193af107 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -815,6 +815,7 @@ "camino_1.2.2": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"bincode\",\"req\":\"^1\"},{\"name\":\"proptest\",\"optional\":true,\"req\":\"^1.0.0\"},{\"features\":[\"derive\"],\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0.223\"},{\"kind\":\"dev\",\"name\":\"serde_bytes\",\"req\":\"^0.11.8\"},{\"name\":\"serde_core\",\"optional\":true,\"req\":\"^1\"}],\"features\":{\"proptest1\":[\"dep:proptest\"],\"serde1\":[\"dep:serde_core\"]}}", "cargo-platform_0.3.2": "{\"dependencies\":[{\"name\":\"serde\",\"req\":\"^1.0.228\",\"target\":\"cfg(any())\"},{\"name\":\"serde_core\",\"req\":\"^1.0.228\"}],\"features\":{}}", "cargo_metadata_0.23.1": "{\"dependencies\":[{\"features\":[\"serde1\"],\"name\":\"camino\",\"req\":\"^1.1.10\"},{\"name\":\"cargo-platform\",\"req\":\"^0.3.0\"},{\"name\":\"derive_builder\",\"optional\":true,\"req\":\"^0.20\"},{\"features\":[\"serde\"],\"name\":\"semver\",\"req\":\"^1.0.26\"},{\"features\":[\"derive\"],\"name\":\"serde\",\"req\":\"^1.0.219\"},{\"features\":[\"unbounded_depth\"],\"name\":\"serde_json\",\"req\":\"^1.0.142\"},{\"name\":\"thiserror\",\"req\":\"^2.0.12\"}],\"features\":{\"builder\":[\"derive_builder\"],\"default\":[],\"unstable\":[]}}", + "caseless_0.2.2": "{\"dependencies\":[{\"name\":\"unicode-normalization\",\"req\":\"^0.1\"}],\"features\":{}}", "castaway_0.2.4": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"paste\",\"req\":\"^1\"},{\"name\":\"rustversion\",\"req\":\"^1\"}],\"features\":{\"alloc\":[],\"default\":[\"std\"],\"std\":[\"alloc\"]}}", "cbc_0.1.2": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"aes\",\"req\":\"^0.8\"},{\"name\":\"cipher\",\"req\":\"^0.4.2\"},{\"features\":[\"dev\"],\"kind\":\"dev\",\"name\":\"cipher\",\"req\":\"^0.4.2\"},{\"kind\":\"dev\",\"name\":\"hex-literal\",\"req\":\"^0.3.3\"}],\"features\":{\"alloc\":[\"cipher/alloc\"],\"block-padding\":[\"cipher/block-padding\"],\"default\":[\"block-padding\"],\"std\":[\"cipher/std\",\"alloc\"],\"zeroize\":[\"cipher/zeroize\"]}}", "cc_1.2.55": "{\"dependencies\":[{\"name\":\"find-msvc-tools\",\"req\":\"^0.1.9\"},{\"default_features\":false,\"name\":\"jobserver\",\"optional\":true,\"req\":\"^0.1.30\"},{\"default_features\":false,\"name\":\"libc\",\"optional\":true,\"req\":\"^0.2.62\",\"target\":\"cfg(unix)\"},{\"name\":\"shlex\",\"req\":\"^1.3.0\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3\"}],\"features\":{\"jobserver\":[],\"parallel\":[\"dep:libc\",\"dep:jobserver\"]}}", diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 4e9ca35a9a..80f65d58ce 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1750,6 +1750,15 @@ dependencies = [ "displaydoc", ] +[[package]] +name = "caseless" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b6fd507454086c8edfd769ca6ada439193cdb209c7681712ef6275cccbfe5d8" +dependencies = [ + "unicode-normalization", +] + [[package]] name = "castaway" version = "0.2.4" @@ -2090,10 +2099,18 @@ dependencies = [ name = "codex-agent-message-board-extension" version = "0.0.0" dependencies = [ + "caseless", "chrono", "codex-protocol", + "codex-state", "futures", + "pretty_assertions", "serde", + "serde_json", + "sqlx", + "tempfile", + "tokio", + "tracing", "uuid", ] diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index cf8b23bfe1..94846e19c1 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -333,6 +333,7 @@ bitflags = "2.13.1" blake3 = "1.8.2" bm25 = "2.3.2" bytes = "1.10.1" +caseless = "0.2.2" chardetng = "0.1.17" chrono = "0.4.43" clap = "4" diff --git a/codex-rs/ext/agent-message-board/Cargo.toml b/codex-rs/ext/agent-message-board/Cargo.toml index 077fba027d..cbfd802062 100644 --- a/codex-rs/ext/agent-message-board/Cargo.toml +++ b/codex-rs/ext/agent-message-board/Cargo.toml @@ -14,8 +14,19 @@ test = false workspace = true [dependencies] +caseless = { workspace = true } chrono = { workspace = true, features = ["serde"] } codex-protocol = { workspace = true } +codex-state = { workspace = true } futures = { workspace = true } serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +sqlx = { workspace = true } +tokio = { workspace = true, features = ["fs", "rt", "sync"] } +tracing = { workspace = true } uuid = { workspace = true, features = ["serde", "v7"] } + +[dev-dependencies] +pretty_assertions = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/codex-rs/ext/agent-message-board/src/lib.rs b/codex-rs/ext/agent-message-board/src/lib.rs index fd50b37455..3ee2d314b5 100644 --- a/codex-rs/ext/agent-message-board/src/lib.rs +++ b/codex-rs/ext/agent-message-board/src/lib.rs @@ -5,6 +5,7 @@ mod api; mod host; +mod local; mod types; pub use api::AgentMessageBoard; @@ -24,6 +25,7 @@ pub use api::ThreadQuery; pub use api::ThreadSort; pub use host::MessageBoardHost; pub use host::NotificationDelivery; +pub use local::LocalAgentMessageBoard; pub use types::ChannelSummary; pub use types::Page; pub use types::PostContent; diff --git a/codex-rs/ext/agent-message-board/src/local.rs b/codex-rs/ext/agent-message-board/src/local.rs new file mode 100644 index 0000000000..7346dd6d06 --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/local.rs @@ -0,0 +1,463 @@ +//! SQLite-backed boards. The tree ID scopes every read and write. +//! +//! Immediate transactions serialize mutations across independently opened handles. +//! Accepted posts survive runtime unload and process restart. + +use crate::ChannelSummary; +use crate::CreateChannelRequest; +use crate::MessageBoardHost; +use crate::PostContent; +use crate::PostDestination; +use crate::PostMetadata; +use crate::PostRequest; +use crate::ReadPostRequest; +use crate::SubscriptionChange; +use crate::SubscriptionRequest; +use crate::SubscriptionState; +use crate::SubscriptionTarget; +use caseless::default_case_fold_str; +use chrono::DateTime; +use chrono::Utc; +use codex_protocol::AgentPath; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result; +use codex_state::SqliteConfig; +use futures::StreamExt; +use serde::Deserialize; +use serde::Serialize; +use sqlx::Row; +use sqlx::SqliteConnection; +use sqlx::SqlitePool; +use std::collections::HashSet; +use std::sync::Arc; +use uuid::Uuid; + +const MAX_POST_BYTES: usize = 64 * 1024; +const MAX_CHANNEL_BYTES: usize = 128; +const MAX_READ_CHARS: usize = 20_000; + +const SCHEMA: &str = " +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, + PRIMARY KEY(board,name) +); +CREATE TABLE IF NOT EXISTS posts ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + board TEXT NOT NULL, id TEXT NOT NULL, channel TEXT NOT NULL, root TEXT NOT NULL, + author TEXT NOT NULL, timestamp INTEGER NOT NULL, body_search TEXT NOT NULL, + payload TEXT NOT NULL, request_id TEXT NOT NULL, request TEXT NOT NULL, + UNIQUE(board,id), UNIQUE(board,request_id) +); +CREATE INDEX IF NOT EXISTS posts_board_channel ON posts(board,channel,seq); +CREATE INDEX IF NOT EXISTS posts_board_root ON posts(board,root,seq); +CREATE TABLE IF NOT EXISTS subscriptions ( + board TEXT NOT NULL, target TEXT NOT NULL, agent TEXT NOT NULL, + PRIMARY KEY(board,target,agent) +);"; + +#[derive(Clone)] +pub struct LocalAgentMessageBoard { + identity: SessionId, + pool: SqlitePool, + host: Arc, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct StoredPost { + metadata: PostMetadata, + text: String, +} + +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. + pub async fn open( + sqlite: &SqliteConfig, + identity: SessionId, + host: Arc, + ) -> Result { + tokio::fs::create_dir_all(sqlite.home()).await?; + let pool = sqlite + .open_read_write_pool(&sqlite.home().join("agent_message_board_1.sqlite")) + .await + .map_err(storage_error)?; + sqlx::raw_sql(SCHEMA) + .execute(&pool) + .await + .map_err(storage_error)?; + Ok(Self { + identity, + pool, + host, + }) + } + + pub async fn create_channel( + &self, + caller: ThreadId, + request: CreateChannelRequest, + ) -> Result { + validate_channel(&request.channel_name)?; + let author = self.host.agent_path(caller).await?; + let now = self.host.current_time(caller).await?; + let mut tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(storage_error)?; + self.insert_channel(&mut tx, &request.channel_name, &author, now) + .await?; + if request.subscription == SubscriptionChange::Subscribe { + self.subscribe( + &mut tx, + &SubscriptionTarget::Channel(request.channel_name.clone()), + caller, + ) + .await?; + } + let summary = self.channel_summary(&mut tx, &request.channel_name).await?; + tx.commit().await.map_err(storage_error)?; + Ok(summary) + } + + /// Once started, finish the accepted write and fanout even if the tool caller + /// disconnects. Delivery is attempted once; retries never replay old notices + /// into a recipient's later turn. Once committed, notification failures are + /// logged without failing the post; the persisted content remains readable. + pub async fn post(&self, caller: ThreadId, request: PostRequest) -> Result { + let board = self.clone(); + tokio::spawn(async move { board.post_inner(caller, request).await }) + .await + .map_err(storage_error)? + } + + async fn post_inner(&self, caller: ThreadId, request: PostRequest) -> Result { + if request.text.len() > MAX_POST_BYTES + || request.text.is_empty() + || request.request_id.is_empty() + || request.request_id.len() > 512 + || request.agents_to_notify.len() > 256 + { + return Err(invalid( + "post text, request ID or recipient count exceeds the board limits", + )); + } + let author = self.host.agent_path(caller).await?; + let request_id = format!("{caller}:{}", request.request_id); + let request_json = serde_json::to_string(&request).map_err(storage_error)?; + let mut conn = self.pool.acquire().await.map_err(storage_error)?; + if let Some(post) = self + .existing_post(&mut conn, &request_id, &request_json) + .await? + { + return Ok(post.metadata); + } + drop(conn); + let mut recipients = HashSet::new(); + for path in &request.agents_to_notify { + recipients.insert(self.host.resolve_agent(path.clone()).await?); + } + let now = self.host.current_time(caller).await?; + let mut tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(storage_error)?; + if let Some(post) = self + .existing_post(&mut tx, &request_id, &request_json) + .await? + { + return Ok(post.metadata); + } + let id = Uuid::now_v7(); + let (channel, root, target) = match &request.destination { + PostDestination::Channel(channel) => { + let exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM channels WHERE board=? AND name=?)", + ) + .bind(self.identity.to_string()) + .bind(channel) + .fetch_one(&mut *tx) + .await + .map_err(storage_error)?; + if !exists { + return Err(invalid("channel not found in this board")); + } + ( + channel.clone(), + id, + SubscriptionTarget::Channel(channel.clone()), + ) + } + PostDestination::NewChannel(channel) => { + validate_channel(channel)?; + self.insert_channel(&mut tx, channel, &author, now).await?; + ( + channel.clone(), + id, + SubscriptionTarget::Channel(channel.clone()), + ) + } + PostDestination::Thread(root) => { + let post = self.load_post(&mut tx, *root).await?; + if post.metadata.thread_id != *root { + return Err(invalid("thread_id must identify a top-level post")); + } + ( + post.metadata.channel_name, + *root, + SubscriptionTarget::Thread(*root), + ) + } + }; + let subscribed = sqlx::query_scalar::<_, String>( + "SELECT agent FROM subscriptions WHERE board=? AND target=?", + ) + .bind(self.identity.to_string()) + .bind(target_key(&target)?) + .fetch_all(&mut *tx) + .await + .map_err(storage_error)?; + for recipient in subscribed { + recipients.insert(ThreadId::from_string(&recipient).map_err(storage_error)?); + } + let post = StoredPost { + metadata: PostMetadata { + message_id: id, + channel_name: channel.clone(), + author: author.clone(), + thread_id: root, + created_at: now, + }, + text: request.text.clone(), + }; + 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(serde_json::to_string(&post).map_err(storage_error)?).bind(request_id).bind(request_json) + .execute(&mut *tx).await.map_err(storage_error)?; + self.subscribe(&mut tx, &SubscriptionTarget::Thread(root), caller) + .await?; + tx.commit().await.map_err(storage_error)?; + + // A committed post succeeds even if a best-effort notice cannot be delivered. + futures::stream::iter(recipients) + .for_each_concurrent(/*limit*/ 16, |recipient| { + let metadata = post.metadata.clone(); + async move { + if let Err(error) = self.host.notify(recipient, metadata).await { + tracing::warn!(%recipient, %error, "Failed to deliver message-board notification"); + } + } + }) + .await; + Ok(post.metadata) + } + + pub async fn set_subscription( + &self, + caller: ThreadId, + request: SubscriptionRequest, + ) -> Result { + let caller_path = self.host.agent_path(caller).await?; + let target_path = request.target_agent.unwrap_or(caller_path); + let target_agent = self.host.resolve_agent(target_path.clone()).await?; + let mut tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(storage_error)?; + let (channel, root, last) = match &request.target { + SubscriptionTarget::Channel(name) => { + let summary = self.channel_summary(&mut tx, name).await?; + (name.clone(), None, summary.last_message_id) + } + SubscriptionTarget::Thread(root) => { + let post = self.load_post(&mut tx, *root).await?; + if post.metadata.thread_id != *root { + return Err(invalid("thread_id must identify a top-level post")); + } + let last: String = sqlx::query_scalar("SELECT id FROM posts WHERE board=? AND root=? ORDER BY timestamp DESC, seq DESC LIMIT 1") + .bind(self.identity.to_string()).bind(root.to_string()).fetch_one(&mut *tx).await.map_err(storage_error)?; + ( + post.metadata.channel_name, + Some(*root), + Some(Uuid::parse_str(&last).map_err(storage_error)?), + ) + } + }; + let enabled = request.change == SubscriptionChange::Subscribe; + match request.change { + SubscriptionChange::Subscribe => { + self.subscribe(&mut tx, &request.target, target_agent) + .await? + } + SubscriptionChange::Unsubscribe => { + sqlx::query("DELETE FROM subscriptions WHERE board=? AND target=? AND agent=?") + .bind(self.identity.to_string()) + .bind(target_key(&request.target)?) + .bind(target_agent.to_string()) + .execute(&mut *tx) + .await + .map_err(storage_error)?; + } + } + tx.commit().await.map_err(storage_error)?; + Ok(SubscriptionState { + channel_name: channel, + thread_id: root, + target_agent: target_path, + enabled, + last_message_id: last, + }) + } + + pub async fn read_post( + &self, + caller: ThreadId, + request: ReadPostRequest, + ) -> Result { + self.host.agent_path(caller).await?; + let mut conn = self.pool.acquire().await.map_err(storage_error)?; + let post = self.load_post(&mut conn, request.message_id).await?; + let n_chars = post.text.chars().count(); + let offset = (request.offset_chars as usize).min(n_chars); + let text: String = post + .text + .chars() + .skip(offset) + .take((request.limit_chars.get() as usize).min(MAX_READ_CHARS)) + .collect(); + let next_offset_chars = offset + text.chars().count(); + Ok(PostContent { + metadata: post.metadata, + text, + n_chars, + next_offset_chars, + }) + } + + async fn insert_channel( + &self, + conn: &mut SqliteConnection, + name: &str, + author: &AgentPath, + now: DateTime, + ) -> Result<()> { + let inserted = sqlx::query("INSERT OR IGNORE INTO channels(board,name,name_search,created_at,timestamp,author) VALUES(?,?,?,?,?,?)") + .bind(self.identity.to_string()).bind(name).bind(default_case_fold_str(name)).bind(now.to_rfc3339()).bind(now.timestamp_micros()).bind(author.to_string()) + .execute(conn).await.map_err(storage_error)?.rows_affected(); + if inserted == 0 { + return Err(invalid("channel already exists")); + } + Ok(()) + } + + async fn subscribe( + &self, + conn: &mut SqliteConnection, + target: &SubscriptionTarget, + agent: ThreadId, + ) -> Result<()> { + sqlx::query("INSERT OR IGNORE INTO subscriptions(board,target,agent) VALUES(?,?,?)") + .bind(self.identity.to_string()) + .bind(target_key(target)?) + .bind(agent.to_string()) + .execute(conn) + .await + .map_err(storage_error)?; + Ok(()) + } + + async fn load_post(&self, conn: &mut SqliteConnection, id: Uuid) -> Result { + let payload: Option = + sqlx::query_scalar("SELECT payload FROM posts WHERE board=? AND id=?") + .bind(self.identity.to_string()) + .bind(id.to_string()) + .fetch_optional(conn) + .await + .map_err(storage_error)?; + serde_json::from_str(&payload.ok_or_else(|| invalid("post not found in this board"))?) + .map_err(storage_error) + } + + async fn existing_post( + &self, + conn: &mut SqliteConnection, + request_id: &str, + request: &str, + ) -> Result> { + let row = sqlx::query("SELECT payload,request FROM posts WHERE board=? AND request_id=?") + .bind(self.identity.to_string()) + .bind(request_id) + .fetch_optional(conn) + .await + .map_err(storage_error)?; + row.map(|row| { + if row.get::("request") != request { + return Err(invalid("request ID was already used for a different post")); + } + serde_json::from_str(row.get("payload")).map_err(storage_error) + }) + .transpose() + } + + async fn channel_summary( + &self, + conn: &mut SqliteConnection, + name: &str, + ) -> Result { + let row = sqlx::query( + "SELECT c.created_at,c.author, + (SELECT COUNT(*) FROM posts p WHERE p.board=c.board AND p.channel=c.name) AS message_count, + (SELECT p.id FROM posts p WHERE p.board=c.board AND p.channel=c.name ORDER BY p.timestamp DESC,p.seq DESC LIMIT 1) AS last_message_id + FROM channels c WHERE c.board=? AND c.name=?", + ) + .bind(self.identity.to_string()) + .bind(name) + .fetch_optional(conn) + .await + .map_err(storage_error)? + .ok_or_else(|| invalid("channel not found in this board"))?; + Ok(ChannelSummary { + channel_name: name.to_string(), + created_at: DateTime::parse_from_rfc3339(row.get("created_at")) + .map_err(storage_error)? + .with_timezone(&Utc), + created_by: AgentPath::try_from(row.get::("author")).map_err(invalid)?, + message_count: row.get::("message_count") as usize, + last_message_id: row + .get::, _>("last_message_id") + .map(|id| Uuid::parse_str(&id)) + .transpose() + .map_err(storage_error)?, + }) + } +} + +fn validate_channel(name: &str) -> Result<()> { + if name.is_empty() + || name.len() > MAX_CHANNEL_BYTES + || name.trim() != name + || name.chars().any(char::is_control) + { + return Err(invalid( + "channel names must contain 1–128 bytes without edge whitespace or control characters", + )); + } + Ok(()) +} + +fn invalid(message: impl Into) -> CodexErr { + CodexErr::InvalidRequest(message.into()) +} + +fn storage_error(error: impl std::fmt::Display) -> CodexErr { + CodexErr::Io(std::io::Error::other(error.to_string())) +} + +fn target_key(target: &SubscriptionTarget) -> Result { + serde_json::to_string(target).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 new file mode 100644 index 0000000000..ac012b3092 --- /dev/null +++ b/codex-rs/ext/agent-message-board/tests/local_board.rs @@ -0,0 +1,276 @@ +//! Exercises real local storage through independent handles and host capabilities. + +use std::collections::HashMap; +use std::num::NonZeroU32; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; + +use chrono::DateTime; +use chrono::Utc; +use codex_agent_message_board_extension::*; +use codex_protocol::AgentPath; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result; +use codex_state::SqliteConfig; +use futures::future::BoxFuture; +use pretty_assertions::assert_eq; + +struct Host { + members: HashMap, + active: AtomicBool, + fail_notifications: AtomicBool, + notifications: Mutex>, +} + +impl MessageBoardHost for Host { + fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result> { + Box::pin(async move { + self.members + .get(&caller) + .cloned() + .ok_or_else(|| CodexErr::ThreadNotFound(caller)) + }) + } + + fn resolve_agent(&self, path: AgentPath) -> BoxFuture<'_, Result> { + Box::pin(async move { + self.members + .iter() + .find_map(|(id, member)| (*member == path).then_some(*id)) + .ok_or_else(|| CodexErr::InvalidRequest("unknown agent".into())) + }) + } + + fn current_time(&self, _caller: ThreadId) -> BoxFuture<'_, Result>> { + Box::pin(async { + Ok(DateTime::parse_from_rfc3339("2026-09-18T12:00:00Z") + .map_err(|error| CodexErr::Io(std::io::Error::other(error)))? + .with_timezone(&Utc)) + }) + } + + fn notify( + &self, + recipient: ThreadId, + post: PostMetadata, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + if self.fail_notifications.load(Ordering::SeqCst) { + return Err(CodexErr::Io(std::io::Error::other( + "notification transport failed", + ))); + } + if !self.active.load(Ordering::SeqCst) { + return Ok(NotificationDelivery::SkippedInactive); + } + self.notifications + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push((recipient, post)); + Ok(NotificationDelivery::Accepted) + }) + } +} + +#[tokio::test] +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 root = ThreadId::new(); + let child = ThreadId::new(); + let child_path = AgentPath::root().join("worker").unwrap(); + let host = Arc::new(Host { + members: [(root, AgentPath::root()), (child, child_path.clone())].into(), + fail_notifications: AtomicBool::new(false), + active: AtomicBool::new(true), + notifications: Mutex::default(), + }); + let tree = SessionId::from(root); + let first = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + .await + .unwrap(); + first + .create_channel( + child, + CreateChannelRequest { + channel_name: "proofs".into(), + subscription: SubscriptionChange::Subscribe, + }, + ) + .await + .unwrap(); + let request = PostRequest { + request_id: "call-1".into(), + destination: PostDestination::Channel("proofs".into()), + text: "é🦀 proof".into(), + agents_to_notify: vec![child_path.clone(), child_path.clone()], + }; + let second = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + .await + .unwrap(); + let (metadata, duplicate) = tokio::join!( + first.post(root, request.clone()), + second.post(root, request.clone()), + ); + let metadata = metadata.unwrap(); + assert_eq!(duplicate.unwrap(), metadata); + drop(second); + assert_eq!( + *host.notifications.lock().unwrap(), + vec![(child, metadata.clone())] + ); + drop(first); + + let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + .await + .unwrap(); + assert_eq!(resumed.post(root, request).await.unwrap(), metadata); + assert_eq!(host.notifications.lock().unwrap().len(), 1); + let content = resumed + .read_post( + child, + ReadPostRequest { + message_id: metadata.message_id, + offset_chars: 1, + limit_chars: NonZeroU32::new(2).unwrap(), + }, + ) + .await + .unwrap(); + assert_eq!( + content, + PostContent { + metadata: metadata.clone(), + text: "🦀 ".into(), + n_chars: 8, + next_offset_chars: 3, + } + ); + resumed + .set_subscription( + root, + SubscriptionRequest { + target: SubscriptionTarget::Channel("proofs".into()), + target_agent: Some(child_path.clone()), + change: SubscriptionChange::Unsubscribe, + }, + ) + .await + .unwrap(); + resumed + .set_subscription( + root, + SubscriptionRequest { + target: SubscriptionTarget::Thread(metadata.message_id), + target_agent: Some(child_path), + change: SubscriptionChange::Subscribe, + }, + ) + .await + .unwrap(); + let reply = resumed + .post( + root, + PostRequest { + request_id: "reply".into(), + destination: PostDestination::Thread(metadata.message_id), + text: "done".into(), + agents_to_notify: Vec::new(), + }, + ) + .await + .unwrap(); + let mut received = host.notifications.lock().unwrap().clone(); + received.sort_by_key(|(id, post)| (id.to_string(), post.message_id)); + let mut expected = vec![ + (child, metadata.clone()), + (root, reply.clone()), + (child, reply), + ]; + expected.sort_by_key(|(id, post)| (id.to_string(), post.message_id)); + assert_eq!(received, expected); + + let other = LocalAgentMessageBoard::open(&sqlite, SessionId::new(), host.clone()) + .await + .unwrap(); + assert!( + other + .read_post( + root, + ReadPostRequest { + message_id: metadata.message_id, + offset_chars: 0, + limit_chars: NonZeroU32::new(20).unwrap(), + } + ) + .await + .is_err() + ); +} + +#[tokio::test] +async fn failed_requests_do_not_create_channels_or_notify_inactive_agents() { + let dir = tempfile::tempdir().unwrap(); + let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let root = ThreadId::new(); + let host = Arc::new(Host { + members: [(root, AgentPath::root())].into(), + fail_notifications: AtomicBool::new(false), + active: AtomicBool::new(false), + notifications: Mutex::default(), + }); + let board = LocalAgentMessageBoard::open(&sqlite, SessionId::from(root), host.clone()) + .await + .unwrap(); + let mut request = PostRequest { + request_id: "post".into(), + destination: PostDestination::NewChannel("work".into()), + text: "first".into(), + agents_to_notify: vec![AgentPath::root().join("unknown").unwrap()], + }; + assert!(board.post(root, request.clone()).await.is_err()); + request.agents_to_notify = vec![AgentPath::root()]; + let posted = board.post(root, request.clone()).await.unwrap(); + assert_eq!(*host.notifications.lock().unwrap(), Vec::new()); + host.active.store(true, Ordering::SeqCst); + assert_eq!(board.post(root, request.clone()).await.unwrap(), posted); + assert_eq!(*host.notifications.lock().unwrap(), Vec::new()); + request.text = "different".into(); + assert!(board.post(root, request).await.is_err()); + + // Notification failure must not turn a committed write into a failed post. + host.fail_notifications.store(true, Ordering::SeqCst); + let reply = PostRequest { + request_id: "reply".into(), + destination: PostDestination::Thread(posted.thread_id), + text: "saved even if the notice fails".into(), + agents_to_notify: vec![], + }; + let saved = board.post(root, reply.clone()).await.unwrap(); + host.fail_notifications.store(false, Ordering::SeqCst); + assert_eq!(board.post(root, reply).await.unwrap(), saved); + assert_eq!(*host.notifications.lock().unwrap(), Vec::new()); + assert_eq!( + board + .read_post( + root, + ReadPostRequest { + message_id: saved.message_id, + offset_chars: 0, + limit_chars: NonZeroU32::new(100).unwrap(), + } + ) + .await + .unwrap(), + PostContent { + metadata: saved, + text: "saved even if the notice fails".into(), + n_chars: 30, + next_offset_chars: 30, + } + ); +}