From b9b934e99b57aa1908a7cd83adee9faf6d0b8f7d Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 6 Jul 2026 20:41:57 -0700 Subject: [PATCH] refactor(protocol): map canonical tool items to legacy events (#31296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description This PR adds legacy `EventMsg` mappings for the `TurnItem` types introduced in [#30282](https://github.com/openai/codex/pull/30282): - `CommandExecution` - `DynamicToolCall` - `CollabAgentToolCall` - `SubAgentActivity` When their producers move to canonical `ItemStarted` / `ItemCompleted`, raw core event consumers can still receive the existing begin/end-style events. The canonical item lifecycle remains the live source of truth. We also record the mapped legacy events in rollout trace so the producer migration preserves the existing tool-runtime trace entries. ## Why This is the compatibility layer for the follow-up producer migrations. Splitting it out first keeps each producer PR small and keeps the legacy mapping in one place. ## What changed - Added `TurnItem` → legacy `EventMsg` mappings in `protocol/src/legacy_events.rs`. - Added the command execution status conversion used by the exec mapping. - Added focused coverage for command execution and dynamic tool mappings. --- codex-rs/core/src/session/mod.rs | 3 + codex-rs/protocol/src/items.rs | 11 + codex-rs/protocol/src/legacy_events.rs | 304 +++++++++++++++++++++++++ codex-rs/protocol/src/protocol.rs | 137 +++++++++++ 4 files changed, 455 insertions(+) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 91a21ef731..fa240b890b 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1794,6 +1794,9 @@ impl Session { let show_raw_agent_reasoning = self.show_raw_agent_reasoning(); for legacy in legacy_source.as_legacy_events(show_raw_agent_reasoning) { + self.services + .rollout_thread_trace + .record_tool_call_event(turn_context.sub_id.clone(), &legacy); let legacy_event = Event { id: turn_context.sub_id.clone(), msg: legacy, diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index 820c3b774b..2c9d073f0d 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -13,6 +13,7 @@ use crate::parse_command::ParsedCommand; use crate::protocol::AgentStatus; use crate::protocol::CollabAgentRef; use crate::protocol::ExecCommandSource; +use crate::protocol::ExecCommandStatus; use crate::protocol::FileChange; use crate::protocol::PatchApplyStatus; use crate::protocol::SubAgentActivityKind; @@ -137,6 +138,16 @@ pub enum CommandExecutionStatus { Declined, } +impl From for CommandExecutionStatus { + fn from(value: ExecCommandStatus) -> Self { + match value { + ExecCommandStatus::Completed => Self::Completed, + ExecCommandStatus::Failed => Self::Failed, + ExecCommandStatus::Declined => Self::Declined, + } + } +} + #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] pub struct CommandExecutionItem { pub id: String, diff --git a/codex-rs/protocol/src/legacy_events.rs b/codex-rs/protocol/src/legacy_events.rs index f4a511d367..c12a2747a2 100644 --- a/codex-rs/protocol/src/legacy_events.rs +++ b/codex-rs/protocol/src/legacy_events.rs @@ -1,10 +1,20 @@ +use crate::ThreadId; +use crate::dynamic_tools::DynamicToolCallRequest; use crate::items::AgentMessageContent; use crate::items::AgentMessageItem; +use crate::items::CollabAgentTool; +use crate::items::CollabAgentToolCallItem; +use crate::items::CollabAgentToolCallStatus; +use crate::items::CommandExecutionItem; +use crate::items::CommandExecutionStatus; use crate::items::ContextCompactionItem; +use crate::items::DynamicToolCallItem; +use crate::items::DynamicToolCallStatus; use crate::items::FileChangeItem; use crate::items::ImageGenerationItem; use crate::items::McpToolCallItem; use crate::items::ReasoningItem; +use crate::items::SubAgentActivityItem; use crate::items::TurnItem; use crate::items::UserMessageItem; use crate::items::WebSearchItem; @@ -12,8 +22,24 @@ use crate::protocol::AgentMessageContentDeltaEvent; use crate::protocol::AgentMessageEvent; use crate::protocol::AgentReasoningEvent; use crate::protocol::AgentReasoningRawContentEvent; +use crate::protocol::AgentStatus; +use crate::protocol::CollabAgentInteractionBeginEvent; +use crate::protocol::CollabAgentInteractionEndEvent; +use crate::protocol::CollabAgentSpawnBeginEvent; +use crate::protocol::CollabAgentSpawnEndEvent; +use crate::protocol::CollabAgentStatusEntry; +use crate::protocol::CollabCloseBeginEvent; +use crate::protocol::CollabCloseEndEvent; +use crate::protocol::CollabResumeBeginEvent; +use crate::protocol::CollabResumeEndEvent; +use crate::protocol::CollabWaitingBeginEvent; +use crate::protocol::CollabWaitingEndEvent; use crate::protocol::ContextCompactedEvent; +use crate::protocol::DynamicToolCallResponseEvent; use crate::protocol::EventMsg; +use crate::protocol::ExecCommandBeginEvent; +use crate::protocol::ExecCommandEndEvent; +use crate::protocol::ExecCommandStatus; use crate::protocol::ImageGenerationBeginEvent; use crate::protocol::ImageGenerationEndEvent; use crate::protocol::ItemCompletedEvent; @@ -26,6 +52,7 @@ use crate::protocol::PatchApplyEndEvent; use crate::protocol::PatchApplyStatus; use crate::protocol::ReasoningContentDeltaEvent; use crate::protocol::ReasoningRawContentDeltaEvent; +use crate::protocol::SubAgentActivityEvent; use crate::protocol::UserMessageEvent; use crate::protocol::ViewImageToolCallEvent; use crate::protocol::WebSearchBeginEvent; @@ -97,6 +124,258 @@ impl ReasoningItem { } } +impl CommandExecutionItem { + pub(crate) fn as_legacy_begin_event(&self, turn_id: String, started_at_ms: i64) -> EventMsg { + EventMsg::ExecCommandBegin(ExecCommandBeginEvent { + call_id: self.id.clone(), + process_id: self.process_id.clone(), + turn_id, + started_at_ms, + command: self.command.clone(), + cwd: self.cwd.clone(), + parsed_cmd: self.parsed_cmd.clone(), + source: self.source, + interaction_input: self.interaction_input.clone(), + }) + } + + pub(crate) fn as_legacy_end_event( + &self, + turn_id: String, + completed_at_ms: i64, + ) -> Option { + let status = match self.status { + CommandExecutionStatus::InProgress => return None, + CommandExecutionStatus::Completed => ExecCommandStatus::Completed, + CommandExecutionStatus::Failed => ExecCommandStatus::Failed, + CommandExecutionStatus::Declined => ExecCommandStatus::Declined, + }; + Some(EventMsg::ExecCommandEnd(ExecCommandEndEvent { + call_id: self.id.clone(), + process_id: self.process_id.clone(), + turn_id, + completed_at_ms, + command: self.command.clone(), + cwd: self.cwd.clone(), + parsed_cmd: self.parsed_cmd.clone(), + source: self.source, + interaction_input: self.interaction_input.clone(), + stdout: self.stdout.clone().unwrap_or_default(), + stderr: self.stderr.clone().unwrap_or_default(), + aggregated_output: self.aggregated_output.clone().unwrap_or_default(), + exit_code: self.exit_code.unwrap_or_default(), + duration: self.duration.unwrap_or_default(), + formatted_output: self.formatted_output.clone().unwrap_or_default(), + status, + })) + } +} + +impl DynamicToolCallItem { + pub(crate) fn as_legacy_request_event(&self, turn_id: String, started_at_ms: i64) -> EventMsg { + EventMsg::DynamicToolCallRequest(DynamicToolCallRequest { + call_id: self.id.clone(), + turn_id, + started_at_ms, + namespace: self.namespace.clone(), + tool: self.tool.clone(), + arguments: self.arguments.clone(), + }) + } + + pub(crate) fn as_legacy_response_event( + &self, + turn_id: String, + completed_at_ms: i64, + ) -> Option { + if matches!(self.status, DynamicToolCallStatus::InProgress) { + return None; + } + Some(EventMsg::DynamicToolCallResponse( + DynamicToolCallResponseEvent { + call_id: self.id.clone(), + turn_id, + completed_at_ms, + namespace: self.namespace.clone(), + tool: self.tool.clone(), + arguments: self.arguments.clone(), + content_items: self.content_items.clone().unwrap_or_default(), + success: self.success.unwrap_or(false), + error: self.error.clone(), + duration: self.duration.unwrap_or_default(), + }, + )) + } +} + +impl CollabAgentToolCallItem { + pub(crate) fn as_legacy_begin_event(&self, started_at_ms: i64) -> Option { + let receiver_thread_id = self.receiver_thread_ids.first().copied(); + match self.tool { + CollabAgentTool::SpawnAgent => Some(EventMsg::CollabAgentSpawnBegin( + CollabAgentSpawnBeginEvent { + call_id: self.id.clone(), + started_at_ms, + sender_thread_id: self.sender_thread_id, + prompt: self.prompt.clone().unwrap_or_default(), + model: self.model.clone().unwrap_or_default(), + reasoning_effort: self.reasoning_effort.clone().unwrap_or_default(), + }, + )), + CollabAgentTool::SendInput => receiver_thread_id.map(|receiver_thread_id| { + EventMsg::CollabAgentInteractionBegin(CollabAgentInteractionBeginEvent { + call_id: self.id.clone(), + started_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_id, + prompt: self.prompt.clone().unwrap_or_default(), + }) + }), + CollabAgentTool::ResumeAgent => receiver_thread_id.map(|receiver_thread_id| { + let (receiver_agent_nickname, receiver_agent_role) = + self.receiver_agent_identity(receiver_thread_id); + EventMsg::CollabResumeBegin(CollabResumeBeginEvent { + call_id: self.id.clone(), + started_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_id, + receiver_agent_nickname, + receiver_agent_role, + }) + }), + CollabAgentTool::Wait => Some(EventMsg::CollabWaitingBegin(CollabWaitingBeginEvent { + started_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_ids: self.receiver_thread_ids.clone(), + receiver_agents: self.receiver_agents.clone(), + call_id: self.id.clone(), + })), + CollabAgentTool::CloseAgent => receiver_thread_id.map(|receiver_thread_id| { + EventMsg::CollabCloseBegin(CollabCloseBeginEvent { + call_id: self.id.clone(), + started_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_id, + }) + }), + } + } + + pub(crate) fn as_legacy_end_event(&self, completed_at_ms: i64) -> Option { + if matches!(self.status, CollabAgentToolCallStatus::InProgress) { + return None; + } + let receiver_thread_id = self.receiver_thread_ids.first().copied(); + match self.tool { + CollabAgentTool::SpawnAgent => { + let (new_agent_nickname, new_agent_role) = receiver_thread_id + .map(|thread_id| self.receiver_agent_identity(thread_id)) + .unwrap_or_default(); + Some(EventMsg::CollabAgentSpawnEnd(CollabAgentSpawnEndEvent { + call_id: self.id.clone(), + completed_at_ms, + sender_thread_id: self.sender_thread_id, + new_thread_id: receiver_thread_id, + new_agent_nickname, + new_agent_role, + prompt: self.prompt.clone().unwrap_or_default(), + model: self.model.clone().unwrap_or_default(), + reasoning_effort: self.reasoning_effort.clone().unwrap_or_default(), + status: receiver_thread_id + .map(|thread_id| self.agent_status(thread_id)) + .unwrap_or(AgentStatus::NotFound), + })) + } + CollabAgentTool::SendInput => receiver_thread_id.map(|receiver_thread_id| { + let (receiver_agent_nickname, receiver_agent_role) = + self.receiver_agent_identity(receiver_thread_id); + EventMsg::CollabAgentInteractionEnd(CollabAgentInteractionEndEvent { + call_id: self.id.clone(), + completed_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_id, + receiver_agent_nickname, + receiver_agent_role, + prompt: self.prompt.clone().unwrap_or_default(), + status: self.agent_status(receiver_thread_id), + }) + }), + CollabAgentTool::ResumeAgent => receiver_thread_id.map(|receiver_thread_id| { + let (receiver_agent_nickname, receiver_agent_role) = + self.receiver_agent_identity(receiver_thread_id); + EventMsg::CollabResumeEnd(CollabResumeEndEvent { + call_id: self.id.clone(), + completed_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_id, + receiver_agent_nickname, + receiver_agent_role, + status: self.agent_status(receiver_thread_id), + }) + }), + CollabAgentTool::Wait => Some(EventMsg::CollabWaitingEnd(CollabWaitingEndEvent { + sender_thread_id: self.sender_thread_id, + call_id: self.id.clone(), + completed_at_ms, + agent_statuses: self + .receiver_agents + .iter() + .map(|agent| CollabAgentStatusEntry { + thread_id: agent.thread_id, + agent_nickname: agent.agent_nickname.clone(), + agent_role: agent.agent_role.clone(), + status: self.agent_status(agent.thread_id), + }) + .collect(), + statuses: self.agents_states.clone(), + })), + CollabAgentTool::CloseAgent => receiver_thread_id.map(|receiver_thread_id| { + let (receiver_agent_nickname, receiver_agent_role) = + self.receiver_agent_identity(receiver_thread_id); + EventMsg::CollabCloseEnd(CollabCloseEndEvent { + call_id: self.id.clone(), + completed_at_ms, + sender_thread_id: self.sender_thread_id, + receiver_thread_id, + receiver_agent_nickname, + receiver_agent_role, + status: self.agent_status(receiver_thread_id), + }) + }), + } + } + + fn receiver_agent_identity(&self, thread_id: ThreadId) -> (Option, Option) { + let receiver_agent = self + .receiver_agents + .iter() + .find(|agent| agent.thread_id == thread_id); + ( + receiver_agent.and_then(|agent| agent.agent_nickname.clone()), + receiver_agent.and_then(|agent| agent.agent_role.clone()), + ) + } + + fn agent_status(&self, thread_id: ThreadId) -> AgentStatus { + self.agents_states + .get(&thread_id) + .cloned() + .unwrap_or(AgentStatus::NotFound) + } +} + +impl SubAgentActivityItem { + pub(crate) fn as_legacy_event(&self, occurred_at_ms: i64) -> EventMsg { + EventMsg::SubAgentActivity(SubAgentActivityEvent { + event_id: self.id.clone(), + occurred_at_ms, + agent_thread_id: self.agent_thread_id, + agent_path: self.agent_path.clone(), + kind: self.kind, + }) + } +} + impl WebSearchItem { pub fn as_legacy_event(&self) -> EventMsg { EventMsg::WebSearchEnd(WebSearchEndEvent { @@ -234,6 +513,16 @@ impl HasLegacyEvent for ItemStartedEvent { } TurnItem::FileChange(item) => vec![item.as_legacy_begin_event(self.turn_id.clone())], TurnItem::McpToolCall(item) => vec![item.as_legacy_begin_event()], + TurnItem::CommandExecution(item) => { + vec![item.as_legacy_begin_event(self.turn_id.clone(), self.started_at_ms)] + } + TurnItem::DynamicToolCall(item) => { + vec![item.as_legacy_request_event(self.turn_id.clone(), self.started_at_ms)] + } + TurnItem::CollabAgentToolCall(item) => item + .as_legacy_begin_event(self.started_at_ms) + .into_iter() + .collect(), _ => Vec::new(), } } @@ -246,6 +535,21 @@ impl HasLegacyEvent for ItemCompletedEvent { .as_legacy_end_event(self.turn_id.clone()) .into_iter() .collect(), + TurnItem::CommandExecution(item) => item + .as_legacy_end_event(self.turn_id.clone(), self.completed_at_ms) + .into_iter() + .collect(), + TurnItem::DynamicToolCall(item) => item + .as_legacy_response_event(self.turn_id.clone(), self.completed_at_ms) + .into_iter() + .collect(), + TurnItem::CollabAgentToolCall(item) => item + .as_legacy_end_event(self.completed_at_ms) + .into_iter() + .collect(), + TurnItem::SubAgentActivity(item) => { + vec![item.as_legacy_event(self.completed_at_ms)] + } _ => self.item.as_legacy_events(show_raw_agent_reasoning), } } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index ca509a5a17..b1b845fe28 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -4309,6 +4309,10 @@ pub struct CollabResumeEndEvent { #[cfg(test)] mod tests { use super::*; + use crate::items::CommandExecutionItem; + use crate::items::CommandExecutionStatus; + use crate::items::DynamicToolCallItem; + use crate::items::DynamicToolCallStatus; use crate::items::FileChangeItem; use crate::items::ImageGenerationItem; use crate::items::McpToolCallItem; @@ -5290,6 +5294,139 @@ mod tests { } } + #[test] + fn command_execution_item_lifecycle_emits_legacy_exec_events() { + let cwd = PathUri::from_abs_path(&test_path_buf("/tmp").abs()); + let started = ItemStartedEvent { + thread_id: ThreadId::new(), + turn_id: "turn-1".into(), + started_at_ms: 10, + item: TurnItem::CommandExecution(CommandExecutionItem { + id: "exec-1".into(), + process_id: Some("pid-1".into()), + command: vec!["echo".into(), "done".into()], + cwd: cwd.clone(), + parsed_cmd: vec![ParsedCommand::Unknown { + cmd: "echo done".into(), + }], + source: ExecCommandSource::Agent, + interaction_input: None, + status: CommandExecutionStatus::InProgress, + stdout: None, + stderr: None, + aggregated_output: None, + exit_code: None, + duration: None, + formatted_output: None, + }), + }; + let completed = ItemCompletedEvent { + thread_id: ThreadId::new(), + turn_id: "turn-1".into(), + completed_at_ms: 20, + item: TurnItem::CommandExecution(CommandExecutionItem { + id: "exec-1".into(), + process_id: Some("pid-1".into()), + command: vec!["echo".into(), "done".into()], + cwd, + parsed_cmd: vec![ParsedCommand::Unknown { + cmd: "echo done".into(), + }], + source: ExecCommandSource::Agent, + interaction_input: None, + status: CommandExecutionStatus::Completed, + stdout: Some("done\n".into()), + stderr: Some(String::new()), + aggregated_output: Some("done\n".into()), + exit_code: Some(0), + duration: Some(Duration::from_millis(5)), + formatted_output: Some("done\n".into()), + }), + }; + + assert!(matches!( + started.as_legacy_events(/*show_raw_agent_reasoning*/ false).as_slice(), + [EventMsg::ExecCommandBegin(ExecCommandBeginEvent { + call_id, + turn_id, + started_at_ms: 10, + .. + })] if call_id == "exec-1" && turn_id == "turn-1" + )); + assert!(matches!( + completed + .as_legacy_events(/*show_raw_agent_reasoning*/ false) + .as_slice(), + [EventMsg::ExecCommandEnd(ExecCommandEndEvent { + call_id, + turn_id, + completed_at_ms: 20, + aggregated_output, + .. + })] if call_id == "exec-1" && turn_id == "turn-1" && aggregated_output == "done\n" + )); + } + + #[test] + fn dynamic_tool_call_item_lifecycle_emits_legacy_dynamic_tool_events() { + let started = ItemStartedEvent { + thread_id: ThreadId::new(), + turn_id: "turn-1".into(), + started_at_ms: 10, + item: TurnItem::DynamicToolCall(DynamicToolCallItem { + id: "dynamic-1".into(), + namespace: Some("apps".into()), + tool: "lookup".into(), + arguments: json!({"id": "123"}), + status: DynamicToolCallStatus::InProgress, + content_items: None, + success: None, + error: None, + duration: None, + }), + }; + let completed = ItemCompletedEvent { + thread_id: ThreadId::new(), + turn_id: "turn-1".into(), + completed_at_ms: 20, + item: TurnItem::DynamicToolCall(DynamicToolCallItem { + id: "dynamic-1".into(), + namespace: Some("apps".into()), + tool: "lookup".into(), + arguments: json!({"id": "123"}), + status: DynamicToolCallStatus::Completed, + content_items: Some(vec![DynamicToolCallOutputContentItem::InputText { + text: "ok".into(), + }]), + success: Some(true), + error: None, + duration: Some(Duration::from_millis(5)), + }), + }; + + assert!(matches!( + started.as_legacy_events(/*show_raw_agent_reasoning*/ false).as_slice(), + [EventMsg::DynamicToolCallRequest(DynamicToolCallRequest { + call_id, + turn_id, + started_at_ms: 10, + .. + })] if call_id == "dynamic-1" && turn_id == "turn-1" + )); + assert!(matches!( + completed + .as_legacy_events(/*show_raw_agent_reasoning*/ false) + .as_slice(), + [EventMsg::DynamicToolCallResponse(DynamicToolCallResponseEvent { + call_id, + turn_id, + completed_at_ms: 20, + success: true, + .. + })] if call_id == "dynamic-1" && turn_id == "turn-1" + )); + } + #[test] fn item_started_event_requires_started_at_ms() { let mut value = serde_json::to_value(ItemStartedEvent {