openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-23 18:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add silent append support to message boards (#56461)

Browse Source 
## 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
This commit is contained in:
felixxia-oai
2026-10-23 12:18:18 +00:00
committed by copyberry
parent bca6cff1e1
commit b719a7fb7f
13 changed files with 517 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/agent-message-board-client/src/client.rs
+13
View File
@@ -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<PostMetadata>> {
// 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<ThreadSummary>);
operation!(search_posts, SearchPosts, PostQuery, Page<PostPreview>);
operation!(read_thread, ReadThread, ReadThreadRequest, ThreadPage);
codex-rs/agent-message-board-client/tests/remote_board.rs
+21
View File
@@ -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"))
codex-rs/message-board/src/api.rs
+24
View File
@@ -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<PostMetadata>>;
/// 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<PostMetadata>>;
fn list_threads(
&self,
caller: ThreadId,
@@ -183,6 +192,21 @@ pub struct PostRequest {
pub agents_to_notify: Vec<AgentPath>,
}
#[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 {
codex-rs/message-board/src/in_memory.rs
+53 -8
View File
@@ -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<usize>,
@@ -136,12 +140,13 @@ impl State {
&self,
caller: ThreadId,
request: &PostRequest,
behavior: PostBehavior,
) -> Result<Option<PostMetadata>> {
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<PostMetadata> {
self.write(caller, request, PostBehavior::Discussion).await
}
async fn append(&self, caller: ThreadId, request: AppendRequest) -> Result<PostMetadata> {
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<PostMetadata> {
// 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<PostMetadata> {
async fn post_inner(
&self,
caller: ThreadId,
request: PostRequest,
behavior: PostBehavior,
) -> Result<PostMetadata> {
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);
}
codex-rs/message-board/src/in_memory/queries.rs
+9
View File
@@ -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<PostMetadata>> {
Box::pin(self.append(caller, request))
}
fn set_subscription(
&self,
caller: ThreadId,
codex-rs/message-board/src/lib.rs
+1
View File
@@ -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;
codex-rs/message-board/src/local.rs
+75 -24
View File
@@ -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<PostMetadata> {
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<PostMetadata> {
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<PostMetadata> {
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<PostMetadata> {
async fn post_inner(
&self,
caller: ThreadId,
request: PostRequest,
behavior: PostBehavior,
) -> Result<PostMetadata> {
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<String> = 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<String> =
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)
codex-rs/message-board/src/local/queries.rs
+9
View File
@@ -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<PostMetadata>> {
Box::pin(LocalAgentMessageBoard::append(self, caller, request))
}
fn set_subscription(
&self,
caller: ThreadId,
codex-rs/message-board/src/tools.rs
+30 -6
View File
@@ -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
})
}
}
}
codex-rs/message-board/src/tools/arguments.rs
+9
View File
@@ -80,3 +80,12 @@ pub(super) struct Post {
pub(super) thread_id: Option<Uuid>,
pub(super) agents_to_notify: Option<Vec<String>>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct Append {
pub(super) text: String,
pub(super) channel_name: Option<String>,
pub(super) new_channel_name: Option<String>,
pub(super) thread_id: Option<Uuid>,
}
codex-rs/message-board/src/tools/selection.rs
+3
View File
@@ -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",
}
}
}
codex-rs/message-board/src/tools/spec.rs
+9 -1
View File
@@ -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."})
}
codex-rs/message-board/tests/local_board.rs
+261 -10
View File
@@ -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::<PostContent>(&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<dyn std::error::Error>> {
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;