[pull] main from openai:main

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
                 },
             )
         })

codex-rs/core/src/session/input_queue.rs

@@ -120,13 +120,13 @@ impl InputQueue {
         Option<InputQueueActivity>,
     ) {
         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<Option<ActiveTurn>>,
+        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<InputQueueActivity> {
+        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,

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -2110,19 +2110,26 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() ->
 
     skip_if_no_network!(Ok(()));
 
-    let server = start_websocket_server(vec![
-        vec![],
-        vec![vec![json!({
-            "type": "session.updated",
-            "session": { "id": "sess_env", "instructions": "backend prompt" }
-        })]],
-    ])
+    // Startup prewarm can reconnect after the empty connection closes. Keep its
+    // Responses requests from consuming the realtime session's scripted reply.
+    let startup_server = start_websocket_server(vec![vec![]]).await;
+    let server = start_websocket_server(vec![vec![vec![json!({
+        "type": "session.updated",
+        "session": { "id": "sess_env", "instructions": "backend prompt" }
+    })]]])
     .await;
 
-    let mut builder = test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing());
-    let test = builder.build_with_websocket_server(&server).await?;
+    let mut builder = test_codex()
+        .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
+        .with_config({
+            let realtime_base_url = server.uri().to_string();
+            move |config| {
+                config.experimental_realtime_ws_base_url = Some(realtime_base_url);
+            }
+        });
+    let test = builder.build_with_websocket_server(&startup_server).await?;
     assert!(
-        server
+        startup_server
             .wait_for_handshakes(/*expected*/ 1, Duration::from_secs(2))
             .await
     );
@@ -2174,8 +2181,9 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() ->
     .await;
     assert_eq!(session_updated, "sess_env");
 
+    assert_eq!(server.handshakes().len(), 1);
     assert_eq!(
-        server.handshakes()[1].header("authorization").as_deref(),
+        server.handshakes()[0].header("authorization").as_deref(),
         Some("Bearer env-realtime-key")
     );
 
@@ -2186,6 +2194,7 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() ->
     })
     .await;
 
+    startup_server.shutdown().await;
     server.shutdown().await;
     Ok(())
 }

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::<Value>(&body))
+        .collect::<serde_json::Result<Vec<_>>>()?;
+    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::<Vec<_>>();
+        assert_eq!(
+            requests.len(),
+            1,
+            "recipient must not sample again after its final"
+        );
+        assert_eq!(
+            requests[0]["input"].to_string().contains(&notification),
+            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)]

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<AgentPath>>;

codex-rs/thread-store/src/local/mod.rs

@@ -38,6 +38,9 @@ mod pending_thread_metadata_tests;
 mod read_thread_tests;
 #[cfg(test)]
 mod test_support;
+#[cfg(test)]
+#[path = "timestamp_metadata_tests.rs"]
+mod timestamp_metadata_tests;
 
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::ThreadHistoryMode;