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 <noreply@openai.com>
This commit is contained in:
Friel
2026-03-28 11:35:15 -07:00
co-authored by Codex
parent 0e91619094
commit 4d0a1431f8
5 changed files with 182 additions and 8 deletions
+5 -3
View File
@@ -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;
+168
View File
@@ -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;
+6 -3
View File
@@ -4741,9 +4741,12 @@ mod handlers {
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
let regular_task = sess.take_startup_regular_task().await.unwrap_or_default();
sess.spawn_task(Arc::clone(&current_context), turn_input, regular_task)
.await;
sess.spawn_task(
Arc::clone(&current_context),
turn_input,
crate::tasks::RegularTask::new(),
)
.await;
if pending_items.is_empty() {
return;
+1 -1
View File
@@ -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()
};
@@ -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