openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge ce349df98a into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-04-02 09:08:19 -07:00
committed by GitHub
parent 6577870a3b ce349df98a
commit 300fec2bfb
20 changed files with 830 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+64 -11
View File
@@ -294,6 +294,7 @@ use crate::shell_snapshot::spawn_stale_snapshot_cleanup;
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;
@@ -4025,6 +4026,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())
}
@@ -4046,6 +4048,42 @@ impl Session {
}
}
pub(crate) async fn defer_mailbox_delivery_to_next_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;
};
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<Arc<tokio::sync::Mutex<crate::state::TurnState>>> {
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<u64> {
self.mailbox.subscribe()
}
@@ -4071,16 +4109,22 @@ impl Session {
}
pub async fn get_pending_input(&self) -> Vec<ResponseInputItem> {
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
@@ -4120,17 +4164,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(
codex-rs/core/src/codex_tests.rs
+210
View File
@@ -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;
@@ -4785,6 +4787,214 @@ 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]
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;
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;
codex-rs/core/src/memories/mod.rs
+1 -1
View File
@@ -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.
codex-rs/core/src/state/mod.rs
+2
View File
@@ -5,5 +5,7 @@ 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;
pub(crate) use turn::TurnState;
codex-rs/core/src/state/turn.rs
+34
View File
@@ -29,6 +29,27 @@ pub(crate) struct ActiveTurn {
pub(crate) turn_state: Arc<Mutex<TurnState>>,
}
/// 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.
#[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 +102,7 @@ pub(crate) struct TurnState {
pending_elicitations: HashMap<(String, RequestId), oneshot::Sender<ElicitationResponse>>,
pending_dynamic_tools: HashMap<String, oneshot::Sender<DynamicToolResponse>>,
pending_input: Vec<ResponseInputItem>,
mailbox_delivery_phase: MailboxDeliveryPhase,
granted_permissions: Option<PermissionProfile>,
pub(crate) tool_calls: u64,
pub(crate) token_usage_at_turn_start: TokenUsage,
@@ -202,6 +224,18 @@ impl TurnState {
!self.pending_input.is_empty()
}
pub(crate) fn accept_mailbox_delivery_for_current_turn(&mut self) {
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));
codex-rs/core/src/stream_events_utils.rs
+30
View File
@@ -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;
}
@@ -205,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,
@@ -426,6 +438,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<ResponseItem> {
match input {
ResponseInputItem::FunctionCallOutput { call_id, output } => {
codex-rs/core/src/stream_events_utils_tests.rs
+39 -1
View File
@@ -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, /*phase*/ None)
}
fn assistant_output_text_with_phase(text: &str, phase: Option<MessagePhase>) -> 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");
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+17 -17
View File
@@ -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 {
codex-rs/core/src/tools/handlers/multi_agents_v2.rs
+2 -2
View File
@@ -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;
codex-rs/core/src/tools/handlers/multi_agents_v2/assign_task.rs → codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs
+2 -2
View File
@@ -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<Self::Output, FunctionCallError> {
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,
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+4 -4
View File
@@ -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<String, FunctionCallError> {
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,
codex-rs/core/src/tools/spec.rs
+4 -4
View File
@@ -47,8 +47,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;
@@ -123,9 +123,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));
}
@@ -141,6 +138,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());
}
codex-rs/exec/src/lib.rs
+2 -404
View File
@@ -1657,407 +1657,5 @@ fn build_review_request(args: &ReviewArgs) -> anyhow::Result<ReviewRequest> {
}
#[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<stdin>\nmy output\n</stdin>"
);
}
#[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<stdin>\nmy output\n</stdin>"
);
}
#[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;
codex-rs/exec/src/lib_tests.rs
+402
View File
@@ -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<stdin>\nmy output\n</stdin>"
);
}
#[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<stdin>\nmy output\n</stdin>"
);
}
#[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
);
}
codex-rs/tools/src/agent_tool.rs
+2 -2
View File
@@ -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,
codex-rs/tools/src/agent_tool_tests.rs
+5 -5
View File
@@ -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"])
);
}
codex-rs/tools/src/lib.rs
+1 -1
View File
@@ -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;
codex-rs/tools/src/tool_registry_plan.rs
+3 -3
View File
@@ -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);
codex-rs/tools/src/tool_registry_plan_tests.rs
+5 -5
View File
@@ -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"));
codex-rs/tools/src/tool_registry_plan_types.rs
+1 -1
View File
@@ -12,12 +12,12 @@ use std::collections::HashMap;
pub enum ToolHandlerKind {
AgentJobs,
ApplyPatch,
AssignTaskV2,
CloseAgentV1,
CloseAgentV2,
CodeModeExecute,
CodeModeWait,
DynamicTool,
FollowupTaskV2,
JsRepl,
JsReplReset,
ListAgentsV2,