diff --git a/codex-rs/agent-message-board-client/src/client.rs b/codex-rs/agent-message-board-client/src/client.rs index d09804dc8b..178848b71d 100644 --- a/codex-rs/agent-message-board-client/src/client.rs +++ b/codex-rs/agent-message-board-client/src/client.rs @@ -363,8 +363,8 @@ impl AgentMessageBoard for RemoteAgentMessageBoard { ); } -/// One active turn's live receiver. The host must also check the turn atomically -/// when injecting a returned notification, since the turn can end after `next`. +/// One active turn's live receiver. Frames are checked against this receiver's +/// identity before accepted notices enter the host's ordinary message delivery. pub struct BoardNotifications { caller: ThreadId, turn_id: String, diff --git a/codex-rs/agent-message-board-client/src/notifications.rs b/codex-rs/agent-message-board-client/src/notifications.rs index e6e534559a..1a5f96e3c1 100644 --- a/codex-rs/agent-message-board-client/src/notifications.rs +++ b/codex-rs/agent-message-board-client/src/notifications.rs @@ -112,7 +112,7 @@ impl TurnLifecycleContributor for Notifications { } /// Installs live notification delivery. The host resolves the existing remote -/// client and a delivery host bound to this exact turn; idle agents are never woken. +/// client and delivery host. Accepted notices use the host's ordinary message policy. /// The turn store owns the receiver and cancels it when the turn ends. pub fn install_notifications( registry: &mut ExtensionRegistryBuilder, diff --git a/codex-rs/core/src/agent/control/residency.rs b/codex-rs/core/src/agent/control/residency.rs index d0601c8900..3ea01481ab 100644 --- a/codex-rs/core/src/agent/control/residency.rs +++ b/codex-rs/core/src/agent/control/residency.rs @@ -171,15 +171,15 @@ impl LocalAgentRuntime { /// Pins and touches the registered runtime without waiting for unrelated eviction. pub(crate) async fn pin_v2_residency( &self, - state: &ThreadManagerState, - thread: &Arc, + thread: &CodexThread, ) -> CodexResult>> { if !is_resident_candidate(thread) { return Ok(None); } let guard = Arc::clone(&thread.residency_gate).read_owned().await; let thread_id = thread.session.thread_id; - if !Arc::ptr_eq(thread, &state.get_thread(thread_id).await?) { + let registered = self.upgrade()?.get_thread(thread_id).await?; + if !std::ptr::eq(thread, registered.as_ref()) { return Err(CodexErr::ThreadNotFound(thread_id)); } self.residency.touch(thread_id); diff --git a/codex-rs/core/src/agent/control/residency_tests.rs b/codex-rs/core/src/agent/control/residency_tests.rs index 5bf450e456..e098af6440 100644 --- a/codex-rs/core/src/agent/control/residency_tests.rs +++ b/codex-rs/core/src/agent/control/residency_tests.rs @@ -7,11 +7,15 @@ use crate::config::test_config; use crate::thread_manager::ThreadManagerState; use codex_features::Feature; use codex_login::CodexAuth; +use codex_protocol::AgentPath; use codex_protocol::ThreadId; use codex_protocol::error::CodexErrorDetails; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; @@ -58,6 +62,29 @@ async fn residency_slot_reservation_unloads_oldest_idle_v2_agent() { first_slot.commit(first.thread_id); mark_thread_completed(first.thread.as_ref()).await; + // Public message submission must honor the same eviction gate as MAv2 sends. + let eviction_gate = Arc::clone(&first.thread.residency_gate).write_owned().await; + let communication = InterAgentCommunication::new( + AgentPath::root(), + AgentPath::try_from("/root/worker_1").unwrap(), + Vec::new(), + "queued notice".into(), + /*trigger_turn*/ false, + ); + let delivery = first.thread.submit(Op::InterAgentCommunication { + communication: communication.clone(), + start_options: Default::default(), + }); + tokio::pin!(delivery); + assert!(futures::poll!(delivery.as_mut()).is_pending()); + drop(eviction_gate); + delivery.await.expect("submit after eviction gate releases"); + first + .thread + .update_thread_settings(ThreadSettingsOverrides::default()) + .await + .expect("dispatch queued notice before eviction"); + let second_slot = control .reserve_v2_residency_slot( &state, @@ -74,6 +101,14 @@ async fn residency_slot_reservation_unloads_oldest_idle_v2_agent() { }, Ok(_) => panic!("expected evicted thread to be missing"), } + let retained = first + .thread + .session + .services + .local_agent_runtime + .mailboxes + .take(first.thread_id); + assert_eq!(retained, vec![communication.into()]); let second = spawn_v2_subagent(&control, &state, config, root.thread_id, "worker-2").await; second_slot.commit(second.thread_id); diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 9210827567..a4ac17f71a 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -264,7 +264,7 @@ impl LocalAgentControl { if owner_thread_id.is_none() && let Ok(thread) = state.get_thread(thread_id).await { - return self.runtime.pin_v2_residency(&state, &thread).await; + return self.runtime.pin_v2_residency(&thread).await; } if self .runtime @@ -328,7 +328,7 @@ impl LocalAgentControl { } if let Ok(thread) = state.get_thread(thread_id).await { self.validate_loaded_v2_child(&thread, parent_thread_id)?; - return self.runtime.pin_v2_residency(&state, &thread).await; + return self.runtime.pin_v2_residency(&thread).await; } } let parent = if let Some(parent) = parent { @@ -589,9 +589,7 @@ impl LocalAgentControl { // Finish registration even if the sender is cancelled while acquiring its pin. tokio::spawn(async move { let _membership = membership; - let residency_pin = runtime - .pin_v2_residency(&state, &reloaded_thread.thread) - .await?; + let residency_pin = runtime.pin_v2_residency(&reloaded_thread.thread).await?; runtime.registry.clear_evicted_environments(thread_id); residency_slot.commit(reloaded_thread.thread_id); // Register before listeners can forward events from the resumed thread. @@ -616,7 +614,7 @@ impl LocalAgentControl { } self.runtime.registry.clear_evicted_environments(thread_id); drop(residency_slot); - return self.runtime.pin_v2_residency(&state, &thread).await; + return self.runtime.pin_v2_residency(&thread).await; } Err(err) } diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index 12cd8ec42b..c9dc09505f 100644 --- a/codex-rs/core/src/agent_message_board.rs +++ b/codex-rs/core/src/agent_message_board.rs @@ -2,7 +2,7 @@ //! //! The extension owns backend construction, storage and tool registration. This adapter //! selects configuration and supplies host capabilities. It never starts or restores -//! recipients and never queues a notification for an idle agent. +//! recipients during lookup. Notifications use ordinary agent-message scheduling. use crate::CodexThread; use crate::ThreadManager; @@ -28,6 +28,7 @@ use codex_protocol::error::CodexErr; use codex_protocol::error::CodexErrorDetails; use codex_protocol::error::Result; use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::Op; use futures::future::BoxFuture; use std::sync::Arc; use std::sync::Weak; @@ -44,7 +45,7 @@ pub fn install_agent_message_board( Some(( board, Arc::new(LocalBoardHost { - expected_turn_id: Some(input.turn_id.into()), + remote_notifications: true, ..host.as_ref().clone() }), )) @@ -91,7 +92,7 @@ pub fn install_agent_message_board( manager: manager.clone(), tree, caller, - expected_turn_id: None, + remote_notifications: false, connection, }, )) @@ -104,7 +105,7 @@ struct LocalBoardHost { manager: Weak, tree: SessionId, caller: ThreadId, - expected_turn_id: Option, + remote_notifications: bool, connection: Option>, } @@ -237,7 +238,7 @@ impl MessageBoardHost for LocalBoardHost { ); // Remote metadata is untrusted. Skip an oversized notice without closing // the receiver; the post remains available through the board tools. - if self.expected_turn_id.is_some() + if self.remote_notifications && communication.content.len() + communication.author.as_str().len() + communication.recipient.as_str().len() @@ -247,22 +248,13 @@ impl MessageBoardHost for LocalBoardHost { "remote board notice exceeds the context budget".into(), )); } - Ok( - if recipient - .session - .input_queue - .deliver_mailbox_communication_to_current_turn( - &recipient.session.active_turn, - communication, - self.expected_turn_id.as_deref(), - ) - .await - { - NotificationDelivery::Accepted - } else { - NotificationDelivery::SkippedInactive - }, - ) + recipient + .submit(Op::InterAgentCommunication { + communication, + start_options: Default::default(), + }) + .await?; + Ok(NotificationDelivery::Accepted) }) } } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 60de98ddbc..ff23071182 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -242,7 +242,7 @@ impl CodexThread { } pub async fn submit(&self, op: Op) -> CodexResult { - self.io.submit(op).await + self.submit_with_trace(op, /*trace*/ None).await } /// Returns the session telemetry handle for thread-scoped production instrumentation. @@ -453,10 +453,24 @@ impl CodexThread { op: Op, trace: Option, ) -> CodexResult { + // Use the same admission guard as MAv2 sends, retaining it through dispatch. + let residency_guard = if matches!(op, Op::InterAgentCommunication { .. }) { + self.session + .services + .local_agent_runtime + .pin_v2_residency(self) + .await? + } else { + None + }; self.io .submit_with_trace( - op, trace, /*parent_turn_id*/ None, /*root_turn_id*/ None, - /*residency_guard*/ None, /*message_delivery*/ None, + op, + trace, + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + residency_guard, + /*message_delivery*/ None, ) .await } diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index de016d7ac9..b1100b12e8 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -7,7 +7,6 @@ use codex_diagnostics::GaugeGuard; use codex_history::ResponseItemEnvelope; use codex_protocol::ThreadId; use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::turn_input::TurnStartOptions; use codex_protocol::user_input::UserInput; use serde::Deserialize; @@ -202,38 +201,6 @@ impl InputQueue { (activity_rx, pending_activity) } - #[expect( - clippy::await_holding_invalid_type, - reason = "active turn checks and turn state updates must remain atomic" - )] - pub(crate) async fn deliver_mailbox_communication_to_current_turn( - &self, - active_turn: &Mutex>, - communication: InterAgentCommunication, - expected_turn_id: Option<&str>, - ) -> bool { - let active = active_turn.lock().await; - let Some(active_turn) = active.as_ref().filter(|turn| { - turn.task.as_ref().is_some_and(|task| { - expected_turn_id.is_none_or(|id| { - task.turn_context.sub_id == id && !task.cancellation_token.is_cancelled() - }) - }) - }) else { - return false; - }; - let mut turn_state = active_turn.turn_state.lock().await; - if !turn_state.accepts_mailbox_delivery_for_current_turn() { - return false; - } - turn_state - .pending_input - .items - .push(TurnInput::InterAgentCommunication(communication.into())); - self.activity_tx.send_replace(InputQueueActivity::Mailbox); - true - } - pub(crate) async fn enqueue_mailbox_communication( &self, communication: crate::PendingAgentMessage, @@ -556,6 +523,7 @@ mod tests { use super::*; use codex_history::CodexHarnessMetadata; use codex_protocol::AgentPath; + use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::user_input::UserInput; use pretty_assertions::assert_eq; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index f07c7d18a0..5a5e48422c 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1886,7 +1886,7 @@ impl ThreadManagerState { .session .services .local_agent_runtime - .pin_v2_residency(self, &thread) + .pin_v2_residency(&thread) .await? } else { None 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 dd1fbe82f5..88fa3f45d2 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -1,4 +1,4 @@ -//! Exercises board tools through the runtime, including resume and active-only notices. +//! Exercises board tools through the runtime, including resume and native mailbox notifications. use anyhow::Context; use codex_core::TurnInputRequest; @@ -7,6 +7,7 @@ use codex_message_board::PostMetadata; use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::UserInput; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; @@ -250,7 +251,7 @@ async fn board_post_and_reads_reach_model_context_without_self_notices( #[test_case::test_case(false; "disk")] #[test_case::test_case(true; "in_memory")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn board_is_shared_with_children_and_skips_idle_notices( +async fn board_is_shared_with_children_and_queues_idle_notices( in_memory: bool, ) -> anyhow::Result<()> { let server = responses::start_mock_server().await; @@ -341,11 +342,7 @@ async fn board_is_shared_with_children_and_skips_idle_notices( ) .await; root.submit_turn("Read the worker's decision.").await?; - assert!( - read.requests() - .iter() - .all(|request| !request.body_contains_text("Message Type: CHANNEL_POST")) - ); + assert!(read.requests()[1].body_contains_text("Message Type: CHANNEL_POST")); let output = read.requests()[1] .function_call_output_text("root-read") .expect("read result"); @@ -454,7 +451,8 @@ async fn board_notifications_do_not_reopen_a_final_answer( gate: Some(child_gate), body: done(), }], - response(done()), + // Native mail is drained after the new task's first model step. + response(tool("next-step", "get_channels", json!({}))), response(done()), ]) .await; @@ -550,18 +548,22 @@ async fn board_notifications_do_not_reopen_a_final_answer( let next_turn = requests.last().context("next root request")?["client_metadata"]["turn_id"] .as_str() .context("next root turn ID")?; - for (turn, notified) in [(target_turn.as_str(), false), (next_turn, !after_final)] { + for (turn, expected_requests, notified) in + [(target_turn.as_str(), 1, false), (next_turn, 2, true)] + { let requests = requests .iter() .filter(|request| request["client_metadata"]["turn_id"] == turn) .collect::>(); assert_eq!( requests.len(), - 1, - "recipient must not sample again after its final" + expected_requests, + "unexpected sampling count for root turn {turn}" ); assert_eq!( - requests[0]["input"].to_string().contains(¬ification), + requests.last().context("root request")?["input"] + .to_string() + .contains(¬ification), notified ); } @@ -955,6 +957,8 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { let server = responses::start_mock_server().await; let response = |body| vec![StreamingSseChunk { gate: None, body }]; let (release, gate) = oneshot::channel(); + let mut started = ev_message_item_added("watching", ""); + started["item"]["phase"] = json!("commentary"); let mut working = ev_assistant_message("watching", "Watching the policy board."); working["item"]["phase"] = json!("commentary"); let (streaming, _) = start_streaming_sse_server(vec![ @@ -984,11 +988,15 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { vec![ StreamingSseChunk { gate: None, - body: sse(vec![ev_response_created("watching"), working]), + body: sse(vec![ev_response_created("watching"), started]), }, StreamingSseChunk { gate: Some(gate), - body: tool("checkpoint", "get_channels", json!({})), + body: sse(vec![ + working, + ev_function_call_with_namespace("checkpoint", "collaboration", "get_channels", "{}"), + ev_completed("checkpoint"), + ]), }, ], // Separate responses order the writes before snapshotting last_message_id. @@ -1015,6 +1023,8 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { }), )), response(done()), + // Queued board notices preempt the commentary before its planned checkpoint. + response(tool("checkpoint", "get_channels", json!({}))), response(done()), ]) .await; @@ -1132,8 +1142,13 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { text_elements: vec![], }])) .await?; + let guardian_turn = wait_for_event_match(guardian, |event| match event { + EventMsg::TurnStarted(event) => Some(event.turn_id.clone()), + _ => None, + }) + .await; wait_for_event(guardian, |event| { - matches!(event, EventMsg::ItemCompleted(event) + matches!(event, EventMsg::ItemStarted(event) if matches!(&event.item, TurnItem::AgentMessage(message) if message.id == "watching")) }) .await; @@ -1144,6 +1159,16 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { }])) .await?; wait_for_event(writer, |event| matches!(event, EventMsg::TurnComplete(_))).await; + // Acknowledgement is ordered after both notices in the recipient's submission queue. + guardian + .update_thread_settings(ThreadSettingsOverrides::default()) + .await?; + assert_eq!( + guardian + .wait_for_input_activity(&guardian_turn, std::time::Duration::from_secs(/*secs*/ 10)) + .await, + codex_core::InputWaitOutcome::MailboxActivity, + ); release .send(()) .expect("release Guardian after notification delivery"); @@ -1154,6 +1179,20 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { .iter() .any(|body| String::from_utf8_lossy(body).contains("Message Type: CHANNEL_POST")) ); + let resumed: Value = serde_json::from_slice(&requests[16])?; + assert!( + resumed["input"] + .to_string() + .contains("Message Type: CHANNEL_POST") + ); + assert!( + !resumed["input"] + .as_array() + .context("resumed input")? + .iter() + .any(|item| item["call_id"] == "checkpoint"), + "queued board mail must be read before the planned checkpoint runs", + ); let last: Value = serde_json::from_slice(requests.last().unwrap())?; let checkpoint = last["input"] .as_array() @@ -1169,7 +1208,7 @@ async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { .iter() .map(|body| serde_json::from_slice::(body)) .collect::>>()?; - assert_eq!(bodies.len(), 17); + assert_eq!(bodies.len(), 18); for body in &mut bodies { for item in body["input"].as_array_mut().context("request input")? { if let Some(Value::String(output)) = item.get_mut("output") { diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs index 06667f15e9..0d54d86443 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs @@ -200,15 +200,14 @@ async fn http_board_is_shared_with_children(mode: BoardMode) -> anyhow::Result<( .map_or(output.as_str(), |(_, json)| json); serde_json::from_str(json).with_context(|| format!("board result: {output}")) }; - let posted = responses::mount_sse_sequence( - &server, - board_turn( - "worker-post", - "post", - json!({"new_channel_name":"design","text":"Worker's durable decision."}), - ), - ) - .await; + let mut post_arguments = + json!({"new_channel_name":"design","text":"Worker's durable decision."}); + if mode == BoardMode::Feature { + post_arguments["agents_to_notify"] = json!(["/root"]); + } + let posted = + responses::mount_sse_sequence(&server, board_turn("worker-post", "post", post_arguments)) + .await; child .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { text: "Find the board tools and post your decision.".into(), @@ -261,6 +260,9 @@ async fn http_board_is_shared_with_children(mode: BoardMode) -> anyhow::Result<( if let Some(writer) = writer { writer.rollback().await?; } + if mode == BoardMode::Feature { + assert!(read.requests()[1].body_contains_text("Message Type: CHANNEL_POST")); + } let result = board_result( read.requests().last().context("parent follow-up")?, "root-read", diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap index f4be7df78a..145c06b61d 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap @@ -116,7 +116,8 @@ Scenario: Independent Codex runtimes share a local board: a writer publishes, an 21:function_call_output:{"channel_name":"policy","enabled":true,"last_message_id":"","target_agent":"/root/00000000000040008000000000000001","thread_id":null} 22:message/assistant: Watching the policy board. -23:function_call/collaboration.get_channels:{} -24:function_call_output:{"board_permissions":{"create_channel":false,"edit_metadata":false,"manage_permissions":false},"has_more...ata":false,"post":false,"read":true,"reply":false}}]} [hash=7cc8dc25fc1c52b2] -25:agent_message -26:agent_message +23:agent_message +24:agent_message +-- request 18 (request) -- +25:function_call/collaboration.get_channels:{} +26:function_call_output:{"board_permissions":{"create_channel":false,"edit_metadata":false,"manage_permissions":false},"has_more...ata":false,"post":false,"read":true,"reply":false}}]} [hash=7cc8dc25fc1c52b2] diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_feature_context.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_feature_context.snap index 98ac248bb0..932c573fd5 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_feature_context.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_feature_context.snap @@ -45,7 +45,7 @@ Scenario: The config flag starts an in-memory MCP board shared by a parent and c 08:tool_search_call 09:tool_search_output -- request 3 (request) -- -10:function_call/mcp__codex_message_board.post:{"new_channel_name":"design","text":"Worker's durable decision."} +10:function_call/mcp__codex_message_board.post:{"agents_to_notify":["/root"],"new_channel_name":"design","text":"Worker's durable decision."} 11:function_call_output:Wall time: seconds Output: {"author":"/root/worker","channel_name":"design","created_at":"","message_id":"","thread_id":""} @@ -95,8 +95,9 @@ Scenario: The config flag starts an in-memory MCP board shared by a parent and c 11:tool_search_output 12:agent_message 13:agent_message +14:agent_message -- request 6 (request) -- -14:function_call/mcp__codex_message_board.search_posts:{"channel_name":"design"} -15:function_call_output:Wall time: seconds +15:function_call/mcp__codex_message_board.search_posts:{"channel_name":"design"} +16:function_call_output:Wall time: seconds Output: {"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root/worker","channel_name":"...ecision.","thread_id":"","truncated":false}]} [hash=0322071c1fb9baa5] diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_permissions_direct.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_permissions_direct.snap index e07c147301..041aeed072 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_permissions_direct.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_permissions_direct.snap @@ -6,7 +6,7 @@ Scenario: Permission discovery preserves legacy output, exposes reply-only acces ## Window 1 -- request 1 (request) -- -00:additional_tools/developer (3; hash=e49a28343a9fe4ed): +00:additional_tools/developer (3; hash=788c4a5014dcfe4f): - namespace/functions; hash=c8b8fa91706a7620 - function/exec_command - function/write_stdin @@ -17,7 +17,7 @@ Scenario: Permission discovery preserves legacy output, exposes reply-only acces - namespace/clock: Tools for reading and waiting on time.; hash=a80435c7a3a3943f - function/curr_time - function/sleep - - namespace/collaboration: Tools for spawning and managing sub-agents.; hash=ba5664aaab81db64 + - namespace/collaboration: Tools for spawning and managing sub-agents.; hash=5c5c29d91fa33d4a - function/create_channel - function/followup_task - function/get_channels diff --git a/codex-rs/message-board/src/host.rs b/codex-rs/message-board/src/host.rs index c915a41b8d..494d862b09 100644 --- a/codex-rs/message-board/src/host.rs +++ b/codex-rs/message-board/src/host.rs @@ -18,10 +18,9 @@ pub enum NotificationDelivery { /// /// The host remains authoritative for membership even when runtimes are unloaded. /// Clock reads use the posting agent's configured source, including simulated -/// time. Notifications are accepted only while a running turn still accepts -/// mailbox delivery; otherwise they are skipped without being queued. -/// Accepted notifications must not start a new turn or reopen a finalized answer, -/// but may remain in conversation history for a later turn. +/// time. Notifications follow the host's ordinary queue-only agent-message +/// scheduling and wake policy, and may be buffered for an idle recipient. +/// Acceptance does not guarantee that the model has consumed the notification. /// Backends may implement these capabilities with local handles or remote RPCs. pub trait MessageBoardHost: Send + Sync { fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result>; diff --git a/codex-rs/message-board/src/tools/spec.rs b/codex-rs/message-board/src/tools/spec.rs index 88ebb1a9ac..e9691e5ffd 100644 --- a/codex-rs/message-board/src/tools/spec.rs +++ b/codex-rs/message-board/src/tools/spec.rs @@ -66,7 +66,7 @@ pub(super) fn tool( &["message_id"], ), MessageBoardTool::Subscribe => ( - "Subscribe yourself or another agent to a channel for new top-level posts, or to a thread for replies. Provide exactly one of channel_name or thread_id. Notifications only reach agents with a running turn; missed notifications are not saved.", + "Subscribe yourself or another agent to a channel for new top-level posts, or to a thread for replies. Provide exactly one of channel_name or thread_id. Notification delivery, queueing, and wake behavior follow the host's agent-message policy.", &["channel_name", "thread_id", "target_agent"], &[], ), @@ -76,7 +76,7 @@ pub(super) fn tool( &[], ), MessageBoardTool::Post => ( - "Start a thread in an existing or new channel, or reply using the first post's message ID as thread_id. Exactly one destination is required. Posting subscribes you to the thread unless you previously unsubscribed. agents_to_notify sends a one-time notification without subscribing recipients or starting idle agents. Returns metadata, not the post text.", + "Start a thread in an existing or new channel, or reply using the first post's message ID as thread_id. Exactly one destination is required. Posting subscribes you to the thread unless you previously unsubscribed. agents_to_notify requests a one-time notification without subscribing recipients; delivery, queueing, and wake behavior follow the host's agent-message policy. Returns metadata, not the post text.", &[ "text", "channel_name",