From 857146b328009c259f65b871c1c3b1f6494c2cb2 Mon Sep 17 00:00:00 2001 From: Andrey Mishchenko Date: Tue, 28 Apr 2026 23:19:48 -0700 Subject: [PATCH] Delete multi_agent_v2 followup_task interrupt parameter (#20139) 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. --- .../src/tools/handlers/multi_agents_tests.rs | 370 ------------------ .../handlers/multi_agents_v2/followup_task.rs | 1 - .../handlers/multi_agents_v2/message_tool.rs | 21 +- .../handlers/multi_agents_v2/send_message.rs | 1 - codex-rs/tools/src/agent_tool.rs | 11 +- codex-rs/tools/src/agent_tool_tests.rs | 21 - 6 files changed, 3 insertions(+), 422 deletions(-) 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 2fecb7780..0b53fb973 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -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::(text) - .ok() - .as_ref() - == Some(expected) - } - ContentItem::InputText { .. } | ContentItem::InputImage { .. } => false, - }) - }) -} - -async fn wait_for_turn_aborted( - thread: &Arc, - 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, - 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, - _session: Arc, - _ctx: Arc, - _input: Vec, - cancellation_token: CancellationToken, - ) -> Option { - cancellation_token.cancelled().await; - None - } -} - fn expect_text_output(output: T) -> (String, Option) 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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs index e618f5da2..bcb3f49de 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs @@ -25,7 +25,6 @@ impl ToolHandler for Handler { MessageDeliveryMode::TriggerTurn, args.target, args.message, - args.interrupt, ) .await } 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 c9c2fda51..a42cde8f6 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 @@ -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 { @@ -62,16 +60,8 @@ pub(crate) async fn handle_message_string_tool( mode: MessageDeliveryMode, target: String, message: String, - interrupt: bool, ) -> Result { - 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 { 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, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs index 0d142623a..b327ccf52 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs @@ -25,7 +25,6 @@ impl ToolHandler for Handler { MessageDeliveryMode::QueueOnly, args.target, args.message, - /*interrupt*/ false, ) .await } diff --git a/codex-rs/tools/src/agent_tool.rs b/codex-rs/tools/src/agent_tool.rs index 42ba9c4d2..7f83e6cad 100644 --- a/codex-rs/tools/src/agent_tool.rs +++ b/codex-rs/tools/src/agent_tool.rs @@ -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, diff --git a/codex-rs/tools/src/agent_tool_tests.rs b/codex-rs/tools/src/agent_tool_tests.rs index eb82636fc..3157cfc54 100644 --- a/codex-rs/tools/src/agent_tool_tests.rs +++ b/codex-rs/tools/src/agent_tool_tests.rs @@ -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()])