diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index e3b6333a46bf..c004e78dd008 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 c70ff7500816..2212dc09f663 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/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 4a9d557c6aff..745439ebf361 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/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(()) } 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 d110775998d3..a0fcbfaee7c6 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 0a4d24d92637..a9952c1aa0a0 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>; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 24f1e54990d8..05f2d9061da5 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/timestamp_metadata_tests.rs b/codex-rs/thread-store/src/local/timestamp_metadata_tests.rs new file mode 100644 index 000000000000..0e48bbce7b6b --- /dev/null +++ b/codex-rs/thread-store/src/local/timestamp_metadata_tests.rs @@ -0,0 +1,126 @@ +//! Verifies narrow SQLite timestamp updates preserve metadata and repair fallbacks. + +use super::test_support::test_config; +use super::test_support::write_session_file_with_history_mode; +use super::*; +use chrono::Utc; +use pretty_assertions::assert_eq; + +#[tokio::test] +async fn timestamp_updates_repair_missing_rows_then_touch_only_timestamp_columns() { + for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] { + let home = tempfile::TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("initialize state db"); + let pool = config + .sqlite + .open_read_write_pool(&config.sqlite.state_db_path()) + .await + .expect("open SQL observer"); + let store = LocalThreadStore::new(config, Some(runtime.clone())); + let id = uuid::Uuid::new_v4(); + let thread_id = ThreadId::from_string(&id.to_string()).expect("thread id"); + let path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-00", + id, + history_mode, + ) + .expect("write canonical history"); + let params = UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + updated_at: Some(Utc::now()), + ..Default::default() + }, + include_archived: false, + }; + // Missing rows must still be reconstructed from their canonical history. + store + .update_thread_metadata(params.clone()) + .await + .expect("reconstruct missing metadata"); + let mut expected = runtime + .get_thread(thread_id) + .await + .expect("read repaired metadata") + .expect("repaired row"); + assert_eq!( + (&expected.rollout_path, expected.history_mode), + (&path, history_mode) + ); + sqlx::raw_sql( + r#" + CREATE TABLE timestamp_writes(kind TEXT); + CREATE TRIGGER count_timestamp AFTER UPDATE OF updated_at_ms ON threads + BEGIN INSERT INTO timestamp_writes VALUES ('timestamp'); END; + CREATE TRIGGER count_full_row AFTER UPDATE OF title ON threads + BEGIN INSERT INTO timestamp_writes VALUES ('full'); END; + "#, + ) + .execute(&pool) + .await + .expect("install write counters"); + + store + .update_thread_metadata(params.clone()) + .await + .expect("touch existing metadata"); + let actual = runtime + .get_thread(thread_id) + .await + .expect("read touched metadata") + .expect("touched row"); + assert!(actual.updated_at > expected.updated_at); + expected.updated_at = actual.updated_at; + assert_eq!(actual, expected); + let writes: Vec = + sqlx::query_scalar("SELECT kind FROM timestamp_writes ORDER BY rowid") + .fetch_all(&pool) + .await + .expect("read write counters"); + assert_eq!(writes, vec!["timestamp"]); + + store + .stage_pending_thread_metadata( + thread_id, + ThreadMetadataPatch { + model: Some("staged-model".to_string()), + ..Default::default() + }, + ) + .await + .expect("stage metadata"); + store + .update_thread_metadata(params) + .await + .expect("apply staged metadata"); + let actual = runtime + .get_thread(thread_id) + .await + .expect("read staged metadata") + .expect("updated row"); + expected.model = Some("staged-model".to_string()); + expected.updated_at = actual.updated_at; + assert_eq!(actual, expected); + assert!( + store + .pending_thread_metadata + .lock(thread_id) + .await + .is_none() + ); + let mut writes: Vec = sqlx::query_scalar("SELECT kind FROM timestamp_writes") + .fetch_all(&pool) + .await + .expect("read write counters"); + writes.sort(); + assert_eq!(writes, vec!["full", "timestamp", "timestamp"]); + pool.close().await; + } +} diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 8cfb38bf2f24..76bc877d5ea5 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -360,6 +360,22 @@ async fn apply_metadata_update( .map_err(|err| ThreadStoreError::Internal { message: format!("failed to read thread metadata for {thread_id}: {err}"), })?; + // Existing timestamp-only observations must not rewrite metadata or its indexes. + // Missing rows and changed rollout paths still need the normal repair path. + if let Some(updated_at) = patch.updated_at + && patch.is_empty_except_updated_at() + && existing.as_ref().is_some_and(|metadata| { + rollout_path.as_ref().is_none_or(|path| path == &metadata.rollout_path) + }) + && state_db + .touch_thread_updated_at(thread_id, updated_at) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to update thread timestamp for {thread_id}: {err}"), + })? + { + return Ok(()); + } let project_id = if existing.is_none() && let Some(Some(project_id)) = patch.project_id.as_ref() && state_db diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 692cb1029ca0..5030f8864d3c 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -914,6 +914,10 @@ impl ThreadMetadataPatch { } pub fn is_empty(&self) -> bool { + self.updated_at.is_none() && self.is_empty_except_updated_at() + } + + pub(crate) fn is_empty_except_updated_at(&self) -> bool { self.name.is_none() && self.rollout_path.is_none() && self.preview.is_none() @@ -922,7 +926,6 @@ impl ThreadMetadataPatch { && self.model.is_none() && self.reasoning_effort.is_none() && self.created_at.is_none() - && self.updated_at.is_none() && self.advance_recency_at.is_none() && self.source.is_none() && self.originator.is_none()