Rename AgentControl to LocalAgentControl (#46513)

GitOrigin-RevId: a8e8a5c818a1751c5d2667f5af87e481482fc7db
This commit is contained in:
jif
2026-09-18 23:39:30 +00:00
committed by copyberry
parent 0f7c8608bd
commit 7fb599de8c
24 changed files with 69 additions and 69 deletions
+1 -1
View File
@@ -164,7 +164,7 @@ fn reject_full_fork_agent_type_override(agent_type: Option<&str>) -> Result<(),
Ok(())
}
/// Copies runtime-only turn state onto a child config before it is handed to `AgentControl`.
/// Copies runtime-only turn state onto a child config before it is handed to `LocalAgentControl`.
///
/// These values are chosen by the live turn rather than persisted config, so leaving them stale can
/// make a child agent disagree with its parent about approval policy, cwd, or sandboxing.
+7 -7
View File
@@ -123,13 +123,13 @@ pub(crate) struct ListedAgent {
}
/// Control-plane handle for multi-agent operations.
/// `AgentControl` is held by each session (via `SessionServices`). It provides capability to
/// `LocalAgentControl` is held by each session (via `SessionServices`). It provides capability to
/// spawn new agents and the inter-agent communication layer.
/// An `AgentControl` instance is intended to be created at most once per root thread/session
/// tree. That same `AgentControl` is then shared with every sub-agent spawned from that root,
/// An `LocalAgentControl` instance is intended to be created at most once per root thread/session
/// tree. That same `LocalAgentControl` is then shared with every sub-agent spawned from that root,
/// which keeps the registry scoped to that root thread rather than the entire `ThreadManager`.
#[derive(Clone)]
pub(crate) struct AgentControl {
pub(crate) struct LocalAgentControl {
/// session_id is equal to the root thread's ID.
session_id: SessionId,
/// Weak handle back to the global thread registry/state.
@@ -147,7 +147,7 @@ pub(crate) struct AgentControl {
root_service_tier: Arc<ArcSwapOption<String>>,
}
impl Default for AgentControl {
impl Default for LocalAgentControl {
fn default() -> Self {
Self::new(
Weak::default(),
@@ -157,8 +157,8 @@ impl Default for AgentControl {
}
}
impl AgentControl {
/// Construct a new `AgentControl` that can spawn/message agents via the given manager state.
impl LocalAgentControl {
/// Construct a new `LocalAgentControl` that can spawn/message agents via the given manager state.
pub(crate) fn new(
manager: Weak<ThreadManagerState>,
thread_id_generator: ThreadIdGenerator,
+2 -2
View File
@@ -2,7 +2,7 @@
//!
//! Target checks precede reload, and queue-only messages retain their non-waking semantics.
use super::AgentControl;
use super::LocalAgentControl;
use crate::TurnStartOptions;
use crate::agent::child_config::build_agent_resume_config;
use crate::agent_communication::AgentCommunicationContext;
@@ -69,7 +69,7 @@ pub(crate) enum MessageDeliveryError {
Agent(CodexErr),
}
impl AgentControl {
impl LocalAgentControl {
/// Checks and delivers to a resolved target, restoring an evicted runtime when necessary.
///
/// The caller resolves tool-facing names separately so it can attribute failures and
+2 -2
View File
@@ -1,4 +1,4 @@
use super::AgentControl;
use super::LocalAgentControl;
use crate::codex_thread::CodexThread;
use codex_protocol::error::CodexErr;
use codex_protocol::error::CodexErrorDetails;
@@ -26,7 +26,7 @@ impl Drop for AgentExecutionGuard {
}
}
impl AgentControl {
impl LocalAgentControl {
pub(crate) async fn ensure_execution_capacity_for_turn_start(
&self,
thread: &CodexThread,
@@ -1,12 +1,12 @@
use crate::agent::AgentControl;
use crate::agent::LocalAgentControl;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use pretty_assertions::assert_eq;
fn control_with_limit(max_threads: usize) -> AgentControl {
let control = AgentControl::default();
fn control_with_limit(max_threads: usize) -> LocalAgentControl {
let control = LocalAgentControl::default();
control.agent_execution_limiter.initialize(max_threads);
control
}
+2 -2
View File
@@ -3,7 +3,7 @@
//! Root and self targets are rejected. An unloaded or already-dead runtime is a successful
//! interruption; this operation never reloads it.
use super::AgentControl;
use super::LocalAgentControl;
use crate::agent::AgentStatus;
use codex_protocol::AgentPath;
use codex_protocol::ThreadId;
@@ -22,7 +22,7 @@ pub(crate) enum AgentInterruptError {
Agent(CodexErr),
}
impl AgentControl {
impl LocalAgentControl {
/// Interrupts a spawned agent's current task, preserving the status observed before dispatch.
pub(crate) async fn interrupt_spawned_agent(
&self,
+1 -1
View File
@@ -2,7 +2,7 @@ use super::*;
use codex_protocol::error::CodexErrorDetails;
use codex_thread_store::PersistContext;
impl AgentControl {
impl LocalAgentControl {
/// Submit a shutdown request for a live agent without marking it explicitly closed in
/// persisted spawn-edge state.
pub(crate) async fn shutdown_live_agent(&self, agent_id: ThreadId) -> CodexResult<String> {
+2 -2
View File
@@ -1,4 +1,4 @@
use super::AgentControl;
use super::LocalAgentControl;
use crate::agent::AgentStatus;
use crate::codex_thread::CodexThread;
use crate::config::Config;
@@ -45,7 +45,7 @@ impl Drop for V2ResidencySlot {
}
}
impl AgentControl {
impl LocalAgentControl {
pub(super) async fn reserve_v2_residency_slot(
&self,
state: &Arc<ThreadManagerState>,
@@ -1,6 +1,6 @@
use crate::StartThreadOptions;
use crate::ThreadManager;
use crate::agent::AgentControl;
use crate::agent::LocalAgentControl;
use crate::codex_thread::CodexThread;
use crate::config::Config;
use crate::config::test_config;
@@ -134,7 +134,7 @@ async fn interrupted_v2_agent_is_lost_after_residency_eviction() {
}
async fn spawn_v2_subagent(
control: &AgentControl,
control: &LocalAgentControl,
state: &Arc<ThreadManagerState>,
config: Config,
parent_thread_id: ThreadId,
@@ -10,9 +10,9 @@ use codex_history::SenderUserMessages;
use codex_protocol::ThreadId;
use codex_protocol::models::ResponseItem;
use super::AgentControl;
use super::LocalAgentControl;
impl AgentControl {
impl LocalAgentControl {
pub(crate) async fn capture_sender_user_messages(
&self,
item: &ResponseItem,
@@ -1,9 +1,9 @@
//! Shares the root user's selected routing tier across the entire agent tree.
use super::AgentControl;
use super::LocalAgentControl;
use std::sync::Arc;
impl AgentControl {
impl LocalAgentControl {
/// Returns the latest user-selected tier for this root and all its descendants.
pub(crate) fn root_service_tier(&self) -> Option<String> {
self.root_service_tier
+2 -2
View File
@@ -172,7 +172,7 @@ async fn load_agent_model_context(
}
}
impl AgentControl {
impl LocalAgentControl {
/// Restore persisted V2 agent identities without reopening their runtimes.
pub(crate) async fn restore_v2_agent_metadata(
&self,
@@ -689,7 +689,7 @@ impl AgentControl {
};
let notification_source = session_source.clone();
// The same `AgentControl` is sent to spawn the thread.
// The same `LocalAgentControl` is sent to spawn the thread.
let new_thread = match (session_source, options.fork_mode.as_ref(), inheritance) {
(Some(session_source), Some(_), inheritance) => {
Box::pin(self.spawn_forked_thread(
@@ -5,7 +5,7 @@
use std::borrow::Cow;
use super::AgentControl;
use super::LocalAgentControl;
use crate::codex_thread::GuardianRootMessage;
use crate::codex_thread::GuardianRootSnapshot;
use crate::compact::is_summary_message;
@@ -29,7 +29,7 @@ use codex_protocol::protocol::MultiAgentVersion;
const MAX_ROOT_MESSAGES: usize = 8;
impl AgentControl {
impl LocalAgentControl {
/// Returns bounded root conversation and authorization state for a MultiAgent V2 worker.
pub(crate) async fn root_user_authorization(
&self,
+8 -8
View File
@@ -161,7 +161,7 @@ fn assistant_message(text: &str, phase: Option<MessagePhase>) -> ResponseItem {
#[test]
fn register_session_root_skips_threads_with_explicit_parent() {
let control = AgentControl::default();
let control = LocalAgentControl::default();
control.register_session_root(ThreadId::new(), Some(ThreadId::new()));
@@ -185,7 +185,7 @@ struct AgentControlHarness {
config: Config,
state_db: Option<StateDbHandle>,
manager: ThreadManager,
control: AgentControl,
control: LocalAgentControl,
}
impl AgentControlHarness {
@@ -416,7 +416,7 @@ async fn persist_thread_for_tree_resume(thread: &Arc<CodexThread>, message: &str
}
async fn wait_for_live_thread_spawn_children(
control: &AgentControl,
control: &LocalAgentControl,
parent_thread_id: ThreadId,
expected_children: &[ThreadId],
) {
@@ -455,7 +455,7 @@ async fn assert_thread_not_loaded(manager: &ThreadManager, thread_id: ThreadId)
#[tokio::test]
async fn send_input_errors_when_manager_dropped() {
let control = AgentControl::default();
let control = LocalAgentControl::default();
let err = control
.send_input(
ThreadId::new(),
@@ -475,7 +475,7 @@ async fn send_input_errors_when_manager_dropped() {
#[tokio::test]
async fn get_status_returns_not_found_without_manager() {
let control = AgentControl::default();
let control = LocalAgentControl::default();
let got = control.get_status(ThreadId::new()).await;
assert_eq!(got, AgentStatus::NotFound);
}
@@ -553,7 +553,7 @@ async fn on_event_updates_status_from_shutdown_complete() {
#[tokio::test]
async fn spawn_agent_errors_when_manager_dropped() {
let control = AgentControl::default();
let control = LocalAgentControl::default();
let (_home, config) = test_config().await;
let err = control
.spawn_agent(config, text_input("hello"), /*session_source*/ None)
@@ -567,7 +567,7 @@ async fn spawn_agent_errors_when_manager_dropped() {
#[tokio::test]
async fn resume_agent_errors_when_manager_dropped() {
let control = AgentControl::default();
let control = LocalAgentControl::default();
let (_home, config) = test_config().await;
let err = control
.resume_agent_from_rollout(config, ThreadId::new(), SessionSource::Exec)
@@ -750,7 +750,7 @@ enum V2ReloadRoute {
}
async fn spawn_v2_reload_test_child(
control: &AgentControl,
control: &LocalAgentControl,
config: Config,
parent: &CodexThread,
task_name: &str,
+1 -1
View File
@@ -6,7 +6,7 @@ pub(crate) mod role;
pub(crate) mod status;
pub(crate) use codex_protocol::protocol::AgentStatus;
pub(crate) use control::AgentControl;
pub(crate) use control::LocalAgentControl;
pub(crate) use registry::exceeds_thread_spawn_depth_limit;
pub(crate) use registry::next_thread_spawn_depth;
pub(crate) use status::agent_status_from_event;
+1 -1
View File
@@ -19,7 +19,7 @@ use std::sync::atomic::Ordering;
/// the current implementation, it limits:
/// * Total number of sub-agents (i.e. threads) per user session
///
/// This structure is shared by all agents in the same user session (because the `AgentControl`
/// This structure is shared by all agents in the same user session (because the `LocalAgentControl`
/// is).
#[derive(Default)]
pub(crate) struct AgentRegistry {
+2 -2
View File
@@ -10,8 +10,8 @@ use std::sync::atomic::AtomicU64;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use crate::agent::AgentControl;
use crate::agent::AgentStatus;
use crate::agent::LocalAgentControl;
use crate::agent::agent_status_from_event;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
@@ -460,7 +460,7 @@ pub(crate) struct SessionSpawnArgs {
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) thread_source: Option<ThreadSource>,
pub(crate) originator: String,
pub(crate) agent_control: AgentControl,
pub(crate) agent_control: LocalAgentControl,
pub(crate) dynamic_tools: Vec<DynamicToolSpec>,
pub(crate) metrics_service_name: Option<String>,
pub(crate) inherited_exec_policy: Option<Arc<ExecPolicyManager>>,
+1 -1
View File
@@ -758,7 +758,7 @@ impl Session {
extensions: Arc<codex_extension_api::ExtensionRegistry<crate::config::Config>>,
mut thread_extension_init: ExtensionDataInit,
client_mcp_extensions: ClientMcpExtensions,
agent_control: AgentControl,
agent_control: LocalAgentControl,
reserved_thread_id: Option<ThreadId>,
environment_manager: Arc<EnvironmentManager>,
inherited_environments: Option<TurnEnvironmentSnapshot>,
+8 -8
View File
@@ -5922,7 +5922,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() {
Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()),
codex_extension_api::ExtensionDataInit::default(),
ClientMcpExtensions::default(),
AgentControl::default(),
LocalAgentControl::default(),
/*reserved_thread_id*/ None,
environment_manager,
/*inherited_environments*/ None,
@@ -6005,7 +6005,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
auth_manager.clone(),
config.model_provider.clone(),
);
let agent_control = AgentControl::default();
let agent_control = LocalAgentControl::default();
let exec_policy = Arc::new(ExecPolicyManager::default());
let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit);
let model = get_model_offline_for_tests(config.model.as_deref());
@@ -6421,7 +6421,7 @@ async fn make_session_with_config_and_rx(
Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()),
codex_extension_api::ExtensionDataInit::default(),
ClientMcpExtensions::default(),
AgentControl::default(),
LocalAgentControl::default(),
/*reserved_thread_id*/ None,
environment_manager,
/*inherited_environments*/ None,
@@ -6446,7 +6446,7 @@ async fn make_session_with_config_and_rx(
async fn make_session_with_history_source_and_agent_control_and_rx(
initial_history: InitialHistory,
session_source: SessionSource,
agent_control: AgentControl,
agent_control: LocalAgentControl,
) -> anyhow::Result<(Arc<Session>, async_channel::Receiver<Event>)> {
let codex_home = tempfile::tempdir().expect("create temp dir");
let mut config = build_test_config(codex_home.path()).await;
@@ -6593,7 +6593,7 @@ async fn resumed_root_session_uses_thread_id_as_session_id() {
rollout_path: None,
}),
SessionSource::Exec,
AgentControl::default(),
LocalAgentControl::default(),
)
.await
.expect("resume should succeed");
@@ -6636,7 +6636,7 @@ async fn resumed_subagent_session_restores_persisted_session_id() {
rollout_path: None,
}),
session_source,
AgentControl::default(),
LocalAgentControl::default(),
)
.await
.expect("resume should succeed");
@@ -6689,7 +6689,7 @@ async fn resumed_copied_fork_ignores_source_history_base() {
rollout_path: None,
}),
SessionSource::Exec,
AgentControl::default(),
LocalAgentControl::default(),
)
.await
.expect("resume should succeed");
@@ -8249,7 +8249,7 @@ where
auth_manager.clone(),
config.model_provider.clone(),
);
let agent_control = AgentControl::default();
let agent_control = LocalAgentControl::default();
let exec_policy = Arc::new(ExecPolicyManager::default());
let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit);
let model = get_model_offline_for_tests(config.model.as_deref());
@@ -1325,7 +1325,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
parent_thread_id: None,
thread_source: None,
originator: "test_originator".to_string(),
agent_control: AgentControl::default(),
agent_control: LocalAgentControl::default(),
dynamic_tools: Vec::new(),
metrics_service_name: None,
inherited_environments: None,
+2 -2
View File
@@ -1,7 +1,7 @@
use std::path::PathBuf;
use std::sync::Arc;
use crate::agent::AgentControl;
use crate::agent::LocalAgentControl;
use crate::agents_md_manager::AgentsMdManager;
use crate::attestation::AttestationProvider;
use crate::client::ModelClient;
@@ -82,7 +82,7 @@ pub(crate) struct SessionServices {
/// current executor environments before using them.
pub(crate) selected_capability_roots: Vec<SelectedCapabilityRoot>,
pub(crate) mcp_thread_init: ExtensionDataInit,
pub(crate) agent_control: AgentControl,
pub(crate) agent_control: LocalAgentControl,
pub(crate) network_proxy: ArcSwapOption<StartedNetworkProxy>,
pub(crate) network_proxy_audit_metadata: NetworkProxyAuditMetadata,
pub(crate) managed_network_requirements_configured: bool,
+13 -13
View File
@@ -1,7 +1,7 @@
mod managed;
use crate::CodexAppsToolsCache;
use crate::agent::AgentControl;
use crate::agent::LocalAgentControl;
use crate::agents_md_manager::SessionInstructions;
use crate::attestation::AttestationProvider;
use crate::codex_thread::CodexThread;
@@ -240,7 +240,7 @@ pub struct ThreadManager {
pub struct InternalSessionParent {
pub(crate) thread_id: ThreadId,
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) agent_control: AgentControl,
pub(crate) agent_control: LocalAgentControl,
pub(crate) originator: String,
pub(crate) inherited_instructions: Option<SessionInstructions>,
}
@@ -312,7 +312,7 @@ struct ThreadSpawnRequest {
startup: Option<Arc<crate::session::startup::SessionStartup>>,
options: StartThreadOptions,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
agent_control: LocalAgentControl,
parent_thread_id: Option<ThreadId>,
parent_originator: Option<String>,
forked_from_thread_id: Option<ThreadId>,
@@ -327,7 +327,7 @@ impl ThreadSpawnRequest {
fn new(
options: StartThreadOptions,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
agent_control: LocalAgentControl,
) -> Self {
Self {
startup: None,
@@ -379,7 +379,7 @@ fn effective_originator_value(
pub(crate) struct ResumeThreadWithHistoryOptions {
pub(crate) config: Config,
pub(crate) initial_history: InitialHistory,
pub(crate) agent_control: AgentControl,
pub(crate) agent_control: LocalAgentControl,
pub(crate) session_source: SessionSource,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) environment_selections: Option<Vec<TurnEnvironmentSelection>>,
@@ -390,7 +390,7 @@ pub(crate) struct ResumeThreadWithHistoryOptions {
}
/// Shared, `Arc`-owned state for [`ThreadManager`]. This `Arc` is required to have a single
/// `Arc` reference that can be downgraded to by `AgentControl` while preventing every single
/// `Arc` reference that can be downgraded to by `LocalAgentControl` while preventing every single
/// function to require an `Arc<&Self>`.
pub(crate) struct ThreadManagerState {
threads: Arc<RwLock<HashMap<ThreadId, Arc<CodexThread>>>>,
@@ -1531,16 +1531,16 @@ impl ThreadManager {
Box::pin(self.state.spawn_thread(request)).await
}
pub(crate) fn agent_control(&self) -> AgentControl {
AgentControl::new(
pub(crate) fn agent_control(&self) -> LocalAgentControl {
LocalAgentControl::new(
Arc::downgrade(&self.state),
self.state.thread_id_generator.clone(),
/*rollout_budget*/ None,
)
}
fn agent_control_for_config(&self, config: &Config) -> AgentControl {
AgentControl::new(
fn agent_control_for_config(&self, config: &Config) -> LocalAgentControl {
LocalAgentControl::new(
Arc::downgrade(&self.state),
self.state.thread_id_generator.clone(),
config.rollout_budget.clone(),
@@ -1832,7 +1832,7 @@ impl ThreadManagerState {
pub(crate) async fn spawn_new_thread(
&self,
config: Config,
agent_control: AgentControl,
agent_control: LocalAgentControl,
) -> CodexResult<NewThread> {
Box::pin(self.spawn_new_thread_with_source(
config,
@@ -1854,7 +1854,7 @@ impl ThreadManagerState {
pub(crate) async fn spawn_new_thread_with_source(
&self,
config: Config,
agent_control: AgentControl,
agent_control: LocalAgentControl,
session_source: SessionSource,
history_mode: Option<ThreadHistoryMode>,
parent_thread_id: Option<ThreadId>,
@@ -1933,7 +1933,7 @@ impl ThreadManagerState {
config: Config,
initial_history: InitialHistory,
history_mode: Option<ThreadHistoryMode>,
agent_control: AgentControl,
agent_control: LocalAgentControl,
session_source: SessionSource,
thread_source: Option<ThreadSource>,
parent_thread_id: Option<ThreadId>,
+1 -1
View File
@@ -281,7 +281,7 @@ async fn thread_analytics_opt_out_overrides_shared_client() {
/// Controls without a custom allocation policy still produce distinct thread identifiers.
#[test]
fn thread_id_generator_defaults_to_standard_ids() {
let agent_control = AgentControl::default();
let agent_control = LocalAgentControl::default();
assert_ne!(
agent_control.generate_thread_id(),
@@ -1,6 +1,6 @@
//! Implements the collaboration tool surface for spawning and managing sub-agents.
//!
//! This handler translates model tool calls into `AgentControl` operations and keeps spawned
//! This handler translates model tool calls into `LocalAgentControl` operations and keeps spawned
//! agents aligned with the live turn that created them. Sub-agents start from the turn's effective
//! config, inherit runtime-only state such as provider, approval policy, sandbox, and cwd, and
//! then optionally layer role-specific config on top.