From b58b05e062b8ce7354f641f1e0ee059cfc5b7e01 Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 22 Sep 2026 09:24:51 +0000 Subject: [PATCH] Suppress agent message board notifications to the post author (#47236) ## Why Channel and thread subscriptions, or explicit `agents_to_notify`, could cause agents to receive notifications for their own posts. ## What changed Exclude the author from notification recipients, even when explicitly targeted. Preserve the author's subscriptions so replies from other agents still reach them, and document this requirement in the board API. ## Testing Update board tests to cover explicit self-targeting, subscribed authors, and delivery of other agents' replies. Update the model-context scenario and snapshot to verify posts remain searchable without a `CHANNEL_POST` self-notification. GitOrigin-RevId: 0b2558e7df79122a6da06ad592f9735ba260cce1 --- .../suite/scenarios_agent_message_board.rs | 33 +++------------- ...ge_board__agent_message_board_context.snap | 9 ++--- codex-rs/ext/agent-message-board/src/api.rs | 3 +- codex-rs/ext/agent-message-board/src/local.rs | 1 + .../agent-message-board/tests/local_board.rs | 38 +++++++++++-------- 5 files changed, 35 insertions(+), 49 deletions(-) diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs index e95533b5da..9f5f7dc1cd 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -124,7 +124,7 @@ async fn board_requires_persistent_v2_runtime( } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn board_active_notice_and_reads_reach_model_context() -> anyhow::Result<()> { +async fn board_post_and_reads_reach_model_context_without_self_notices() -> anyhow::Result<()> { let server = responses::start_mock_server().await; let mock = responses::mount_sse_sequence(&server, vec![ tool("post-decision", "post", json!({"new_channel_name":"design", "text":"A shared decision.", "agents_to_notify":["/root"]})), @@ -163,31 +163,10 @@ async fn board_active_notice_and_reads_reach_model_context() -> anyhow::Result<( .expect("search result"), )?; assert_eq!(result["results"][0]["text_preview"], "A shared decision."); - let input = requests[1].body_json(); - let notices = input["input"] - .as_array() - .context("request input")? - .iter() - .filter(|item| item["type"] == "agent_message") - .map(|item| { - ( - item["author"].clone(), - item["recipient"].clone(), - item["content"].clone(), - ) - }) - .collect::>(); - let message_id = post["message_id"].as_str().context("message ID")?; - assert_eq!( - notices, - vec![( - json!("/root"), - json!("/root"), - json!([{ - "type": "input_text", - "text": format!("Message Type: CHANNEL_POST\nSender: /root\nChannel: design\nMessage ID: {message_id}\nThread ID: {message_id}\nPayload:\nA shared decision."), - }]) - )] + assert!( + requests + .iter() + .all(|request| !request.body_contains_text("Message Type: CHANNEL_POST")) ); // Cargo and Bazel can enable different serde_json ordering features. Normalize only // the snapshot copies so the assertion compares JSON content, not object key order. @@ -208,7 +187,7 @@ async fn board_active_notice_and_reads_reach_model_context() -> anyhow::Result<( insta::assert_snapshot!( "agent_message_board_context", context_snapshot::format_context_snapshot( - "An active agent posts a shared decision, receives an attributed preview, and fetches the text.", + "An active agent posts a shared decision and fetches the text without a self-notification.", &bodies.iter().map(SnapshotEntry::body).collect::>(), &ContextSnapshotOptions::default().rewrite_known_segments(), ) diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap index 858c890816..a68b08e292 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap @@ -1,9 +1,9 @@ --- source: core/tests/suite/scenarios_agent_message_board.rs assertion_line: 88 -expression: "context_snapshot::format_context_snapshot(\"An active agent posts a shared decision, receives an attributed preview, and fetches the text.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(), &ContextSnapshotOptions::default().rewrite_known_segments(),)" +expression: "context_snapshot::format_context_snapshot(\"An active agent posts a shared decision and fetches the text without a self-notification.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(), &ContextSnapshotOptions::default().rewrite_known_segments(),)" --- -Scenario: An active agent posts a shared decision, receives an attributed preview, and fetches the text. +Scenario: An active agent posts a shared decision and fetches the text without a self-notification. ## Window 1 -- request 1 (request) -- @@ -36,7 +36,6 @@ Scenario: An active agent posts a shared decision, receives an attributed previe -- request 2 (request) -- 05:function_call/collaboration.post:{"agents_to_notify":["/root"],"new_channel_name":"design","text":"A shared decision."} 06:function_call_output:{"author":"/root","channel_name":"design","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} -07:agent_message -- request 3 (request) -- -08:function_call/collaboration.search_posts:{"channel_name":"design","query":"decision"} -09:function_call_output:{"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root","channel_name":"design"...ecision.","thread_id":"","truncated":false}]} [hash=facdb1e4f66cc5b9] +07:function_call/collaboration.search_posts:{"channel_name":"design","query":"decision"} +08:function_call_output:{"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root","channel_name":"design"...ecision.","thread_id":"","truncated":false}]} [hash=facdb1e4f66cc5b9] diff --git a/codex-rs/ext/agent-message-board/src/api.rs b/codex-rs/ext/agent-message-board/src/api.rs index 36e7c2093c..29b9eb2733 100644 --- a/codex-rs/ext/agent-message-board/src/api.rs +++ b/codex-rs/ext/agent-message-board/src/api.rs @@ -28,7 +28,8 @@ const _: Option<&dyn AgentMessageBoard> = None; /// Mutation success acknowledges acceptance, not that recipients read a post. /// Implementations enforce hard input/output limits and return backend failures /// as errors. They own atomic subscription changes, posting and recipient -/// selection. Notification delivery must neither wake finalized agents nor +/// selection, excluding the post author even when explicitly targeted. +/// Notification delivery must neither wake finalized agents nor /// leave notifications for a later turn. /// /// Boxed Send futures support an Arc, like AgentControl. diff --git a/codex-rs/ext/agent-message-board/src/local.rs b/codex-rs/ext/agent-message-board/src/local.rs index 20370dbb5f..b4e0563fd5 100644 --- a/codex-rs/ext/agent-message-board/src/local.rs +++ b/codex-rs/ext/agent-message-board/src/local.rs @@ -263,6 +263,7 @@ impl LocalAgentMessageBoard { for recipient in subscribed { recipients.insert(ThreadId::from_string(&recipient).map_err(storage_error)?); } + recipients.remove(&caller); let post = StoredPost { metadata: PostMetadata { message_id: id, diff --git a/codex-rs/ext/agent-message-board/tests/local_board.rs b/codex-rs/ext/agent-message-board/tests/local_board.rs index 85a51ba924..6fae5a40e3 100644 --- a/codex-rs/ext/agent-message-board/tests/local_board.rs +++ b/codex-rs/ext/agent-message-board/tests/local_board.rs @@ -115,7 +115,7 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { request_id: "call-1".into(), destination: PostDestination::Channel("proofs".into()), text: "é🦀 proof".into(), - agents_to_notify: vec![child_path.clone(), child_path.clone()], + agents_to_notify: vec![AgentPath::root(), child_path.clone(), child_path.clone()], }; let second = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) .await @@ -209,11 +209,24 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { ) .await .unwrap(); + // Excluding the author must keep its subscription for other agents' replies. + let child_reply = resumed + .post( + child, + PostRequest { + request_id: "child-reply".into(), + destination: PostDestination::Thread(metadata.message_id), + text: "acknowledged".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()), + (root, child_reply), (child, reply), ]; expected.sort_by_key(|(id, post)| (id.to_string(), post.message_id)); @@ -242,10 +255,12 @@ 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 child = ThreadId::new(); + let child_path = AgentPath::root().join("worker").unwrap(); let host = Arc::new(Host { clock: AtomicI64::default(), agent_path_calls: AtomicUsize::default(), - members: [(root, AgentPath::root())].into(), + members: [(root, AgentPath::root()), (child, child_path.clone())].into(), fail_notifications: AtomicBool::new(false), active: AtomicBool::new(false), notifications: Mutex::default(), @@ -260,7 +275,7 @@ async fn failed_requests_do_not_create_channels_or_notify_inactive_agents() { 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()]; + request.agents_to_notify = vec![child_path.clone()]; let posted = board.post(root, request.clone()).await.unwrap(); assert_eq!(*host.notifications.lock().unwrap(), Vec::new()); host.active.store(true, Ordering::SeqCst); @@ -275,7 +290,7 @@ async fn failed_requests_do_not_create_channels_or_notify_inactive_agents() { request_id: "reply".into(), destination: PostDestination::Thread(posted.thread_id), text: "saved even if the notice fails".into(), - agents_to_notify: vec![], + agents_to_notify: vec![child_path], }; let saved = board.post(root, reply.clone()).await.unwrap(); host.fail_notifications.store(false, Ordering::SeqCst); @@ -744,17 +759,8 @@ async fn tools_cover_channel_discussions_subscriptions_and_escaped_previews() { let reply: PostMetadata = serde_json::from_str(&replied.log_output()).unwrap(); last = Some((post, reply)); } - // Default create_channel subscribes the author to roots; post subscribes it - // to replies. The unsubscribed child receives neither. - assert_eq!( - host.notifications - .lock() - .unwrap() - .iter() - .map(|(id, _)| *id) - .collect::>(), - vec![root; 14] - ); + // The subscribed author receives no self-notices; the child is unsubscribed. + assert_eq!(*host.notifications.lock().unwrap(), Vec::new()); let (last, last_reply) = last.unwrap(); assert!( tool("subscribe")