diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index e3b6333a46..c004e78dd0 100644 --- a/codex-rs/core/src/agent_message_board.rs +++ b/codex-rs/core/src/agent_message_board.rs @@ -187,12 +187,18 @@ impl MessageBoardHost for LocalBoardHost { /*trigger_turn*/ false, ); Ok( - match recipient - .inject_if_running(vec![communication.to_model_input_item()]) + if recipient + .session + .input_queue + .deliver_mailbox_communication_to_current_turn( + &recipient.session.active_turn, + communication, + ) .await { - Ok(()) => NotificationDelivery::Accepted, - Err(_) => NotificationDelivery::SkippedInactive, + NotificationDelivery::Accepted + } else { + NotificationDelivery::SkippedInactive }, ) }) diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index c70ff75008..2212dc09f6 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -120,13 +120,13 @@ impl InputQueue { Option, ) { let activity_rx = self.activity_tx.subscribe(); - let has_pending_steer = if let Some(turn_state) = turn_state { - turn_state.lock().await.pending_input.has_pending_input() + let turn_activity = if let Some(turn_state) = turn_state { + turn_state.lock().await.pending_input.pending_activity() } else { - false + None }; - let pending_activity = if has_pending_steer { - Some(InputQueueActivity::Steer) + let pending_activity = if let Some(activity) = turn_activity { + Some(activity) } else if self.has_pending_mailbox_items().await { Some(InputQueueActivity::Mailbox) } else { @@ -135,6 +135,31 @@ 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, + ) -> bool { + let active = active_turn.lock().await; + let Some(active_turn) = active.as_ref().filter(|turn| turn.task.is_some()) 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)); + self.activity_tx.send_replace(InputQueueActivity::Mailbox); + true + } + pub(crate) async fn enqueue_mailbox_communication( &self, communication: InterAgentCommunication, @@ -398,13 +423,23 @@ impl TurnInputQueue { self.items.is_empty() } - fn has_pending_input(&self) -> bool { - self.items.iter().any(|input| { + fn pending_activity(&self) -> Option { + if self.items.iter().any(|input| { matches!( input, TurnInput::UserInput { .. } | TurnInput::FunctionCallOutput(_) ) - }) + }) { + Some(InputQueueActivity::Steer) + } else if self + .items + .iter() + .any(|input| matches!(input, TurnInput::InterAgentCommunication(_))) + { + Some(InputQueueActivity::Mailbox) + } else { + None + } } } @@ -557,6 +592,22 @@ mod tests { input_queue.subscribe_activity(Some(&turn_state)).await.1, None ); + let communication = make_mail( + AgentPath::root(), + AgentPath::try_from("/root/worker").expect("agent path"), + "already pending mail", + /*trigger_turn*/ false, + ); + input_queue + .extend_pending_input_for_turn_state( + &turn_state, + vec![TurnInput::InterAgentCommunication(communication)], + ) + .await; + assert_eq!( + input_queue.subscribe_activity(Some(&turn_state)).await.1, + Some(InputQueueActivity::Mailbox) + ); input_queue .extend_pending_input_and_accept_mailbox_delivery_for_turn_state( &turn_state, 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 d110775998..a0fcbfaee7 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -16,6 +16,7 @@ use core_test_support::responses::ResponsesRequest; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call_with_namespace; +use core_test_support::responses::ev_message_item_added; use core_test_support::responses::ev_response_created; use core_test_support::responses::sse; use core_test_support::streaming_sse::StreamingSseChunk; @@ -352,6 +353,181 @@ async fn board_is_shared_with_children_and_skips_idle_notices( Ok(()) } +#[test_case::test_case(false; "before_final")] +#[test_case::test_case(true; "after_final")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn board_notifications_do_not_reopen_a_final_answer(after_final: bool) -> anyhow::Result<()> { + let response = |body| vec![StreamingSseChunk { gate: None, body }]; + let (release_final, final_gate) = oneshot::channel(); + let (release_child, child_gate) = oneshot::channel(); + let final_phase = |mut event: Value| { + event["item"]["phase"] = json!("final_answer"); + event + }; + let started = final_phase(ev_message_item_added("root-final", "")); + let finished = final_phase(ev_assistant_message("root-final", "Finished.")); + let (before_gate, after_gate) = if after_final { + // Starting an empty item confirms the answer was processed without adding mailbox work. + ( + vec![ + started, + finished, + final_phase(ev_message_item_added("after-final", "")), + ], + vec![final_phase(ev_assistant_message("after-final", ""))], + ) + } else { + (vec![started], vec![finished]) + }; + let (streaming, _) = start_streaming_sse_server(vec![ + response(tool( + "create", + "create_channel", + json!({"channel_name":"design"}), + )), + response(tool( + "spawn", + "spawn_agent", + json!({"task_name":"worker","message":"Say ready.","fork_turns":"none"}), + )), + response(done()), + response(done()), + vec![ + StreamingSseChunk { + gate: None, + body: sse([vec![ev_response_created("root")], before_gate].concat()), + }, + StreamingSseChunk { + gate: Some(final_gate), + body: sse([after_gate, vec![ev_completed("root")]].concat()), + }, + ], + response(tool( + "worker-post", + "post", + json!({"channel_name":"design","text":"Late update."}), + )), + vec![StreamingSseChunk { + gate: Some(child_gate), + body: done(), + }], + response(done()), + response(done()), + ]) + .await; + let server = responses::start_mock_server().await; + let base_url = format!("{}/v1", streaming.uri()); + let root = test_codex() + .with_config(move |config| { + configure(config); + config.model_provider.base_url = Some(base_url); + config + .features + .disable(Feature::EnableRequestCompression) + .expect("read raw requests"); + }) + .with_model_info_override("gpt-5.5", |model| { + model.multi_agent_version = Some(MultiAgentVersion::V2); + }) + .build_with_auto_env(&server) + .await?; + root.submit_turn("Create the channel and spawn a worker.") + .await?; + let child_id = root + .thread_manager + .list_thread_ids() + .await + .into_iter() + .find(|id| *id != root.session_configured.thread_id) + .context("child runtime")?; + let child = root.thread_manager.get_thread(child_id).await?; + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + root.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Finish your answer.".into(), + text_elements: vec![], + }])) + .await?; + let target_turn = wait_for_event_match(&root.codex, |event| match event { + EventMsg::TurnStarted(event) => Some(event.turn_id.clone()), + _ => None, + }) + .await; + wait_for_event(&root.codex, |event| match event { + EventMsg::ItemStarted(event) => match &event.item { + TurnItem::AgentMessage(message) => { + message.id + == if after_final { + "after-final" + } else { + "root-final" + } + } + _ => false, + }, + _ => false, + }) + .await; + child + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Post an update.".into(), + text_elements: vec![], + }])) + .await?; + // The child's follow-up confirms fanout completed, but its final would separately notify root. + streaming.wait_for_request_count(/*count*/ 7).await; + let requests = streaming.requests().await; + let request: Value = serde_json::from_slice(requests.last().context("child follow-up")?)?; + let post = request["input"] + .as_array() + .context("child follow-up input")? + .iter() + .find(|item| item["type"] == "function_call_output" && item["call_id"] == "worker-post") + .context("child post result")?; + let post: PostMetadata = serde_json::from_str(post["output"].as_str().context("post output")?)?; + assert_eq!(post.author.as_str(), "/root/worker"); + release_final.send(()).expect("release root response"); + wait_for_event(&root.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + release_child.send(()).expect("release child response"); + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + root.submit_turn("Start another turn.").await?; + + let requests = streaming + .requests() + .await + .into_iter() + .map(|body| serde_json::from_slice::(&body)) + .collect::>>()?; + let notification = format!("Message ID: {}", post.message_id); + 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)] { + 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" + ); + assert_eq!( + requests[0]["input"].to_string().contains(¬ification), + notified + ); + } + child.shutdown_and_wait().await?; + root.codex.shutdown_and_wait().await?; + streaming.shutdown().await; + Ok(()) +} + #[test_case::test_case(false; "direct_messages")] #[test_case::test_case(true; "channels_only")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] diff --git a/codex-rs/ext/agent-message-board/src/host.rs b/codex-rs/ext/agent-message-board/src/host.rs index 0a4d24d926..a9952c1aa0 100644 --- a/codex-rs/ext/agent-message-board/src/host.rs +++ b/codex-rs/ext/agent-message-board/src/host.rs @@ -18,8 +18,10 @@ 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. Notification admission is atomic with turn completion: inactive agents -/// are skipped, and no notification may start work or survive into a later turn. +/// 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. /// Backends may implement these capabilities with local handles or remote RPCs. pub trait MessageBoardHost: Send + Sync { fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result>;