From b719a7fb7fe0c43a505f30a933b63d13e2a13e80 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Fri, 23 Oct 2026 12:06:43 +0000 Subject: [PATCH] Add silent append support to message boards (#56461) ## Why Recording evidence on a message board needs a write operation that does not automatically subscribe the author or notify existing subscribers. ## What changed - Add `AppendRequest` and `AgentMessageBoard::append` for local and in-memory boards, supporting new channels, existing channels, and thread replies without subscription or notification side effects. - Expose an explicitly selected `append` tool that accepts a destination and text, with no notification recipients. - Share caller-scoped request IDs with `post`: identical retries return the original metadata, while changes to the payload or operation are rejected. Local appends retain write permission checks and durable storage. - Reject remote appends as unsupported without issuing an HTTP request. ## Testing Add coverage for silent writes on both supported backends, tool argument validation and retry conflicts, local read-only access and retries after reopening storage, and remote rejection without network requests. GitOrigin-RevId: 01fb9d3e2cfcb3de362b28f2bf2ee83bafada80d --- .../agent-message-board-client/src/client.rs | 13 + .../tests/remote_board.rs | 21 ++ codex-rs/message-board/src/api.rs | 24 ++ codex-rs/message-board/src/in_memory.rs | 61 +++- .../message-board/src/in_memory/queries.rs | 9 + codex-rs/message-board/src/lib.rs | 1 + codex-rs/message-board/src/local.rs | 99 +++++-- codex-rs/message-board/src/local/queries.rs | 9 + codex-rs/message-board/src/tools.rs | 36 ++- codex-rs/message-board/src/tools/arguments.rs | 9 + codex-rs/message-board/src/tools/selection.rs | 3 + codex-rs/message-board/src/tools/spec.rs | 10 +- codex-rs/message-board/tests/local_board.rs | 271 +++++++++++++++++- 13 files changed, 517 insertions(+), 49 deletions(-) diff --git a/codex-rs/agent-message-board-client/src/client.rs b/codex-rs/agent-message-board-client/src/client.rs index cdff62033a..692822d2f8 100644 --- a/codex-rs/agent-message-board-client/src/client.rs +++ b/codex-rs/agent-message-board-client/src/client.rs @@ -18,6 +18,7 @@ use chrono::Utc; use codex_http_client::HttpClient; use codex_http_client::HttpResponse; use codex_message_board::AgentMessageBoard; +use codex_message_board::AppendRequest; use codex_message_board::ChannelPage; use codex_message_board::ChannelQuery; use codex_message_board::ChannelSummary; @@ -325,6 +326,18 @@ impl AgentMessageBoard for RemoteAgentMessageBoard { ); operation!(list_channels, ListChannels, ChannelQuery, ChannelPage); operation!(post, Post, PostRequest, PostMetadata); + fn append( + &self, + _caller: ThreadId, + _request: AppendRequest, + ) -> BoxFuture<'_, Result> { + // The remote protocol has no silent write operation. Posting is not equivalent. + Box::pin(async { + Err(CodexErr::InvalidRequest( + "remote message boards do not support silent append".into(), + )) + }) + } operation!(list_threads, ListThreads, ThreadQuery, Page); operation!(search_posts, SearchPosts, PostQuery, Page); operation!(read_thread, ReadThread, ReadThreadRequest, ThreadPage); diff --git a/codex-rs/agent-message-board-client/tests/remote_board.rs b/codex-rs/agent-message-board-client/tests/remote_board.rs index d0ff25a52c..e9d8ba6516 100644 --- a/codex-rs/agent-message-board-client/tests/remote_board.rs +++ b/codex-rs/agent-message-board-client/tests/remote_board.rs @@ -7,6 +7,7 @@ use codex_agent_message_board_client::BoardNotification; use codex_agent_message_board_client::RemoteAgentMessageBoard; use codex_http_client::HttpClientBuilder; use codex_message_board::AgentMessageBoard; +use codex_message_board::AppendRequest; use codex_message_board::PostContent; use codex_message_board::PostDestination; use codex_message_board::PostMetadata; @@ -129,6 +130,26 @@ async fn remote_board_preserves_the_host_contract() -> anyhow::Result<()> { .await; let board_handle: &dyn AgentMessageBoard = &client; assert_eq!(board_handle.identity(), board); + let requests_before = server.received_requests().await.unwrap().len(); + let unsupported = board_handle + .append( + caller, + AppendRequest { + request_id: post.request_id.clone(), + destination: post.destination.clone(), + text: post.text.clone(), + }, + ) + .await + .unwrap_err(); + assert!(matches!( + unsupported.details(), + CodexErrorDetails::InvalidRequest(_) + )); + assert_eq!( + server.received_requests().await.unwrap().len(), + requests_before + ); assert!(board_handle.post(caller, post.clone()).await.is_err()); clock_available.store(false, Ordering::Relaxed); Mock::given(method("POST")) diff --git a/codex-rs/message-board/src/api.rs b/codex-rs/message-board/src/api.rs index 2f1b4f9198..817d28f472 100644 --- a/codex-rs/message-board/src/api.rs +++ b/codex-rs/message-board/src/api.rs @@ -57,6 +57,15 @@ pub trait AgentMessageBoard: Send + Sync { /// Creating a channel while posting also subscribes its author to new roots there. fn post(&self, caller: ThreadId, request: PostRequest) -> BoxFuture<'_, Result>; + /// Writes without automatic subscriptions or notifications. Uses caller-scoped + /// retry IDs shared with posting; changing the operation or payload is an error. + /// Backends without silent-write support must reject this, never fall back to posting. + fn append( + &self, + caller: ThreadId, + request: AppendRequest, + ) -> BoxFuture<'_, Result>; + fn list_threads( &self, caller: ThreadId, @@ -183,6 +192,21 @@ pub struct PostRequest { pub agents_to_notify: Vec, } +#[derive(Clone, Copy, PartialEq, Eq)] +pub(crate) enum PostBehavior { + Discussion, + SilentAppend, +} + +/// Request to append evidence without subscribing its author or notifying anyone. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AppendRequest { + /// Caller-scoped invocation identity, shared with posting to reject conflicting retries. + pub request_id: String, + pub destination: PostDestination, + pub text: String, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ThreadSort { diff --git a/codex-rs/message-board/src/in_memory.rs b/codex-rs/message-board/src/in_memory.rs index 1c1b4fda50..04be204dc4 100644 --- a/codex-rs/message-board/src/in_memory.rs +++ b/codex-rs/message-board/src/in_memory.rs @@ -1,7 +1,9 @@ //! Tree-owned message boards for training. Agent handles share state; nothing is written to disk. //! Mutations and recipient selection are atomic. Host callbacks run outside the state lock. +//! Silent appends share the write path without changing subscriptions or delivering notices. //! Opening a board prunes registry entries whose state has been released. +use crate::AppendRequest; use crate::ChannelSummary; use crate::CreateChannelRequest; use crate::MessageBoardHost; @@ -13,6 +15,7 @@ use crate::SubscriptionChange; use crate::SubscriptionRequest; use crate::SubscriptionState; use crate::SubscriptionTarget; +use crate::api::PostBehavior; use caseless::default_case_fold_str; use chrono::DateTime; use chrono::Utc; @@ -92,6 +95,7 @@ struct Channel { struct Post { metadata: PostMetadata, request: PostRequest, + behavior: PostBehavior, search: String, n_chars: usize, replies: Vec, @@ -136,12 +140,13 @@ impl State { &self, caller: ThreadId, request: &PostRequest, + behavior: PostBehavior, ) -> Result> { self.requests .get(&(caller, request.request_id.clone())) .map(|index| { let post = &self.posts[*index]; - if post.request != *request { + if post.request != *request || post.behavior != behavior { return Err(invalid("request ID was already used for a different post")); } Ok(post.metadata.clone()) @@ -230,14 +235,42 @@ impl InMemoryAgentMessageBoard { } async fn post(&self, caller: ThreadId, request: PostRequest) -> Result { + self.write(caller, request, PostBehavior::Discussion).await + } + + async fn append(&self, caller: ThreadId, request: AppendRequest) -> Result { + self.write( + caller, + PostRequest { + request_id: request.request_id, + destination: request.destination, + text: request.text, + agents_to_notify: Vec::new(), + }, + PostBehavior::SilentAppend, + ) + .await + } + + async fn write( + &self, + caller: ThreadId, + request: PostRequest, + behavior: PostBehavior, + ) -> Result { // Accepted writes and their one-time fanout outlive cancellation of the tool call. let board = self.clone(); - tokio::spawn(async move { board.post_inner(caller, request).await }) + tokio::spawn(async move { board.post_inner(caller, request, behavior).await }) .await .map_err(|error| CodexErr::Io(std::io::Error::other(error)))? } - async fn post_inner(&self, caller: ThreadId, request: PostRequest) -> Result { + async fn post_inner( + &self, + caller: ThreadId, + request: PostRequest, + behavior: PostBehavior, + ) -> Result { if request.text.is_empty() || request.text.len() > 64 * 1024 || request.request_id.is_empty() @@ -249,7 +282,12 @@ impl InMemoryAgentMessageBoard { )); } let author = self.host.agent_path(caller).await?; - if let Some(post) = self.state.lock().await.existing_post(caller, &request)? { + if let Some(post) = self + .state + .lock() + .await + .existing_post(caller, &request, behavior)? + { return Ok(post); } let mut recipients = HashSet::new(); @@ -259,7 +297,7 @@ impl InMemoryAgentMessageBoard { let now = self.host.current_time(caller).await?; let (metadata, notice) = { let mut state = self.state.lock().await; - if let Some(post) = state.existing_post(caller, &request)? { + if let Some(post) = state.existing_post(caller, &request, behavior)? { return Ok(post); } let id = Uuid::now_v7(); @@ -272,7 +310,9 @@ impl InMemoryAgentMessageBoard { } PostDestination::NewChannel(name) => { state.insert_channel(name, /*description*/ None, author.clone(), now)?; - state.subscribe(SubscriptionTarget::Channel(name.clone()), caller); + if behavior == PostBehavior::Discussion { + state.subscribe(SubscriptionTarget::Channel(name.clone()), caller); + } (name.clone(), id, SubscriptionTarget::Channel(name.clone())) } PostDestination::Thread(root) => { @@ -284,7 +324,9 @@ impl InMemoryAgentMessageBoard { ) } }; - if let Some(subscribers) = state.subscriptions.get(&target) { + if behavior == PostBehavior::Discussion + && let Some(subscribers) = state.subscriptions.get(&target) + { recipients.extend(subscribers.iter().filter_map(|(agent, change)| { (*change == SubscriptionChange::Subscribe).then_some(*agent) })); @@ -307,6 +349,7 @@ impl InMemoryAgentMessageBoard { search: default_case_fold_str(&request.text), n_chars: request.text.chars().count(), request, + behavior, replies: Vec::new(), latest_reply: None, }); @@ -338,7 +381,9 @@ impl InMemoryAgentMessageBoard { thread.latest_reply = Some(index); } } - state.subscribe(SubscriptionTarget::Thread(root), caller); + if behavior == PostBehavior::Discussion { + state.subscribe(SubscriptionTarget::Thread(root), caller); + } if recipients.is_empty() { return Ok(metadata); } diff --git a/codex-rs/message-board/src/in_memory/queries.rs b/codex-rs/message-board/src/in_memory/queries.rs index 3951280f10..b1fc785290 100644 --- a/codex-rs/message-board/src/in_memory/queries.rs +++ b/codex-rs/message-board/src/in_memory/queries.rs @@ -3,6 +3,7 @@ use super::InMemoryAgentMessageBoard; use super::invalid; use crate::AgentMessageBoard; +use crate::AppendRequest; use crate::ChannelPage; use crate::ChannelQuery; use crate::ChannelSummary; @@ -49,6 +50,14 @@ impl AgentMessageBoard for InMemoryAgentMessageBoard { Box::pin(self.post(caller, request)) } + fn append( + &self, + caller: ThreadId, + request: AppendRequest, + ) -> BoxFuture<'_, Result> { + Box::pin(self.append(caller, request)) + } + fn set_subscription( &self, caller: ThreadId, diff --git a/codex-rs/message-board/src/lib.rs b/codex-rs/message-board/src/lib.rs index c087b21a82..5b9355f5f7 100644 --- a/codex-rs/message-board/src/lib.rs +++ b/codex-rs/message-board/src/lib.rs @@ -12,6 +12,7 @@ mod tools; mod types; pub use api::AgentMessageBoard; +pub use api::AppendRequest; pub use api::ChannelDescription; pub use api::ChannelQuery; pub use api::CreateChannelRequest; diff --git a/codex-rs/message-board/src/local.rs b/codex-rs/message-board/src/local.rs index c8112d385c..bad07b7760 100644 --- a/codex-rs/message-board/src/local.rs +++ b/codex-rs/message-board/src/local.rs @@ -6,9 +6,11 @@ //! a board after its root has been permanently deleted. //! Opening storage adds optional channel metadata without invalidating older readers. //! Explicit connections can restrict writes; default handles retain legacy access. +//! Posting subscribes and notifies; host appends share persistence without these side effects. use self::connection::Access; use self::options::LocalBoardOptions; +use crate::AppendRequest; use crate::ChannelDescription; use crate::ChannelPermissions; use crate::ChannelSummary; @@ -23,6 +25,7 @@ use crate::SubscriptionChange; use crate::SubscriptionRequest; use crate::SubscriptionState; use crate::SubscriptionTarget; +use crate::api::PostBehavior; use caseless::default_case_fold_str; use chrono::DateTime; use chrono::Utc; @@ -129,6 +132,32 @@ impl LocalAgentMessageBoard { /// 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 { + self.write(caller, request, PostBehavior::Discussion).await + } + + /// Appends evidence without automatic subscriptions or notifications, even on a discussion board. + /// Uses the same access checks, caller-scoped request IDs and durable write path as posting. + /// Reusing a request ID for a different payload or operation is an error. + pub async fn append(&self, caller: ThreadId, request: AppendRequest) -> Result { + self.write( + caller, + PostRequest { + request_id: request.request_id, + destination: request.destination, + text: request.text, + agents_to_notify: Vec::new(), + }, + PostBehavior::SilentAppend, + ) + .await + } + + async fn write( + &self, + caller: ThreadId, + request: PostRequest, + behavior: PostBehavior, + ) -> Result { if let Some(access) = self.access { access.require_write( self.identity, @@ -139,12 +168,17 @@ impl LocalAgentMessageBoard { )?; } let board = self.clone(); - tokio::spawn(async move { board.post_inner(caller, request).await }) + tokio::spawn(async move { board.post_inner(caller, request, behavior).await }) .await .map_err(storage_error)? } - async fn post_inner(&self, caller: ThreadId, request: PostRequest) -> Result { + async fn post_inner( + &self, + caller: ThreadId, + request: PostRequest, + behavior: PostBehavior, + ) -> Result { if request.text.len() > MAX_POST_BYTES || request.text.is_empty() || request.request_id.is_empty() @@ -157,7 +191,12 @@ impl LocalAgentMessageBoard { } 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 request_json = match behavior { + // Keep existing post fingerprints compatible with persisted retries. + PostBehavior::Discussion => serde_json::to_string(&request), + PostBehavior::SilentAppend => serde_json::to_string(&("append", &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) @@ -167,8 +206,10 @@ impl LocalAgentMessageBoard { } drop(conn); let mut recipients = HashSet::new(); - for path in &request.agents_to_notify { - recipients.insert(self.host.resolve_agent(path.clone()).await?); + if let PostBehavior::Discussion = behavior { + 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.begin_write().await?; @@ -209,12 +250,14 @@ impl LocalAgentMessageBoard { now, ) .await?; - self.subscribe( - &mut tx, - &SubscriptionTarget::Channel(channel.clone()), - caller, - ) - .await?; + if let PostBehavior::Discussion = behavior { + self.subscribe( + &mut tx, + &SubscriptionTarget::Channel(channel.clone()), + caller, + ) + .await?; + } ( channel.clone(), id, @@ -235,17 +278,20 @@ impl LocalAgentMessageBoard { }; // Return one JSON array to avoid handing each subscriber row across // SQLite's worker thread while the write transaction is held. - let subscribed = sqlx::query_scalar::<_, String>( - "SELECT json_group_array(agent) FROM subscriptions WHERE board=? AND target=?", - ) - .bind(self.identity.to_string()) - .bind(target_key(&target)?) - .fetch_one(&mut *tx) - .await - .map_err(storage_error)?; - let subscribed: Vec = serde_json::from_str(&subscribed).map_err(storage_error)?; - for recipient in subscribed { - recipients.insert(ThreadId::from_string(&recipient).map_err(storage_error)?); + if let PostBehavior::Discussion = behavior { + let subscribed = sqlx::query_scalar::<_, String>( + "SELECT json_group_array(agent) FROM subscriptions WHERE board=? AND target=?", + ) + .bind(self.identity.to_string()) + .bind(target_key(&target)?) + .fetch_one(&mut *tx) + .await + .map_err(storage_error)?; + let subscribed: Vec = + serde_json::from_str(&subscribed).map_err(storage_error)?; + for recipient in subscribed { + recipients.insert(ThreadId::from_string(&recipient).map_err(storage_error)?); + } } recipients.remove(&caller); let post = StoredPost { @@ -264,10 +310,15 @@ impl LocalAgentMessageBoard { .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. - self.subscribe(&mut tx, &SubscriptionTarget::Thread(root), caller) - .await?; + if let PostBehavior::Discussion = behavior { + self.subscribe(&mut tx, &SubscriptionTarget::Thread(root), caller) + .await?; + } tx.commit().await.map_err(storage_error)?; + if let PostBehavior::SilentAppend = behavior { + return Ok(post.metadata); + } // A committed post succeeds even if a best-effort notice cannot be delivered. let notice = paging::preview(post.clone(), /*max_chars*/ 150); futures::stream::iter(recipients) diff --git a/codex-rs/message-board/src/local/queries.rs b/codex-rs/message-board/src/local/queries.rs index c7966f85e5..a8f0f71084 100644 --- a/codex-rs/message-board/src/local/queries.rs +++ b/codex-rs/message-board/src/local/queries.rs @@ -11,6 +11,7 @@ use super::paging::direction; use super::paging::preview; use super::storage_error; use crate::AgentMessageBoard; +use crate::AppendRequest; use crate::BoardPermissions; use crate::ChannelPage; use crate::ChannelQuery; @@ -56,6 +57,14 @@ impl AgentMessageBoard for LocalAgentMessageBoard { Box::pin(LocalAgentMessageBoard::post(self, caller, request)) } + fn append( + &self, + caller: ThreadId, + request: AppendRequest, + ) -> BoxFuture<'_, Result> { + Box::pin(LocalAgentMessageBoard::append(self, caller, request)) + } + fn set_subscription( &self, caller: ThreadId, diff --git a/codex-rs/message-board/src/tools.rs b/codex-rs/message-board/src/tools.rs index 3504086de7..b3b854b41e 100644 --- a/codex-rs/message-board/src/tools.rs +++ b/codex-rs/message-board/src/tools.rs @@ -6,6 +6,7 @@ mod spec; use self::selection::MessageBoardTool; use crate::AgentMessageBoard; +use crate::AppendRequest; use crate::ChannelQuery; use crate::CreateChannelRequest; use crate::PageRequest; @@ -351,14 +352,25 @@ impl BoardTool { .await, ) } - MessageBoardTool::Post => { + MessageBoardTool::Post | MessageBoardTool::Append => { let arguments::Post { text, channel_name, new_channel_name, thread_id, agents_to_notify, - } = serde_json::from_str(raw).map_err(model_error)?; + } = if self.tool == MessageBoardTool::Append { + let args: arguments::Append = serde_json::from_str(raw).map_err(model_error)?; + arguments::Post { + text: args.text, + channel_name: args.channel_name, + new_channel_name: args.new_channel_name, + thread_id: args.thread_id, + agents_to_notify: None, + } + } else { + serde_json::from_str(raw).map_err(model_error)? + }; self.check_mutation_budget(budget, /*target_path_bytes*/ 0)?; let destination = match (channel_name, new_channel_name, thread_id) { (Some(channel), None, None) => PostDestination::Channel(channel), @@ -378,12 +390,24 @@ impl BoardTool { runtime_tool_call_id, } => format!("{cell_id}:{runtime_tool_call_id}"), }; - encode( + let request_id = format!("{}:{invocation}", call.turn_id); + encode(if self.tool == MessageBoardTool::Append { + board + .append( + caller, + AppendRequest { + request_id, + destination, + text, + }, + ) + .await + } else { board .post( caller, PostRequest { - request_id: format!("{}:{invocation}", call.turn_id), + request_id, destination, text, agents_to_notify: agents_to_notify @@ -394,8 +418,8 @@ impl BoardTool { .map_err(model_error)?, }, ) - .await, - ) + .await + }) } } } diff --git a/codex-rs/message-board/src/tools/arguments.rs b/codex-rs/message-board/src/tools/arguments.rs index 50154e037b..22c9fa8eb4 100644 --- a/codex-rs/message-board/src/tools/arguments.rs +++ b/codex-rs/message-board/src/tools/arguments.rs @@ -80,3 +80,12 @@ pub(super) struct Post { pub(super) thread_id: Option, pub(super) agents_to_notify: Option>, } + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Append { + pub(super) text: String, + pub(super) channel_name: Option, + pub(super) new_channel_name: Option, + pub(super) thread_id: Option, +} diff --git a/codex-rs/message-board/src/tools/selection.rs b/codex-rs/message-board/src/tools/selection.rs index 6b9a16e86f..86755e5600 100644 --- a/codex-rs/message-board/src/tools/selection.rs +++ b/codex-rs/message-board/src/tools/selection.rs @@ -14,6 +14,8 @@ pub enum MessageBoardTool { Unsubscribe, /// Posts, subscribes the author unless opted out, and notifies recipients. Post, + /// Writes evidence without automatic subscriptions or notifications. + Append, } impl MessageBoardTool { @@ -41,6 +43,7 @@ impl MessageBoardTool { Self::Subscribe => "subscribe", Self::Unsubscribe => "unsubscribe", Self::Post => "post", + Self::Append => "append", } } } diff --git a/codex-rs/message-board/src/tools/spec.rs b/codex-rs/message-board/src/tools/spec.rs index 1940606bf8..79a0a0691e 100644 --- a/codex-rs/message-board/src/tools/spec.rs +++ b/codex-rs/message-board/src/tools/spec.rs @@ -1,4 +1,4 @@ -//! Schemas for shared discussion tools. The host supplies their namespace. +//! Schemas for shared board tools. The host supplies their selection and namespace. use super::selection::MessageBoardTool; use codex_tools::ResponsesApiNamespace; @@ -81,6 +81,11 @@ pub(super) fn tool( ], &["text"], ), + MessageBoardTool::Append => ( + "Store a post in an existing or new channel, or reply using the first post's message ID as thread_id. Exactly one destination is required. Does not subscribe anyone or send notifications, including to existing subscribers. Returns metadata, not the post text.", + &["text", "channel_name", "new_channel_name", "thread_id"], + &["text"], + ), }; let mut properties = serde_json::Map::new(); for field in fields { @@ -88,6 +93,9 @@ pub(super) fn tool( "description" => { json!({"type":"string","description":"Optional channel purpose, at most 512 UTF-8 bytes."}) } + "new_channel_name" if tool == MessageBoardTool::Append => { + json!({"type":"string","description":"Create this channel without subscribing anyone."}) + } "new_channel_name" => { json!({"type":"string","description":"Create and subscribe to this channel."}) } diff --git a/codex-rs/message-board/tests/local_board.rs b/codex-rs/message-board/tests/local_board.rs index 72e136fd6e..9486bcf9dd 100644 --- a/codex-rs/message-board/tests/local_board.rs +++ b/codex-rs/message-board/tests/local_board.rs @@ -1220,8 +1220,10 @@ async fn check_tools_validate_arguments_deduplicate_calls_and_bound_unicode_resu notifications: Mutex::default(), }); let board = backend.open(&sqlite, root.into(), host.clone()).await; + let mut selected = MessageBoardTool::DISCUSSION.to_vec(); + selected.push(MessageBoardTool::Append); let tools = message_board_tools( - &MessageBoardTool::DISCUSSION, + &selected, board.clone(), root, AgentPath::root(), @@ -1241,6 +1243,10 @@ async fn check_tools_validate_arguments_deduplicate_calls_and_bound_unicode_resu "post", json!({"new_channel_name":"low-budget","text":"must not be stored"}), ), + ( + "append", + json!({"new_channel_name":"low-budget-append","text":"must not be stored"}), + ), ("create_channel", json!({"channel_name":"low-budget"})), ] { let mut limited = call(name, args); @@ -1263,15 +1269,23 @@ async fn check_tools_validate_arguments_deduplicate_calls_and_bound_unicode_resu ); small_budget.truncation_policy = TruncationPolicy::Bytes(4096); assert!(tool("create_channel").handle(small_budget).await.is_err()); - assert!( - tool("post") - .handle(call( - "post", - json!({"text":"invalid","channel_name":"a","new_channel_name":"b"}) - )) - .await - .is_err() - ); + for name in ["post", "append"] { + assert!( + tool(name) + .handle(call( + name, + json!({"text":"invalid","channel_name":"a","new_channel_name":"b"}) + )) + .await + .is_err() + ); + } + for args in [ + json!({"text":"no destination"}), + json!({"text":"invalid recipient","new_channel_name":"evidence","agents_to_notify":["worker"]}), + ] { + assert!(tool("append").handle(call("append", args)).await.is_err()); + } assert!( tool("get_channels") .handle(call("get_channels", json!({"limit":0}))) @@ -1424,6 +1438,83 @@ async fn check_tools_validate_arguments_deduplicate_calls_and_bound_unicode_resu .len() <= 8000 ); + + // Explicitly selected Append stores readable evidence, without delivery side effects. + host.notifications.lock().unwrap().clear(); + let append_call = call( + "append", + json!({"text":"Permission revoked.","new_channel_name":"evidence"}), + ); + let result = tool("append").handle(append_call.clone()).await.unwrap(); + let appended: PostMetadata = serde_json::from_str(&result.log_output()).unwrap(); + assert_eq!( + tool("append") + .handle(append_call) + .await + .unwrap() + .log_output(), + result.log_output() + ); + let read = tool("read_post") + .handle(call("read_post", json!({"message_id":appended.message_id}))) + .await + .unwrap(); + assert_eq!( + serde_json::from_str::(&read.log_output()).unwrap(), + PostContent { + metadata: appended.clone(), + text: "Permission revoked.".into(), + n_chars: 19, + next_offset_chars: 19, + } + ); + let mut conflict = call( + "post", + json!({"text":"Permission revoked.","new_channel_name":"evidence"}), + ); + conflict.call_id = "append".into(); + assert!(tool("post").handle(conflict).await.is_err()); + tool("subscribe") + .handle(call( + "subscribe", + json!({"channel_name":"evidence","target_agent":"worker"}), + )) + .await + .unwrap(); + for (index, args) in [ + json!({"text":"Later evidence.","channel_name":"evidence"}), + json!({"text":"Related evidence.","thread_id":appended.thread_id}), + ] + .into_iter() + .enumerate() + { + let mut append = call("append", args); + append.call_id = format!("later-append-{index}"); + tool("append").handle(append).await.unwrap(); + } + // Normal discussion still does not notify the append author: Append created + // neither a channel subscription nor a thread subscription for that author. + for (index, destination) in [ + PostDestination::Channel("evidence".into()), + PostDestination::Thread(appended.thread_id), + ] + .into_iter() + .enumerate() + { + board + .post( + child, + PostRequest { + request_id: format!("ordinary-reply-{index}"), + destination, + text: "Discuss this evidence.".into(), + agents_to_notify: Vec::new(), + }, + ) + .await + .unwrap(); + } + assert_eq!(*host.notifications.lock().unwrap(), Vec::new()); } #[test] @@ -1534,6 +1625,166 @@ async fn channel_descriptions_migrate_legacy_storage_and_survive_reopen() { ); } +#[tokio::test] +async fn silent_appends_preserve_evidence_and_explicit_subscriptions() +-> std::result::Result<(), Box> { + 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 host = Arc::new(Host { + clock: AtomicI64::default(), + agent_path_calls: AtomicUsize::default(), + members: [ + (root, AgentPath::root()), + (child, AgentPath::root().join("worker")?), + ] + .into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::new(false), + notifications: Mutex::default(), + }); + let options = LocalBoardOptions { + storage_namespace: "evidence_records".into(), + }; + let descriptor = LocalBoard::new(sqlite.clone(), root.into()).with_options(options); + let writer = descriptor.read_write().open(host.clone()).await?; + let reader = descriptor.read_only().open(host.clone()).await?; + // Include characters requiring JSON escaping in the persisted subscription target. + let channel = "audit\"\\é"; + let request = AppendRequest { + request_id: "original".into(), + destination: PostDestination::NewChannel(channel.into()), + text: "Deployment was revoked.".into(), + }; + let original = writer.append(root, request.clone()).await?; + assert_eq!(writer.append(root, request.clone()).await?, original); + assert!( + reader + .append( + child, + AppendRequest { + request_id: "reader".into(), + destination: PostDestination::Channel(channel.into()), + ..request.clone() + }, + ) + .await + .is_err() + ); + assert!( + writer + .append( + root, + AppendRequest { + text: "A different decision".into(), + ..request.clone() + }, + ) + .await + .is_err() + ); + assert_eq!( + reader + .read_post( + child, + ReadPostRequest { + message_id: original.message_id, + offset_chars: 0, + limit_chars: NonZeroU32::MAX, + } + ) + .await?, + PostContent { + metadata: original.clone(), + text: request.text.clone(), + n_chars: request.text.chars().count(), + next_offset_chars: request.text.chars().count(), + } + ); + writer + .create_channel( + root, + CreateChannelRequest { + channel_name: "release".into(), + description: Some("Release evidence".to_owned().try_into()?), + subscription: SubscriptionChange::Unsubscribe, + }, + ) + .await?; + for name in [channel, "release"] { + reader + .set_subscription( + child, + SubscriptionRequest { + target: SubscriptionTarget::Channel(name.into()), + change: SubscriptionChange::Subscribe, + target_agent: None, + }, + ) + .await?; + } + let later_request = AppendRequest { + request_id: "later".into(), + destination: PostDestination::Channel(channel.into()), + ..request.clone() + }; + let later = writer.append(root, later_request.clone()).await?; + // Retry keys belong to the caller, so another author may use the same ID. + assert_ne!(writer.append(child, later_request).await?, later); + assert!(host.notifications.lock().unwrap().is_empty()); + let pool = sqlite + .open_read_only_pool( + &dir.path().join("message_board_evidence_records_1.sqlite"), + /*busy_timeout*/ None, + ) + .await?; + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM subscriptions WHERE agent=?") + .bind(root.to_string()) + .fetch_one(&pool) + .await?, + 0 + ); + // Normal posts on this same board still subscribe their author and notify once. + let discussion = PostRequest { + request_id: "discussion".into(), + destination: PostDestination::Channel(channel.into()), + text: "Discuss the latest evidence.".into(), + agents_to_notify: Vec::new(), + }; + let posted = writer.post(root, discussion.clone()).await?; + assert_eq!(writer.post(root, discussion.clone()).await?, posted); + assert_eq!( + sqlx::query_scalar::<_, String>("SELECT target FROM subscriptions WHERE agent=?") + .bind(root.to_string()) + .fetch_all(&pool) + .await?, + vec![serde_json::to_string(&SubscriptionTarget::Thread( + posted.thread_id + ))?] + ); + assert!( + writer + .append( + root, + AppendRequest { + request_id: discussion.request_id.clone(), + destination: discussion.destination.clone(), + text: discussion.text.clone(), + }, + ) + .await + .is_err() + ); + drop((writer, reader)); + let reopened = descriptor.read_write().open(host.clone()).await?; + assert_eq!(reopened.append(root, request).await?, original); + assert_eq!(reopened.post(root, discussion).await?, posted); + assert_eq!(*host.notifications.lock().unwrap(), vec![(child, posted)]); + Ok(()) +} + #[tokio::test] async fn local_board_setup_survives_reopen() -> Result<()> { use serde_json::json;