From 4d0a1431f8445981f0ea74b8921cba075d6448ef Mon Sep 17 00:00:00 2001 From: Friel Date: Sun, 22 Mar 2026 21:51:52 -0700 Subject: [PATCH] fix(core): adapt inbox delivery rebase to current main Keep the rebased inbox-delivery branch on current core APIs by moving the inbox-specific coverage into current main's control tests and updating the turn-restart helper to use RegularTask. Co-authored-by: Codex --- codex-rs/core/src/agent/control.rs | 8 +- codex-rs/core/src/agent/control_tests.rs | 168 ++++++++++++++++++ codex-rs/core/src/codex.rs | 9 +- codex-rs/core/src/config/config_tests.rs | 2 +- .../tools/handlers/multi_agents/send_input.rs | 3 +- 5 files changed, 182 insertions(+), 8 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 9b17791acd..fca716d119 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -543,7 +543,7 @@ impl AgentControl { vec![UserInput::Text { text: prompt, text_elements: Vec::new(), - }], + }].into(), ) .await } @@ -596,7 +596,7 @@ impl AgentControl { self.send_agent_message(agent_id, sender_thread_id, message) .await } - (None, Some(items)) => self.send_input(agent_id, items).await, + (None, Some(items)) => self.send_input(agent_id, items.into()).await, _ => Err(CodexErr::UnsupportedOperation( "invalid agent input".to_string(), )), @@ -1209,7 +1209,9 @@ fn build_agent_inbox_items( #[cfg(test)] #[path = "control_tests.rs"] mod tests; -#[cfg(test)] +// Keep inbox coverage in `control_tests.rs`. The large inline test module below is a stale +// replay artifact from older pre-refactor rebases and no longer matches current core test APIs. +#[cfg(any())] mod inbox_tests { use super::*; use crate::CodexAuth; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 2def29af9a..952d0f1120 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -14,7 +14,10 @@ use codex_features::Feature; use codex_protocol::AgentPath; use codex_protocol::config_types::ModeKind; use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::AGENT_INBOX_KIND; +use codex_protocol::protocol::AgentInboxPayload; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InterAgentCommunication; @@ -414,6 +417,171 @@ async fn send_input_submits_user_message() { assert_eq!(captured, Some(expected)); } +#[test] +fn build_agent_inbox_items_emits_function_call_and_output() { + let sender_thread_id = ThreadId::new(); + let items = build_agent_inbox_items(sender_thread_id, "watchdog update".to_string(), false) + .expect("tool role should build inbox items"); + + assert_eq!(items.len(), 2); + + let call_id = match &items[0] { + ResponseInputItem::FunctionCall { + name, + arguments, + call_id, + } => { + assert_eq!(name, AGENT_INBOX_KIND); + assert_eq!(arguments, "{}"); + call_id.clone() + } + other => panic!("expected function call item, got {other:?}"), + }; + + match &items[1] { + ResponseInputItem::FunctionCallOutput { + call_id: output_call_id, + output, + } => { + assert_eq!(output_call_id, &call_id); + let output_text = output + .body + .to_text() + .expect("payload should convert to text"); + let payload: AgentInboxPayload = + serde_json::from_str(&output_text).expect("payload should be valid json"); + assert!(payload.injected); + assert_eq!(payload.kind, AGENT_INBOX_KIND); + assert_eq!(payload.sender_thread_id, sender_thread_id); + assert_eq!(payload.message, "watchdog update"); + } + other => panic!("expected function call output item, got {other:?}"), + } +} + +#[test] +fn build_agent_inbox_items_prepends_empty_user_message_when_requested() { + let sender_thread_id = ThreadId::new(); + let items = build_agent_inbox_items(sender_thread_id, "watchdog update".to_string(), true) + .expect("tool role should build inbox items"); + + assert_eq!(items.len(), 3); + assert_eq!( + items[0], + ResponseInputItem::Message { + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: String::new(), + }], + } + ); + assert_matches!(&items[1], ResponseInputItem::FunctionCall { .. }); + assert_matches!(&items[2], ResponseInputItem::FunctionCallOutput { .. }); +} + +#[tokio::test] +async fn send_agent_message_to_root_thread_defaults_to_user_input() { + let harness = AgentControlHarness::new().await; + let (receiver_thread_id, _thread) = harness.start_thread().await; + let sender_thread_id = ThreadId::new(); + + let submission_id = harness + .control + .send_agent_message( + receiver_thread_id, + sender_thread_id, + "watchdog update".to_string(), + ) + .await + .expect("send_agent_message should succeed"); + assert!(!submission_id.is_empty()); + + let expected = ( + receiver_thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "watchdog update".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + + assert_eq!(captured, Some(expected)); +} + +#[tokio::test] +async fn send_agent_message_to_root_thread_injects_response_items_when_enabled() { + let mut harness = AgentControlHarness::new().await; + harness.config.agent_use_function_call_inbox = true; + let (receiver_thread_id, _thread) = harness.start_thread().await; + let sender_thread_id = ThreadId::new(); + + let submission_id = harness + .control + .send_agent_message( + receiver_thread_id, + sender_thread_id, + "watchdog update".to_string(), + ) + .await + .expect("send_agent_message should succeed"); + assert!(!submission_id.is_empty()); + + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|(thread_id, op)| { + *thread_id == receiver_thread_id && matches!(op, Op::InjectResponseItems { .. }) + }) + .expect("expected injected agent inbox op"); + + let Op::InjectResponseItems { items } = captured.1 else { + unreachable!("matched above"); + }; + assert_eq!(items.len(), 3); + match &items[0] { + ResponseInputItem::Message { role, content } => { + assert_eq!(role, "user"); + assert_eq!( + content, + &vec![ContentItem::InputText { + text: String::new(), + }] + ); + } + other => panic!("expected prepended user message, got {other:?}"), + } + match &items[1] { + ResponseInputItem::FunctionCall { + name, arguments, .. + } => { + assert_eq!(name, AGENT_INBOX_KIND); + assert_eq!(arguments, "{}"); + } + other => panic!("expected function call item, got {other:?}"), + } + match &items[2] { + ResponseInputItem::FunctionCallOutput { output, .. } => { + let output_text = output + .body + .to_text() + .expect("payload should convert to text"); + let payload: AgentInboxPayload = + serde_json::from_str(&output_text).expect("payload should be valid json"); + assert_eq!(payload.sender_thread_id, sender_thread_id); + assert_eq!(payload.message, "watchdog update"); + } + other => panic!("expected function call output item, got {other:?}"), + } +} + #[tokio::test] async fn send_inter_agent_communication_without_turn_queues_message_without_triggering_turn() { let harness = AgentControlHarness::new().await; diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index e347ef0772..1b328c766f 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -4741,9 +4741,12 @@ mod handlers { sess.refresh_mcp_servers_if_requested(¤t_context) .await; - let regular_task = sess.take_startup_regular_task().await.unwrap_or_default(); - sess.spawn_task(Arc::clone(¤t_context), turn_input, regular_task) - .await; + sess.spawn_task( + Arc::clone(¤t_context), + turn_input, + crate::tasks::RegularTask::new(), + ) + .await; if pending_items.is_empty() { return; diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 6a683e36e8..91fb71dc67 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -1721,7 +1721,7 @@ fn feature_table_enables_agent_function_call_inbox() -> std::io::Result<()> { let mut entries = BTreeMap::new(); entries.insert("agent_function_call_inbox".to_string(), true); let cfg = ConfigToml { - features: Some(crate::features::FeaturesToml { entries }), + features: Some(codex_features::FeaturesToml { entries }), ..Default::default() }; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs index 6d6b7c587f..d16d60ad92 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs @@ -1,4 +1,5 @@ use super::*; +use crate::agent::agent_resolver::resolve_agent_target; use crate::agent::control::render_input_preview; pub(crate) struct Handler; @@ -27,7 +28,7 @@ impl ToolHandler for Handler { let args: SendInputArgs = parse_arguments(&arguments)?; let receiver_thread_id = resolve_agent_target(&session, &turn, &args.target).await?; let input_items = parse_collab_input(args.message.clone(), args.items.clone())?; - let prompt = input_preview(&input_items); + let prompt = render_input_preview(&input_items); let receiver_agent = session .services .agent_control