From 1b1835f751ebdc0cfc50b3fe55d4571dbb294563 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 28 Sep 2026 11:55:54 +0000 Subject: [PATCH 1/3] Isolate realtime auth fallback test from startup prewarm (#48973) ## Why Startup prewarm can reconnect after its empty WebSocket connection closes, consuming the scripted realtime session reply in the environment-key fallback test with ChatGPT authentication. ## What changed Use separate WebSocket servers for startup and realtime traffic, configuring `experimental_realtime_ws_base_url` to point to the realtime server. Assert that it receives exactly one handshake and that the handshake uses the environment key for authorization. GitOrigin-RevId: 0e8ba4bdd66ef8aa3c619e36454dded9dcf55444 --- .../core/tests/suite/realtime_conversation.rs | 31 ++++++++++++------- 1 file changed, 20 insertions(+), 11 deletions(-) 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(()) } From 06971ec9aad037d7c32b7466031fbb8b3b407103 Mon Sep 17 00:00:00 2001 From: eknight-oai Date: Mon, 28 Sep 2026 13:26:12 +0000 Subject: [PATCH 2/3] Prevent message-board notifications from reopening final answers (#48982) ## Why Message-board notifications delivered as ordinary injected input could cause an agent to sample again after its final answer. ## What changed Route notifications through the current turn's mailbox as `InterAgentCommunication`, checking atomically that the turn is running and still accepts mailbox delivery. Skip notifications when delivery is closed. Accepted notifications may remain in conversation history for a later turn without reopening the finalized answer or starting a new turn. Report pending inter-agent communications as mailbox activity, while preserving priority for steering input. ## Testing Add regression scenarios for notifications arriving before and after finalization, asserting no extra sampling and the expected notification presence in the next turn. Extend input-queue coverage to recognize already-pending mailbox activity. GitOrigin-RevId: 41d786041b4eccc5ccf926f2f190473676a3ac2f --- codex-rs/core/src/agent_message_board.rs | 14 +- codex-rs/core/src/session/input_queue.rs | 67 ++++++- .../suite/scenarios_agent_message_board.rs | 176 ++++++++++++++++++ codex-rs/ext/agent-message-board/src/host.rs | 6 +- 4 files changed, 249 insertions(+), 14 deletions(-) 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/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>; From c0d26949be4144c751894ae96e28d3db2208b764 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 28 Sep 2026 13:28:36 +0000 Subject: [PATCH 3/3] Avoid full metadata rewrites for thread timestamp updates (#48983) ## Why Updates that only change `updated_at` should not rewrite unrelated thread metadata or its indexes. ## What changed Use `touch_thread_updated_at` for timestamp-only patches when the SQLite row exists and the rollout path is unchanged. Preserve the full update path for missing rows, changed rollout paths, and patches containing other metadata, including staged changes. ## Testing Add regression coverage for legacy and paginated histories that verifies missing-row reconstruction, timestamp-only writes using SQLite triggers, preservation of other metadata, and application and removal of staged metadata. GitOrigin-RevId: d5bc70d57e591e3eb8e9521f0c31a7ef75f244d3 --- codex-rs/thread-store/src/local/mod.rs | 3 + .../src/local/timestamp_metadata_tests.rs | 126 ++++++++++++++++++ .../src/local/update_thread_metadata.rs | 16 +++ codex-rs/thread-store/src/types.rs | 5 +- 4 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 codex-rs/thread-store/src/local/timestamp_metadata_tests.rs 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()