diff --git a/codex-rs/core/src/agent/bus.rs b/codex-rs/core/src/agent/bus.rs deleted file mode 100644 index 0b15f415f1..0000000000 --- a/codex-rs/core/src/agent/bus.rs +++ /dev/null @@ -1,61 +0,0 @@ -use codex_protocol::ConversationId; -use codex_protocol::protocol::EventMsg; -use std::collections::HashMap; -use std::sync::Arc; -use tokio::sync::RwLock; - -/// Status store for globally-tracked agents. -#[derive(Clone, Default)] -pub(crate) struct AgentBus { - /// In-memory map of conversation id to the latest derived status. - statuses: Arc>>, -} - -#[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) enum AgentStatus { - PendingInit, - Running, - Completed(Option), - Errored(String), - Shutdown, - #[allow(dead_code)] // Used by upcoming multi-agent tooling. - NotFound, -} - -impl AgentBus { - /// Fetch the last known status for `agent_id`, returning `NotFound` if unseen. - #[allow(dead_code)] // Used by upcoming multi-agent tooling. - pub(crate) async fn status(&self, agent_id: ConversationId) -> AgentStatus { - let statuses = self.statuses.read().await; - statuses - .get(&agent_id) - .cloned() - .unwrap_or(AgentStatus::NotFound) - } - - /// Derive and record agent status from a single emitted event. - pub(crate) async fn on_event(&self, conversation_id: ConversationId, msg: &EventMsg) { - let next_status = match msg { - EventMsg::TaskStarted(_) => Some(AgentStatus::Running), - EventMsg::TaskComplete(ev) => { - Some(AgentStatus::Completed(ev.last_agent_message.clone())) - } - EventMsg::TurnAborted(ev) => Some(AgentStatus::Errored(format!("{:?}", ev.reason))), - EventMsg::Error(ev) => Some(AgentStatus::Errored(ev.message.clone())), - EventMsg::ShutdownComplete => Some(AgentStatus::Shutdown), - _ => None, - }; - if let Some(status) = next_status { - self.record_status(&conversation_id, status).await; - } - } - - /// Force-set the tracked status for an agent conversation. - pub(crate) async fn record_status( - &self, - conversation_id: &ConversationId, - status: AgentStatus, - ) { - self.statuses.write().await.insert(*conversation_id, status); - } -} diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 158a5e59b9..468f580a21 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -1,5 +1,4 @@ use crate::CodexConversation; -use crate::agent::AgentBus; use crate::agent::AgentStatus; use crate::conversation_manager::ConversationManagerState; use crate::error::CodexErr; @@ -20,8 +19,6 @@ pub(crate) struct AgentControl { /// This is `Weak` to avoid reference cycles and shadow persistence of the form /// `ConversationManagerState -> CodexConversation -> Session -> SessionServices -> ConversationManagerState`. manager: Weak, - /// Shared agent status store updated from emitted events. - pub(crate) bus: AgentBus, } impl AgentControl { @@ -44,10 +41,6 @@ impl AgentControl { let state = self.upgrade()?; let new_conversation = state.spawn_new_conversation(config, self.clone()).await?; - self.bus - .record_status(&new_conversation.conversation_id, AgentStatus::PendingInit) - .await; - if headless { spawn_headless_drain(Arc::clone(&new_conversation.conversation)); } @@ -55,10 +48,6 @@ impl AgentControl { self.send_prompt(new_conversation.conversation_id, prompt) .await?; - self.bus - .record_status(&new_conversation.conversation_id, AgentStatus::Running) - .await; - Ok(new_conversation.conversation_id) } diff --git a/codex-rs/core/src/agent/mod.rs b/codex-rs/core/src/agent/mod.rs index bde8821379..d6348b38b3 100644 --- a/codex-rs/core/src/agent/mod.rs +++ b/codex-rs/core/src/agent/mod.rs @@ -1,4 +1,3 @@ -pub(crate) mod bus; pub(crate) mod control; pub(crate) mod status;