Add SQLite storage for local agent message boards (#46966)

## What changed Add `LocalAgentMessageBoard` with persistent channels, posts, and subscriptions scoped by `SessionId`. Support channel creation, posting and replies, subscription changes, and bounded reads using Unicode character offsets. Serialize mutations across independent handles with immediate transactions. Deduplicate retries by caller and request ID, reject changed requests that reuse an ID, and return existing post metadata without repeating notifications. Deliver deduplicated notifications after commit; notification failures leave committed posts successful and readable. ## Testing Add integration tests covering concurrent retries across independent handles, reopening persisted boards, subscription routing, board isolation, Unicode reads, invalid recipients, inactive recipients, and notification failures. GitOrigin-RevId: 0cdf621d19517bf57d527fd729290e08ac99f165
2026-10-20 00:18:49 +00:00 · 2026-09-21 08:23:58 +00:00
parent 0ad18769ad
commit 293de27177
7 changed files with 771 additions and 0 deletions
@@ -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\"]}}",
@@ -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",
 ]

@@ -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"
@@ -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"] }
@@ -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;
@@ -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<dyn MessageBoardHost>,
+}
+
+#[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<dyn MessageBoardHost>,
+    ) -> Result<Self> {
+        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<ChannelSummary> {
+        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<PostMetadata> {
+        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<PostMetadata> {
+        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<SubscriptionState> {
+        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<PostContent> {
+        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<Utc>,
+    ) -> 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<StoredPost> {
+        let payload: Option<String> =
+            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<Option<StoredPost>> {
+        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::<String, _>("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<ChannelSummary> {
+        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::<String, _>("author")).map_err(invalid)?,
+            message_count: row.get::<i64, _>("message_count") as usize,
+            last_message_id: row
+                .get::<Option<String>, _>("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<String>) -> 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<String> {
+    serde_json::to_string(target).map_err(storage_error)
+}
@@ -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<ThreadId, AgentPath>,
+    active: AtomicBool,
+    fail_notifications: AtomicBool,
+    notifications: Mutex<Vec<(ThreadId, PostMetadata)>>,
+}
+
+impl MessageBoardHost for Host {
+    fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result<AgentPath>> {
+        Box::pin(async move {
+            self.members
+                .get(&caller)
+                .cloned()
+                .ok_or_else(|| CodexErr::ThreadNotFound(caller))
+        })
+    }
+
+    fn resolve_agent(&self, path: AgentPath) -> BoxFuture<'_, Result<ThreadId>> {
+        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<DateTime<Utc>>> {
+        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<NotificationDelivery>> {
+        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,
+        }
+    );
+}