mirror of
https://github.com/openai/codex.git
synced 2026-11-01 19:33:35 +00:00
Remove unrelated multi-agent stack changes
This commit is contained in:
Eric Traut
@@ -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
|
||||
|
||||
@@ -112,37 +112,30 @@ impl ToolHandler for BatchJobHandler {
|
||||
}
|
||||
|
||||
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
|
||||
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}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -14,84 +14,81 @@ impl ToolHandler for Handler {
|
||||
}
|
||||
|
||||
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -16,111 +16,108 @@ impl ToolHandler for Handler {
|
||||
}
|
||||
|
||||
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
|
||||
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 })
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -22,165 +22,162 @@ impl ToolHandler for Handler {
|
||||
}
|
||||
|
||||
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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::<ThreadId>::new());
|
||||
assert_eq!(shutdown_report.timed_out, Vec::<ThreadId>::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::<ThreadId>::new());
|
||||
assert_eq!(shutdown_report.timed_out, Vec::<ThreadId>::new());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user