mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Add agent inspection without restoring unloaded runtimes (#47010)
## Why Agent inspection needs to distinguish registered but unloaded agents from missing identities and backend failures without restoring a runtime. ## What changed - Add inspection returning `AgentInfo` with metadata, status, and configuration for loaded agents, or metadata for unloaded agents. - Use inspection for configuration snapshots and to capture an agent's state before interruption. - Add `AgentInfo` accessors and share V2 error mapping between interruption and messaging tools, preserving the distinction between invalid requests and a dropped thread manager. ## Testing Extend the V2 reload test helper to assert that inspecting a registered, unloaded agent returns `AgentInfo::Unloaded` and leaves its runtime unloaded. GitOrigin-RevId: fa897264ff3ff3ec6cc76259d8482be55dec0c0a
This commit is contained in:
@@ -170,6 +170,22 @@ pub enum AgentInfo {
|
||||
Unloaded(AgentMetadata),
|
||||
}
|
||||
|
||||
impl AgentInfo {
|
||||
pub fn metadata(&self) -> &AgentMetadata {
|
||||
match self {
|
||||
Self::Loaded { agent, .. } => &agent.metadata,
|
||||
Self::Unloaded(metadata) => metadata,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn status(&self) -> Option<&AgentStatus> {
|
||||
match self {
|
||||
Self::Loaded { agent, .. } => Some(&agent.status),
|
||||
Self::Unloaded(_) => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// User input starts or steers a turn; agent messages retain their sender and wake mode.
|
||||
pub enum AgentInput {
|
||||
UserInput(Vec<UserInput>),
|
||||
|
||||
@@ -67,14 +67,13 @@ use tracing::warn;
|
||||
use uuid::Uuid;
|
||||
|
||||
use self::execution::AgentExecutionLimiter;
|
||||
pub(crate) use self::interrupt::AgentInterruptError;
|
||||
pub(crate) use self::interrupt::AgentInterruptOutcome;
|
||||
use self::residency::V2Residency;
|
||||
|
||||
mod budget;
|
||||
mod completion;
|
||||
mod delivery;
|
||||
mod execution;
|
||||
mod inspection;
|
||||
mod interrupt;
|
||||
mod legacy;
|
||||
mod residency;
|
||||
@@ -430,13 +429,10 @@ impl LocalAgentControl {
|
||||
&self,
|
||||
agent_id: ThreadId,
|
||||
) -> Option<ThreadConfigSnapshot> {
|
||||
let Ok(state) = self.upgrade() else {
|
||||
return None;
|
||||
};
|
||||
let Ok(thread) = state.get_thread(agent_id).await else {
|
||||
return None;
|
||||
};
|
||||
Some(thread.config_snapshot().await)
|
||||
match self.inspect_agent(agent_id).await.ok()? {
|
||||
crate::agent::api::AgentInfo::Loaded { config, .. } => Some(*config),
|
||||
crate::agent::api::AgentInfo::Unloaded(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribe to status updates for `agent_id`, yielding the latest value and changes.
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
//! Reads registered agent identity, status and settings without restoring a runtime.
|
||||
//! Known unloaded agents stay distinct from missing identities and backend failures.
|
||||
|
||||
use super::LocalAgentControl;
|
||||
use crate::agent::api::AgentInfo;
|
||||
use crate::agent::api::AgentTarget;
|
||||
use crate::agent::types::LiveAgent;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::error::CodexErrorDetails;
|
||||
use codex_protocol::error::Result as CodexResult;
|
||||
|
||||
impl LocalAgentControl {
|
||||
pub(crate) async fn inspect(
|
||||
&self,
|
||||
caller: ThreadId,
|
||||
target: AgentTarget,
|
||||
) -> CodexResult<AgentInfo> {
|
||||
let target = self.resolve_target(caller, &target)?;
|
||||
self.inspect_agent(target).await
|
||||
}
|
||||
|
||||
pub(super) async fn inspect_agent(&self, thread_id: ThreadId) -> CodexResult<AgentInfo> {
|
||||
let manager = self.upgrade()?;
|
||||
let thread = match manager.get_thread(thread_id).await {
|
||||
Ok(thread) => thread,
|
||||
Err(err) if matches!(err.details(), CodexErrorDetails::ThreadNotFound(_)) => {
|
||||
return Ok(AgentInfo::Unloaded(self.ensure_agent_known(thread_id)?));
|
||||
}
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
Ok(AgentInfo::Loaded {
|
||||
agent: LiveAgent {
|
||||
thread_id,
|
||||
metadata: self.get_agent_metadata(thread_id).unwrap_or_default(),
|
||||
status: thread.agent_status().await,
|
||||
},
|
||||
config: Box::new(thread.config_snapshot().await),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -4,23 +4,13 @@
|
||||
//! interruption; this operation never reloads it.
|
||||
|
||||
use super::LocalAgentControl;
|
||||
use crate::agent::AgentStatus;
|
||||
use crate::agent::api::AgentInfo;
|
||||
use crate::agent::api::AgentTarget;
|
||||
use codex_protocol::AgentPath;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::error::CodexErr;
|
||||
use codex_protocol::error::CodexErrorDetails;
|
||||
|
||||
pub(crate) struct AgentInterruptOutcome {
|
||||
pub(crate) agent_path: AgentPath,
|
||||
pub(crate) previous_status: AgentStatus,
|
||||
}
|
||||
|
||||
/// Keeps request validation distinct from runtime failures for the tool adapter's error mapping.
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum AgentInterruptError {
|
||||
InvalidRequest(String),
|
||||
Agent(CodexErr),
|
||||
}
|
||||
use codex_protocol::error::Result as CodexResult;
|
||||
|
||||
impl LocalAgentControl {
|
||||
/// Interrupts a spawned agent's current task, preserving the status observed before dispatch.
|
||||
@@ -28,29 +18,27 @@ impl LocalAgentControl {
|
||||
&self,
|
||||
caller: ThreadId,
|
||||
target: ThreadId,
|
||||
) -> Result<AgentInterruptOutcome, AgentInterruptError> {
|
||||
let receiver_agent = self
|
||||
.ensure_agent_known(target)
|
||||
.map_err(AgentInterruptError::Agent)?;
|
||||
) -> CodexResult<AgentInfo> {
|
||||
let receiver_agent = self.ensure_agent_known(target)?;
|
||||
if receiver_agent
|
||||
.agent_path
|
||||
.as_ref()
|
||||
.is_some_and(AgentPath::is_root)
|
||||
{
|
||||
return Err(AgentInterruptError::InvalidRequest(
|
||||
return Err(CodexErr::UnsupportedOperation(
|
||||
"root is not a spawned agent".to_string(),
|
||||
));
|
||||
}
|
||||
if target == caller {
|
||||
return Err(AgentInterruptError::InvalidRequest(
|
||||
return Err(CodexErr::UnsupportedOperation(
|
||||
"an agent cannot interrupt itself; return your result and let the parent interrupt you if needed"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
let agent_path = receiver_agent.agent_path.clone().ok_or_else(|| {
|
||||
AgentInterruptError::InvalidRequest("target agent is missing an agent_path".to_string())
|
||||
receiver_agent.agent_path.as_ref().ok_or_else(|| {
|
||||
CodexErr::UnsupportedOperation("target agent is missing an agent_path".to_string())
|
||||
})?;
|
||||
let previous_status = self.get_status(target).await;
|
||||
let snapshot = self.inspect(caller, AgentTarget::Id(target)).await?;
|
||||
match self.interrupt_agent(target).await {
|
||||
Ok(_) => {}
|
||||
Err(err)
|
||||
@@ -58,11 +46,8 @@ impl LocalAgentControl {
|
||||
err.details(),
|
||||
CodexErrorDetails::ThreadNotFound(_) | CodexErrorDetails::InternalAgentDied
|
||||
) => {}
|
||||
Err(err) => return Err(AgentInterruptError::Agent(err)),
|
||||
Err(err) => return Err(err),
|
||||
}
|
||||
Ok(AgentInterruptOutcome {
|
||||
agent_path,
|
||||
previous_status,
|
||||
})
|
||||
Ok(snapshot)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,8 @@ use crate::CodexThread;
|
||||
use crate::StateDbHandle;
|
||||
use crate::ThreadManager;
|
||||
use crate::agent::agent_status_from_event;
|
||||
use crate::agent::api::AgentInfo;
|
||||
use crate::agent::api::AgentTarget;
|
||||
use crate::agent::next_thread_spawn_depth;
|
||||
use crate::agent::types::AgentMessage;
|
||||
use crate::agent::types::LiveAgent;
|
||||
@@ -881,6 +883,14 @@ async fn check_v2_agent_reload(route: V2ReloadRoute) {
|
||||
.await
|
||||
.is_some()
|
||||
);
|
||||
assert_matches!(
|
||||
control
|
||||
.inspect(parent_thread_id, AgentTarget::Id(spawned_agent.thread_id))
|
||||
.await
|
||||
.expect("inspect registered unloaded agent"),
|
||||
AgentInfo::Unloaded(_)
|
||||
);
|
||||
// Inspection must not restore the evicted runtime.
|
||||
match harness.manager.get_thread(spawned_agent.thread_id).await {
|
||||
Err(err) => match err.details() {
|
||||
CodexErrorDetails::ThreadNotFound(id) => assert_eq!(*id, spawned_agent.thread_id),
|
||||
|
||||
@@ -87,6 +87,16 @@ pub(crate) fn collab_agent_error(agent_id: ThreadId, err: CodexErr) -> FunctionC
|
||||
}
|
||||
}
|
||||
|
||||
/// Keeps V2 request validation distinct from a dropped local manager.
|
||||
pub(crate) fn collab_v2_agent_error(agent_id: ThreadId, err: CodexErr) -> FunctionCallError {
|
||||
match err.details() {
|
||||
CodexErrorDetails::UnsupportedOperation(message) if message != "thread manager dropped" => {
|
||||
FunctionCallError::RespondToModel(message.clone())
|
||||
}
|
||||
_ => collab_agent_error(agent_id, err),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn thread_spawn_source(
|
||||
parent_thread_id: ThreadId,
|
||||
parent_session_source: &SessionSource,
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
use super::analytics::ToolCallAnalytics;
|
||||
use super::*;
|
||||
use crate::agent::control::AgentInterruptError;
|
||||
use crate::agent::control::AgentInterruptOutcome;
|
||||
use crate::tools::handlers::multi_agents_spec::create_interrupt_agent_tool_v2;
|
||||
use codex_tools::ToolSpec;
|
||||
|
||||
@@ -45,20 +43,16 @@ async fn handle_interrupt_agent(
|
||||
let args: InterruptAgentArgs = parse_arguments(&arguments)?;
|
||||
let agent_id = resolve_agent_target(&session, &turn, &args.target).await?;
|
||||
analytics.set_receiver(agent_id);
|
||||
let AgentInterruptOutcome {
|
||||
agent_path,
|
||||
previous_status,
|
||||
} = session
|
||||
let snapshot = session
|
||||
.services
|
||||
.agent_control
|
||||
.interrupt_spawned_agent(session.thread_id, agent_id)
|
||||
.await
|
||||
.map_err(|err| match err {
|
||||
AgentInterruptError::InvalidRequest(message) => {
|
||||
FunctionCallError::RespondToModel(message)
|
||||
}
|
||||
AgentInterruptError::Agent(err) => collab_agent_error(agent_id, err),
|
||||
})?;
|
||||
.map_err(|err| collab_v2_agent_error(agent_id, err))?;
|
||||
let agent_path = snapshot.metadata().agent_path.clone().ok_or_else(|| {
|
||||
FunctionCallError::RespondToModel("target agent is missing an agent_path".to_string())
|
||||
})?;
|
||||
let previous_status = snapshot.status().cloned().unwrap_or(AgentStatus::NotFound);
|
||||
emit_sub_agent_activity(
|
||||
&session,
|
||||
&turn,
|
||||
|
||||
@@ -12,7 +12,6 @@ use crate::agent::api::SendRequest;
|
||||
use crate::agent::child_config::build_agent_resume_config;
|
||||
use crate::agent::types::MessageDeliveryMode;
|
||||
use crate::tools::context::FunctionToolOutput;
|
||||
use codex_protocol::error::CodexErrorDetails;
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
@@ -80,14 +79,7 @@ pub(super) async fn handle_message_string_tool(
|
||||
},
|
||||
})
|
||||
.await
|
||||
.map_err(|err| match err.details() {
|
||||
CodexErrorDetails::UnsupportedOperation(message)
|
||||
if message != "thread manager dropped" =>
|
||||
{
|
||||
FunctionCallError::RespondToModel(message.clone())
|
||||
}
|
||||
_ => collab_agent_error(receiver_thread_id, err),
|
||||
})?;
|
||||
.map_err(|err| collab_v2_agent_error(receiver_thread_id, err))?;
|
||||
let receiver_agent_path = receipt.metadata.agent_path.ok_or_else(|| {
|
||||
FunctionCallError::RespondToModel("target agent is missing an agent_path".to_string())
|
||||
})?;
|
||||
|
||||
Reference in New Issue
Block a user