From 483f3787c251e288a0704cfcb0b7f4ba20ab839a Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Sat, 2 May 2026 12:09:36 -0700 Subject: [PATCH] Remove unrelated multi-agent stack changes --- codex-rs/core/src/agent/control.rs | 20 +- .../core/src/tools/handlers/agent_jobs.rs | 51 +-- .../handlers/multi_agents/close_agent.rs | 147 ++++--- .../handlers/multi_agents/resume_agent.rs | 195 +++++---- .../src/tools/handlers/multi_agents/spawn.rs | 309 +++++++------- .../src/tools/handlers/multi_agents_tests.rs | 388 +++++++++--------- 6 files changed, 534 insertions(+), 576 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index a7bd55410d..705d2d168f 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -487,12 +487,13 @@ impl AgentControl { agent_nickname: None, agent_role: None, }); - match Box::pin(self.resume_single_agent_from_rollout( - config.clone(), - child_thread_id, - child_session_source, - )) - .await + match self + .resume_single_agent_from_rollout( + config.clone(), + child_thread_id, + child_session_source, + ) + .await { Ok(_) => true, Err(err) => { @@ -704,14 +705,9 @@ impl AgentControl { thread.codex.session.ensure_rollout_materialized().await; thread.codex.session.flush_rollout().await?; if matches!(thread.agent_status().await, AgentStatus::Shutdown) { - thread.wait_until_terminated().await; Ok(String::new()) } else { - let result = state.send_op(agent_id, Op::Shutdown {}).await; - if result.is_ok() { - thread.wait_until_terminated().await; - } - result + state.send_op(agent_id, Op::Shutdown {}).await } } else { state.send_op(agent_id, Op::Shutdown {}).await diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs index b3b785cdf3..d5f719febf 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs.rs @@ -112,37 +112,30 @@ impl ToolHandler for BatchJobHandler { } async fn handle(&self, invocation: ToolInvocation) -> Result { - Box::pin(async move { - let ToolInvocation { - session, - turn, - tool_name, - payload, - .. - } = invocation; + let ToolInvocation { + session, + turn, + tool_name, + payload, + .. + } = invocation; - let arguments = match payload { - ToolPayload::Function { arguments } => arguments, - _ => { - return Err(FunctionCallError::RespondToModel( - "agent jobs handler received unsupported payload".to_string(), - )); - } - }; - - match tool_name.name.as_str() { - "spawn_agents_on_csv" => { - Box::pin(spawn_agents_on_csv::handle(session, turn, arguments)).await - } - "report_agent_job_result" => { - Box::pin(report_agent_job_result::handle(session, arguments)).await - } - other => Err(FunctionCallError::RespondToModel(format!( - "unsupported agent job tool {other}" - ))), + let arguments = match payload { + ToolPayload::Function { arguments } => arguments, + _ => { + return Err(FunctionCallError::RespondToModel( + "agent jobs handler received unsupported payload".to_string(), + )); } - }) - .await + }; + + match tool_name.name.as_str() { + "spawn_agents_on_csv" => spawn_agents_on_csv::handle(session, turn, arguments).await, + "report_agent_job_result" => report_agent_job_result::handle(session, arguments).await, + other => Err(FunctionCallError::RespondToModel(format!( + "unsupported agent job tool {other}" + ))), + } } } diff --git a/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs index eb211f0175..8c00b0a13c 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs @@ -14,84 +14,81 @@ impl ToolHandler for Handler { } async fn handle(&self, invocation: ToolInvocation) -> Result { - Box::pin(async move { - let ToolInvocation { - session, - turn, - payload, - call_id, - .. - } = invocation; - let arguments = function_arguments(payload)?; - let args: CloseAgentArgs = parse_arguments(&arguments)?; - let agent_id = parse_agent_id_target(&args.target)?; - let receiver_agent = session - .services - .agent_control - .get_agent_metadata(agent_id) - .unwrap_or_default(); - session - .send_event( - &turn, - CollabCloseBeginEvent { - call_id: call_id.clone(), - sender_thread_id: session.conversation_id, - receiver_thread_id: agent_id, - } - .into(), - ) - .await; - let status = match session - .services - .agent_control - .subscribe_status(agent_id) - .await - { - Ok(mut status_rx) => status_rx.borrow_and_update().clone(), - Err(err) => { - let status = session.services.agent_control.get_status(agent_id).await; - session - .send_event( - &turn, - CollabCloseEndEvent { - call_id: call_id.clone(), - sender_thread_id: session.conversation_id, - receiver_thread_id: agent_id, - receiver_agent_nickname: receiver_agent.agent_nickname.clone(), - receiver_agent_role: receiver_agent.agent_role.clone(), - status, - } - .into(), - ) - .await; - return Err(collab_agent_error(agent_id, err)); + let ToolInvocation { + session, + turn, + payload, + call_id, + .. + } = invocation; + let arguments = function_arguments(payload)?; + let args: CloseAgentArgs = parse_arguments(&arguments)?; + let agent_id = parse_agent_id_target(&args.target)?; + let receiver_agent = session + .services + .agent_control + .get_agent_metadata(agent_id) + .unwrap_or_default(); + session + .send_event( + &turn, + CollabCloseBeginEvent { + call_id: call_id.clone(), + sender_thread_id: session.conversation_id, + receiver_thread_id: agent_id, } - }; - let result = Box::pin(session.services.agent_control.close_agent(agent_id)) - .await - .map_err(|err| collab_agent_error(agent_id, err)) - .map(|_| ()); - session - .send_event( - &turn, - CollabCloseEndEvent { - call_id, - sender_thread_id: session.conversation_id, - receiver_thread_id: agent_id, - receiver_agent_nickname: receiver_agent.agent_nickname, - receiver_agent_role: receiver_agent.agent_role, - status: status.clone(), - } - .into(), - ) - .await; - result?; + .into(), + ) + .await; + let status = match session + .services + .agent_control + .subscribe_status(agent_id) + .await + { + Ok(mut status_rx) => status_rx.borrow_and_update().clone(), + Err(err) => { + let status = session.services.agent_control.get_status(agent_id).await; + session + .send_event( + &turn, + CollabCloseEndEvent { + call_id: call_id.clone(), + sender_thread_id: session.conversation_id, + receiver_thread_id: agent_id, + receiver_agent_nickname: receiver_agent.agent_nickname.clone(), + receiver_agent_role: receiver_agent.agent_role.clone(), + status, + } + .into(), + ) + .await; + return Err(collab_agent_error(agent_id, err)); + } + }; + let result = Box::pin(session.services.agent_control.close_agent(agent_id)) + .await + .map_err(|err| collab_agent_error(agent_id, err)) + .map(|_| ()); + session + .send_event( + &turn, + CollabCloseEndEvent { + call_id, + sender_thread_id: session.conversation_id, + receiver_thread_id: agent_id, + receiver_agent_nickname: receiver_agent.agent_nickname, + receiver_agent_role: receiver_agent.agent_role, + status: status.clone(), + } + .into(), + ) + .await; + result?; - Ok(CloseAgentResult { - previous_status: status, - }) + Ok(CloseAgentResult { + previous_status: status, }) - .await } } diff --git a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs index ca5b5b26b9..2d4f2c3f47 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs @@ -16,111 +16,108 @@ impl ToolHandler for Handler { } async fn handle(&self, invocation: ToolInvocation) -> Result { - Box::pin(async move { - let ToolInvocation { - session, - turn, - payload, - call_id, - .. - } = invocation; - let arguments = function_arguments(payload)?; - let args: ResumeAgentArgs = parse_arguments(&arguments)?; - let receiver_thread_id = ThreadId::from_string(&args.id).map_err(|err| { - FunctionCallError::RespondToModel(format!("invalid agent id {}: {err:?}", args.id)) - })?; - let receiver_agent = session - .services - .agent_control - .get_agent_metadata(receiver_thread_id) - .unwrap_or_default(); - let child_depth = next_thread_spawn_depth(&turn.session_source); - let max_depth = turn.config.agent_max_depth; - if exceeds_thread_spawn_depth_limit(child_depth, max_depth) { - return Err(FunctionCallError::RespondToModel( - "Agent depth limit reached. Solve the task yourself.".to_string(), - )); - } + let ToolInvocation { + session, + turn, + payload, + call_id, + .. + } = invocation; + let arguments = function_arguments(payload)?; + let args: ResumeAgentArgs = parse_arguments(&arguments)?; + let receiver_thread_id = ThreadId::from_string(&args.id).map_err(|err| { + FunctionCallError::RespondToModel(format!("invalid agent id {}: {err:?}", args.id)) + })?; + let receiver_agent = session + .services + .agent_control + .get_agent_metadata(receiver_thread_id) + .unwrap_or_default(); + let child_depth = next_thread_spawn_depth(&turn.session_source); + let max_depth = turn.config.agent_max_depth; + if exceeds_thread_spawn_depth_limit(child_depth, max_depth) { + return Err(FunctionCallError::RespondToModel( + "Agent depth limit reached. Solve the task yourself.".to_string(), + )); + } - session - .send_event( - &turn, - CollabResumeBeginEvent { - call_id: call_id.clone(), - sender_thread_id: session.conversation_id, - receiver_thread_id, - receiver_agent_nickname: receiver_agent.agent_nickname.clone(), - receiver_agent_role: receiver_agent.agent_role.clone(), - } - .into(), - ) - .await; - - let mut status = session - .services - .agent_control - .get_status(receiver_thread_id) - .await; - let (receiver_agent, error) = if matches!(status, AgentStatus::NotFound) { - match Box::pin(try_resume_closed_agent( - &session, - &turn, + session + .send_event( + &turn, + CollabResumeBeginEvent { + call_id: call_id.clone(), + sender_thread_id: session.conversation_id, receiver_thread_id, - child_depth, - )) - .await - { - Ok(()) => { - status = session - .services - .agent_control - .get_status(receiver_thread_id) - .await; - ( - session - .services - .agent_control - .get_agent_metadata(receiver_thread_id) - .unwrap_or(receiver_agent), - None, - ) - } - Err(err) => { - status = session - .services - .agent_control - .get_status(receiver_thread_id) - .await; - (receiver_agent, Some(err)) - } + receiver_agent_nickname: receiver_agent.agent_nickname.clone(), + receiver_agent_role: receiver_agent.agent_role.clone(), } - } else { - (receiver_agent, None) - }; - session - .send_event( - &turn, - CollabResumeEndEvent { - call_id, - sender_thread_id: session.conversation_id, - receiver_thread_id, - receiver_agent_nickname: receiver_agent.agent_nickname, - receiver_agent_role: receiver_agent.agent_role, - status: status.clone(), - } - .into(), - ) - .await; + .into(), + ) + .await; - if let Some(err) = error { - return Err(err); + let mut status = session + .services + .agent_control + .get_status(receiver_thread_id) + .await; + let (receiver_agent, error) = if matches!(status, AgentStatus::NotFound) { + match Box::pin(try_resume_closed_agent( + &session, + &turn, + receiver_thread_id, + child_depth, + )) + .await + { + Ok(()) => { + status = session + .services + .agent_control + .get_status(receiver_thread_id) + .await; + ( + session + .services + .agent_control + .get_agent_metadata(receiver_thread_id) + .unwrap_or(receiver_agent), + None, + ) + } + Err(err) => { + status = session + .services + .agent_control + .get_status(receiver_thread_id) + .await; + (receiver_agent, Some(err)) + } } - turn.session_telemetry - .counter("codex.multi_agent.resume", /*inc*/ 1, &[]); + } else { + (receiver_agent, None) + }; + session + .send_event( + &turn, + CollabResumeEndEvent { + call_id, + sender_thread_id: session.conversation_id, + receiver_thread_id, + receiver_agent_nickname: receiver_agent.agent_nickname, + receiver_agent_role: receiver_agent.agent_role, + status: status.clone(), + } + .into(), + ) + .await; - Ok(ResumeAgentResult { status }) - }) - .await + if let Some(err) = error { + return Err(err); + } + turn.session_telemetry + .counter("codex.multi_agent.resume", /*inc*/ 1, &[]); + + Ok(ResumeAgentResult { status }) } } diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 7394998c3f..777cb9be1c 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -22,165 +22,162 @@ impl ToolHandler for Handler { } async fn handle(&self, invocation: ToolInvocation) -> Result { - Box::pin(async move { - let ToolInvocation { - session, - turn, - payload, - call_id, - .. - } = invocation; - let arguments = function_arguments(payload)?; - let args: SpawnAgentArgs = parse_arguments(&arguments)?; - let role_name = args - .agent_type - .as_deref() - .map(str::trim) - .filter(|role| !role.is_empty()); - let input_items = parse_collab_input(args.message, args.items)?; - let prompt = render_input_preview(&input_items); - let session_source = turn.session_source.clone(); - let child_depth = next_thread_spawn_depth(&session_source); - let max_depth = turn.config.agent_max_depth; - if exceeds_thread_spawn_depth_limit(child_depth, max_depth) { - return Err(FunctionCallError::RespondToModel( - "Agent depth limit reached. Solve the task yourself.".to_string(), - )); - } - session - .send_event( - &turn, - CollabAgentSpawnBeginEvent { - call_id: call_id.clone(), - sender_thread_id: session.conversation_id, - prompt: prompt.clone(), - model: args.model.clone().unwrap_or_default(), - reasoning_effort: args.reasoning_effort.unwrap_or_default(), - } - .into(), - ) - .await; - let mut config = - build_agent_spawn_config(&session.get_base_instructions().await, turn.as_ref())?; - if args.fork_context { - reject_full_fork_spawn_overrides( - role_name, - args.model.as_deref(), - args.reasoning_effort, - )?; - } else { - apply_requested_spawn_agent_model_overrides( - &session, - turn.as_ref(), - &mut config, - args.model.as_deref(), - args.reasoning_effort, - ) - .await?; - apply_role_to_config(&mut config, role_name) - .await - .map_err(FunctionCallError::RespondToModel)?; - } - apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?; - apply_spawn_agent_overrides(&mut config, child_depth); - - let result = Box::pin( - session.services.agent_control.spawn_agent_with_metadata( - config, - input_items, - Some(thread_spawn_source( - session.conversation_id, - &turn.session_source, - child_depth, - role_name, - /*task_name*/ None, - )?), - SpawnAgentOptions { - fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), - fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), - environments: Some( - turn.environments - .iter() - .map(TurnEnvironment::selection) - .collect(), - ), - }, - ), - ) - .await - .map_err(collab_spawn_error); - let (new_thread_id, new_agent_metadata, status) = match &result { - Ok(spawned_agent) => ( - Some(spawned_agent.thread_id), - Some(spawned_agent.metadata.clone()), - spawned_agent.status.clone(), - ), - Err(_) => (None, None, AgentStatus::NotFound), - }; - let agent_snapshot = match new_thread_id { - Some(thread_id) => { - session - .services - .agent_control - .get_agent_config_snapshot(thread_id) - .await + let ToolInvocation { + session, + turn, + payload, + call_id, + .. + } = invocation; + let arguments = function_arguments(payload)?; + let args: SpawnAgentArgs = parse_arguments(&arguments)?; + let role_name = args + .agent_type + .as_deref() + .map(str::trim) + .filter(|role| !role.is_empty()); + let input_items = parse_collab_input(args.message, args.items)?; + let prompt = render_input_preview(&input_items); + let session_source = turn.session_source.clone(); + let child_depth = next_thread_spawn_depth(&session_source); + let max_depth = turn.config.agent_max_depth; + if exceeds_thread_spawn_depth_limit(child_depth, max_depth) { + return Err(FunctionCallError::RespondToModel( + "Agent depth limit reached. Solve the task yourself.".to_string(), + )); + } + session + .send_event( + &turn, + CollabAgentSpawnBeginEvent { + call_id: call_id.clone(), + sender_thread_id: session.conversation_id, + prompt: prompt.clone(), + model: args.model.clone().unwrap_or_default(), + reasoning_effort: args.reasoning_effort.unwrap_or_default(), } - None => None, - }; - let (_new_agent_path, new_agent_nickname, new_agent_role) = - match (&agent_snapshot, new_agent_metadata) { - (Some(snapshot), _) => ( - snapshot.session_source.get_agent_path().map(String::from), - snapshot.session_source.get_nickname(), - snapshot.session_source.get_agent_role(), - ), - (None, Some(metadata)) => ( - metadata.agent_path.map(String::from), - metadata.agent_nickname, - metadata.agent_role, - ), - (None, None) => (None, None, None), - }; - let effective_model = agent_snapshot - .as_ref() - .map(|snapshot| snapshot.model.clone()) - .unwrap_or_else(|| args.model.clone().unwrap_or_default()); - let effective_reasoning_effort = agent_snapshot - .as_ref() - .and_then(|snapshot| snapshot.reasoning_effort) - .unwrap_or(args.reasoning_effort.unwrap_or_default()); - let nickname = new_agent_nickname.clone(); - session - .send_event( - &turn, - CollabAgentSpawnEndEvent { - call_id, - sender_thread_id: session.conversation_id, - new_thread_id, - new_agent_nickname, - new_agent_role, - prompt, - model: effective_model, - reasoning_effort: effective_reasoning_effort, - status, - } - .into(), - ) - .await; - let new_thread_id = result?.thread_id; - let role_tag = role_name.unwrap_or(DEFAULT_ROLE_NAME); - turn.session_telemetry.counter( - "codex.multi_agent.spawn", - /*inc*/ 1, - &[("role", role_tag)], - ); + .into(), + ) + .await; + let mut config = + build_agent_spawn_config(&session.get_base_instructions().await, turn.as_ref())?; + if args.fork_context { + reject_full_fork_spawn_overrides( + role_name, + args.model.as_deref(), + args.reasoning_effort, + )?; + } else { + apply_requested_spawn_agent_model_overrides( + &session, + turn.as_ref(), + &mut config, + args.model.as_deref(), + args.reasoning_effort, + ) + .await?; + apply_role_to_config(&mut config, role_name) + .await + .map_err(FunctionCallError::RespondToModel)?; + } + apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?; + apply_spawn_agent_overrides(&mut config, child_depth); - Ok(SpawnAgentResult { - agent_id: new_thread_id.to_string(), - nickname, - }) - }) + let result = Box::pin( + session.services.agent_control.spawn_agent_with_metadata( + config, + input_items, + Some(thread_spawn_source( + session.conversation_id, + &turn.session_source, + child_depth, + role_name, + /*task_name*/ None, + )?), + SpawnAgentOptions { + fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), + fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), + environments: Some( + turn.environments + .iter() + .map(TurnEnvironment::selection) + .collect(), + ), + }, + ), + ) .await + .map_err(collab_spawn_error); + let (new_thread_id, new_agent_metadata, status) = match &result { + Ok(spawned_agent) => ( + Some(spawned_agent.thread_id), + Some(spawned_agent.metadata.clone()), + spawned_agent.status.clone(), + ), + Err(_) => (None, None, AgentStatus::NotFound), + }; + let agent_snapshot = match new_thread_id { + Some(thread_id) => { + session + .services + .agent_control + .get_agent_config_snapshot(thread_id) + .await + } + None => None, + }; + let (_new_agent_path, new_agent_nickname, new_agent_role) = + match (&agent_snapshot, new_agent_metadata) { + (Some(snapshot), _) => ( + snapshot.session_source.get_agent_path().map(String::from), + snapshot.session_source.get_nickname(), + snapshot.session_source.get_agent_role(), + ), + (None, Some(metadata)) => ( + metadata.agent_path.map(String::from), + metadata.agent_nickname, + metadata.agent_role, + ), + (None, None) => (None, None, None), + }; + let effective_model = agent_snapshot + .as_ref() + .map(|snapshot| snapshot.model.clone()) + .unwrap_or_else(|| args.model.clone().unwrap_or_default()); + let effective_reasoning_effort = agent_snapshot + .as_ref() + .and_then(|snapshot| snapshot.reasoning_effort) + .unwrap_or(args.reasoning_effort.unwrap_or_default()); + let nickname = new_agent_nickname.clone(); + session + .send_event( + &turn, + CollabAgentSpawnEndEvent { + call_id, + sender_thread_id: session.conversation_id, + new_thread_id, + new_agent_nickname, + new_agent_role, + prompt, + model: effective_model, + reasoning_effort: effective_reasoning_effort, + status, + } + .into(), + ) + .await; + let new_thread_id = result?.thread_id; + let role_tag = role_name.unwrap_or(DEFAULT_ROLE_NAME); + turn.session_telemetry.counter( + "codex.multi_agent.spawn", + /*inc*/ 1, + &[("role", role_tag)], + ); + + Ok(SpawnAgentResult { + agent_id: new_thread_id.to_string(), + nickname, + }) } } 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 0129838487..61dc77eb36 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -3146,220 +3146,198 @@ async fn close_agent_submits_shutdown_and_returns_previous_status() { assert_eq!(status_after, AgentStatus::NotFound); } -#[test] -fn tool_handlers_cascade_close_and_resume_and_keep_explicitly_closed_subtrees_closed() { - std::thread::Builder::new() - .name("cascade_close_resume_test".to_string()) - .stack_size(16 * 1024 * 1024) - .spawn(|| { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("tokio runtime") - .block_on(async { - let (_session, turn) = make_session_and_context().await; - let manager = thread_manager(); - let mut config = turn.config.as_ref().clone(); - config.agent_max_depth = 3; - config - .features - .enable(Feature::Sqlite) - .expect("test config should allow sqlite"); +#[tokio::test] +async fn tool_handlers_cascade_close_and_resume_and_keep_explicitly_closed_subtrees_closed() { + let (_session, turn) = make_session_and_context().await; + let manager = thread_manager(); + let mut config = turn.config.as_ref().clone(); + config.agent_max_depth = 3; + config + .features + .enable(Feature::Sqlite) + .expect("test config should allow sqlite"); - let parent = manager - .start_thread(config.clone()) - .await - .expect("parent thread should start"); - let parent_thread_id = parent.thread_id; - let parent_session = parent.thread.codex.session.clone(); + let parent = manager + .start_thread(config.clone()) + .await + .expect("parent thread should start"); + let parent_thread_id = parent.thread_id; + let parent_session = parent.thread.codex.session.clone(); - let child_spawn_output = SpawnAgentHandler - .handle(invocation( - parent_session.clone(), - parent_session.new_default_turn().await, - "spawn_agent", - function_payload(json!({"message": "hello child"})), - )) - .await - .expect("child spawn should succeed"); - let (child_content, child_success) = expect_text_output(child_spawn_output); - let child_result: serde_json::Value = serde_json::from_str(&child_content) - .expect("child spawn result should be json"); - let child_thread_id = parse_agent_id( - child_result - .get("agent_id") - .and_then(serde_json::Value::as_str) - .expect("child spawn result should include agent_id"), - ); - assert_eq!(child_success, Some(true)); + let child_spawn_output = SpawnAgentHandler + .handle(invocation( + parent_session.clone(), + parent_session.new_default_turn().await, + "spawn_agent", + function_payload(json!({"message": "hello child"})), + )) + .await + .expect("child spawn should succeed"); + let (child_content, child_success) = expect_text_output(child_spawn_output); + let child_result: serde_json::Value = + serde_json::from_str(&child_content).expect("child spawn result should be json"); + let child_thread_id = parse_agent_id( + child_result + .get("agent_id") + .and_then(serde_json::Value::as_str) + .expect("child spawn result should include agent_id"), + ); + assert_eq!(child_success, Some(true)); - let child_thread = manager - .get_thread(child_thread_id) - .await - .expect("child thread should exist"); - let child_session = child_thread.codex.session.clone(); - let grandchild_spawn_output = SpawnAgentHandler - .handle(invocation( - child_session.clone(), - child_session.new_default_turn().await, - "spawn_agent", - function_payload(json!({"message": "hello grandchild"})), - )) - .await - .expect("grandchild spawn should succeed"); - let (grandchild_content, grandchild_success) = - expect_text_output(grandchild_spawn_output); - let grandchild_result: serde_json::Value = - serde_json::from_str(&grandchild_content) - .expect("grandchild spawn result should be json"); - let grandchild_thread_id = parse_agent_id( - grandchild_result - .get("agent_id") - .and_then(serde_json::Value::as_str) - .expect("grandchild spawn result should include agent_id"), - ); - assert_eq!(grandchild_success, Some(true)); + let child_thread = manager + .get_thread(child_thread_id) + .await + .expect("child thread should exist"); + let child_session = child_thread.codex.session.clone(); + let grandchild_spawn_output = SpawnAgentHandler + .handle(invocation( + child_session.clone(), + child_session.new_default_turn().await, + "spawn_agent", + function_payload(json!({"message": "hello grandchild"})), + )) + .await + .expect("grandchild spawn should succeed"); + let (grandchild_content, grandchild_success) = expect_text_output(grandchild_spawn_output); + let grandchild_result: serde_json::Value = + serde_json::from_str(&grandchild_content).expect("grandchild spawn result should be json"); + let grandchild_thread_id = parse_agent_id( + grandchild_result + .get("agent_id") + .and_then(serde_json::Value::as_str) + .expect("grandchild spawn result should include agent_id"), + ); + assert_eq!(grandchild_success, Some(true)); - let close_output = CloseAgentHandler - .handle(invocation( - parent_session.clone(), - parent_session.new_default_turn().await, - "close_agent", - function_payload(json!({"target": child_thread_id.to_string()})), - )) - .await - .expect("close_agent should close the child subtree"); - let (close_content, close_success) = expect_text_output(close_output); - let close_result: close_agent::CloseAgentResult = - serde_json::from_str(&close_content) - .expect("close_agent result should be json"); - assert_ne!(close_result.previous_status, AgentStatus::NotFound); - assert_eq!(close_success, Some(true)); - assert_eq!( - manager.agent_control().get_status(child_thread_id).await, - AgentStatus::NotFound - ); - assert_eq!( - manager - .agent_control() - .get_status(grandchild_thread_id) - .await, - AgentStatus::NotFound - ); + let close_output = CloseAgentHandler + .handle(invocation( + parent_session.clone(), + parent_session.new_default_turn().await, + "close_agent", + function_payload(json!({"target": child_thread_id.to_string()})), + )) + .await + .expect("close_agent should close the child subtree"); + let (close_content, close_success) = expect_text_output(close_output); + let close_result: close_agent::CloseAgentResult = + serde_json::from_str(&close_content).expect("close_agent result should be json"); + assert_ne!(close_result.previous_status, AgentStatus::NotFound); + assert_eq!(close_success, Some(true)); + assert_eq!( + manager.agent_control().get_status(child_thread_id).await, + AgentStatus::NotFound + ); + assert_eq!( + manager + .agent_control() + .get_status(grandchild_thread_id) + .await, + AgentStatus::NotFound + ); - let child_resume_output = ResumeAgentHandler - .handle(invocation( - parent_session.clone(), - parent_session.new_default_turn().await, - "resume_agent", - function_payload(json!({"id": child_thread_id.to_string()})), - )) - .await - .expect("resume_agent should reopen the child subtree"); - let (child_resume_content, child_resume_success) = - expect_text_output(child_resume_output); - let child_resume_result: resume_agent::ResumeAgentResult = - serde_json::from_str(&child_resume_content) - .expect("resume result should be json"); - assert_ne!(child_resume_result.status, AgentStatus::NotFound); - assert_eq!(child_resume_success, Some(true)); - assert_ne!( - manager.agent_control().get_status(child_thread_id).await, - AgentStatus::NotFound - ); - assert_ne!( - manager - .agent_control() - .get_status(grandchild_thread_id) - .await, - AgentStatus::NotFound - ); + let child_resume_output = ResumeAgentHandler + .handle(invocation( + parent_session.clone(), + parent_session.new_default_turn().await, + "resume_agent", + function_payload(json!({"id": child_thread_id.to_string()})), + )) + .await + .expect("resume_agent should reopen the child subtree"); + let (child_resume_content, child_resume_success) = expect_text_output(child_resume_output); + let child_resume_result: resume_agent::ResumeAgentResult = + serde_json::from_str(&child_resume_content).expect("resume result should be json"); + assert_ne!(child_resume_result.status, AgentStatus::NotFound); + assert_eq!(child_resume_success, Some(true)); + assert_ne!( + manager.agent_control().get_status(child_thread_id).await, + AgentStatus::NotFound + ); + assert_ne!( + manager + .agent_control() + .get_status(grandchild_thread_id) + .await, + AgentStatus::NotFound + ); - let close_again_output = CloseAgentHandler - .handle(invocation( - parent_session.clone(), - parent_session.new_default_turn().await, - "close_agent", - function_payload(json!({"target": child_thread_id.to_string()})), - )) - .await - .expect("close_agent should be repeatable for the child subtree"); - let (close_again_content, close_again_success) = - expect_text_output(close_again_output); - let close_again_result: close_agent::CloseAgentResult = - serde_json::from_str(&close_again_content) - .expect("second close_agent result should be json"); - assert_ne!(close_again_result.previous_status, AgentStatus::NotFound); - assert_eq!(close_again_success, Some(true)); - assert_eq!( - manager.agent_control().get_status(child_thread_id).await, - AgentStatus::NotFound - ); - assert_eq!( - manager - .agent_control() - .get_status(grandchild_thread_id) - .await, - AgentStatus::NotFound - ); + let close_again_output = CloseAgentHandler + .handle(invocation( + parent_session.clone(), + parent_session.new_default_turn().await, + "close_agent", + function_payload(json!({"target": child_thread_id.to_string()})), + )) + .await + .expect("close_agent should be repeatable for the child subtree"); + let (close_again_content, close_again_success) = expect_text_output(close_again_output); + let close_again_result: close_agent::CloseAgentResult = + serde_json::from_str(&close_again_content) + .expect("second close_agent result should be json"); + assert_ne!(close_again_result.previous_status, AgentStatus::NotFound); + assert_eq!(close_again_success, Some(true)); + assert_eq!( + manager.agent_control().get_status(child_thread_id).await, + AgentStatus::NotFound + ); + assert_eq!( + manager + .agent_control() + .get_status(grandchild_thread_id) + .await, + AgentStatus::NotFound + ); - let operator = manager - .start_thread(config.clone()) - .await - .expect("operator thread should start"); - let operator_session = operator.thread.codex.session.clone(); - let _ = manager - .agent_control() - .shutdown_live_agent(parent_thread_id) - .await - .expect("parent shutdown should succeed"); - assert_eq!( - manager.agent_control().get_status(parent_thread_id).await, - AgentStatus::NotFound - ); + let operator = manager + .start_thread(config.clone()) + .await + .expect("operator thread should start"); + let operator_session = operator.thread.codex.session.clone(); + let _ = manager + .agent_control() + .shutdown_live_agent(parent_thread_id) + .await + .expect("parent shutdown should succeed"); + assert_eq!( + manager.agent_control().get_status(parent_thread_id).await, + AgentStatus::NotFound + ); - let parent_resume_output = ResumeAgentHandler - .handle(invocation( - operator_session, - operator.thread.codex.session.new_default_turn().await, - "resume_agent", - function_payload(json!({"id": parent_thread_id.to_string()})), - )) - .await - .expect("resume_agent should reopen the parent thread"); - let (parent_resume_content, parent_resume_success) = - expect_text_output(parent_resume_output); - let parent_resume_result: resume_agent::ResumeAgentResult = - serde_json::from_str(&parent_resume_content) - .expect("parent resume result should be json"); - assert_ne!(parent_resume_result.status, AgentStatus::NotFound); - assert_eq!(parent_resume_success, Some(true)); - assert_ne!( - manager.agent_control().get_status(parent_thread_id).await, - AgentStatus::NotFound - ); - assert_eq!( - manager.agent_control().get_status(child_thread_id).await, - AgentStatus::NotFound - ); - assert_eq!( - manager - .agent_control() - .get_status(grandchild_thread_id) - .await, - AgentStatus::NotFound - ); + let parent_resume_output = ResumeAgentHandler + .handle(invocation( + operator_session, + operator.thread.codex.session.new_default_turn().await, + "resume_agent", + function_payload(json!({"id": parent_thread_id.to_string()})), + )) + .await + .expect("resume_agent should reopen the parent thread"); + let (parent_resume_content, parent_resume_success) = expect_text_output(parent_resume_output); + let parent_resume_result: resume_agent::ResumeAgentResult = + serde_json::from_str(&parent_resume_content).expect("parent resume result should be json"); + assert_ne!(parent_resume_result.status, AgentStatus::NotFound); + assert_eq!(parent_resume_success, Some(true)); + assert_ne!( + manager.agent_control().get_status(parent_thread_id).await, + AgentStatus::NotFound + ); + assert_eq!( + manager.agent_control().get_status(child_thread_id).await, + AgentStatus::NotFound + ); + assert_eq!( + manager + .agent_control() + .get_status(grandchild_thread_id) + .await, + AgentStatus::NotFound + ); - let shutdown_report = manager - .shutdown_all_threads_bounded(Duration::from_secs(5)) - .await; - assert_eq!(shutdown_report.submit_failed, Vec::::new()); - assert_eq!(shutdown_report.timed_out, Vec::::new()); - }); - }) - .expect("spawn cascade close/resume test thread") - .join() - .expect("cascade close/resume test thread should finish"); + let shutdown_report = manager + .shutdown_all_threads_bounded(Duration::from_secs(5)) + .await; + assert_eq!(shutdown_report.submit_failed, Vec::::new()); + assert_eq!(shutdown_report.timed_out, Vec::::new()); } #[tokio::test]