openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-16 12:13:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: emit events around collab tools

Browse Source
This commit is contained in:
jif-oai
2026-01-12 14:53:09 +00:00
parent 9659583559
commit e3cf74885a
8 changed files with 161 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
codex-rs/core/src/codex.rs
View File

@@ -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<Event>,
agent_status: watch::Sender<AgentStatus>,
state: Mutex<SessionState>,

3
codex-rs/core/src/rollout/policy.rs
View File

@@ -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,
}
}

112
codex-rs/core/src/tools/handlers/collab.rs
View File

@@ -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<Session>,
turn: Arc<TurnContext>,
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<Session>,
turn: Arc<TurnContext>,
arguments: String,
) -> Result<ToolOutput, FunctionCallError> {
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<Session>,
turn: Arc<TurnContext>,
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<Session>,
turn: Arc<TurnContext>,
arguments: String,
) -> Result<ToolOutput, FunctionCallError> {
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<Session>,
turn: Arc<TurnContext>,
arguments: String,
) -> Result<ToolOutput, FunctionCallError> {
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<Session>,
turn: Arc<TurnContext>,
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<ThreadId, FunctionCallError> {

3
codex-rs/exec/src/event_processor_with_human_output.rs
View File

@@ -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(_)

1
codex-rs/mcp-server/src/codex_tool_runner.rs
View File

@@ -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

53
codex-rs/protocol/src/protocol.rs
View File

@@ -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::*;

3
codex-rs/tui/src/chatwidget.rs
View File

@@ -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(_)

3
codex-rs/tui2/src/chatwidget.rs
View File

@@ -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(_)