From 97df35c74fcb042ab5568d6f8fa8c88ca3cea8ed Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 14:48:43 +0200 Subject: [PATCH 1/9] chore: memories mini model (#16559) --- codex-rs/core/src/memories/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codex-rs/core/src/memories/mod.rs b/codex-rs/core/src/memories/mod.rs index 2a7eef4804..ac49c640ca 100644 --- a/codex-rs/core/src/memories/mod.rs +++ b/codex-rs/core/src/memories/mod.rs @@ -32,7 +32,7 @@ mod artifacts { /// Phase 1 (startup extraction). mod phase_one { /// Default model used for phase 1. - pub(super) const MODEL: &str = "gpt-5.1-codex-mini"; + pub(super) const MODEL: &str = "gpt-5.4-mini"; /// Default reasoning effort used for phase 1. pub(super) const REASONING_EFFORT: super::ReasoningEffort = super::ReasoningEffort::Low; /// Prompt used for phase 1. From 627299c5511cc3ce7513dfa6a3751c2f04b89c9e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 15:31:30 +0200 Subject: [PATCH 2/9] fix: race pending (#16561) --- codex-rs/core/src/codex.rs | 57 +++++++-- codex-rs/core/src/codex_tests.rs | 115 ++++++++++++++++++ codex-rs/core/src/state/turn.rs | 24 ++++ codex-rs/core/src/stream_events_utils.rs | 26 ++++ .../core/src/stream_events_utils_tests.rs | 40 +++++- 5 files changed, 250 insertions(+), 12 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 92d75390d4..86105c67fc 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -4023,6 +4023,7 @@ impl Session { let mut turn_state = active_turn.turn_state.lock().await; turn_state.push_pending_input(input.into()); + turn_state.accept_mailbox_delivery_for_current_turn(); Ok(active_turn_id.clone()) } @@ -4044,6 +4045,25 @@ impl Session { } } + pub(crate) async fn defer_mailbox_delivery_to_next_turn(&self, sub_id: &str) { + let turn_state = { + let active = self.active_turn.lock().await; + active.as_ref().and_then(|active_turn| { + active_turn + .tasks + .contains_key(sub_id) + .then(|| Arc::clone(&active_turn.turn_state)) + }) + }; + let Some(turn_state) = turn_state else { + return; + }; + turn_state + .lock() + .await + .defer_mailbox_delivery_to_next_turn(); + } + pub(crate) fn subscribe_mailbox_seq(&self) -> watch::Receiver { self.mailbox.subscribe() } @@ -4069,16 +4089,22 @@ impl Session { } pub async fn get_pending_input(&self) -> Vec { - let pending_input = { + let (pending_input, accepts_mailbox_delivery) = { let mut active = self.active_turn.lock().await; match active.as_mut() { Some(at) => { let mut ts = at.turn_state.lock().await; - ts.take_pending_input() + ( + ts.take_pending_input(), + ts.accepts_mailbox_delivery_for_current_turn(), + ) } - None => Vec::new(), + None => (Vec::new(), true), } }; + if !accepts_mailbox_delivery { + return pending_input; + } let mailbox_items = { let mut mailbox_rx = self.mailbox_rx.lock().await; mailbox_rx @@ -4118,17 +4144,26 @@ impl Session { } pub async fn has_pending_input(&self) -> bool { - if self.mailbox_rx.lock().await.has_pending() { + let (has_turn_pending_input, accepts_mailbox_delivery) = { + let active = self.active_turn.lock().await; + match active.as_ref() { + Some(at) => { + let ts = at.turn_state.lock().await; + ( + ts.has_pending_input(), + ts.accepts_mailbox_delivery_for_current_turn(), + ) + } + None => (false, true), + } + }; + if has_turn_pending_input { return true; } - let active = self.active_turn.lock().await; - match active.as_ref() { - Some(at) => { - let ts = at.turn_state.lock().await; - ts.has_pending_input() - } - None => false, + if !accepts_mailbox_delivery { + return false; } + self.mailbox_rx.lock().await.has_pending() } pub async fn list_resources( diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 8f1bba647c..441095a5a5 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -18,6 +18,7 @@ use crate::tools::format_exec_output_str; use codex_features::Features; use codex_login::CodexAuth; use codex_mcp::mcp_connection_manager::ToolInfo; +use codex_protocol::AgentPath; use codex_protocol::ThreadId; use codex_protocol::models::FunctionCallOutputBody; use codex_protocol::models::FunctionCallOutputPayload; @@ -69,6 +70,7 @@ use codex_protocol::protocol::ConversationAudioParams; use codex_protocol::protocol::CreditsSnapshot; use codex_protocol::protocol::GranularApprovalConfig; use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::NetworkApprovalProtocol; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; @@ -4781,6 +4783,119 @@ async fn queued_response_items_for_next_turn_move_into_next_active_turn() { assert_eq!(sess.get_pending_input().await, vec![queued_item]); } +#[tokio::test] +async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { + let (sess, tc, _rx) = make_session_and_context_with_rx().await; + let communication = InterAgentCommunication::new( + AgentPath::try_from("/root/worker").expect("worker path should parse"), + AgentPath::root(), + Vec::new(), + "late queue-only update".to_string(), + /*trigger_turn*/ false, + ); + sess.spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + + sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await; + sess.enqueue_mailbox_communication(communication.clone()); + + assert!( + !sess.has_pending_input().await, + "queue-only mailbox mail should stay buffered once the current turn emitted its answer" + ); + assert_eq!(sess.get_pending_input().await, Vec::new()); + + sess.abort_all_tasks(TurnAbortReason::Replaced).await; + + assert_eq!( + sess.get_pending_input().await, + vec![communication.to_response_input_item()], + ); +} + +#[tokio::test] +async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() { + let (sess, tc, _rx) = make_session_and_context_with_rx().await; + sess.spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + + sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await; + sess.enqueue_mailbox_communication(InterAgentCommunication::new( + AgentPath::try_from("/root/worker").expect("worker path should parse"), + AgentPath::root(), + Vec::new(), + "late trigger update".to_string(), + /*trigger_turn*/ true, + )); + + assert!( + !sess.has_pending_input().await, + "trigger-turn mailbox mail should not extend the current turn after its answer boundary" + ); + + sess.abort_all_tasks(TurnAbortReason::Replaced).await; + + assert!(sess.has_trigger_turn_mailbox_items().await); +} + +#[tokio::test] +async fn steered_input_reopens_mailbox_delivery_for_current_turn() { + let (sess, tc, _rx) = make_session_and_context_with_rx().await; + let communication = InterAgentCommunication::new( + AgentPath::try_from("/root/worker").expect("worker path should parse"), + AgentPath::root(), + Vec::new(), + "queued child update".to_string(), + /*trigger_turn*/ false, + ); + sess.spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + + sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await; + sess.enqueue_mailbox_communication(communication.clone()); + sess.steer_input( + vec![UserInput::Text { + text: "follow up".to_string(), + text_elements: Vec::new(), + }], + Some(&tc.sub_id), + ) + .await + .expect("steered input should be accepted"); + + assert_eq!( + sess.get_pending_input().await, + vec![ + ResponseInputItem::from(vec![UserInput::Text { + text: "follow up".to_string(), + text_elements: Vec::new(), + }]), + communication.to_response_input_item(), + ], + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn abort_review_task_emits_exited_then_aborted_and_records_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index b573123ad4..3cc6200a31 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -29,6 +29,17 @@ pub(crate) struct ActiveTurn { pub(crate) turn_state: Arc>, } +/// Whether mailbox deliveries should still be folded into the current turn. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(crate) enum MailboxDeliveryPhase { + /// Incoming mailbox messages can still be consumed by the current turn. + #[default] + CurrentTurn, + /// The current turn already emitted visible final answer text; mailbox + /// messages should remain queued for a later turn. + NextTurn, +} + impl Default for ActiveTurn { fn default() -> Self { Self { @@ -81,6 +92,7 @@ pub(crate) struct TurnState { pending_elicitations: HashMap<(String, RequestId), oneshot::Sender>, pending_dynamic_tools: HashMap>, pending_input: Vec, + mailbox_delivery_phase: MailboxDeliveryPhase, granted_permissions: Option, pub(crate) tool_calls: u64, pub(crate) token_usage_at_turn_start: TokenUsage, @@ -202,6 +214,18 @@ impl TurnState { !self.pending_input.is_empty() } + pub(crate) fn defer_mailbox_delivery_to_next_turn(&mut self) { + self.mailbox_delivery_phase = MailboxDeliveryPhase::NextTurn; + } + + pub(crate) fn accept_mailbox_delivery_for_current_turn(&mut self) { + self.mailbox_delivery_phase = MailboxDeliveryPhase::CurrentTurn; + } + + pub(crate) fn accepts_mailbox_delivery_for_current_turn(&self) -> bool { + self.mailbox_delivery_phase == MailboxDeliveryPhase::CurrentTurn + } + pub(crate) fn record_granted_permissions(&mut self, permissions: PermissionProfile) { self.granted_permissions = merge_permission_profiles(self.granted_permissions.as_ref(), Some(&permissions)); diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 5cc9574db3..90a3409580 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -23,6 +23,7 @@ use crate::tools::router::ToolRouter; use codex_protocol::models::DeveloperInstructions; use codex_protocol::models::FunctionCallOutputBody; use codex_protocol::models::FunctionCallOutputPayload; +use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_rollout::state_db; @@ -129,6 +130,13 @@ pub(crate) async fn record_completed_response_item( ) { sess.record_conversation_items(turn_context, std::slice::from_ref(item)) .await; + if completed_item_defers_mailbox_delivery_to_next_turn( + item, + turn_context.collaboration_mode.mode == ModeKind::Plan, + ) { + sess.defer_mailbox_delivery_to_next_turn(&turn_context.sub_id) + .await; + } maybe_mark_thread_memory_mode_polluted_from_web_search(sess, turn_context, item).await; record_stage1_output_usage_for_completed_item(turn_context, item).await; } @@ -426,6 +434,24 @@ pub(crate) fn last_assistant_message_from_item( None } +fn completed_item_defers_mailbox_delivery_to_next_turn( + item: &ResponseItem, + plan_mode: bool, +) -> bool { + match item { + ResponseItem::Message { role, phase, .. } => { + if role != "assistant" || matches!(phase, Some(MessagePhase::Commentary)) { + return false; + } + // Treat `None` like final-answer text so untagged providers default + // to the safer "defer mailbox mail" behavior. + last_assistant_message_from_item(item, plan_mode).is_some() + } + ResponseItem::ImageGenerationCall { .. } => true, + _ => false, + } +} + pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Option { match input { ResponseInputItem::FunctionCallOutput { call_id, output } => { diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index 23f52a068a..5d61870b31 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/codex-rs/core/src/stream_events_utils_tests.rs @@ -1,3 +1,4 @@ +use super::completed_item_defers_mailbox_delivery_to_next_turn; use super::handle_non_tool_response_item; use super::image_generation_artifact_path; use super::last_assistant_message_from_item; @@ -6,10 +7,15 @@ use crate::codex::make_session_and_context; use crate::error::CodexErr; use codex_protocol::items::TurnItem; use codex_protocol::models::ContentItem; +use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; use pretty_assertions::assert_eq; fn assistant_output_text(text: &str) -> ResponseItem { + assistant_output_text_with_phase(text, None) +} + +fn assistant_output_text_with_phase(text: &str, phase: Option) -> ResponseItem { ResponseItem::Message { id: Some("msg-1".to_string()), role: "assistant".to_string(), @@ -17,7 +23,7 @@ fn assistant_output_text(text: &str) -> ResponseItem { text: text.to_string(), }], end_turn: Some(true), - phase: None, + phase, } } @@ -87,6 +93,38 @@ fn last_assistant_message_from_item_returns_none_for_plan_only_hidden_message() ); } +#[test] +fn completed_item_defers_mailbox_delivery_for_unknown_phase_messages() { + let item = assistant_output_text("final answer"); + + assert!(completed_item_defers_mailbox_delivery_to_next_turn( + &item, /*plan_mode*/ false, + )); +} + +#[test] +fn completed_item_keeps_mailbox_delivery_open_for_commentary_messages() { + let item = assistant_output_text_with_phase("still working", Some(MessagePhase::Commentary)); + + assert!(!completed_item_defers_mailbox_delivery_to_next_turn( + &item, /*plan_mode*/ false, + )); +} + +#[test] +fn completed_item_defers_mailbox_delivery_for_image_generation_calls() { + let item = ResponseItem::ImageGenerationCall { + id: "ig-1".to_string(), + status: "completed".to_string(), + revised_prompt: None, + result: "Zm9v".to_string(), + }; + + assert!(completed_item_defers_mailbox_delivery_to_next_turn( + &item, /*plan_mode*/ false, + )); +} + #[tokio::test] async fn save_image_generation_result_saves_base64_to_png_in_codex_home() { let codex_home = tempfile::tempdir().expect("create codex home"); From bd50496411a827a0f48d50d34ef950bd03817f96 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 15:41:18 +0200 Subject: [PATCH 3/9] nit: lint (#16564) --- codex-rs/core/src/stream_events_utils_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index 5d61870b31..f2b0012780 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/codex-rs/core/src/stream_events_utils_tests.rs @@ -12,7 +12,7 @@ use codex_protocol::models::ResponseItem; use pretty_assertions::assert_eq; fn assistant_output_text(text: &str) -> ResponseItem { - assistant_output_text_with_phase(text, None) + assistant_output_text_with_phase(text, /*phase*/ None) } fn assistant_output_text_with_phase(text: &str, phase: Option) -> ResponseItem { From e47ed5e57f299df15a3e40e413171ed331def8cd Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 15:55:55 +0200 Subject: [PATCH 4/9] fix: races in end of turn (#16566) --- codex-rs/core/src/codex.rs | 16 ++++++-- codex-rs/core/src/codex_tests.rs | 49 ++++++++++++++++++++++++ codex-rs/core/src/state/mod.rs | 1 + codex-rs/core/src/state/turn.rs | 8 +++- codex-rs/core/src/stream_events_utils.rs | 4 ++ 5 files changed, 72 insertions(+), 6 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 86105c67fc..17f6701063 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -293,6 +293,7 @@ use crate::shell_snapshot::ShellSnapshot; use crate::skills_watcher::SkillsWatcher; use crate::skills_watcher::SkillsWatcherEvent; use crate::state::ActiveTurn; +use crate::state::MailboxDeliveryPhase; use crate::state::SessionServices; use crate::state::SessionState; use crate::tasks::GhostSnapshotTask; @@ -4046,6 +4047,16 @@ impl Session { } pub(crate) async fn defer_mailbox_delivery_to_next_turn(&self, sub_id: &str) { + self.set_mailbox_delivery_phase(sub_id, MailboxDeliveryPhase::NextTurn) + .await; + } + + pub(crate) async fn accept_mailbox_delivery_for_current_turn(&self, sub_id: &str) { + self.set_mailbox_delivery_phase(sub_id, MailboxDeliveryPhase::CurrentTurn) + .await; + } + + async fn set_mailbox_delivery_phase(&self, sub_id: &str, phase: MailboxDeliveryPhase) { let turn_state = { let active = self.active_turn.lock().await; active.as_ref().and_then(|active_turn| { @@ -4058,10 +4069,7 @@ impl Session { let Some(turn_state) = turn_state else { return; }; - turn_state - .lock() - .await - .defer_mailbox_delivery_to_next_turn(); + turn_state.lock().await.set_mailbox_delivery_phase(phase); } pub(crate) fn subscribe_mailbox_seq(&self) -> watch::Receiver { diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 441095a5a5..f7d343a81e 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -4896,6 +4896,55 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { ); } +#[tokio::test] +async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { + let (sess, tc, _rx) = make_session_and_context_with_rx().await; + let communication = InterAgentCommunication::new( + AgentPath::try_from("/root/worker").expect("worker path should parse"), + AgentPath::root(), + Vec::new(), + "queued child update".to_string(), + /*trigger_turn*/ false, + ); + sess.spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + + sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await; + sess.enqueue_mailbox_communication(communication.clone()); + + let item = ResponseItem::FunctionCall { + id: None, + name: "test_tool".to_string(), + namespace: None, + arguments: "{}".to_string(), + call_id: "call-1".to_string(), + }; + let mut ctx = HandleOutputCtx { + sess: Arc::clone(&sess), + turn_context: Arc::clone(&tc), + tool_runtime: test_tool_runtime(Arc::clone(&sess), Arc::clone(&tc)), + cancellation_token: CancellationToken::new(), + }; + + let output = handle_output_item_done(&mut ctx, item, /*previously_active_item*/ None) + .await + .expect("tool call should be handled"); + + assert!(output.needs_follow_up); + assert!(output.tool_future.is_some()); + assert_eq!( + sess.get_pending_input().await, + vec![communication.to_response_input_item()], + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn abort_review_task_emits_exited_then_aborted_and_records_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/state/mod.rs b/codex-rs/core/src/state/mod.rs index 642433a786..af318d8017 100644 --- a/codex-rs/core/src/state/mod.rs +++ b/codex-rs/core/src/state/mod.rs @@ -5,5 +5,6 @@ mod turn; pub(crate) use service::SessionServices; pub(crate) use session::SessionState; pub(crate) use turn::ActiveTurn; +pub(crate) use turn::MailboxDeliveryPhase; pub(crate) use turn::RunningTask; pub(crate) use turn::TaskKind; diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 3cc6200a31..223edbe7a6 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -215,17 +215,21 @@ impl TurnState { } pub(crate) fn defer_mailbox_delivery_to_next_turn(&mut self) { - self.mailbox_delivery_phase = MailboxDeliveryPhase::NextTurn; + self.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn); } pub(crate) fn accept_mailbox_delivery_for_current_turn(&mut self) { - self.mailbox_delivery_phase = MailboxDeliveryPhase::CurrentTurn; + self.set_mailbox_delivery_phase(MailboxDeliveryPhase::CurrentTurn); } pub(crate) fn accepts_mailbox_delivery_for_current_turn(&self) -> bool { self.mailbox_delivery_phase == MailboxDeliveryPhase::CurrentTurn } + pub(crate) fn set_mailbox_delivery_phase(&mut self, phase: MailboxDeliveryPhase) { + self.mailbox_delivery_phase = phase; + } + pub(crate) fn record_granted_permissions(&mut self, permissions: PermissionProfile) { self.granted_permissions = merge_permission_profiles(self.granted_permissions.as_ref(), Some(&permissions)); diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 90a3409580..931b8429c8 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -213,6 +213,10 @@ pub(crate) async fn handle_output_item_done( match ToolRouter::build_tool_call(ctx.sess.as_ref(), item.clone()).await { // The model emitted a tool call; log it, persist the item immediately, and queue the tool execution. Ok(Some(call)) => { + ctx.sess + .accept_mailbox_delivery_for_current_turn(&ctx.turn_context.sub_id) + .await; + let payload_preview = call.payload.log_payload().into_owned(); tracing::info!( thread_id = %ctx.sess.conversation_id, From ab6cce62b8a7c12c35a2f91a3373b66d2a828a1d Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 16:15:28 +0200 Subject: [PATCH 5/9] chore: rework state machine further (#16567) --- codex-rs/core/src/codex.rs | 50 +++++++++++++++++++------------- codex-rs/core/src/codex_tests.rs | 46 +++++++++++++++++++++++++++++ codex-rs/core/src/state/mod.rs | 1 + codex-rs/core/src/state/turn.rs | 4 --- 4 files changed, 77 insertions(+), 24 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 17f6701063..73f5c6ed7d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -4047,29 +4047,39 @@ impl Session { } pub(crate) async fn defer_mailbox_delivery_to_next_turn(&self, sub_id: &str) { - self.set_mailbox_delivery_phase(sub_id, MailboxDeliveryPhase::NextTurn) - .await; - } - - pub(crate) async fn accept_mailbox_delivery_for_current_turn(&self, sub_id: &str) { - self.set_mailbox_delivery_phase(sub_id, MailboxDeliveryPhase::CurrentTurn) - .await; - } - - async fn set_mailbox_delivery_phase(&self, sub_id: &str, phase: MailboxDeliveryPhase) { - let turn_state = { - let active = self.active_turn.lock().await; - active.as_ref().and_then(|active_turn| { - active_turn - .tasks - .contains_key(sub_id) - .then(|| Arc::clone(&active_turn.turn_state)) - }) - }; + let turn_state = self.turn_state_for_sub_id(sub_id).await; let Some(turn_state) = turn_state else { return; }; - turn_state.lock().await.set_mailbox_delivery_phase(phase); + let mut turn_state = turn_state.lock().await; + if turn_state.has_pending_input() { + return; + } + turn_state.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn); + } + + pub(crate) async fn accept_mailbox_delivery_for_current_turn(&self, sub_id: &str) { + let turn_state = self.turn_state_for_sub_id(sub_id).await; + let Some(turn_state) = turn_state else { + return; + }; + turn_state + .lock() + .await + .set_mailbox_delivery_phase(MailboxDeliveryPhase::CurrentTurn); + } + + async fn turn_state_for_sub_id( + &self, + sub_id: &str, + ) -> Option>> { + let active = self.active_turn.lock().await; + active.as_ref().and_then(|active_turn| { + active_turn + .tasks + .contains_key(sub_id) + .then(|| Arc::clone(&active_turn.turn_state)) + }) } pub(crate) fn subscribe_mailbox_seq(&self) -> watch::Receiver { diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index f7d343a81e..4ff31e8bd0 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -4896,6 +4896,52 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { ); } +#[tokio::test] +async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { + let (sess, tc, _rx) = make_session_and_context_with_rx().await; + let communication = InterAgentCommunication::new( + AgentPath::try_from("/root/worker").expect("worker path should parse"), + AgentPath::root(), + Vec::new(), + "queued child update".to_string(), + /*trigger_turn*/ false, + ); + sess.spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + + sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await; + sess.enqueue_mailbox_communication(communication.clone()); + sess.steer_input( + vec![UserInput::Text { + text: "follow up".to_string(), + text_elements: Vec::new(), + }], + Some(&tc.sub_id), + ) + .await + .expect("steered input should be accepted"); + + sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await; + + assert_eq!( + sess.get_pending_input().await, + vec![ + ResponseInputItem::from(vec![UserInput::Text { + text: "follow up".to_string(), + text_elements: Vec::new(), + }]), + communication.to_response_input_item(), + ], + ); +} + #[tokio::test] async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/state/mod.rs b/codex-rs/core/src/state/mod.rs index af318d8017..f3ebc7225d 100644 --- a/codex-rs/core/src/state/mod.rs +++ b/codex-rs/core/src/state/mod.rs @@ -8,3 +8,4 @@ pub(crate) use turn::ActiveTurn; pub(crate) use turn::MailboxDeliveryPhase; pub(crate) use turn::RunningTask; pub(crate) use turn::TaskKind; +pub(crate) use turn::TurnState; diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 223edbe7a6..8cfd8555e6 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -214,10 +214,6 @@ impl TurnState { !self.pending_input.is_empty() } - pub(crate) fn defer_mailbox_delivery_to_next_turn(&mut self) { - self.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn); - } - pub(crate) fn accept_mailbox_delivery_for_current_turn(&mut self) { self.set_mailbox_delivery_phase(MailboxDeliveryPhase::CurrentTurn); } From ea27d861b21c0540b323bf925feba634eafdd6ef Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 16:18:53 +0200 Subject: [PATCH 6/9] nit: state machine desc (#16569) --- codex-rs/core/src/state/turn.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 8cfd8555e6..a8e3e167b5 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -30,6 +30,16 @@ pub(crate) struct ActiveTurn { } /// Whether mailbox deliveries should still be folded into the current turn. +/// +/// State machine: +/// - A turn starts in `CurrentTurn`, so queued child mail can join the next +/// model request for that turn. +/// - After user-visible terminal output is recorded, we switch to `NextTurn` +/// to leave late child mail queued instead of extending an already shown +/// answer. +/// - If the same task later gets explicit same-turn work again (a steered user +/// prompt or a tool call after an untagged preamble), we reopen `CurrentTurn` +/// so that pending child mail is drained into that follow-up request. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub(crate) enum MailboxDeliveryPhase { /// Incoming mailbox messages can still be consumed by the current turn. From 7fc36249b5f8661fd067018282e68c12a3ae0912 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 2 Apr 2026 16:51:17 +0200 Subject: [PATCH 7/9] chore: rename assign_task for followup_task (#16571) --- .../src/tools/handlers/multi_agents_tests.rs | 34 +++++++++---------- .../src/tools/handlers/multi_agents_v2.rs | 4 +-- .../{assign_task.rs => followup_task.rs} | 4 +-- .../handlers/multi_agents_v2/message_tool.rs | 8 ++--- codex-rs/core/src/tools/spec.rs | 8 ++--- codex-rs/tools/src/agent_tool.rs | 4 +-- codex-rs/tools/src/agent_tool_tests.rs | 10 +++--- codex-rs/tools/src/lib.rs | 2 +- codex-rs/tools/src/tool_registry_plan.rs | 6 ++-- .../tools/src/tool_registry_plan_tests.rs | 10 +++--- .../tools/src/tool_registry_plan_types.rs | 2 +- 11 files changed, 46 insertions(+), 46 deletions(-) rename codex-rs/core/src/tools/handlers/multi_agents_v2/{assign_task.rs => followup_task.rs} (88%) diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 4fe6ac0e90..ee81b961e3 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -9,8 +9,8 @@ use crate::state::TaskKind; use crate::tasks::SessionTask; use crate::tasks::SessionTaskContext; use crate::tools::context::ToolOutput; -use crate::tools::handlers::multi_agents_v2::AssignTaskHandler as AssignTaskHandlerV2; use crate::tools::handlers::multi_agents_v2::CloseAgentHandler as CloseAgentHandlerV2; +use crate::tools::handlers::multi_agents_v2::FollowupTaskHandler as FollowupTaskHandlerV2; use crate::tools::handlers::multi_agents_v2::ListAgentsHandler as ListAgentsHandlerV2; use crate::tools::handlers::multi_agents_v2::SendMessageHandler as SendMessageHandlerV2; use crate::tools::handlers::multi_agents_v2::SpawnAgentHandler as SpawnAgentHandlerV2; @@ -710,7 +710,7 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() { } #[tokio::test] -async fn multi_agent_v2_assign_task_rejects_root_target_from_child() { +async fn multi_agent_v2_followup_task_rejects_root_target_from_child() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); let root = manager @@ -758,11 +758,11 @@ async fn multi_agent_v2_assign_task_rejects_root_target_from_child() { agent_role: None, }); - let err = AssignTaskHandlerV2 + let err = FollowupTaskHandlerV2 .handle(invocation( Arc::new(session), Arc::new(turn), - "assign_task", + "followup_task", function_payload(json!({ "target": "/root", "message": "run this", @@ -770,7 +770,7 @@ async fn multi_agent_v2_assign_task_rejects_root_target_from_child() { })), )) .await - .expect_err("assign_task should reject the root target"); + .expect_err("followup_task should reject the root target"); assert_eq!( err, @@ -1161,7 +1161,7 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() { } #[tokio::test] -async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message() { +async fn multi_agent_v2_followup_task_interrupts_busy_child_without_losing_message() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); let root = manager @@ -1213,11 +1213,11 @@ async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message ) .await; - AssignTaskHandlerV2 + FollowupTaskHandlerV2 .handle(invocation( session, turn, - "assign_task", + "followup_task", function_payload(json!({ "target": agent_id.to_string(), "message": "continue", @@ -1225,7 +1225,7 @@ async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message })), )) .await - .expect("interrupting v2 assign_task should succeed"); + .expect("interrupting v2 followup_task should succeed"); let ops = manager.captured_ops(); let ops_for_agent: Vec<&Op> = ops @@ -1281,7 +1281,7 @@ async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message } }) .await - .expect("interrupting v2 assign_task should preserve the redirected message"); + .expect("interrupting v2 followup_task should preserve the redirected message"); let _ = thread .submit(Op::Shutdown {}) @@ -1290,7 +1290,7 @@ async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message } #[tokio::test] -async fn multi_agent_v2_assign_task_completion_notifies_parent_on_every_turn() { +async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); let root = manager @@ -1342,18 +1342,18 @@ async fn multi_agent_v2_assign_task_completion_notifies_parent_on_every_turn() { ) .await; - AssignTaskHandlerV2 + FollowupTaskHandlerV2 .handle(invocation( session, turn, - "assign_task", + "followup_task", function_payload(json!({ "target": agent_id.to_string(), "message": "continue", })), )) .await - .expect("assign_task should succeed"); + .expect("followup_task should succeed"); let second_turn = thread.codex.session.new_default_turn().await; thread @@ -1419,7 +1419,7 @@ async fn multi_agent_v2_assign_task_completion_notifies_parent_on_every_turn() { } #[tokio::test] -async fn multi_agent_v2_assign_task_rejects_legacy_items_field() { +async fn multi_agent_v2_followup_task_rejects_legacy_items_field() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); let root = manager @@ -1455,14 +1455,14 @@ async fn multi_agent_v2_assign_task_rejects_legacy_items_field() { let invocation = invocation( session, turn, - "assign_task", + "followup_task", function_payload(json!({ "target": agent_id.to_string(), "items": [{"type": "text", "text": "continue"}], })), ); - let Err(err) = AssignTaskHandlerV2.handle(invocation).await else { + let Err(err) = FollowupTaskHandlerV2.handle(invocation).await else { panic!("legacy items field should be rejected in v2"); }; let FunctionCallError::RespondToModel(message) = err else { diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs index e572ec0af1..44191ed85e 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs @@ -28,15 +28,15 @@ use serde::Deserialize; use serde::Serialize; use serde_json::Value as JsonValue; -pub(crate) use assign_task::Handler as AssignTaskHandler; pub(crate) use close_agent::Handler as CloseAgentHandler; +pub(crate) use followup_task::Handler as FollowupTaskHandler; pub(crate) use list_agents::Handler as ListAgentsHandler; pub(crate) use send_message::Handler as SendMessageHandler; pub(crate) use spawn::Handler as SpawnAgentHandler; pub(crate) use wait::Handler as WaitAgentHandler; -mod assign_task; mod close_agent; +mod followup_task; mod list_agents; mod message_tool; mod send_message; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/assign_task.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs similarity index 88% rename from codex-rs/core/src/tools/handlers/multi_agents_v2/assign_task.rs rename to codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs index c55c25d05d..d0f87f3426 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/assign_task.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs @@ -1,4 +1,4 @@ -use super::message_tool::AssignTaskArgs; +use super::message_tool::FollowupTaskArgs; use super::message_tool::MessageDeliveryMode; use super::message_tool::MessageToolResult; use super::message_tool::handle_message_string_tool; @@ -20,7 +20,7 @@ impl ToolHandler for Handler { async fn handle(&self, invocation: ToolInvocation) -> Result { let arguments = function_arguments(invocation.payload.clone())?; - let args: AssignTaskArgs = parse_arguments(&arguments)?; + let args: FollowupTaskArgs = parse_arguments(&arguments)?; handle_message_string_tool( invocation, MessageDeliveryMode::TriggerTurn, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index 62835a12ca..3123e93f4c 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -1,6 +1,6 @@ //! Shared argument parsing and dispatch for the v2 text-only agent messaging tools. //! -//! `send_message` and `assign_task` share the same submission path and differ only in whether the +//! `send_message` and `followup_task` share the same submission path and differ only in whether the //! resulting `InterAgentCommunication` should wake the target immediately. use super::*; @@ -38,8 +38,8 @@ pub(crate) struct SendMessageArgs { #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] -/// Input for the MultiAgentV2 `assign_task` tool. -pub(crate) struct AssignTaskArgs { +/// Input for the MultiAgentV2 `followup_task` tool. +pub(crate) struct FollowupTaskArgs { pub(crate) target: String, pub(crate) message: String, #[serde(default)] @@ -79,7 +79,7 @@ fn message_content(message: String) -> Result { Ok(message) } -/// Handles the shared MultiAgentV2 plain-text message flow for both `send_message` and `assign_task`. +/// Handles the shared MultiAgentV2 plain-text message flow for both `send_message` and `followup_task`. pub(crate) async fn handle_message_string_tool( invocation: ToolInvocation, mode: MessageDeliveryMode, diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index 341db24767..90b9a382aa 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/codex-rs/core/src/tools/spec.rs @@ -60,8 +60,8 @@ pub(crate) fn build_specs_with_discoverable_tools( use crate::tools::handlers::multi_agents::SendInputHandler; use crate::tools::handlers::multi_agents::SpawnAgentHandler; use crate::tools::handlers::multi_agents::WaitAgentHandler; - use crate::tools::handlers::multi_agents_v2::AssignTaskHandler as AssignTaskHandlerV2; use crate::tools::handlers::multi_agents_v2::CloseAgentHandler as CloseAgentHandlerV2; + use crate::tools::handlers::multi_agents_v2::FollowupTaskHandler as FollowupTaskHandlerV2; use crate::tools::handlers::multi_agents_v2::ListAgentsHandler as ListAgentsHandlerV2; use crate::tools::handlers::multi_agents_v2::SendMessageHandler as SendMessageHandlerV2; use crate::tools::handlers::multi_agents_v2::SpawnAgentHandler as SpawnAgentHandlerV2; @@ -136,9 +136,6 @@ pub(crate) fn build_specs_with_discoverable_tools( ToolHandlerKind::ApplyPatch => { builder.register_handler(handler.name, apply_patch_handler.clone()); } - ToolHandlerKind::AssignTaskV2 => { - builder.register_handler(handler.name, Arc::new(AssignTaskHandlerV2)); - } ToolHandlerKind::CloseAgentV1 => { builder.register_handler(handler.name, Arc::new(CloseAgentHandler)); } @@ -154,6 +151,9 @@ pub(crate) fn build_specs_with_discoverable_tools( ToolHandlerKind::DynamicTool => { builder.register_handler(handler.name, dynamic_tool_handler.clone()); } + ToolHandlerKind::FollowupTaskV2 => { + builder.register_handler(handler.name, Arc::new(FollowupTaskHandlerV2)); + } ToolHandlerKind::JsRepl => { builder.register_handler(handler.name, js_repl_handler.clone()); } diff --git a/codex-rs/tools/src/agent_tool.rs b/codex-rs/tools/src/agent_tool.rs index b82084d51d..017cc6e5ee 100644 --- a/codex-rs/tools/src/agent_tool.rs +++ b/codex-rs/tools/src/agent_tool.rs @@ -150,7 +150,7 @@ pub fn create_send_message_tool() -> ToolSpec { }) } -pub fn create_assign_task_tool() -> ToolSpec { +pub fn create_followup_task_tool() -> ToolSpec { let properties = BTreeMap::from([ ( "target".to_string(), @@ -178,7 +178,7 @@ pub fn create_assign_task_tool() -> ToolSpec { ]); ToolSpec::Function(ResponsesApiTool { - name: "assign_task".to_string(), + name: "followup_task".to_string(), description: "Add a message to an existing non-root agent and trigger a turn in the target. Use interrupt=true to redirect work immediately. In MultiAgentV2, this tool currently supports text content only." .to_string(), strict: false, diff --git a/codex-rs/tools/src/agent_tool_tests.rs b/codex-rs/tools/src/agent_tool_tests.rs index 2277a5ec19..cd68fbd13a 100644 --- a/codex-rs/tools/src/agent_tool_tests.rs +++ b/codex-rs/tools/src/agent_tool_tests.rs @@ -127,14 +127,14 @@ fn send_message_tool_requires_message_and_uses_submission_output() { } #[test] -fn assign_task_tool_requires_message_and_uses_submission_output() { +fn followup_task_tool_requires_message_and_uses_submission_output() { let ToolSpec::Function(ResponsesApiTool { parameters, output_schema, .. - }) = create_assign_task_tool() + }) = create_followup_task_tool() else { - panic!("assign_task should be a function tool"); + panic!("followup_task should be a function tool"); }; let JsonSchema::Object { properties, @@ -142,7 +142,7 @@ fn assign_task_tool_requires_message_and_uses_submission_output() { .. } = parameters else { - panic!("assign_task should use object params"); + panic!("followup_task should use object params"); }; assert!(properties.contains_key("target")); assert!(properties.contains_key("message")); @@ -153,7 +153,7 @@ fn assign_task_tool_requires_message_and_uses_submission_output() { Some(vec!["target".to_string(), "message".to_string()]) ); assert_eq!( - output_schema.expect("assign_task output schema")["required"], + output_schema.expect("followup_task output schema")["required"], json!(["submission_id"]) ); } diff --git a/codex-rs/tools/src/lib.rs b/codex-rs/tools/src/lib.rs index 91931e8944..16e84700bb 100644 --- a/codex-rs/tools/src/lib.rs +++ b/codex-rs/tools/src/lib.rs @@ -29,9 +29,9 @@ pub use agent_job_tool::create_report_agent_job_result_tool; pub use agent_job_tool::create_spawn_agents_on_csv_tool; pub use agent_tool::SpawnAgentToolOptions; pub use agent_tool::WaitAgentTimeoutOptions; -pub use agent_tool::create_assign_task_tool; pub use agent_tool::create_close_agent_tool_v1; pub use agent_tool::create_close_agent_tool_v2; +pub use agent_tool::create_followup_task_tool; pub use agent_tool::create_list_agents_tool; pub use agent_tool::create_resume_agent_tool; pub use agent_tool::create_send_input_tool_v1; diff --git a/codex-rs/tools/src/tool_registry_plan.rs b/codex-rs/tools/src/tool_registry_plan.rs index 314d93f55f..cc706f24fc 100644 --- a/codex-rs/tools/src/tool_registry_plan.rs +++ b/codex-rs/tools/src/tool_registry_plan.rs @@ -18,11 +18,11 @@ use crate::collect_tool_search_app_infos; use crate::collect_tool_suggest_entries; use crate::create_apply_patch_freeform_tool; use crate::create_apply_patch_json_tool; -use crate::create_assign_task_tool; use crate::create_close_agent_tool_v1; use crate::create_close_agent_tool_v2; use crate::create_code_mode_tool; use crate::create_exec_command_tool; +use crate::create_followup_task_tool; use crate::create_image_generation_tool; use crate::create_js_repl_reset_tool; use crate::create_js_repl_tool; @@ -358,7 +358,7 @@ pub fn build_tool_registry_plan( config.code_mode_enabled, ); plan.push_spec( - create_assign_task_tool(), + create_followup_task_tool(), /*supports_parallel_tool_calls*/ false, config.code_mode_enabled, ); @@ -379,7 +379,7 @@ pub fn build_tool_registry_plan( ); plan.register_handler("spawn_agent", ToolHandlerKind::SpawnAgentV2); plan.register_handler("send_message", ToolHandlerKind::SendMessageV2); - plan.register_handler("assign_task", ToolHandlerKind::AssignTaskV2); + plan.register_handler("followup_task", ToolHandlerKind::FollowupTaskV2); plan.register_handler("wait_agent", ToolHandlerKind::WaitAgentV2); plan.register_handler("close_agent", ToolHandlerKind::CloseAgentV2); plan.register_handler("list_agents", ToolHandlerKind::ListAgentsV2); diff --git a/codex-rs/tools/src/tool_registry_plan_tests.rs b/codex-rs/tools/src/tool_registry_plan_tests.rs index c2c4a0aea1..fd7ab10073 100644 --- a/codex-rs/tools/src/tool_registry_plan_tests.rs +++ b/codex-rs/tools/src/tool_registry_plan_tests.rs @@ -207,7 +207,7 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() { &[ "spawn_agent", "send_message", - "assign_task", + "followup_task", "wait_agent", "close_agent", "list_agents", @@ -269,9 +269,9 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() { Some(&vec!["target".to_string(), "message".to_string()]) ); - let assign_task = find_tool(&tools, "assign_task"); - let ToolSpec::Function(ResponsesApiTool { parameters, .. }) = &assign_task.spec else { - panic!("assign_task should be a function tool"); + let followup_task = find_tool(&tools, "followup_task"); + let ToolSpec::Function(ResponsesApiTool { parameters, .. }) = &followup_task.spec else { + panic!("followup_task should be a function tool"); }; let JsonSchema::Object { properties, @@ -279,7 +279,7 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() { .. } = parameters else { - panic!("assign_task should use object params"); + panic!("followup_task should use object params"); }; assert!(properties.contains_key("target")); assert!(properties.contains_key("message")); diff --git a/codex-rs/tools/src/tool_registry_plan_types.rs b/codex-rs/tools/src/tool_registry_plan_types.rs index bdcb75acd8..d15cf15d5a 100644 --- a/codex-rs/tools/src/tool_registry_plan_types.rs +++ b/codex-rs/tools/src/tool_registry_plan_types.rs @@ -12,12 +12,12 @@ use std::collections::HashMap; pub enum ToolHandlerKind { AgentJobs, ApplyPatch, - AssignTaskV2, CloseAgentV1, CloseAgentV2, CodeModeExecute, CodeModeWait, DynamicTool, + FollowupTaskV2, JsRepl, JsReplReset, ListAgentsV2, From a0988341483676caa019a5e288f69e45c5d5c058 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 2 Apr 2026 08:41:04 -0700 Subject: [PATCH 8/9] ci: upload compact Bazel execution logs for bazel.yml (#16577) ## Why The main Bazel CI lanes need compact execution logs to investigate cache misses and unexpected rebuilds, but local users of the shared wrapper should not pay that log-generation cost by default. ## What Changed - [`.github/scripts/run-bazel-ci.sh`](https://github.com/openai/codex/blob/a6ec239a24fd21507bb75a28f17277653f52e3a2/.github/scripts/run-bazel-ci.sh#L149-L153) now appends `--execution_log_compact_file=...` only when `CODEX_BAZEL_EXECUTION_LOG_COMPACT_DIR` is set; the caller owns creating that directory. - [`.github/workflows/bazel.yml`](https://github.com/openai/codex/blob/a6ec239a24fd21507bb75a28f17277653f52e3a2/.github/workflows/bazel.yml#L66-L174) enables that env var only for the main `test` and `clippy` jobs, creates the temp log directory in each job, and uploads the resulting `*.zst` files from `runner.temp`. ## Verification - `bash -n .github/scripts/run-bazel-ci.sh` - Parsed `.github/workflows/bazel.yml` as YAML. - Ran a local opt-in wrapper smoke test and confirmed it writes `execution-log-cquery-local-*.zst` when the caller pre-creates `CODEX_BAZEL_EXECUTION_LOG_COMPACT_DIR`. --- .github/scripts/run-bazel-ci.sh | 6 ++++++ .github/workflows/bazel.yml | 30 ++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/.github/scripts/run-bazel-ci.sh b/.github/scripts/run-bazel-ci.sh index d08b32e3d4..0e46b71573 100755 --- a/.github/scripts/run-bazel-ci.sh +++ b/.github/scripts/run-bazel-ci.sh @@ -146,6 +146,12 @@ if [[ -n "${BAZEL_REPOSITORY_CACHE:-}" ]]; then post_config_bazel_args+=("--repository_cache=${BAZEL_REPOSITORY_CACHE}") fi +if [[ -n "${CODEX_BAZEL_EXECUTION_LOG_COMPACT_DIR:-}" ]]; then + post_config_bazel_args+=( + "--execution_log_compact_file=${CODEX_BAZEL_EXECUTION_LOG_COMPACT_DIR}/execution-log-${bazel_args[0]}-${GITHUB_JOB:-local}-$$.zst" + ) +fi + if [[ "${RUNNER_OS:-}" == "Windows" ]]; then windows_action_env_vars=( INCLUDE diff --git a/.github/workflows/bazel.yml b/.github/workflows/bazel.yml index a6c9277b61..fa1398c3e9 100644 --- a/.github/workflows/bazel.yml +++ b/.github/workflows/bazel.yml @@ -63,6 +63,12 @@ jobs: shell: bash run: ./scripts/check-module-bazel-lock.sh + - name: Set up Bazel execution logs + shell: bash + run: | + mkdir -p "${RUNNER_TEMP}/bazel-execution-logs" + echo "CODEX_BAZEL_EXECUTION_LOG_COMPACT_DIR=${RUNNER_TEMP}/bazel-execution-logs" >> "${GITHUB_ENV}" + - name: bazel test //... env: BUILDBUDDY_API_KEY: ${{ secrets.BUILDBUDDY_API_KEY }} @@ -87,6 +93,15 @@ jobs: -- \ "${bazel_targets[@]}" + - name: Upload Bazel execution logs + if: always() && !cancelled() + continue-on-error: true + uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7 + with: + name: bazel-execution-logs-test-${{ matrix.target }} + path: ${{ runner.temp }}/bazel-execution-logs + if-no-files-found: ignore + # Save bazel repository cache explicitly; make non-fatal so cache uploading # never fails the overall job. Only save when key wasn't hit. - name: Save bazel repository cache @@ -126,6 +141,12 @@ jobs: with: target: ${{ matrix.target }} + - name: Set up Bazel execution logs + shell: bash + run: | + mkdir -p "${RUNNER_TEMP}/bazel-execution-logs" + echo "CODEX_BAZEL_EXECUTION_LOG_COMPACT_DIR=${RUNNER_TEMP}/bazel-execution-logs" >> "${GITHUB_ENV}" + - name: bazel build --config=clippy //codex-rs/... env: BUILDBUDDY_API_KEY: ${{ secrets.BUILDBUDDY_API_KEY }} @@ -143,6 +164,15 @@ jobs: //codex-rs/... \ -//codex-rs/v8-poc:all + - name: Upload Bazel execution logs + if: always() && !cancelled() + continue-on-error: true + uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7 + with: + name: bazel-execution-logs-clippy-${{ matrix.target }} + path: ${{ runner.temp }}/bazel-execution-logs + if-no-files-found: ignore + # Save bazel repository cache explicitly; make non-fatal so cache uploading # never fails the overall job. Only save when key wasn't hit. - name: Save bazel repository cache From ce349df98acdf14be32811859df0e8c54032e2c4 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 2 Apr 2026 09:08:12 -0700 Subject: [PATCH 9/9] chore: move unit tests in codex-exec into their own files --- codex-rs/exec/src/lib.rs | 406 +-------------------------------- codex-rs/exec/src/lib_tests.rs | 402 ++++++++++++++++++++++++++++++++ 2 files changed, 404 insertions(+), 404 deletions(-) create mode 100644 codex-rs/exec/src/lib_tests.rs diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 6ad3b2047b..5a396a9810 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -1657,407 +1657,5 @@ fn build_review_request(args: &ReviewArgs) -> anyhow::Result { } #[cfg(test)] -mod tests { - use super::*; - use codex_otel::set_parent_from_w3c_trace_context; - use codex_protocol::config_types::ApprovalsReviewer; - use opentelemetry::trace::TraceContextExt; - use opentelemetry::trace::TraceId; - use opentelemetry::trace::TracerProvider as _; - use opentelemetry_sdk::trace::SdkTracerProvider; - use pretty_assertions::assert_eq; - use tempfile::tempdir; - use tracing_opentelemetry::OpenTelemetrySpanExt; - - fn test_tracing_subscriber() -> impl tracing::Subscriber + Send + Sync { - let provider = SdkTracerProvider::builder().build(); - let tracer = provider.tracer("codex-exec-tests"); - tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer)) - } - - #[test] - fn exec_defaults_analytics_to_enabled() { - assert_eq!(DEFAULT_ANALYTICS_ENABLED, true); - } - - #[test] - fn exec_root_span_can_be_parented_from_trace_context() { - let subscriber = test_tracing_subscriber(); - let _guard = tracing::subscriber::set_default(subscriber); - - let parent = codex_protocol::protocol::W3cTraceContext { - traceparent: Some("00-00000000000000000000000000000077-0000000000000088-01".into()), - tracestate: Some("vendor=value".into()), - }; - let exec_span = exec_root_span(); - assert!(set_parent_from_w3c_trace_context(&exec_span, &parent)); - - let trace_id = exec_span.context().span().span_context().trace_id(); - assert_eq!( - trace_id, - TraceId::from_hex("00000000000000000000000000000077").expect("trace id") - ); - } - - #[test] - fn builds_uncommitted_review_request() { - let args = ReviewArgs { - uncommitted: true, - base: None, - commit: None, - commit_title: None, - prompt: None, - }; - let request = build_review_request(&args).expect("builds uncommitted review request"); - - let expected = ReviewRequest { - target: ReviewTarget::UncommittedChanges, - user_facing_hint: None, - }; - - assert_eq!(request, expected); - } - - #[test] - fn builds_commit_review_request_with_title() { - let args = ReviewArgs { - uncommitted: false, - base: None, - commit: Some("123456789".to_string()), - commit_title: Some("Add review command".to_string()), - prompt: None, - }; - let request = build_review_request(&args).expect("builds commit review request"); - - let expected = ReviewRequest { - target: ReviewTarget::Commit { - sha: "123456789".to_string(), - title: Some("Add review command".to_string()), - }, - user_facing_hint: None, - }; - - assert_eq!(request, expected); - } - - #[test] - fn builds_custom_review_request_trims_prompt() { - let args = ReviewArgs { - uncommitted: false, - base: None, - commit: None, - commit_title: None, - prompt: Some(" custom review instructions ".to_string()), - }; - let request = build_review_request(&args).expect("builds custom review request"); - - let expected = ReviewRequest { - target: ReviewTarget::Custom { - instructions: "custom review instructions".to_string(), - }, - user_facing_hint: None, - }; - - assert_eq!(request, expected); - } - - #[test] - fn decode_prompt_bytes_strips_utf8_bom() { - let input = [0xEF, 0xBB, 0xBF, b'h', b'i', b'\n']; - - let out = decode_prompt_bytes(&input).expect("decode utf-8 with BOM"); - - assert_eq!(out, "hi\n"); - } - - #[test] - fn decode_prompt_bytes_decodes_utf16le_bom() { - // UTF-16LE BOM + "hi\n" - let input = [0xFF, 0xFE, b'h', 0x00, b'i', 0x00, b'\n', 0x00]; - - let out = decode_prompt_bytes(&input).expect("decode utf-16le with BOM"); - - assert_eq!(out, "hi\n"); - } - - #[test] - fn decode_prompt_bytes_decodes_utf16be_bom() { - // UTF-16BE BOM + "hi\n" - let input = [0xFE, 0xFF, 0x00, b'h', 0x00, b'i', 0x00, b'\n']; - - let out = decode_prompt_bytes(&input).expect("decode utf-16be with BOM"); - - assert_eq!(out, "hi\n"); - } - - #[test] - fn decode_prompt_bytes_rejects_utf32le_bom() { - // UTF-32LE BOM + "hi\n" - let input = [ - 0xFF, 0xFE, 0x00, 0x00, b'h', 0x00, 0x00, 0x00, b'i', 0x00, 0x00, 0x00, b'\n', 0x00, - 0x00, 0x00, - ]; - - let err = decode_prompt_bytes(&input).expect_err("utf-32le should be rejected"); - - assert_eq!( - err, - PromptDecodeError::UnsupportedBom { - encoding: "UTF-32LE" - } - ); - } - - #[test] - fn decode_prompt_bytes_rejects_utf32be_bom() { - // UTF-32BE BOM + "hi\n" - let input = [ - 0x00, 0x00, 0xFE, 0xFF, 0x00, 0x00, 0x00, b'h', 0x00, 0x00, 0x00, b'i', 0x00, 0x00, - 0x00, b'\n', - ]; - - let err = decode_prompt_bytes(&input).expect_err("utf-32be should be rejected"); - - assert_eq!( - err, - PromptDecodeError::UnsupportedBom { - encoding: "UTF-32BE" - } - ); - } - - #[test] - fn decode_prompt_bytes_rejects_invalid_utf8() { - // Invalid UTF-8 sequence: 0xC3 0x28 - let input = [0xC3, 0x28]; - - let err = decode_prompt_bytes(&input).expect_err("invalid utf-8 should fail"); - - assert_eq!(err, PromptDecodeError::InvalidUtf8 { valid_up_to: 0 }); - } - - #[test] - fn prompt_with_stdin_context_wraps_stdin_block() { - let combined = prompt_with_stdin_context("Summarize this concisely", "my output"); - - assert_eq!( - combined, - "Summarize this concisely\n\n\nmy output\n" - ); - } - - #[test] - fn prompt_with_stdin_context_preserves_trailing_newline() { - let combined = prompt_with_stdin_context("Summarize this concisely", "my output\n"); - - assert_eq!( - combined, - "Summarize this concisely\n\n\nmy output\n" - ); - } - - #[test] - fn lagged_event_warning_message_is_explicit() { - assert_eq!( - lagged_event_warning_message(/*skipped*/ 7), - "in-process app-server event stream lagged; dropped 7 events".to_string() - ); - } - - #[tokio::test] - async fn resume_lookup_model_providers_filters_only_last_lookup() { - let codex_home = tempdir().expect("create temp codex home"); - let cwd = tempdir().expect("create temp cwd"); - let mut config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(cwd.path().to_path_buf())) - .build() - .await - .expect("build default config"); - config.model_provider_id = "test-provider".to_string(); - - let last_args = crate::cli::ResumeArgs { - session_id: None, - last: true, - all: false, - images: vec![], - prompt: None, - }; - let named_args = crate::cli::ResumeArgs { - session_id: Some("named-session".to_string()), - last: false, - all: false, - images: vec![], - prompt: None, - }; - - assert_eq!( - resume_lookup_model_providers(&config, &last_args), - Some(vec!["test-provider".to_string()]) - ); - assert_eq!(resume_lookup_model_providers(&config, &named_args), None); - } - - #[test] - fn turn_items_for_thread_returns_matching_turn_items() { - let thread = AppServerThread { - id: "thread-1".to_string(), - preview: String::new(), - ephemeral: false, - model_provider: "openai".to_string(), - created_at: 0, - updated_at: 0, - status: codex_app_server_protocol::ThreadStatus::Idle, - path: None, - cwd: PathBuf::from("/tmp/project"), - cli_version: "0.0.0-test".to_string(), - source: codex_app_server_protocol::SessionSource::Exec, - agent_nickname: None, - agent_role: None, - git_info: None, - name: None, - turns: vec![ - codex_app_server_protocol::Turn { - id: "turn-1".to_string(), - items: vec![AppServerThreadItem::AgentMessage { - id: "msg-1".to_string(), - text: "hello".to_string(), - phase: None, - memory_citation: None, - }], - status: codex_app_server_protocol::TurnStatus::Completed, - error: None, - }, - codex_app_server_protocol::Turn { - id: "turn-2".to_string(), - items: vec![AppServerThreadItem::Plan { - id: "plan-1".to_string(), - text: "ship it".to_string(), - }], - status: codex_app_server_protocol::TurnStatus::Completed, - error: None, - }, - ], - }; - - assert_eq!( - turn_items_for_thread(&thread, "turn-1"), - Some(vec![AppServerThreadItem::AgentMessage { - id: "msg-1".to_string(), - text: "hello".to_string(), - phase: None, - memory_citation: None, - }]) - ); - assert_eq!(turn_items_for_thread(&thread, "missing-turn"), None); - } - - #[test] - fn canceled_mcp_server_elicitation_response_uses_cancel_action() { - let value = canceled_mcp_server_elicitation_response() - .expect("mcp elicitation cancel response should serialize"); - let response: McpServerElicitationRequestResponse = - serde_json::from_value(value).expect("cancel response should deserialize"); - - assert_eq!( - response, - McpServerElicitationRequestResponse { - action: McpServerElicitationAction::Cancel, - content: None, - meta: None, - } - ); - } - - #[tokio::test] - async fn thread_start_params_include_review_policy_when_review_policy_is_manual_only() { - let codex_home = tempdir().expect("create temp codex home"); - let cwd = tempdir().expect("create temp cwd"); - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .harness_overrides(ConfigOverrides { - approvals_reviewer: Some(ApprovalsReviewer::User), - ..Default::default() - }) - .fallback_cwd(Some(cwd.path().to_path_buf())) - .build() - .await - .expect("build config with manual-only review policy"); - - let params = thread_start_params_from_config(&config); - - assert_eq!( - params.approvals_reviewer, - Some(codex_app_server_protocol::ApprovalsReviewer::User) - ); - } - - #[tokio::test] - async fn thread_start_params_include_review_policy_when_auto_review_is_enabled() { - let codex_home = tempdir().expect("create temp codex home"); - let cwd = tempdir().expect("create temp cwd"); - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .harness_overrides(ConfigOverrides { - approvals_reviewer: Some(ApprovalsReviewer::GuardianSubagent), - ..Default::default() - }) - .fallback_cwd(Some(cwd.path().to_path_buf())) - .build() - .await - .expect("build config with guardian review policy"); - - let params = thread_start_params_from_config(&config); - - assert_eq!( - params.approvals_reviewer, - Some(codex_app_server_protocol::ApprovalsReviewer::GuardianSubagent) - ); - } - - #[test] - fn session_configured_from_thread_response_uses_review_policy_from_response() { - let response = ThreadStartResponse { - thread: codex_app_server_protocol::Thread { - id: "67e55044-10b1-426f-9247-bb680e5fe0c8".to_string(), - preview: String::new(), - ephemeral: false, - model_provider: "openai".to_string(), - created_at: 0, - updated_at: 0, - status: codex_app_server_protocol::ThreadStatus::Idle, - path: Some(PathBuf::from("/tmp/rollout.jsonl")), - cwd: PathBuf::from("/tmp"), - cli_version: "0.0.0".to_string(), - source: codex_app_server_protocol::SessionSource::Cli, - agent_nickname: None, - agent_role: None, - git_info: None, - name: Some("thread".to_string()), - turns: vec![], - }, - model: "gpt-5.4".to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: PathBuf::from("/tmp"), - approval_policy: codex_app_server_protocol::AskForApproval::OnRequest, - approvals_reviewer: codex_app_server_protocol::ApprovalsReviewer::GuardianSubagent, - sandbox: codex_app_server_protocol::SandboxPolicy::WorkspaceWrite { - writable_roots: vec![], - read_only_access: codex_app_server_protocol::ReadOnlyAccess::FullAccess, - network_access: false, - exclude_tmpdir_env_var: false, - exclude_slash_tmp: false, - }, - reasoning_effort: None, - }; - - let event = session_configured_from_thread_start_response(&response) - .expect("build bootstrap session configured event"); - - assert_eq!( - event.approvals_reviewer, - ApprovalsReviewer::GuardianSubagent - ); - } -} +#[path = "lib_tests.rs"] +mod tests; diff --git a/codex-rs/exec/src/lib_tests.rs b/codex-rs/exec/src/lib_tests.rs new file mode 100644 index 0000000000..b223ba13cc --- /dev/null +++ b/codex-rs/exec/src/lib_tests.rs @@ -0,0 +1,402 @@ +use super::*; +use codex_otel::set_parent_from_w3c_trace_context; +use codex_protocol::config_types::ApprovalsReviewer; +use opentelemetry::trace::TraceContextExt; +use opentelemetry::trace::TraceId; +use opentelemetry::trace::TracerProvider as _; +use opentelemetry_sdk::trace::SdkTracerProvider; +use pretty_assertions::assert_eq; +use tempfile::tempdir; +use tracing_opentelemetry::OpenTelemetrySpanExt; + +fn test_tracing_subscriber() -> impl tracing::Subscriber + Send + Sync { + let provider = SdkTracerProvider::builder().build(); + let tracer = provider.tracer("codex-exec-tests"); + tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer)) +} + +#[test] +fn exec_defaults_analytics_to_enabled() { + assert_eq!(DEFAULT_ANALYTICS_ENABLED, true); +} + +#[test] +fn exec_root_span_can_be_parented_from_trace_context() { + let subscriber = test_tracing_subscriber(); + let _guard = tracing::subscriber::set_default(subscriber); + + let parent = codex_protocol::protocol::W3cTraceContext { + traceparent: Some("00-00000000000000000000000000000077-0000000000000088-01".into()), + tracestate: Some("vendor=value".into()), + }; + let exec_span = exec_root_span(); + assert!(set_parent_from_w3c_trace_context(&exec_span, &parent)); + + let trace_id = exec_span.context().span().span_context().trace_id(); + assert_eq!( + trace_id, + TraceId::from_hex("00000000000000000000000000000077").expect("trace id") + ); +} + +#[test] +fn builds_uncommitted_review_request() { + let args = ReviewArgs { + uncommitted: true, + base: None, + commit: None, + commit_title: None, + prompt: None, + }; + let request = build_review_request(&args).expect("builds uncommitted review request"); + + let expected = ReviewRequest { + target: ReviewTarget::UncommittedChanges, + user_facing_hint: None, + }; + + assert_eq!(request, expected); +} + +#[test] +fn builds_commit_review_request_with_title() { + let args = ReviewArgs { + uncommitted: false, + base: None, + commit: Some("123456789".to_string()), + commit_title: Some("Add review command".to_string()), + prompt: None, + }; + let request = build_review_request(&args).expect("builds commit review request"); + + let expected = ReviewRequest { + target: ReviewTarget::Commit { + sha: "123456789".to_string(), + title: Some("Add review command".to_string()), + }, + user_facing_hint: None, + }; + + assert_eq!(request, expected); +} + +#[test] +fn builds_custom_review_request_trims_prompt() { + let args = ReviewArgs { + uncommitted: false, + base: None, + commit: None, + commit_title: None, + prompt: Some(" custom review instructions ".to_string()), + }; + let request = build_review_request(&args).expect("builds custom review request"); + + let expected = ReviewRequest { + target: ReviewTarget::Custom { + instructions: "custom review instructions".to_string(), + }, + user_facing_hint: None, + }; + + assert_eq!(request, expected); +} + +#[test] +fn decode_prompt_bytes_strips_utf8_bom() { + let input = [0xEF, 0xBB, 0xBF, b'h', b'i', b'\n']; + + let out = decode_prompt_bytes(&input).expect("decode utf-8 with BOM"); + + assert_eq!(out, "hi\n"); +} + +#[test] +fn decode_prompt_bytes_decodes_utf16le_bom() { + // UTF-16LE BOM + "hi\n" + let input = [0xFF, 0xFE, b'h', 0x00, b'i', 0x00, b'\n', 0x00]; + + let out = decode_prompt_bytes(&input).expect("decode utf-16le with BOM"); + + assert_eq!(out, "hi\n"); +} + +#[test] +fn decode_prompt_bytes_decodes_utf16be_bom() { + // UTF-16BE BOM + "hi\n" + let input = [0xFE, 0xFF, 0x00, b'h', 0x00, b'i', 0x00, b'\n']; + + let out = decode_prompt_bytes(&input).expect("decode utf-16be with BOM"); + + assert_eq!(out, "hi\n"); +} + +#[test] +fn decode_prompt_bytes_rejects_utf32le_bom() { + // UTF-32LE BOM + "hi\n" + let input = [ + 0xFF, 0xFE, 0x00, 0x00, b'h', 0x00, 0x00, 0x00, b'i', 0x00, 0x00, 0x00, b'\n', 0x00, 0x00, + 0x00, + ]; + + let err = decode_prompt_bytes(&input).expect_err("utf-32le should be rejected"); + + assert_eq!( + err, + PromptDecodeError::UnsupportedBom { + encoding: "UTF-32LE" + } + ); +} + +#[test] +fn decode_prompt_bytes_rejects_utf32be_bom() { + // UTF-32BE BOM + "hi\n" + let input = [ + 0x00, 0x00, 0xFE, 0xFF, 0x00, 0x00, 0x00, b'h', 0x00, 0x00, 0x00, b'i', 0x00, 0x00, 0x00, + b'\n', + ]; + + let err = decode_prompt_bytes(&input).expect_err("utf-32be should be rejected"); + + assert_eq!( + err, + PromptDecodeError::UnsupportedBom { + encoding: "UTF-32BE" + } + ); +} + +#[test] +fn decode_prompt_bytes_rejects_invalid_utf8() { + // Invalid UTF-8 sequence: 0xC3 0x28 + let input = [0xC3, 0x28]; + + let err = decode_prompt_bytes(&input).expect_err("invalid utf-8 should fail"); + + assert_eq!(err, PromptDecodeError::InvalidUtf8 { valid_up_to: 0 }); +} + +#[test] +fn prompt_with_stdin_context_wraps_stdin_block() { + let combined = prompt_with_stdin_context("Summarize this concisely", "my output"); + + assert_eq!( + combined, + "Summarize this concisely\n\n\nmy output\n" + ); +} + +#[test] +fn prompt_with_stdin_context_preserves_trailing_newline() { + let combined = prompt_with_stdin_context("Summarize this concisely", "my output\n"); + + assert_eq!( + combined, + "Summarize this concisely\n\n\nmy output\n" + ); +} + +#[test] +fn lagged_event_warning_message_is_explicit() { + assert_eq!( + lagged_event_warning_message(/*skipped*/ 7), + "in-process app-server event stream lagged; dropped 7 events".to_string() + ); +} + +#[tokio::test] +async fn resume_lookup_model_providers_filters_only_last_lookup() { + let codex_home = tempdir().expect("create temp codex home"); + let cwd = tempdir().expect("create temp cwd"); + let mut config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .fallback_cwd(Some(cwd.path().to_path_buf())) + .build() + .await + .expect("build default config"); + config.model_provider_id = "test-provider".to_string(); + + let last_args = crate::cli::ResumeArgs { + session_id: None, + last: true, + all: false, + images: vec![], + prompt: None, + }; + let named_args = crate::cli::ResumeArgs { + session_id: Some("named-session".to_string()), + last: false, + all: false, + images: vec![], + prompt: None, + }; + + assert_eq!( + resume_lookup_model_providers(&config, &last_args), + Some(vec!["test-provider".to_string()]) + ); + assert_eq!(resume_lookup_model_providers(&config, &named_args), None); +} + +#[test] +fn turn_items_for_thread_returns_matching_turn_items() { + let thread = AppServerThread { + id: "thread-1".to_string(), + preview: String::new(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 0, + updated_at: 0, + status: codex_app_server_protocol::ThreadStatus::Idle, + path: None, + cwd: PathBuf::from("/tmp/project"), + cli_version: "0.0.0-test".to_string(), + source: codex_app_server_protocol::SessionSource::Exec, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: vec![ + codex_app_server_protocol::Turn { + id: "turn-1".to_string(), + items: vec![AppServerThreadItem::AgentMessage { + id: "msg-1".to_string(), + text: "hello".to_string(), + phase: None, + memory_citation: None, + }], + status: codex_app_server_protocol::TurnStatus::Completed, + error: None, + }, + codex_app_server_protocol::Turn { + id: "turn-2".to_string(), + items: vec![AppServerThreadItem::Plan { + id: "plan-1".to_string(), + text: "ship it".to_string(), + }], + status: codex_app_server_protocol::TurnStatus::Completed, + error: None, + }, + ], + }; + + assert_eq!( + turn_items_for_thread(&thread, "turn-1"), + Some(vec![AppServerThreadItem::AgentMessage { + id: "msg-1".to_string(), + text: "hello".to_string(), + phase: None, + memory_citation: None, + }]) + ); + assert_eq!(turn_items_for_thread(&thread, "missing-turn"), None); +} + +#[test] +fn canceled_mcp_server_elicitation_response_uses_cancel_action() { + let value = canceled_mcp_server_elicitation_response() + .expect("mcp elicitation cancel response should serialize"); + let response: McpServerElicitationRequestResponse = + serde_json::from_value(value).expect("cancel response should deserialize"); + + assert_eq!( + response, + McpServerElicitationRequestResponse { + action: McpServerElicitationAction::Cancel, + content: None, + meta: None, + } + ); +} + +#[tokio::test] +async fn thread_start_params_include_review_policy_when_review_policy_is_manual_only() { + let codex_home = tempdir().expect("create temp codex home"); + let cwd = tempdir().expect("create temp cwd"); + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .harness_overrides(ConfigOverrides { + approvals_reviewer: Some(ApprovalsReviewer::User), + ..Default::default() + }) + .fallback_cwd(Some(cwd.path().to_path_buf())) + .build() + .await + .expect("build config with manual-only review policy"); + + let params = thread_start_params_from_config(&config); + + assert_eq!( + params.approvals_reviewer, + Some(codex_app_server_protocol::ApprovalsReviewer::User) + ); +} + +#[tokio::test] +async fn thread_start_params_include_review_policy_when_auto_review_is_enabled() { + let codex_home = tempdir().expect("create temp codex home"); + let cwd = tempdir().expect("create temp cwd"); + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .harness_overrides(ConfigOverrides { + approvals_reviewer: Some(ApprovalsReviewer::GuardianSubagent), + ..Default::default() + }) + .fallback_cwd(Some(cwd.path().to_path_buf())) + .build() + .await + .expect("build config with guardian review policy"); + + let params = thread_start_params_from_config(&config); + + assert_eq!( + params.approvals_reviewer, + Some(codex_app_server_protocol::ApprovalsReviewer::GuardianSubagent) + ); +} + +#[test] +fn session_configured_from_thread_response_uses_review_policy_from_response() { + let response = ThreadStartResponse { + thread: codex_app_server_protocol::Thread { + id: "67e55044-10b1-426f-9247-bb680e5fe0c8".to_string(), + preview: String::new(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 0, + updated_at: 0, + status: codex_app_server_protocol::ThreadStatus::Idle, + path: Some(PathBuf::from("/tmp/rollout.jsonl")), + cwd: PathBuf::from("/tmp"), + cli_version: "0.0.0".to_string(), + source: codex_app_server_protocol::SessionSource::Cli, + agent_nickname: None, + agent_role: None, + git_info: None, + name: Some("thread".to_string()), + turns: vec![], + }, + model: "gpt-5.4".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: PathBuf::from("/tmp"), + approval_policy: codex_app_server_protocol::AskForApproval::OnRequest, + approvals_reviewer: codex_app_server_protocol::ApprovalsReviewer::GuardianSubagent, + sandbox: codex_app_server_protocol::SandboxPolicy::WorkspaceWrite { + writable_roots: vec![], + read_only_access: codex_app_server_protocol::ReadOnlyAccess::FullAccess, + network_access: false, + exclude_tmpdir_env_var: false, + exclude_slash_tmp: false, + }, + reasoning_effort: None, + }; + + let event = session_configured_from_thread_start_response(&response) + .expect("build bootstrap session configured event"); + + assert_eq!( + event.approvals_reviewer, + ApprovalsReviewer::GuardianSubagent + ); +}