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")