diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 7f26acf176..00b210909a 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -354,7 +354,7 @@ impl Codex { /// /// A session has at most 1 running task at a time, and can be interrupted by user input. pub(crate) struct Session { - conversation_id: ThreadId, + pub(crate) conversation_id: ThreadId, tx_event: Sender, agent_status: watch::Sender, state: Mutex, diff --git a/codex-rs/core/src/rollout/policy.rs b/codex-rs/core/src/rollout/policy.rs index 224e45dc52..7198e1faa0 100644 --- a/codex-rs/core/src/rollout/policy.rs +++ b/codex-rs/core/src/rollout/policy.rs @@ -90,6 +90,7 @@ pub(crate) fn should_persist_event_msg(ev: &EventMsg) -> bool { | EventMsg::AgentMessageContentDelta(_) | EventMsg::ReasoningContentDelta(_) | EventMsg::ReasoningRawContentDelta(_) - | EventMsg::SkillsUpdateAvailable => false, + | EventMsg::SkillsUpdateAvailable + | EventMsg::CollabInteraction(_) => false, } } diff --git a/codex-rs/core/src/tools/handlers/collab.rs b/codex-rs/core/src/tools/handlers/collab.rs index 0f74ef658a..71a47baeaa 100644 --- a/codex-rs/core/src/tools/handlers/collab.rs +++ b/codex-rs/core/src/tools/handlers/collab.rs @@ -11,6 +11,8 @@ use crate::tools::registry::ToolHandler; use crate::tools::registry::ToolKind; use async_trait::async_trait; use codex_protocol::ThreadId; +use codex_protocol::protocol::CollabInteractionEvent; +use codex_protocol::protocol::EventMsg; use serde::Deserialize; use serde::Serialize; @@ -54,9 +56,9 @@ impl ToolHandler for CollabHandler { match tool_name.as_str() { "spawn_agent" => spawn::handle(session, turn, arguments).await, - "send_input" => send_input::handle(session, arguments).await, - "wait" => wait::handle(session, arguments).await, - "close_agent" => close_agent::handle(session, arguments).await, + "send_input" => send_input::handle(session, turn, arguments).await, + "wait" => wait::handle(session, turn, arguments).await, + "close_agent" => close_agent::handle(session, turn, arguments).await, other => Err(FunctionCallError::RespondToModel(format!( "unsupported collab tool {other}" ))), @@ -89,16 +91,36 @@ mod spawn { let result = session .services .agent_control - .spawn_agent(config, args.message, true) + .spawn_agent(config, args.message.clone(), true) .await .map_err(|err| FunctionCallError::Fatal(err.to_string()))?; + emit_event(session, turn, args.message, result).await; + Ok(ToolOutput::Function { content: format!("agent_id: {result}"), success: Some(true), content_items: None, }) } + + async fn emit_event( + session: Arc, + turn: Arc, + prompt: String, + new_id: ThreadId, + ) { + session + .send_event( + &turn, + EventMsg::CollabInteraction(CollabInteractionEvent::AgentSpawned { + sender_id: session.conversation_id, + new_id, + prompt, + }), + ) + .await + } } mod send_input { @@ -114,6 +136,7 @@ mod send_input { pub async fn handle( session: Arc, + turn: Arc, arguments: String, ) -> Result { let args: SendInputArgs = parse_arguments(&arguments)?; @@ -126,7 +149,7 @@ mod send_input { let content = session .services .agent_control - .send_prompt(agent_id, args.message) + .send_prompt(agent_id, args.message.clone()) .await .map_err(|err| match err { CodexErr::ThreadNotFound(id) => { @@ -135,12 +158,32 @@ mod send_input { err => FunctionCallError::Fatal(err.to_string()), })?; + emit_event(session, turn, agent_id, args.message).await; + Ok(ToolOutput::Function { content, success: Some(true), content_items: None, }) } + + async fn emit_event( + session: Arc, + turn: Arc, + receiver_id: ThreadId, + prompt: String, + ) { + session + .send_event( + &turn, + EventMsg::CollabInteraction(CollabInteractionEvent::AgentInteraction { + sender_id: session.conversation_id, + receiver_id, + prompt, + }), + ) + .await + } } mod wait { @@ -166,6 +209,7 @@ mod wait { pub async fn handle( session: Arc, + turn: Arc, arguments: String, ) -> Result { let args: WaitArgs = parse_arguments(&arguments)?; @@ -194,6 +238,18 @@ mod wait { err => FunctionCallError::Fatal(err.to_string()), })?; + let waiting_id = format!("collab-waiting-{}", uuid::Uuid::new_v4()); + session + .send_event( + &turn, + EventMsg::CollabInteraction(CollabInteractionEvent::WaitingBegin { + sender_id: session.conversation_id, + receiver_id: agent_id, + waiting_id: waiting_id.clone(), + }), + ) + .await; + // Get last known status. let mut status = status_rx.borrow_and_update().clone(); let deadline = Instant::now() + Duration::from_millis(timeout_ms as u64); @@ -218,6 +274,18 @@ mod wait { } }; + session + .send_event( + &turn, + EventMsg::CollabInteraction(CollabInteractionEvent::WaitingEnd { + sender_id: session.conversation_id, + receiver_id: agent_id, + waiting_id, + status: status.clone(), + }), + ) + .await; + if matches!(status, AgentStatus::NotFound) { return Err(FunctionCallError::RespondToModel(format!( "agent with id {agent_id} not found" @@ -250,22 +318,12 @@ pub mod close_agent { pub async fn handle( session: Arc, + turn: Arc, arguments: String, ) -> Result { let args: CloseAgentArgs = parse_arguments(&arguments)?; let agent_id = agent_id(&args.id)?; - let mut status_rx = session - .services - .agent_control - .subscribe_status(agent_id) - .await - .map_err(|err| match err { - CodexErr::ThreadNotFound(id) => { - FunctionCallError::RespondToModel(format!("agent with id {id} not found")) - } - err => FunctionCallError::Fatal(err.to_string()), - })?; - let status = status_rx.borrow_and_update().clone(); + let status = session.services.agent_control.get_status(agent_id).await; if !matches!(status, AgentStatus::Shutdown) { let _ = session @@ -281,6 +339,8 @@ pub mod close_agent { })?; } + emit_event(session, turn, agent_id, status.clone()).await; + let content = serde_json::to_string(&CloseAgentResult { status }).map_err(|err| { FunctionCallError::Fatal(format!("failed to serialize close_agent result: {err}")) })?; @@ -291,6 +351,24 @@ pub mod close_agent { content_items: None, }) } + + async fn emit_event( + session: Arc, + turn: Arc, + receiver_id: ThreadId, + status: AgentStatus, + ) { + session + .send_event( + &turn, + EventMsg::CollabInteraction(CollabInteractionEvent::Close { + sender_id: session.conversation_id, + receiver_id, + status, + }), + ) + .await + } } fn agent_id(id: &str) -> Result { diff --git a/codex-rs/exec/src/event_processor_with_human_output.rs b/codex-rs/exec/src/event_processor_with_human_output.rs index f1cba0b9f7..732d832d84 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/codex-rs/exec/src/event_processor_with_human_output.rs @@ -571,6 +571,9 @@ impl EventProcessor for EventProcessorWithHumanOutput { EventMsg::ContextCompacted(_) => { ts_msg!(self, "context compacted"); } + EventMsg::CollabInteraction(_) => { + // TODO(jif) handle collab tools. + } EventMsg::ShutdownComplete => return CodexStatus::Shutdown, EventMsg::WebSearchBegin(_) | EventMsg::ExecApprovalRequest(_) diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index 1ee4cbd7f6..08f6564492 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -306,6 +306,7 @@ async fn run_codex_tool_session_inner( | EventMsg::ExitedReviewMode(_) | EventMsg::ContextCompacted(_) | EventMsg::ThreadRolledBack(_) + | EventMsg::CollabInteraction(_) | EventMsg::DeprecationNotice(_) => { // For now, we do not do anything extra for these // events. Note that diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 4f6314104d..bddd28ccb3 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -683,6 +683,9 @@ pub enum EventMsg { AgentMessageContentDelta(AgentMessageContentDeltaEvent), ReasoningContentDelta(ReasoningContentDeltaEvent), ReasoningRawContentDelta(ReasoningRawContentDeltaEvent), + + /// Collab interaction. + CollabInteraction(CollabInteractionEvent), } /// Agent lifecycle status, derived from emitted events. @@ -1933,6 +1936,56 @@ pub enum TurnAbortReason { ReviewEnded, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "snake_case")] +pub enum CollabInteractionEvent { + AgentSpawned { + /// Thread ID of the sender. + sender_id: ThreadId, + /// Thread ID of the newly spawned agent. + new_id: ThreadId, + /// Initial prompt sent to the agent. Can be empty to prevent CoT leaking at the + /// beginning. + prompt: String, + }, + AgentInteraction { + /// Thread ID of the sender. + sender_id: ThreadId, + /// Thread ID of the receiver. + receiver_id: ThreadId, + /// Prompt sent from the sender to the receiver. Can be empty to prevent CoT + /// leaking at the beginning. + prompt: String, + }, + WaitingBegin { + /// Thread ID of the sender. + sender_id: ThreadId, + /// Thread ID of the receiver. + receiver_id: ThreadId, + /// ID of the waiting call. + waiting_id: String, + }, + WaitingEnd { + /// Thread ID of the sender. + sender_id: ThreadId, + /// Thread ID of the receiver. + receiver_id: ThreadId, + /// ID of the waiting call. + waiting_id: String, + /// Final status of the receiver agent reported to the sender agent. + status: AgentStatus, + }, + Close { + /// Thread ID of the sender. + sender_id: ThreadId, + /// Thread ID of the receiver. + receiver_id: ThreadId, + /// Last known status of the receiver agent reported to the sender agent before + /// the close. + status: AgentStatus, + }, +} + #[cfg(test)] mod tests { use super::*; diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index e3fd7a891e..129e8c2667 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -2182,6 +2182,9 @@ impl ChatWidget { } EventMsg::ExitedReviewMode(review) => self.on_exited_review_mode(review), EventMsg::ContextCompacted(_) => self.on_agent_message("Context compacted".to_owned()), + EventMsg::CollabInteraction(_) => { + // TODO(jif) handle collab tools. + } EventMsg::ThreadRolledBack(_) => {} EventMsg::RawResponseItem(_) | EventMsg::ItemStarted(_) diff --git a/codex-rs/tui2/src/chatwidget.rs b/codex-rs/tui2/src/chatwidget.rs index f5af1bfe67..060db06053 100644 --- a/codex-rs/tui2/src/chatwidget.rs +++ b/codex-rs/tui2/src/chatwidget.rs @@ -1988,6 +1988,9 @@ impl ChatWidget { } EventMsg::ExitedReviewMode(review) => self.on_exited_review_mode(review), EventMsg::ContextCompacted(_) => self.on_agent_message("Context compacted".to_owned()), + EventMsg::CollabInteraction(_) => { + // TODO(jif) handle collab tools. + } EventMsg::RawResponseItem(_) | EventMsg::ThreadRolledBack(_) | EventMsg::ItemStarted(_)