diff --git a/codex-rs/core/src/agent/api.rs b/codex-rs/core/src/agent/api.rs index 6d5f7e2e16..33c1984df5 100644 --- a/codex-rs/core/src/agent/api.rs +++ b/codex-rs/core/src/agent/api.rs @@ -24,108 +24,121 @@ use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TokenUsage; use codex_protocol::turn_input::TurnStartOptions; use codex_protocol::user_input::UserInput; +use codex_rollout_trace::ThreadTraceContext; use futures::future::BoxFuture; use futures::stream::BoxStream; -use std::num::NonZeroU32; -/// Initial observation followed by changes, with no gap between the two. Observation failures -/// are errors, not terminal agent states. Backends own reconnection and reconciliation. +/// An initial runtime snapshot followed by coalesced status changes. The local stream +/// ends when that runtime's status channel closes; it does not follow a later reload. pub type StatusSubscription = BoxStream<'static, Result>; // Keep dynamic dispatch a compile-time property of the contract. const _: Option<&dyn AgentControl> = None; -/// Coordinates one agent tree through a local or host backend. +/// Coordinates agent operations and shared state through a local or host backend. /// -/// Every operation validates tree membership and caller authority, including calls made -/// through stale controller handles. Unknown IDs return `ThreadNotFound`; invalid or -/// unresolved references return `UnsupportedOperation`, preserving existing resolution -/// errors. Backend failures must not be reported as missing targets or successful operations. -/// Mutation success acknowledges acceptance, not completion of the requested agent work. -/// -/// Implementations preserve MAv2 wake modes and keep loading, capacity checks and delivery -/// behind complete operations. Hosts own transport, durable acceptance, retry identities -/// and recovery. Those mechanisms and channel operations are separate implementation work. +/// Implementations preserve the existing wake modes and keep loading and delivery behind +/// complete operations. Successful delivery means accepted, not read by the model. The +/// local backend retains its current best-effort reporting and runtime observation policy; +/// remote ownership, retries and recovery are separate backend work. /// Boxed Send futures allow callers to use `Arc`. pub trait AgentControl: Send + Sync { - fn identity(&self) -> ControlIdentity; + fn identity(&self) -> SessionId; - /// Allocate, register and start a child, then accept its initial input. Return effective - /// settings so callers do not need a second configuration lookup. - fn spawn(&self, request: SpawnRequest) -> BoxFuture<'_, Result>; + /// Start a child and accept its initial input, returning its effective settings. + fn spawn( + &self, + request: SpawnRequest, + ) -> BoxFuture<'_, Result<(LiveAgent, ThreadConfigSnapshot)>>; - /// Explicitly reopen an agent under the caller's authority and captured settings. - /// Reloading required for message delivery belongs inside `send`. + /// Reopen an absent runtime with the captured settings and session source. An already + /// loaded runtime keeps its current settings. Legacy ID-based resume can reopen a + /// stored thread that is not in the controller's registry. fn resume( &self, caller: ThreadId, target: AgentTarget, config: Config, - ) -> BoxFuture<'_, Result>; + source: SessionSource, + ) -> BoxFuture<'_, Result<(LiveAgent, ThreadConfigSnapshot)>>; - /// Resolve, authorize, reload if needed and accept input with its original attribution - /// and wake mode. Queue-only messages cannot start an idle agent; follow-ups cannot - /// target the root. Acceptance does not mean the model has read the input. + /// Resolve, reload if needed and accept input. Agent messages retain their attribution + /// and wake mode: queue-only messages do not start work and follow-ups cannot target + /// the root. Legacy user input can address loaded threads outside the agent registry. fn send(&self, request: SendRequest) -> BoxFuture<'_, Result>; - /// Stop current work without closing the agent. Return its pre-interrupt snapshot. - fn interrupt(&self, caller: ThreadId, target: AgentTarget) -> BoxFuture<'_, Result>; + /// Stop current work and return the pre-interrupt snapshot. V2 rejects root/self + /// targets and tolerates known unloaded agents; other modes retain direct-ID interruption. + fn interrupt( + &self, + caller: ThreadId, + target: AgentTarget, + version: MultiAgentVersion, + ) -> BoxFuture<'_, Result>; - /// Close the agent and its descendants. Return its pre-close snapshot. + /// Close the agent and its live descendants, returning its pre-close snapshot. fn close(&self, caller: ThreadId, target: AgentTarget) -> BoxFuture<'_, Result>; - /// Read identity, runtime state and effective settings without loading a dormant agent. + /// Read runtime settings or known unloaded metadata without loading an agent. fn inspect(&self, caller: ThreadId, target: AgentTarget) -> BoxFuture<'_, Result>; - /// Return a bounded membership page. Callers own model and UI formatting. - fn list(&self, caller: ThreadId, query: AgentQuery) -> BoxFuture<'_, Result>; + /// List loaded agents using the caller's captured source to resolve a path prefix. + /// Callers own model and UI formatting. + fn list<'a>( + &'a self, + source: &'a SessionSource, + path_prefix: Option<&'a str>, + ) -> BoxFuture<'a, Result>>; - /// Subscribe without loading a dormant agent. Unavailability must remain visible. + /// Subscribe to a loaded runtime without restoring it. fn watch( &self, caller: ThreadId, target: AgentTarget, ) -> BoxFuture<'_, Result>; - /// Atomically reserve tree-wide execution capacity. Unlimited turns return no guard. - /// Cancellation during acquisition must release any partial reservation. This contract - /// does not require changing the local policy from rejecting at capacity to waiting. + /// Check capacity before accepting work. This advisory check does not reserve a slot. + fn check_turn_admission( + &self, + version: MultiAgentVersion, + source: &SessionSource, + ) -> Result<()>; + + /// Track a turn's execution until its guard drops. Local admission keeps the existing + /// separate capacity check and running count; it does not atomically reserve capacity. + /// Root and non-V2 turns return no guard. fn admit_turn( &self, - thread_id: ThreadId, - turn_id: String, version: MultiAgentVersion, source: SessionSource, ) -> BoxFuture<'_, Result>>; - /// Account for an inference response, including compaction. Retried reports for the - /// same thread, turn and response must not charge the budget twice. As in the existing - /// rollout-budget path, `SessionBudgetExceeded` means the usage was recorded and the - /// budget is exhausted; retries must preserve that result without charging again. - fn record_usage( + /// Account for one inference response, including compaction. Each call records usage; + /// callers report it once. `SessionBudgetExceeded` means the usage was recorded and + /// the shared budget is now exhausted. + fn record_usage(&self, usage: TokenUsage) -> BoxFuture<'_, Result<()>>; + + /// Report the terminal result to the parent and completion activity to the task + /// initiator. Local delivery remains best effort and uses the reporting runtime's + /// diagnostic trace; it is not deduplicated. + fn turn_finished<'a>( + &'a self, + outcome: AgentTurnOutcome, + trace: &'a ThreadTraceContext, + ) -> BoxFuture<'a, Result<()>>; + + /// Read the latest shared service tier for use at normal runtime config update points. + fn service_tier(&self) -> Option; + + /// Publish a shared setting when the runtime accepts a root-owned config update. + fn propagate_config_update(&self, update: AgentConfigUpdate) -> BoxFuture<'_, Result<()>>; + + /// Read the existing bounded root evidence for a worker. The local backend returns + /// `None` for the root, a non-V2 tree, or an unavailable root runtime. + fn get_guardian_package( &self, - thread_id: ThreadId, - turn_id: String, - response_id: String, - usage: TokenUsage, - ) -> BoxFuture<'_, Result<()>>; - - /// Accept the terminal result and own completion delivery to the parent and task - /// initiator. Retrying the same thread/turn outcome must not duplicate delivery. - fn turn_finished(&self, outcome: AgentTurnOutcome) -> BoxFuture<'_, Result<()>>; - - /// Publish a root-owned update for live agents and later starts/resumes. Only the root - /// may change shared settings; runtimes read their own effective configuration. - fn propagate_config_update( - &self, - caller: ThreadId, - update: AgentConfigUpdate, - ) -> BoxFuture<'_, Result<()>>; - - /// Read authoritative root evidence, including source, completeness and revision. - /// Fetch failures are errors. Approval consumers must revalidate the evidence revision - /// before accepting an approval; this read alone does not grant authorization. - fn get_guardian_package(&self, agent: ThreadId) -> BoxFuture<'_, Result>; + agent: ThreadId, + ) -> BoxFuture<'_, Result>>; fn pending_budget_reminder<'a>( &'a self, @@ -142,15 +155,8 @@ pub trait AgentControl: Send + Sync { ) -> BoxFuture<'a, Result<()>>; } -/// Persistent tree identity and its current owner generation. Clones and reconnects keep -/// the generation; a replacement owner changes it and fences the previous owner. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct ControlIdentity { - pub session_id: SessionId, - pub generation: uuid::Uuid, -} - -/// References resolve relative to the caller. IDs still require membership checks. +/// References resolve relative to the registered caller. IDs retain each operation's +/// existing lookup policy, including legacy access to unregistered loaded threads. #[derive(Clone, Debug)] pub enum AgentTarget { Id(ThreadId), @@ -221,34 +227,6 @@ pub struct DeliveryReceipt { pub submission_id: String, } -#[derive(Clone, Debug)] -pub enum AgentScope { - Tree, - Children(AgentTarget), - /// Includes the target itself. - Subtree(AgentTarget), -} - -#[derive(Clone, Copy, Debug)] -pub enum AgentVisibility { - Known, - Loaded, -} - -pub struct AgentQuery { - pub scope: AgentScope, - pub visibility: AgentVisibility, - /// Opaque backend cursor. Pages are observations, not a locked tree snapshot. - pub cursor: Option, - /// Requested maximum; backends must also enforce their own hard page-size cap. - pub limit: NonZeroU32, -} - -pub struct AgentPage { - pub agents: Vec, - pub next_cursor: Option, -} - pub struct AgentTurnOutcome { pub thread_id: ThreadId, pub turn_id: String, diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index c95f409b3f..1b9fc35424 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -68,6 +68,7 @@ use uuid::Uuid; use self::execution::AgentExecutionLimiter; use self::residency::V2Residency; +mod api; mod budget; mod completion; mod delivery; diff --git a/codex-rs/core/src/agent/control/api.rs b/codex-rs/core/src/agent/control/api.rs new file mode 100644 index 0000000000..7144877eb9 --- /dev/null +++ b/codex-rs/core/src/agent/control/api.rs @@ -0,0 +1,181 @@ +//! Adapts the existing local operations to the shared controller interface. +//! Runtime loading, message delivery and shared state remain in their existing modules. + +use super::LocalAgentControl; +use crate::agent::api::AgentConfigUpdate; +use crate::agent::api::AgentControl; +use crate::agent::api::AgentInfo; +use crate::agent::api::AgentTarget; +use crate::agent::api::AgentTurnOutcome; +use crate::agent::api::DeliveryReceipt; +use crate::agent::api::SendRequest; +use crate::agent::api::SpawnRequest; +use crate::agent::api::StatusSubscription; +use crate::agent::types::AgentExecutionGuard; +use crate::agent::types::LiveAgent; +use crate::codex_thread::GuardianRootSnapshot; +use crate::codex_thread::ThreadConfigSnapshot; +use crate::config::Config; +use crate::rollout_budget::RolloutBudgetReminder; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::Result; +use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::TokenUsage; +use codex_rollout_trace::ThreadTraceContext; +use futures::future::BoxFuture; + +impl AgentControl for LocalAgentControl { + fn identity(&self) -> SessionId { + self.session_id() + } + + fn spawn( + &self, + request: SpawnRequest, + ) -> BoxFuture<'_, Result<(LiveAgent, ThreadConfigSnapshot)>> { + Box::pin(LocalAgentControl::spawn(self, request)) + } + + fn resume( + &self, + caller: ThreadId, + target: AgentTarget, + config: Config, + source: SessionSource, + ) -> BoxFuture<'_, Result<(LiveAgent, ThreadConfigSnapshot)>> { + Box::pin(async move { + let target = self.resolve_target(caller, &target)?; + self.resume_agent(config, target, source).await + }) + } + + fn send(&self, request: SendRequest) -> BoxFuture<'_, Result> { + Box::pin(LocalAgentControl::send(self, request)) + } + + fn interrupt( + &self, + caller: ThreadId, + target: AgentTarget, + version: MultiAgentVersion, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let target = self.resolve_target(caller, &target)?; + match version { + MultiAgentVersion::Disabled | MultiAgentVersion::V1 => { + let snapshot = self.inspect_agent(target).await?; + self.interrupt_agent(target).await?; + Ok(snapshot) + } + MultiAgentVersion::V2 => self.interrupt_spawned_agent(caller, target).await, + } + }) + } + + fn close(&self, caller: ThreadId, target: AgentTarget) -> BoxFuture<'_, Result> { + Box::pin(async move { + let target = self.resolve_target(caller, &target)?; + self.close_agent(target).await + }) + } + + fn inspect(&self, caller: ThreadId, target: AgentTarget) -> BoxFuture<'_, Result> { + Box::pin(LocalAgentControl::inspect(self, caller, target)) + } + + fn list<'a>( + &'a self, + source: &'a SessionSource, + path_prefix: Option<&'a str>, + ) -> BoxFuture<'a, Result>> { + Box::pin(self.list_agents(source, path_prefix)) + } + + fn watch( + &self, + caller: ThreadId, + target: AgentTarget, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let target = self.resolve_target(caller, &target)?; + self.subscribe_status(target).await + }) + } + + fn check_turn_admission( + &self, + version: MultiAgentVersion, + source: &SessionSource, + ) -> Result<()> { + self.ensure_execution_capacity(version, source) + } + + fn admit_turn( + &self, + version: MultiAgentVersion, + source: SessionSource, + ) -> BoxFuture<'_, Result>> { + Box::pin(async move { Ok(self.execution_guard(version, &source)) }) + } + + fn record_usage(&self, usage: TokenUsage) -> BoxFuture<'_, Result<()>> { + Box::pin(async move { self.record_rollout_budget_usage(&usage) }) + } + + fn turn_finished<'a>( + &'a self, + outcome: AgentTurnOutcome, + trace: &'a ThreadTraceContext, + ) -> BoxFuture<'a, Result<()>> { + Box::pin(async move { + self.notify_parent_of_terminal_turn(outcome, trace).await; + Ok(()) + }) + } + + fn service_tier(&self) -> Option { + self.root_service_tier() + } + + fn propagate_config_update(&self, update: AgentConfigUpdate) -> BoxFuture<'_, Result<()>> { + Box::pin(async move { + match update { + AgentConfigUpdate::ServiceTier(tier) => self.set_root_service_tier(tier), + } + Ok(()) + }) + } + + fn get_guardian_package( + &self, + agent: ThreadId, + ) -> BoxFuture<'_, Result>> { + Box::pin(async move { Ok(self.root_user_authorization(agent).await) }) + } + + fn pending_budget_reminder<'a>( + &'a self, + agent: ThreadId, + window: &'a str, + ) -> BoxFuture<'a, Result>> { + Box::pin(async move { + Ok(LocalAgentControl::pending_budget_reminder( + self, agent, window, + )) + }) + } + + fn mark_budget_reminder_delivered<'a>( + &'a self, + agent: ThreadId, + window: &'a str, + reminder: RolloutBudgetReminder, + ) -> BoxFuture<'a, Result<()>> { + Box::pin(async move { + LocalAgentControl::mark_budget_reminder_delivered(self, agent, window, reminder); + Ok(()) + }) + } +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index eb1fc55cdb..3402a27dfb 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -57,13 +57,8 @@ pub use agent::api::AgentConfigUpdate; pub use agent::api::AgentControl; pub use agent::api::AgentInfo; pub use agent::api::AgentInput; -pub use agent::api::AgentPage; -pub use agent::api::AgentQuery; -pub use agent::api::AgentScope; pub use agent::api::AgentTarget; pub use agent::api::AgentTurnOutcome; -pub use agent::api::AgentVisibility; -pub use agent::api::ControlIdentity; pub use agent::api::DeliveryReceipt; pub use agent::api::SendRequest; pub use agent::api::SpawnRequest;