agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Delete multi_agent_v2 followup_task interrupt parameter (#20139)

Browse Source 
Messages sent with `followup_task` already arrive at their target
recipient promptly (at message boundaries while sampling, or after the
pending tool call completes) -- having `interrupt` is not worth the
added complexity.
This commit is contained in:
Andrey Mishchenko
2026-04-28 23:19:48 -07:00
committed by GitHub
Unverified
parent 6ed0440611
commit 857146b328
6 changed files with 3 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
-370
View File
@@ -1,15 +1,10 @@
use super::*;
use crate::CodexThread;
use crate::ThreadManager;
use crate::config::AgentRoleConfig;
use crate::config::DEFAULT_AGENT_MAX_DEPTH;
use crate::context::TurnAborted;
use crate::function_tool::FunctionCallError;
use crate::session::tests::make_session_and_context;
use crate::session_prefix::format_subagent_notification_message;
use crate::state::TaskKind;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
use crate::tools::context::ToolOutput;
use crate::tools::handlers::multi_agents_v2::CloseAgentHandler as CloseAgentHandlerV2;
use crate::tools::handlers::multi_agents_v2::FollowupTaskHandler as FollowupTaskHandlerV2;
@@ -134,121 +129,6 @@ model_reasoning_effort = "minimal"
role_name
}
fn history_contains_inter_agent_communication(
history_items: &[ResponseItem],
expected: &InterAgentCommunication,
) -> bool {
history_items.iter().any(|item| {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
if role != "assistant" {
return false;
}
content.iter().any(|content_item| match content_item {
ContentItem::OutputText { text } => {
serde_json::from_str::<InterAgentCommunication>(text)
.ok()
.as_ref()
== Some(expected)
}
ContentItem::InputText { .. } | ContentItem::InputImage { .. } => false,
})
})
}
async fn wait_for_turn_aborted(
thread: &Arc<CodexThread>,
expected_turn_id: &str,
expected_reason: TurnAbortReason,
) {
timeout(Duration::from_secs(5), async {
loop {
let event = thread
.next_event()
.await
.expect("child thread should emit events");
if matches!(
event.msg,
EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some(ref turn_id),
ref reason,
..
}) if turn_id == expected_turn_id && *reason == expected_reason
) {
break;
}
}
})
.await
.expect("expected child turn to be interrupted");
}
async fn wait_for_redirected_envelope_in_history(
thread: &Arc<CodexThread>,
expected: &InterAgentCommunication,
) {
timeout(Duration::from_secs(5), async {
loop {
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
let saw_envelope =
history_contains_inter_agent_communication(&history_items, expected);
let saw_user_message = history_items.iter().any(|item| {
matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text }
if text == &expected.content
))
)
});
if saw_envelope {
assert!(
!saw_user_message,
"redirected followup should be stored as an assistant envelope, not a plain user message"
);
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("redirected followup envelope should appear in history");
}
#[derive(Clone, Copy)]
struct NeverEndingTask;
impl SessionTask for NeverEndingTask {
fn kind(&self) -> TaskKind {
TaskKind::Regular
}
fn span_name(&self) -> &'static str {
"session_task.multi_agent_never_ending"
}
async fn run(
self: Arc<Self>,
_session: Arc<SessionTaskContext>,
_ctx: Arc<TurnContext>,
_input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> Option<String> {
cancellation_token.cancelled().await;
None
}
}
fn expect_text_output<T>(output: T) -> (String, Option<bool>)
where
T: ToolOutput,
@@ -1086,7 +966,6 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
function_payload(json!({
"target": "/root",
"message": "run this",
"interrupt": true
})),
))
.await
@@ -1485,255 +1364,6 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() {
)));
}
#[tokio::test]
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
.start_thread((*turn.config).clone())
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
session.conversation_id = root.thread_id;
let mut config = turn.config.as_ref().clone();
let _ = config.features.enable(Feature::MultiAgentV2);
turn.config = Arc::new(config);
let session = Arc::new(session);
let turn = Arc::new(turn);
let worker_path = AgentPath::try_from("/root/worker").expect("worker path");
let agent_id = session
.services
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
Op::CleanBackgroundTerminals,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
agent_path: Some(worker_path.clone()),
agent_nickname: None,
agent_role: None,
})),
crate::agent::control::SpawnAgentOptions::default(),
)
.await
.expect("worker spawn should succeed")
.thread_id;
let thread = manager
.get_thread(agent_id)
.await
.expect("worker thread should exist");
let active_turn = thread.codex.session.new_default_turn().await;
let interrupted_turn_id = active_turn.sub_id.clone();
thread
.codex
.session
.spawn_task(
Arc::clone(&active_turn),
vec![UserInput::Text {
text: "working".to_string(),
text_elements: Vec::new(),
}],
NeverEndingTask,
)
.await;
FollowupTaskHandlerV2
.handle(invocation(
session,
turn,
"followup_task",
function_payload(json!({
"target": agent_id.to_string(),
"message": "continue",
"interrupt": true
})),
))
.await
.expect("interrupting v2 followup_task should succeed");
let ops = manager.captured_ops();
let ops_for_agent: Vec<&Op> = ops
.iter()
.filter_map(|(id, op)| (*id == agent_id).then_some(op))
.collect();
assert!(ops_for_agent.iter().any(|op| matches!(op, Op::Interrupt)));
assert!(ops_for_agent.iter().any(|op| {
matches!(
op,
Op::InterAgentCommunication { communication }
if communication.author == AgentPath::root()
&& communication.recipient.as_str() == "/root/worker"
&& communication.other_recipients.is_empty()
&& communication.content == "continue"
)
}));
wait_for_turn_aborted(&thread, &interrupted_turn_id, TurnAbortReason::Interrupted).await;
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
assert!(
history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "developer"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text }
if text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE)
))
)),
"v2 interrupted-turn marker should be recorded as a developer input message"
);
assert!(
!history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } | ContentItem::OutputText { text }
if text.contains(TurnAborted::INTERRUPTED_GUIDANCE)
))
)),
"v2 interrupted-turn marker should not be recorded as a user message"
);
assert!(
!history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "assistant"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } | ContentItem::OutputText { text }
if text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE)
))
)),
"v2 interrupted-turn marker should not be recorded as an assistant message"
);
wait_for_redirected_envelope_in_history(
&thread,
&InterAgentCommunication::new(
AgentPath::root(),
worker_path,
Vec::new(),
"continue".to_string(),
/*trigger_turn*/ true,
),
)
.await;
let _ = thread
.submit(Op::Shutdown {})
.await
.expect("shutdown should submit");
}
#[tokio::test]
async fn multi_agent_v2_followup_task_can_disable_interrupted_marker() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
session.conversation_id = root.thread_id;
let mut config = turn.config.as_ref().clone();
let _ = config.features.enable(Feature::MultiAgentV2);
config.agent_interrupt_message_enabled = false;
turn.config = Arc::new(config);
let session = Arc::new(session);
let turn = Arc::new(turn);
let worker_path = AgentPath::try_from("/root/worker").expect("worker path");
let agent_id = session
.services
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
Op::CleanBackgroundTerminals,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
agent_path: Some(worker_path),
agent_nickname: None,
agent_role: None,
})),
crate::agent::control::SpawnAgentOptions::default(),
)
.await
.expect("worker spawn should succeed")
.thread_id;
let thread = manager
.get_thread(agent_id)
.await
.expect("worker thread should exist");
let active_turn = thread.codex.session.new_default_turn().await;
let interrupted_turn_id = active_turn.sub_id.clone();
thread
.codex
.session
.spawn_task(
Arc::clone(&active_turn),
vec![UserInput::Text {
text: "working".to_string(),
text_elements: Vec::new(),
}],
NeverEndingTask,
)
.await;
FollowupTaskHandlerV2
.handle(invocation(
session,
turn,
"followup_task",
function_payload(json!({
"target": agent_id.to_string(),
"message": "continue",
"interrupt": true
})),
))
.await
.expect("interrupting v2 followup_task should succeed");
wait_for_turn_aborted(&thread, &interrupted_turn_id, TurnAbortReason::Interrupted).await;
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
assert!(
!history_items.iter().any(|item| matches!(
item,
ResponseItem::Message { content, .. }
if content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } | ContentItem::OutputText { text }
if text.contains(TurnAborted::INTERRUPTED_GUIDANCE)
|| text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE)
))
)),
"disabled interrupted-turn marker should not be recorded in history"
);
let _ = thread
.submit(Op::Shutdown {})
.await
.expect("shutdown should submit");
}
#[tokio::test]
async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn() {
let (mut session, mut turn) = make_session_and_context().await;
codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs
-1
View File
@@ -25,7 +25,6 @@ impl ToolHandler for Handler {
MessageDeliveryMode::TriggerTurn,
args.target,
args.message,
args.interrupt,
)
.await
}
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+1 -20
View File
@@ -43,8 +43,6 @@ pub(crate) struct SendMessageArgs {
pub(crate) struct FollowupTaskArgs {
pub(crate) target: String,
pub(crate) message: String,
#[serde(default)]
pub(crate) interrupt: bool,
}
fn message_content(message: String) -> Result<String, FunctionCallError> {
@@ -62,16 +60,8 @@ pub(crate) async fn handle_message_string_tool(
mode: MessageDeliveryMode,
target: String,
message: String,
interrupt: bool,
) -> Result<FunctionToolOutput, FunctionCallError> {
handle_message_submission(
invocation,
mode,
target,
message_content(message)?,
interrupt,
)
.await
handle_message_submission(invocation, mode, target, message_content(message)?).await
}
async fn handle_message_submission(
@@ -79,7 +69,6 @@ async fn handle_message_submission(
mode: MessageDeliveryMode,
target: String,
prompt: String,
interrupt: bool,
) -> Result<FunctionToolOutput, FunctionCallError> {
let ToolInvocation {
session,
@@ -103,14 +92,6 @@ async fn handle_message_submission(
"Tasks can't be assigned to the root agent".to_string(),
));
}
if interrupt {
session
.services
.agent_control
.interrupt_agent(receiver_thread_id)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err))?;
}
session
.send_event(
&turn,
codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs
-1
View File
@@ -25,7 +25,6 @@ impl ToolHandler for Handler {
MessageDeliveryMode::QueueOnly,
args.target,
args.message,
/*interrupt*/ false,
)
.await
}
codex-rs/tools/src/agent_tool.rs
+2 -9
View File
@@ -139,7 +139,7 @@ pub fn create_send_message_tool() -> ToolSpec {
ToolSpec::Function(ResponsesApiTool {
name: "send_message".to_string(),
description: "Send a string message to an existing agent without triggering a new turn."
description: "Send a message to an existing agent. The message will be delivered promptly. Does not trigger a new turn."
.to_string(),
strict: false,
defer_loading: None,
@@ -166,18 +166,11 @@ pub fn create_followup_task_tool() -> ToolSpec {
"Message text to send to the target agent.".to_string(),
)),
),
(
"interrupt".to_string(),
JsonSchema::boolean(Some(
"When true, stop the agent's current task and handle this immediately. When false (default), queue this message; if the target is already running, it starts the target's next turn after the current turn completes."
.to_string(),
)),
),
]);
ToolSpec::Function(ResponsesApiTool {
name: "followup_task".to_string(),
description: "Send a string message to an existing non-root agent and trigger a turn in the target. Use interrupt=true to redirect work immediately. If interrupt=false and the target's turn has not completed, the message is queued and starts the target's next turn after the current turn completes."
description: "Send a message to an existing non-root target agent and trigger a turn in that target. If the target is currently mid-turn, the message is queued and will be used to start the target's next turn, after the current turn completes."
.to_string(),
strict: false,
defer_loading: None,
codex-rs/tools/src/agent_tool_tests.rs
-21
View File
@@ -131,7 +131,6 @@ fn spawn_agent_tool_v1_keeps_legacy_fork_context_field() {
#[test]
fn send_message_tool_requires_message_and_has_no_output_schema() {
let ToolSpec::Function(ResponsesApiTool {
description,
parameters,
output_schema,
..
@@ -151,10 +150,6 @@ fn send_message_tool_requires_message_and_has_no_output_schema() {
assert!(properties.contains_key("message"));
assert!(!properties.contains_key("interrupt"));
assert!(!properties.contains_key("items"));
assert_eq!(
description,
"Send a string message to an existing agent without triggering a new turn."
);
assert_eq!(
properties
.get("target")
@@ -171,7 +166,6 @@ fn send_message_tool_requires_message_and_has_no_output_schema() {
#[test]
fn followup_task_tool_requires_message_and_has_no_output_schema() {
let ToolSpec::Function(ResponsesApiTool {
description,
parameters,
output_schema,
..
@@ -189,22 +183,7 @@ fn followup_task_tool_requires_message_and_has_no_output_schema() {
.expect("followup_task should use object params");
assert!(properties.contains_key("target"));
assert!(properties.contains_key("message"));
assert!(properties.contains_key("interrupt"));
assert!(!properties.contains_key("items"));
assert!(description.contains(
"Send a string message to an existing non-root agent and trigger a turn in the target."
));
assert!(description.contains(
"If interrupt=false and the target's turn has not completed, the message is queued"
));
assert_eq!(
properties
.get("interrupt")
.and_then(|schema| schema.description.as_deref()),
Some(
"When true, stop the agent's current task and handle this immediately. When false (default), queue this message; if the target is already running, it starts the target's next turn after the current turn completes."
)
);
assert_eq!(
parameters.required.as_ref(),
Some(&vec!["target".to_string(), "message".to_string()])