From 3bc6e7499a473dc6324f3ba83403ecc78fdc5375 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 26 Jun 2026 12:36:53 -0700 Subject: [PATCH 1/6] feat(protocol): add projection-friendly turn items --- .../src/protocol/item_builders.rs | 5 +- .../src/protocol/thread_history.rs | 8 + .../src/protocol/v2/item.rs | 122 +++ .../src/protocol/v2/tests.rs | 138 ++++ codex-rs/protocol/src/items.rs | 728 ++++++++++++++++++ codex-rs/protocol/src/protocol.rs | 163 ++++ codex-rs/rollout/src/persistence_metrics.rs | 4 + 7 files changed, 1167 insertions(+), 1 deletion(-) diff --git a/codex-rs/app-server-protocol/src/protocol/item_builders.rs b/codex-rs/app-server-protocol/src/protocol/item_builders.rs index 66019980a3c8..c041fe96217c 100644 --- a/codex-rs/app-server-protocol/src/protocol/item_builders.rs +++ b/codex-rs/app-server-protocol/src/protocol/item_builders.rs @@ -128,7 +128,10 @@ pub fn build_command_execution_end_item(payload: &ExecCommandEndEvent) -> Thread } } -fn command_actions_for_path_uri(parsed_cmd: &[ParsedCommand], cwd: &PathUri) -> Vec { +pub(crate) fn command_actions_for_path_uri( + parsed_cmd: &[ParsedCommand], + cwd: &PathUri, +) -> Vec { // TODO(anp): Carry PathUri into CommandAction so foreign Read actions retain resolved paths. // Until then, omit those actions rather than project a foreign cwd onto the host. let native_cwd = if cwd.infer_path_convention() == Some(PathConvention::native()) { diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index aed87b18ab7e..852167ecd44f 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -594,6 +594,10 @@ impl ThreadHistoryBuilder { | codex_protocol::items::TurnItem::HookPrompt(_) | codex_protocol::items::TurnItem::AgentMessage(_) | codex_protocol::items::TurnItem::Reasoning(_) + | codex_protocol::items::TurnItem::CommandExecution(_) + | codex_protocol::items::TurnItem::DynamicToolCall(_) + | codex_protocol::items::TurnItem::CollabAgentToolCall(_) + | codex_protocol::items::TurnItem::SubAgentActivity(_) | codex_protocol::items::TurnItem::WebSearch(_) | codex_protocol::items::TurnItem::ImageView(_) | codex_protocol::items::TurnItem::ImageGeneration(_) @@ -624,6 +628,10 @@ impl ThreadHistoryBuilder { | codex_protocol::items::TurnItem::HookPrompt(_) | codex_protocol::items::TurnItem::AgentMessage(_) | codex_protocol::items::TurnItem::Reasoning(_) + | codex_protocol::items::TurnItem::CommandExecution(_) + | codex_protocol::items::TurnItem::DynamicToolCall(_) + | codex_protocol::items::TurnItem::CollabAgentToolCall(_) + | codex_protocol::items::TurnItem::SubAgentActivity(_) | codex_protocol::items::TurnItem::WebSearch(_) | codex_protocol::items::TurnItem::ImageView(_) | codex_protocol::items::TurnItem::ImageGeneration(_) diff --git a/codex-rs/app-server-protocol/src/protocol/v2/item.rs b/codex-rs/app-server-protocol/src/protocol/v2/item.rs index 751eab824d9a..cd04752449d1 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/item.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/item.rs @@ -8,12 +8,17 @@ use super::NetworkPolicyAmendment; use super::RequestPermissionProfile; use super::UserInput; use super::shared::v2_enum_from_core; +use crate::protocol::item_builders::command_actions_for_path_uri; use crate::protocol::item_builders::convert_patch_changes; use codex_experimental_api_macros::ExperimentalApi; use codex_protocol::approvals::GuardianAssessmentAction as CoreGuardianAssessmentAction; use codex_protocol::approvals::GuardianAssessmentDecisionSource as CoreGuardianAssessmentDecisionSource; use codex_protocol::approvals::GuardianCommandSource as CoreGuardianCommandSource; use codex_protocol::items::AgentMessageContent as CoreAgentMessageContent; +use codex_protocol::items::CollabAgentTool as CoreCollabAgentTool; +use codex_protocol::items::CollabAgentToolCallStatus as CoreCollabAgentToolCallStatus; +use codex_protocol::items::CommandExecutionStatus as CoreCommandExecutionStatus; +use codex_protocol::items::DynamicToolCallStatus as CoreDynamicToolCallStatus; use codex_protocol::items::McpToolCallStatus as CoreMcpToolCallStatus; use codex_protocol::items::TurnItem as CoreTurnItem; use codex_protocol::memory_citation::MemoryCitation as CoreMemoryCitation; @@ -30,6 +35,7 @@ use codex_protocol::protocol::GuardianUserAuthorization as CoreGuardianUserAutho use codex_protocol::protocol::PatchApplyStatus as CorePatchApplyStatus; use codex_protocol::protocol::ReviewDecision as CoreReviewDecision; use codex_protocol::protocol::SubAgentActivityKind as CoreSubAgentActivityKind; +use codex_shell_command::parse_command::shlex_join; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::LegacyAppPathString; use schemars::JsonSchema; @@ -854,6 +860,64 @@ impl From for ThreadItem { summary: reasoning.summary_text, content: reasoning.raw_content, }, + CoreTurnItem::CommandExecution(command) => ThreadItem::CommandExecution { + id: command.id, + command: shlex_join(&command.command), + cwd: command.cwd.clone().into(), + process_id: command.process_id, + source: command.source.into(), + status: command.status.into(), + command_actions: command_actions_for_path_uri(&command.parsed_cmd, &command.cwd), + aggregated_output: command + .aggregated_output + .filter(|output| !output.is_empty()), + exit_code: command.exit_code, + duration_ms: command + .duration + .and_then(|duration| i64::try_from(duration.as_millis()).ok()), + }, + CoreTurnItem::DynamicToolCall(call) => ThreadItem::DynamicToolCall { + id: call.id, + namespace: call.namespace, + tool: call.tool, + arguments: call.arguments, + status: call.status.into(), + content_items: call.content_items.map(|items| { + items + .into_iter() + .map(DynamicToolCallOutputContentItem::from) + .collect() + }), + success: call.success, + duration_ms: call + .duration + .and_then(|duration| i64::try_from(duration.as_millis()).ok()), + }, + CoreTurnItem::CollabAgentToolCall(call) => ThreadItem::CollabAgentToolCall { + id: call.id, + tool: call.tool.into(), + status: call.status.into(), + sender_thread_id: call.sender_thread_id.to_string(), + receiver_thread_ids: call + .receiver_thread_ids + .into_iter() + .map(String::from) + .collect(), + prompt: call.prompt, + model: call.model, + reasoning_effort: call.reasoning_effort, + agents_states: call + .agents_states + .into_iter() + .map(|(thread_id, status)| (thread_id.to_string(), status.into())) + .collect(), + }, + CoreTurnItem::SubAgentActivity(activity) => ThreadItem::SubAgentActivity { + id: activity.id, + kind: activity.kind.into(), + agent_thread_id: activity.agent_thread_id.to_string(), + agent_path: String::from(activity.agent_path), + }, CoreTurnItem::WebSearch(search) => ThreadItem::WebSearch { id: search.id, query: search.query, @@ -941,6 +1005,17 @@ impl From for CommandExecutionStatus { } } +impl From for CommandExecutionStatus { + fn from(value: CoreCommandExecutionStatus) -> Self { + match value { + CoreCommandExecutionStatus::InProgress => Self::InProgress, + CoreCommandExecutionStatus::Completed => Self::Completed, + CoreCommandExecutionStatus::Failed => Self::Failed, + CoreCommandExecutionStatus::Declined => Self::Declined, + } + } +} + impl From<&CoreExecCommandStatus> for CommandExecutionStatus { fn from(value: &CoreExecCommandStatus) -> Self { match value { @@ -1028,6 +1103,16 @@ impl From for McpToolCallStatus { } } +impl From for DynamicToolCallStatus { + fn from(value: CoreDynamicToolCallStatus) -> Self { + match value { + CoreDynamicToolCallStatus::InProgress => Self::InProgress, + CoreDynamicToolCallStatus::Completed => Self::Completed, + CoreDynamicToolCallStatus::Failed => Self::Failed, + } + } +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -1055,6 +1140,28 @@ pub enum CollabAgentToolCallStatus { Failed, } +impl From for CollabAgentTool { + fn from(value: CoreCollabAgentTool) -> Self { + match value { + CoreCollabAgentTool::SpawnAgent => Self::SpawnAgent, + CoreCollabAgentTool::SendInput => Self::SendInput, + CoreCollabAgentTool::ResumeAgent => Self::ResumeAgent, + CoreCollabAgentTool::Wait => Self::Wait, + CoreCollabAgentTool::CloseAgent => Self::CloseAgent, + } + } +} + +impl From for CollabAgentToolCallStatus { + fn from(value: CoreCollabAgentToolCallStatus) -> Self { + match value { + CoreCollabAgentToolCallStatus::InProgress => Self::InProgress, + CoreCollabAgentToolCallStatus::Completed => Self::Completed, + CoreCollabAgentToolCallStatus::Failed => Self::Failed, + } + } +} + #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -1462,6 +1569,21 @@ pub enum DynamicToolCallOutputContentItem { InputImage { image_url: String }, } +impl From + for DynamicToolCallOutputContentItem +{ + fn from(item: codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem) -> Self { + match item { + codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem::InputText { text } => { + Self::InputText { text } + } + codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem::InputImage { + image_url, + } => Self::InputImage { image_url }, + } + } +} + impl From for codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem { diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 677f4470fd2a..48f92bf785bc 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -4,11 +4,19 @@ use codex_protocol::approvals::ElicitationRequest as CoreElicitationRequest; use codex_protocol::config_types::MultiAgentMode; use codex_protocol::items::AgentMessageContent; use codex_protocol::items::AgentMessageItem; +use codex_protocol::items::CollabAgentTool as CoreCollabAgentTool; +use codex_protocol::items::CollabAgentToolCallItem; +use codex_protocol::items::CollabAgentToolCallStatus as CoreCollabAgentToolCallStatus; +use codex_protocol::items::CommandExecutionItem; +use codex_protocol::items::CommandExecutionStatus as CoreCommandExecutionStatus; +use codex_protocol::items::DynamicToolCallItem; +use codex_protocol::items::DynamicToolCallStatus as CoreDynamicToolCallStatus; use codex_protocol::items::FileChangeItem; use codex_protocol::items::ImageViewItem; use codex_protocol::items::McpToolCallItem; use codex_protocol::items::McpToolCallStatus as CoreMcpToolCallStatus; use codex_protocol::items::ReasoningItem; +use codex_protocol::items::SubAgentActivityItem; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::items::WebSearchItem; @@ -30,8 +38,10 @@ use codex_protocol::permissions::FileSystemSpecialPath as CoreFileSystemSpecialP use codex_protocol::protocol::AgentStatus as CoreAgentStatus; use codex_protocol::protocol::AskForApproval as CoreAskForApproval; use codex_protocol::protocol::ConversationTextRole; +use codex_protocol::protocol::ExecCommandSource as CoreExecCommandSource; use codex_protocol::protocol::GranularApprovalConfig as CoreGranularApprovalConfig; use codex_protocol::protocol::NetworkAccess as CoreNetworkAccess; +use codex_protocol::protocol::SubAgentActivityKind as CoreSubAgentActivityKind; use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile; use codex_protocol::user_input::UserInput as CoreUserInput; use codex_utils_absolute_path::AbsolutePathBuf; @@ -2632,6 +2642,134 @@ fn core_turn_item_into_thread_item_converts_supported_variants() { } ); + let command_item = TurnItem::CommandExecution(CommandExecutionItem { + id: "exec-1".to_string(), + process_id: Some("pid-1".to_string()), + command: vec!["echo".to_string(), "done".to_string()], + cwd: PathUri::from_abs_path(&test_path_buf("/tmp").abs()), + parsed_cmd: vec![codex_protocol::parse_command::ParsedCommand::Unknown { + cmd: "echo done".to_string(), + }], + source: CoreExecCommandSource::Agent, + interaction_input: None, + status: CoreCommandExecutionStatus::Completed, + stdout: Some("done\n".to_string()), + stderr: Some(String::new()), + aggregated_output: Some("done\n".to_string()), + exit_code: Some(0), + duration: Some(Duration::from_millis(5)), + formatted_output: Some("done\n".to_string()), + }); + + assert_eq!( + ThreadItem::from(command_item), + ThreadItem::CommandExecution { + id: "exec-1".to_string(), + command: "echo done".to_string(), + cwd: LegacyAppPathString::from_abs_path(&test_path_buf("/tmp").abs()), + process_id: Some("pid-1".to_string()), + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::Completed, + command_actions: vec![CommandAction::Unknown { + command: "echo done".to_string(), + }], + aggregated_output: Some("done\n".to_string()), + exit_code: Some(0), + duration_ms: Some(5), + } + ); + + let dynamic_tool_call_item = TurnItem::DynamicToolCall(DynamicToolCallItem { + id: "dynamic-1".to_string(), + namespace: Some("apps".to_string()), + tool: "lookup".to_string(), + arguments: json!({"id": "123"}), + status: CoreDynamicToolCallStatus::Completed, + content_items: Some(vec![ + codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem::InputText { + text: "ok".to_string(), + }, + ]), + success: Some(true), + error: None, + duration: Some(Duration::from_millis(5)), + }); + + assert_eq!( + ThreadItem::from(dynamic_tool_call_item), + ThreadItem::DynamicToolCall { + id: "dynamic-1".to_string(), + namespace: Some("apps".to_string()), + tool: "lookup".to_string(), + arguments: json!({"id": "123"}), + status: DynamicToolCallStatus::Completed, + content_items: Some(vec![DynamicToolCallOutputContentItem::InputText { + text: "ok".to_string(), + }]), + success: Some(true), + duration_ms: Some(5), + } + ); + + let sender_thread_id = codex_protocol::ThreadId::default(); + let receiver_thread_id = codex_protocol::ThreadId::default(); + let collab_item = TurnItem::CollabAgentToolCall(CollabAgentToolCallItem { + id: "collab-1".to_string(), + tool: CoreCollabAgentTool::SendInput, + status: CoreCollabAgentToolCallStatus::Completed, + sender_thread_id, + receiver_thread_ids: vec![receiver_thread_id], + receiver_agents: Vec::new(), + prompt: Some("continue".to_string()), + model: None, + reasoning_effort: None, + agents_states: [(receiver_thread_id, CoreAgentStatus::Completed(None))] + .into_iter() + .collect(), + }); + + assert_eq!( + ThreadItem::from(collab_item), + ThreadItem::CollabAgentToolCall { + id: "collab-1".to_string(), + tool: CollabAgentTool::SendInput, + status: CollabAgentToolCallStatus::Completed, + sender_thread_id: sender_thread_id.to_string(), + receiver_thread_ids: vec![receiver_thread_id.to_string()], + prompt: Some("continue".to_string()), + model: None, + reasoning_effort: None, + agents_states: [( + receiver_thread_id.to_string(), + CollabAgentState { + status: CollabAgentStatus::Completed, + message: None, + }, + )] + .into_iter() + .collect(), + } + ); + + let sub_agent_activity_item = TurnItem::SubAgentActivity(SubAgentActivityItem { + id: "activity-1".to_string(), + kind: CoreSubAgentActivityKind::Interrupted, + agent_thread_id: receiver_thread_id, + agent_path: codex_protocol::AgentPath::root() + .join("worker") + .expect("worker path"), + }); + + assert_eq!( + ThreadItem::from(sub_agent_activity_item), + ThreadItem::SubAgentActivity { + id: "activity-1".to_string(), + kind: SubAgentActivityKind::Interrupted, + agent_thread_id: receiver_thread_id.to_string(), + agent_path: "/root/worker".to_string(), + } + ); + let search_item = TurnItem::WebSearch(WebSearchItem { id: "search-1".to_string(), query: "docs".to_string(), diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index c4ee26d96d0e..baaac07a47bd 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -1,3 +1,7 @@ +use crate::AgentPath; +use crate::ThreadId; +use crate::dynamic_tools::DynamicToolCallOutputContentItem; +use crate::dynamic_tools::DynamicToolCallRequest; use crate::mcp::CallToolResult; use crate::memory_citation::MemoryCitation; use crate::models::ContentItem; @@ -5,11 +9,31 @@ use crate::models::ImageDetail; use crate::models::MessagePhase; use crate::models::ResponseItem; use crate::models::WebSearchAction; +use crate::openai_models::ReasoningEffort as ReasoningEffortConfig; +use crate::parse_command::ParsedCommand; 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::CollabAgentRef; +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::ExecCommandSource; +use crate::protocol::ExecCommandStatus; use crate::protocol::FileChange; use crate::protocol::ImageGenerationEndEvent; use crate::protocol::McpInvocation; @@ -18,6 +42,8 @@ use crate::protocol::McpToolCallEndEvent; use crate::protocol::PatchApplyBeginEvent; use crate::protocol::PatchApplyEndEvent; use crate::protocol::PatchApplyStatus; +use crate::protocol::SubAgentActivityEvent; +use crate::protocol::SubAgentActivityKind; use crate::protocol::UserMessageEvent; use crate::protocol::ViewImageToolCallEvent; use crate::protocol::WebSearchEndEvent; @@ -46,6 +72,10 @@ pub enum TurnItem { AgentMessage(AgentMessageItem), Plan(PlanItem), Reasoning(ReasoningItem), + CommandExecution(CommandExecutionItem), + DynamicToolCall(DynamicToolCallItem), + CollabAgentToolCall(CollabAgentToolCallItem), + SubAgentActivity(SubAgentActivityItem), WebSearch(WebSearchItem), ImageView(ImageViewItem), Sleep(SleepItem), @@ -129,6 +159,151 @@ pub struct ReasoningItem { pub raw_content: Vec, } +#[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub enum CommandExecutionStatus { + InProgress, + Completed, + Failed, + 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)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub struct CommandExecutionItem { + pub id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub process_id: Option, + pub command: Vec, + pub cwd: PathUri, + pub parsed_cmd: Vec, + pub source: ExecCommandSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub interaction_input: Option, + pub status: CommandExecutionStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub stdout: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub stderr: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub aggregated_output: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub exit_code: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(type = "string", optional)] + pub duration: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub formatted_output: Option, +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub enum DynamicToolCallStatus { + InProgress, + Completed, + Failed, +} + +#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub struct DynamicToolCallItem { + pub id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub namespace: Option, + pub tool: String, + pub arguments: serde_json::Value, + pub status: DynamicToolCallStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub content_items: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub success: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub error: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(type = "string", optional)] + pub duration: Option, +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub enum CollabAgentTool { + SpawnAgent, + SendInput, + ResumeAgent, + Wait, + CloseAgent, +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub enum CollabAgentToolCallStatus { + InProgress, + Completed, + Failed, +} + +#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub struct CollabAgentToolCallItem { + pub id: String, + pub tool: CollabAgentTool, + pub status: CollabAgentToolCallStatus, + pub sender_thread_id: ThreadId, + #[serde(default)] + pub receiver_thread_ids: Vec, + #[serde(default)] + pub receiver_agents: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub prompt: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub reasoning_effort: Option, + #[serde(default)] + pub agents_states: HashMap, +} + +#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +pub struct SubAgentActivityItem { + pub id: String, + pub kind: SubAgentActivityKind, + pub agent_thread_id: ThreadId, + pub agent_path: AgentPath, +} + #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] pub struct WebSearchItem { pub id: String, @@ -511,6 +686,551 @@ impl ReasoningItem { } } +impl CommandExecutionItem { + pub fn from_exec_command_begin_event(event: ExecCommandBeginEvent) -> Self { + Self { + id: event.call_id, + process_id: event.process_id, + command: event.command, + cwd: event.cwd, + parsed_cmd: event.parsed_cmd, + source: event.source, + interaction_input: event.interaction_input, + status: CommandExecutionStatus::InProgress, + stdout: None, + stderr: None, + aggregated_output: None, + exit_code: None, + duration: None, + formatted_output: None, + } + } + + pub fn from_exec_command_end_event(event: ExecCommandEndEvent) -> Self { + Self { + id: event.call_id, + process_id: event.process_id, + command: event.command, + cwd: event.cwd, + parsed_cmd: event.parsed_cmd, + source: event.source, + interaction_input: event.interaction_input, + status: event.status.into(), + stdout: Some(event.stdout), + stderr: Some(event.stderr), + aggregated_output: Some(event.aggregated_output), + exit_code: Some(event.exit_code), + duration: Some(event.duration), + formatted_output: Some(event.formatted_output), + } + } + + pub 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 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 fn from_dynamic_tool_call_request(event: DynamicToolCallRequest) -> Self { + Self { + id: event.call_id, + namespace: event.namespace, + tool: event.tool, + arguments: event.arguments, + status: DynamicToolCallStatus::InProgress, + content_items: None, + success: None, + error: None, + duration: None, + } + } + + pub fn from_dynamic_tool_call_response(event: DynamicToolCallResponseEvent) -> Self { + Self { + id: event.call_id, + namespace: event.namespace, + tool: event.tool, + arguments: event.arguments, + status: if event.success { + DynamicToolCallStatus::Completed + } else { + DynamicToolCallStatus::Failed + }, + content_items: Some(event.content_items), + success: Some(event.success), + error: event.error, + duration: Some(event.duration), + } + } + + pub 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 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 fn from_collab_agent_spawn_begin_event(event: CollabAgentSpawnBeginEvent) -> Self { + Self { + id: event.call_id, + tool: CollabAgentTool::SpawnAgent, + status: CollabAgentToolCallStatus::InProgress, + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: Vec::new(), + receiver_agents: Vec::new(), + prompt: Some(event.prompt), + model: Some(event.model), + reasoning_effort: Some(event.reasoning_effort), + agents_states: HashMap::new(), + } + } + + pub fn from_collab_agent_spawn_end_event(event: CollabAgentSpawnEndEvent) -> Self { + let receiver_thread_ids = event.new_thread_id.into_iter().collect::>(); + let receiver_agents = receiver_thread_ids + .first() + .copied() + .map(|thread_id| CollabAgentRef { + thread_id, + agent_nickname: event.new_agent_nickname, + agent_role: event.new_agent_role, + }) + .into_iter() + .collect(); + let agents_states = receiver_thread_ids + .first() + .copied() + .map(|thread_id| [(thread_id, event.status.clone())].into_iter().collect()) + .unwrap_or_default(); + Self { + id: event.call_id, + tool: CollabAgentTool::SpawnAgent, + status: collab_tool_call_status(&event.status, !receiver_thread_ids.is_empty()), + sender_thread_id: event.sender_thread_id, + receiver_thread_ids, + receiver_agents, + prompt: Some(event.prompt), + model: Some(event.model), + reasoning_effort: Some(event.reasoning_effort), + agents_states, + } + } + + pub fn from_collab_agent_interaction_begin_event( + event: CollabAgentInteractionBeginEvent, + ) -> Self { + Self { + id: event.call_id, + tool: CollabAgentTool::SendInput, + status: CollabAgentToolCallStatus::InProgress, + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: vec![event.receiver_thread_id], + receiver_agents: Vec::new(), + prompt: Some(event.prompt), + model: None, + reasoning_effort: None, + agents_states: HashMap::new(), + } + } + + pub fn from_collab_agent_interaction_end_event(event: CollabAgentInteractionEndEvent) -> Self { + let receiver_agent = CollabAgentRef { + thread_id: event.receiver_thread_id, + agent_nickname: event.receiver_agent_nickname, + agent_role: event.receiver_agent_role, + }; + Self { + id: event.call_id, + tool: CollabAgentTool::SendInput, + status: collab_tool_call_status(&event.status, true), + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: vec![event.receiver_thread_id], + receiver_agents: vec![receiver_agent], + prompt: Some(event.prompt), + model: None, + reasoning_effort: None, + agents_states: [(event.receiver_thread_id, event.status)] + .into_iter() + .collect(), + } + } + + pub fn from_collab_waiting_begin_event(event: CollabWaitingBeginEvent) -> Self { + Self { + id: event.call_id, + tool: CollabAgentTool::Wait, + status: CollabAgentToolCallStatus::InProgress, + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: event.receiver_thread_ids, + receiver_agents: event.receiver_agents, + prompt: None, + model: None, + reasoning_effort: None, + agents_states: HashMap::new(), + } + } + + pub fn from_collab_waiting_end_event(event: CollabWaitingEndEvent) -> Self { + Self { + id: event.call_id, + tool: CollabAgentTool::Wait, + status: if event + .statuses + .values() + .any(|status| matches!(status, AgentStatus::Errored(_) | AgentStatus::NotFound)) + { + CollabAgentToolCallStatus::Failed + } else { + CollabAgentToolCallStatus::Completed + }, + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: event.statuses.keys().copied().collect(), + receiver_agents: event + .agent_statuses + .iter() + .map(|entry| CollabAgentRef { + thread_id: entry.thread_id, + agent_nickname: entry.agent_nickname.clone(), + agent_role: entry.agent_role.clone(), + }) + .collect(), + prompt: None, + model: None, + reasoning_effort: None, + agents_states: event.statuses, + } + } + + pub fn from_collab_close_begin_event(event: CollabCloseBeginEvent) -> Self { + Self { + id: event.call_id, + tool: CollabAgentTool::CloseAgent, + status: CollabAgentToolCallStatus::InProgress, + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: vec![event.receiver_thread_id], + receiver_agents: Vec::new(), + prompt: None, + model: None, + reasoning_effort: None, + agents_states: HashMap::new(), + } + } + + pub fn from_collab_close_end_event(event: CollabCloseEndEvent) -> Self { + let receiver_agent = CollabAgentRef { + thread_id: event.receiver_thread_id, + agent_nickname: event.receiver_agent_nickname, + agent_role: event.receiver_agent_role, + }; + Self { + id: event.call_id, + tool: CollabAgentTool::CloseAgent, + status: collab_tool_call_status(&event.status, true), + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: vec![event.receiver_thread_id], + receiver_agents: vec![receiver_agent], + prompt: None, + model: None, + reasoning_effort: None, + agents_states: [(event.receiver_thread_id, event.status)] + .into_iter() + .collect(), + } + } + + pub fn from_collab_resume_begin_event(event: CollabResumeBeginEvent) -> Self { + let receiver_agent = CollabAgentRef { + thread_id: event.receiver_thread_id, + agent_nickname: event.receiver_agent_nickname, + agent_role: event.receiver_agent_role, + }; + Self { + id: event.call_id, + tool: CollabAgentTool::ResumeAgent, + status: CollabAgentToolCallStatus::InProgress, + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: vec![event.receiver_thread_id], + receiver_agents: vec![receiver_agent], + prompt: None, + model: None, + reasoning_effort: None, + agents_states: HashMap::new(), + } + } + + pub fn from_collab_resume_end_event(event: CollabResumeEndEvent) -> Self { + let receiver_agent = CollabAgentRef { + thread_id: event.receiver_thread_id, + agent_nickname: event.receiver_agent_nickname, + agent_role: event.receiver_agent_role, + }; + Self { + id: event.call_id, + tool: CollabAgentTool::ResumeAgent, + status: collab_tool_call_status(&event.status, true), + sender_thread_id: event.sender_thread_id, + receiver_thread_ids: vec![event.receiver_thread_id], + receiver_agents: vec![receiver_agent], + prompt: None, + model: None, + reasoning_effort: None, + agents_states: [(event.receiver_thread_id, event.status)] + .into_iter() + .collect(), + } + } + + pub 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 = self.receiver_agent(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 + .as_ref() + .and_then(|agent| agent.agent_nickname.clone()), + receiver_agent_role: receiver_agent.and_then(|agent| agent.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 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 receiver_agent = receiver_thread_id.and_then(|id| self.receiver_agent(id)); + 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: receiver_agent + .as_ref() + .and_then(|agent| agent.agent_nickname.clone()), + new_agent_role: receiver_agent.and_then(|agent| agent.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 = self.receiver_agent(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 + .as_ref() + .and_then(|agent| agent.agent_nickname.clone()), + receiver_agent_role: receiver_agent.and_then(|agent| agent.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 = self.receiver_agent(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 + .as_ref() + .and_then(|agent| agent.agent_nickname.clone()), + receiver_agent_role: receiver_agent.and_then(|agent| agent.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 = self.receiver_agent(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 + .as_ref() + .and_then(|agent| agent.agent_nickname.clone()), + receiver_agent_role: receiver_agent.and_then(|agent| agent.agent_role), + status: self.agent_status(receiver_thread_id), + }) + }), + } + } + + fn receiver_agent(&self, thread_id: ThreadId) -> Option { + self.receiver_agents + .iter() + .find(|agent| agent.thread_id == thread_id) + .cloned() + } + + fn agent_status(&self, thread_id: ThreadId) -> AgentStatus { + self.agents_states + .get(&thread_id) + .cloned() + .unwrap_or(AgentStatus::NotFound) + } +} + +fn collab_tool_call_status(status: &AgentStatus, has_receiver: bool) -> CollabAgentToolCallStatus { + match status { + AgentStatus::Errored(_) | AgentStatus::NotFound => CollabAgentToolCallStatus::Failed, + _ if has_receiver => CollabAgentToolCallStatus::Completed, + _ => CollabAgentToolCallStatus::Failed, + } +} + +impl SubAgentActivityItem { + pub fn from_sub_agent_activity_event(event: SubAgentActivityEvent) -> Self { + Self { + id: event.event_id, + kind: event.kind, + agent_thread_id: event.agent_thread_id, + agent_path: event.agent_path, + } + } + + pub 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 { @@ -611,6 +1331,10 @@ impl TurnItem { TurnItem::AgentMessage(item) => item.id.clone(), TurnItem::Plan(item) => item.id.clone(), TurnItem::Reasoning(item) => item.id.clone(), + TurnItem::CommandExecution(item) => item.id.clone(), + TurnItem::DynamicToolCall(item) => item.id.clone(), + TurnItem::CollabAgentToolCall(item) => item.id.clone(), + TurnItem::SubAgentActivity(item) => item.id.clone(), TurnItem::WebSearch(item) => item.id.clone(), TurnItem::ImageView(item) => item.id.clone(), TurnItem::Sleep(item) => item.id.clone(), @@ -627,6 +1351,10 @@ impl TurnItem { TurnItem::HookPrompt(_) => Vec::new(), TurnItem::AgentMessage(item) => item.as_legacy_events(), TurnItem::Plan(_) => Vec::new(), + TurnItem::CommandExecution(_) + | TurnItem::DynamicToolCall(_) + | TurnItem::CollabAgentToolCall(_) => Vec::new(), + TurnItem::SubAgentActivity(_) => Vec::new(), TurnItem::WebSearch(item) => vec![item.as_legacy_event()], TurnItem::ImageView(item) => { vec![EventMsg::ViewImageToolCall(ViewImageToolCallEvent { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 96a793bcd38f..4c13601b18d6 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1808,6 +1808,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(), } } @@ -1840,6 +1850,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), } } @@ -4377,6 +4402,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; @@ -5358,6 +5387,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 { @@ -5386,6 +5548,7 @@ mod tests { let event = serde_json::from_value::(value).unwrap(); assert_eq!(event.completed_at_ms, 0); } + #[test] fn rollback_failed_error_does_not_affect_turn_status() { let event = ErrorEvent { diff --git a/codex-rs/rollout/src/persistence_metrics.rs b/codex-rs/rollout/src/persistence_metrics.rs index 249a23cc7091..8c2a7e18d238 100644 --- a/codex-rs/rollout/src/persistence_metrics.rs +++ b/codex-rs/rollout/src/persistence_metrics.rs @@ -247,6 +247,10 @@ fn turn_item_type(item: &TurnItem) -> &'static str { TurnItem::AgentMessage(_) => "agent_message", TurnItem::Plan(_) => "plan", TurnItem::Reasoning(_) => "reasoning", + TurnItem::CommandExecution(_) => "command_execution", + TurnItem::DynamicToolCall(_) => "dynamic_tool_call", + TurnItem::CollabAgentToolCall(_) => "collab_agent_tool_call", + TurnItem::SubAgentActivity(_) => "sub_agent_activity", TurnItem::WebSearch(_) => "web_search", TurnItem::ImageView(_) => "image_view", TurnItem::Sleep(_) => "sleep", From 995efb3694413e3328f302488db5baeb9714c639 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 26 Jun 2026 13:49:53 -0700 Subject: [PATCH 2/6] codex: fix CI failure on PR #30282 --- codex-rs/protocol/src/items.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index baaac07a47bd..9899bbeba120 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -909,7 +909,7 @@ impl CollabAgentToolCallItem { Self { id: event.call_id, tool: CollabAgentTool::SendInput, - status: collab_tool_call_status(&event.status, true), + status: collab_tool_call_status(&event.status, /*has_receiver*/ true), sender_thread_id: event.sender_thread_id, receiver_thread_ids: vec![event.receiver_thread_id], receiver_agents: vec![receiver_agent], @@ -992,7 +992,7 @@ impl CollabAgentToolCallItem { Self { id: event.call_id, tool: CollabAgentTool::CloseAgent, - status: collab_tool_call_status(&event.status, true), + status: collab_tool_call_status(&event.status, /*has_receiver*/ true), sender_thread_id: event.sender_thread_id, receiver_thread_ids: vec![event.receiver_thread_id], receiver_agents: vec![receiver_agent], @@ -1034,7 +1034,7 @@ impl CollabAgentToolCallItem { Self { id: event.call_id, tool: CollabAgentTool::ResumeAgent, - status: collab_tool_call_status(&event.status, true), + status: collab_tool_call_status(&event.status, /*has_receiver*/ true), sender_thread_id: event.sender_thread_id, receiver_thread_ids: vec![event.receiver_thread_id], receiver_agents: vec![receiver_agent], From 01baf68dc7a70edaae9f24bbb01accaf18e9bf52 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 26 Jun 2026 14:12:36 -0700 Subject: [PATCH 3/6] codex: move legacy fanout out of schema PR --- codex-rs/protocol/src/items.rs | 572 ------------------------------ codex-rs/protocol/src/protocol.rs | 163 --------- 2 files changed, 735 deletions(-) diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index 9899bbeba120..97c8bd81f62a 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -1,7 +1,6 @@ use crate::AgentPath; use crate::ThreadId; use crate::dynamic_tools::DynamicToolCallOutputContentItem; -use crate::dynamic_tools::DynamicToolCallRequest; use crate::mcp::CallToolResult; use crate::memory_citation::MemoryCitation; use crate::models::ContentItem; @@ -15,25 +14,10 @@ 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::CollabAgentRef; -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::ExecCommandSource; -use crate::protocol::ExecCommandStatus; use crate::protocol::FileChange; use crate::protocol::ImageGenerationEndEvent; use crate::protocol::McpInvocation; @@ -42,7 +26,6 @@ use crate::protocol::McpToolCallEndEvent; use crate::protocol::PatchApplyBeginEvent; use crate::protocol::PatchApplyEndEvent; use crate::protocol::PatchApplyStatus; -use crate::protocol::SubAgentActivityEvent; use crate::protocol::SubAgentActivityKind; use crate::protocol::UserMessageEvent; use crate::protocol::ViewImageToolCallEvent; @@ -169,16 +152,6 @@ 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)] #[serde(rename_all = "camelCase")] #[ts(rename_all = "camelCase")] @@ -686,551 +659,6 @@ impl ReasoningItem { } } -impl CommandExecutionItem { - pub fn from_exec_command_begin_event(event: ExecCommandBeginEvent) -> Self { - Self { - id: event.call_id, - process_id: event.process_id, - command: event.command, - cwd: event.cwd, - parsed_cmd: event.parsed_cmd, - source: event.source, - interaction_input: event.interaction_input, - status: CommandExecutionStatus::InProgress, - stdout: None, - stderr: None, - aggregated_output: None, - exit_code: None, - duration: None, - formatted_output: None, - } - } - - pub fn from_exec_command_end_event(event: ExecCommandEndEvent) -> Self { - Self { - id: event.call_id, - process_id: event.process_id, - command: event.command, - cwd: event.cwd, - parsed_cmd: event.parsed_cmd, - source: event.source, - interaction_input: event.interaction_input, - status: event.status.into(), - stdout: Some(event.stdout), - stderr: Some(event.stderr), - aggregated_output: Some(event.aggregated_output), - exit_code: Some(event.exit_code), - duration: Some(event.duration), - formatted_output: Some(event.formatted_output), - } - } - - pub 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 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 fn from_dynamic_tool_call_request(event: DynamicToolCallRequest) -> Self { - Self { - id: event.call_id, - namespace: event.namespace, - tool: event.tool, - arguments: event.arguments, - status: DynamicToolCallStatus::InProgress, - content_items: None, - success: None, - error: None, - duration: None, - } - } - - pub fn from_dynamic_tool_call_response(event: DynamicToolCallResponseEvent) -> Self { - Self { - id: event.call_id, - namespace: event.namespace, - tool: event.tool, - arguments: event.arguments, - status: if event.success { - DynamicToolCallStatus::Completed - } else { - DynamicToolCallStatus::Failed - }, - content_items: Some(event.content_items), - success: Some(event.success), - error: event.error, - duration: Some(event.duration), - } - } - - pub 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 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 fn from_collab_agent_spawn_begin_event(event: CollabAgentSpawnBeginEvent) -> Self { - Self { - id: event.call_id, - tool: CollabAgentTool::SpawnAgent, - status: CollabAgentToolCallStatus::InProgress, - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: Vec::new(), - receiver_agents: Vec::new(), - prompt: Some(event.prompt), - model: Some(event.model), - reasoning_effort: Some(event.reasoning_effort), - agents_states: HashMap::new(), - } - } - - pub fn from_collab_agent_spawn_end_event(event: CollabAgentSpawnEndEvent) -> Self { - let receiver_thread_ids = event.new_thread_id.into_iter().collect::>(); - let receiver_agents = receiver_thread_ids - .first() - .copied() - .map(|thread_id| CollabAgentRef { - thread_id, - agent_nickname: event.new_agent_nickname, - agent_role: event.new_agent_role, - }) - .into_iter() - .collect(); - let agents_states = receiver_thread_ids - .first() - .copied() - .map(|thread_id| [(thread_id, event.status.clone())].into_iter().collect()) - .unwrap_or_default(); - Self { - id: event.call_id, - tool: CollabAgentTool::SpawnAgent, - status: collab_tool_call_status(&event.status, !receiver_thread_ids.is_empty()), - sender_thread_id: event.sender_thread_id, - receiver_thread_ids, - receiver_agents, - prompt: Some(event.prompt), - model: Some(event.model), - reasoning_effort: Some(event.reasoning_effort), - agents_states, - } - } - - pub fn from_collab_agent_interaction_begin_event( - event: CollabAgentInteractionBeginEvent, - ) -> Self { - Self { - id: event.call_id, - tool: CollabAgentTool::SendInput, - status: CollabAgentToolCallStatus::InProgress, - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: vec![event.receiver_thread_id], - receiver_agents: Vec::new(), - prompt: Some(event.prompt), - model: None, - reasoning_effort: None, - agents_states: HashMap::new(), - } - } - - pub fn from_collab_agent_interaction_end_event(event: CollabAgentInteractionEndEvent) -> Self { - let receiver_agent = CollabAgentRef { - thread_id: event.receiver_thread_id, - agent_nickname: event.receiver_agent_nickname, - agent_role: event.receiver_agent_role, - }; - Self { - id: event.call_id, - tool: CollabAgentTool::SendInput, - status: collab_tool_call_status(&event.status, /*has_receiver*/ true), - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: vec![event.receiver_thread_id], - receiver_agents: vec![receiver_agent], - prompt: Some(event.prompt), - model: None, - reasoning_effort: None, - agents_states: [(event.receiver_thread_id, event.status)] - .into_iter() - .collect(), - } - } - - pub fn from_collab_waiting_begin_event(event: CollabWaitingBeginEvent) -> Self { - Self { - id: event.call_id, - tool: CollabAgentTool::Wait, - status: CollabAgentToolCallStatus::InProgress, - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: event.receiver_thread_ids, - receiver_agents: event.receiver_agents, - prompt: None, - model: None, - reasoning_effort: None, - agents_states: HashMap::new(), - } - } - - pub fn from_collab_waiting_end_event(event: CollabWaitingEndEvent) -> Self { - Self { - id: event.call_id, - tool: CollabAgentTool::Wait, - status: if event - .statuses - .values() - .any(|status| matches!(status, AgentStatus::Errored(_) | AgentStatus::NotFound)) - { - CollabAgentToolCallStatus::Failed - } else { - CollabAgentToolCallStatus::Completed - }, - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: event.statuses.keys().copied().collect(), - receiver_agents: event - .agent_statuses - .iter() - .map(|entry| CollabAgentRef { - thread_id: entry.thread_id, - agent_nickname: entry.agent_nickname.clone(), - agent_role: entry.agent_role.clone(), - }) - .collect(), - prompt: None, - model: None, - reasoning_effort: None, - agents_states: event.statuses, - } - } - - pub fn from_collab_close_begin_event(event: CollabCloseBeginEvent) -> Self { - Self { - id: event.call_id, - tool: CollabAgentTool::CloseAgent, - status: CollabAgentToolCallStatus::InProgress, - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: vec![event.receiver_thread_id], - receiver_agents: Vec::new(), - prompt: None, - model: None, - reasoning_effort: None, - agents_states: HashMap::new(), - } - } - - pub fn from_collab_close_end_event(event: CollabCloseEndEvent) -> Self { - let receiver_agent = CollabAgentRef { - thread_id: event.receiver_thread_id, - agent_nickname: event.receiver_agent_nickname, - agent_role: event.receiver_agent_role, - }; - Self { - id: event.call_id, - tool: CollabAgentTool::CloseAgent, - status: collab_tool_call_status(&event.status, /*has_receiver*/ true), - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: vec![event.receiver_thread_id], - receiver_agents: vec![receiver_agent], - prompt: None, - model: None, - reasoning_effort: None, - agents_states: [(event.receiver_thread_id, event.status)] - .into_iter() - .collect(), - } - } - - pub fn from_collab_resume_begin_event(event: CollabResumeBeginEvent) -> Self { - let receiver_agent = CollabAgentRef { - thread_id: event.receiver_thread_id, - agent_nickname: event.receiver_agent_nickname, - agent_role: event.receiver_agent_role, - }; - Self { - id: event.call_id, - tool: CollabAgentTool::ResumeAgent, - status: CollabAgentToolCallStatus::InProgress, - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: vec![event.receiver_thread_id], - receiver_agents: vec![receiver_agent], - prompt: None, - model: None, - reasoning_effort: None, - agents_states: HashMap::new(), - } - } - - pub fn from_collab_resume_end_event(event: CollabResumeEndEvent) -> Self { - let receiver_agent = CollabAgentRef { - thread_id: event.receiver_thread_id, - agent_nickname: event.receiver_agent_nickname, - agent_role: event.receiver_agent_role, - }; - Self { - id: event.call_id, - tool: CollabAgentTool::ResumeAgent, - status: collab_tool_call_status(&event.status, /*has_receiver*/ true), - sender_thread_id: event.sender_thread_id, - receiver_thread_ids: vec![event.receiver_thread_id], - receiver_agents: vec![receiver_agent], - prompt: None, - model: None, - reasoning_effort: None, - agents_states: [(event.receiver_thread_id, event.status)] - .into_iter() - .collect(), - } - } - - pub 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 = self.receiver_agent(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 - .as_ref() - .and_then(|agent| agent.agent_nickname.clone()), - receiver_agent_role: receiver_agent.and_then(|agent| agent.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 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 receiver_agent = receiver_thread_id.and_then(|id| self.receiver_agent(id)); - 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: receiver_agent - .as_ref() - .and_then(|agent| agent.agent_nickname.clone()), - new_agent_role: receiver_agent.and_then(|agent| agent.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 = self.receiver_agent(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 - .as_ref() - .and_then(|agent| agent.agent_nickname.clone()), - receiver_agent_role: receiver_agent.and_then(|agent| agent.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 = self.receiver_agent(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 - .as_ref() - .and_then(|agent| agent.agent_nickname.clone()), - receiver_agent_role: receiver_agent.and_then(|agent| agent.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 = self.receiver_agent(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 - .as_ref() - .and_then(|agent| agent.agent_nickname.clone()), - receiver_agent_role: receiver_agent.and_then(|agent| agent.agent_role), - status: self.agent_status(receiver_thread_id), - }) - }), - } - } - - fn receiver_agent(&self, thread_id: ThreadId) -> Option { - self.receiver_agents - .iter() - .find(|agent| agent.thread_id == thread_id) - .cloned() - } - - fn agent_status(&self, thread_id: ThreadId) -> AgentStatus { - self.agents_states - .get(&thread_id) - .cloned() - .unwrap_or(AgentStatus::NotFound) - } -} - -fn collab_tool_call_status(status: &AgentStatus, has_receiver: bool) -> CollabAgentToolCallStatus { - match status { - AgentStatus::Errored(_) | AgentStatus::NotFound => CollabAgentToolCallStatus::Failed, - _ if has_receiver => CollabAgentToolCallStatus::Completed, - _ => CollabAgentToolCallStatus::Failed, - } -} - -impl SubAgentActivityItem { - pub fn from_sub_agent_activity_event(event: SubAgentActivityEvent) -> Self { - Self { - id: event.event_id, - kind: event.kind, - agent_thread_id: event.agent_thread_id, - agent_path: event.agent_path, - } - } - - pub 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 { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 4c13601b18d6..96a793bcd38f 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1808,16 +1808,6 @@ 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(), } } @@ -1850,21 +1840,6 @@ 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), } } @@ -4402,10 +4377,6 @@ 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; @@ -5387,139 +5358,6 @@ 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 { @@ -5548,7 +5386,6 @@ mod tests { let event = serde_json::from_value::(value).unwrap(); assert_eq!(event.completed_at_ms, 0); } - #[test] fn rollback_failed_error_does_not_affect_turn_status() { let event = ErrorEvent { From ac158c8b86d79bb5964bee66a18f75d00204dd69 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 26 Jun 2026 14:17:44 -0700 Subject: [PATCH 4/6] codex: use rollout casing for core turn items --- codex-rs/protocol/src/items.rs | 20 ++++---------------- 1 file changed, 4 insertions(+), 16 deletions(-) diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index 97c8bd81f62a..c31957344637 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -143,8 +143,7 @@ pub struct ReasoningItem { } #[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] +#[serde(rename_all = "snake_case")] pub enum CommandExecutionStatus { InProgress, Completed, @@ -153,8 +152,6 @@ pub enum CommandExecutionStatus { } #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] pub struct CommandExecutionItem { pub id: String, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -189,8 +186,7 @@ pub struct CommandExecutionItem { } #[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] +#[serde(rename_all = "snake_case")] pub enum DynamicToolCallStatus { InProgress, Completed, @@ -198,8 +194,6 @@ pub enum DynamicToolCallStatus { } #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] pub struct DynamicToolCallItem { pub id: String, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -223,8 +217,7 @@ pub struct DynamicToolCallItem { } #[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] +#[serde(rename_all = "snake_case")] pub enum CollabAgentTool { SpawnAgent, SendInput, @@ -234,8 +227,7 @@ pub enum CollabAgentTool { } #[derive(Debug, Clone, Copy, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] +#[serde(rename_all = "snake_case")] pub enum CollabAgentToolCallStatus { InProgress, Completed, @@ -243,8 +235,6 @@ pub enum CollabAgentToolCallStatus { } #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] pub struct CollabAgentToolCallItem { pub id: String, pub tool: CollabAgentTool, @@ -268,8 +258,6 @@ pub struct CollabAgentToolCallItem { } #[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)] -#[serde(rename_all = "camelCase")] -#[ts(rename_all = "camelCase")] pub struct SubAgentActivityItem { pub id: String, pub kind: SubAgentActivityKind, From d2d58828cc9caba140e599fd86de57451200519a Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 26 Jun 2026 14:42:55 -0700 Subject: [PATCH 5/6] fix(app-server-protocol): replay canonical tool items --- .../src/protocol/thread_history.rs | 146 +++++++++++------- 1 file changed, 94 insertions(+), 52 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 852167ecd44f..2d9c6b6a1a48 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -574,70 +574,39 @@ impl ThreadHistoryBuilder { } fn handle_item_started(&mut self, payload: &ItemStartedEvent) { - match &payload.item { - codex_protocol::items::TurnItem::Plan(plan) => { - if plan.text.is_empty() { - return; - } - self.upsert_item_in_turn_id( - &payload.turn_id, - ThreadItem::from(payload.item.clone()), - ); - } - codex_protocol::items::TurnItem::Sleep(_) => { - self.upsert_item_in_turn_id( - &payload.turn_id, - ThreadItem::from(payload.item.clone()), - ); - } - codex_protocol::items::TurnItem::UserMessage(_) - | codex_protocol::items::TurnItem::HookPrompt(_) - | codex_protocol::items::TurnItem::AgentMessage(_) - | codex_protocol::items::TurnItem::Reasoning(_) - | codex_protocol::items::TurnItem::CommandExecution(_) - | codex_protocol::items::TurnItem::DynamicToolCall(_) - | codex_protocol::items::TurnItem::CollabAgentToolCall(_) - | codex_protocol::items::TurnItem::SubAgentActivity(_) - | codex_protocol::items::TurnItem::WebSearch(_) - | codex_protocol::items::TurnItem::ImageView(_) - | codex_protocol::items::TurnItem::ImageGeneration(_) - | codex_protocol::items::TurnItem::FileChange(_) - | codex_protocol::items::TurnItem::McpToolCall(_) - | codex_protocol::items::TurnItem::ContextCompaction(_) => {} - } + self.handle_materialized_item_lifecycle(&payload.turn_id, &payload.item); } fn handle_item_completed(&mut self, payload: &ItemCompletedEvent) { - match &payload.item { - codex_protocol::items::TurnItem::Plan(plan) => { - if plan.text.is_empty() { - return; - } - self.upsert_item_in_turn_id( - &payload.turn_id, - ThreadItem::from(payload.item.clone()), - ); - } - codex_protocol::items::TurnItem::Sleep(_) => { - self.upsert_item_in_turn_id( - &payload.turn_id, - ThreadItem::from(payload.item.clone()), - ); - } + self.handle_materialized_item_lifecycle(&payload.turn_id, &payload.item); + } + + fn handle_materialized_item_lifecycle( + &mut self, + turn_id: &str, + item: &codex_protocol::items::TurnItem, + ) { + let should_upsert = match item { + codex_protocol::items::TurnItem::Plan(plan) => !plan.text.is_empty(), + codex_protocol::items::TurnItem::Sleep(_) + | codex_protocol::items::TurnItem::CommandExecution(_) + | codex_protocol::items::TurnItem::DynamicToolCall(_) + | codex_protocol::items::TurnItem::CollabAgentToolCall(_) + | codex_protocol::items::TurnItem::SubAgentActivity(_) => true, codex_protocol::items::TurnItem::UserMessage(_) | codex_protocol::items::TurnItem::HookPrompt(_) | codex_protocol::items::TurnItem::AgentMessage(_) | codex_protocol::items::TurnItem::Reasoning(_) - | codex_protocol::items::TurnItem::CommandExecution(_) - | codex_protocol::items::TurnItem::DynamicToolCall(_) - | codex_protocol::items::TurnItem::CollabAgentToolCall(_) - | codex_protocol::items::TurnItem::SubAgentActivity(_) | codex_protocol::items::TurnItem::WebSearch(_) | codex_protocol::items::TurnItem::ImageView(_) | codex_protocol::items::TurnItem::ImageGeneration(_) | codex_protocol::items::TurnItem::FileChange(_) | codex_protocol::items::TurnItem::McpToolCall(_) - | codex_protocol::items::TurnItem::ContextCompaction(_) => {} + | codex_protocol::items::TurnItem::ContextCompaction(_) => false, + }; + + if should_upsert { + self.upsert_item_in_turn_id(turn_id, ThreadItem::from(item.clone())); } } @@ -1581,6 +1550,8 @@ mod tests { use crate::protocol::v2::CommandExecutionSource; use codex_protocol::ThreadId; use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynamicToolCallOutputContentItem; + use codex_protocol::items::CommandExecutionItem as CoreCommandExecutionItem; + use codex_protocol::items::CommandExecutionStatus as CoreCommandExecutionStatus; use codex_protocol::items::HookPromptFragment as CoreHookPromptFragment; use codex_protocol::items::SleepItem as CoreSleepItem; use codex_protocol::items::TurnItem as CoreTurnItem; @@ -1875,6 +1846,77 @@ mod tests { ); } + #[test] + fn rebuilds_command_execution_item_from_persisted_completion() { + let turn_id = "turn-1"; + let thread_id = ThreadId::new(); + let command_item = CoreTurnItem::CommandExecution(CoreCommandExecutionItem { + id: "exec-1".to_string(), + process_id: Some("pid-1".to_string()), + command: vec!["echo".to_string(), "hello world".to_string()], + cwd: test_path_buf("/tmp").abs().into(), + parsed_cmd: vec![ParsedCommand::Unknown { + cmd: "echo hello world".to_string(), + }], + source: ExecCommandSource::Agent, + interaction_input: None, + status: CoreCommandExecutionStatus::Completed, + stdout: Some("hello world\n".to_string()), + stderr: Some(String::new()), + aggregated_output: Some("hello world\n".to_string()), + exit_code: Some(0), + duration: Some(Duration::from_millis(12)), + formatted_output: Some("hello world\n".to_string()), + }); + let events = vec![ + EventMsg::TurnStarted(TurnStartedEvent { + turn_id: turn_id.to_string(), + trace_id: None, + started_at: None, + model_context_window: None, + collaboration_mode_kind: Default::default(), + }), + EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id, + turn_id: turn_id.to_string(), + item: command_item, + completed_at_ms: 1_000, + }), + EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: turn_id.to_string(), + last_agent_message: None, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }), + ]; + + let items = events + .into_iter() + .map(RolloutItem::EventMsg) + .collect::>(); + let turns = build_turns_from_rollout_items(&items); + + assert_eq!(turns.len(), 1); + assert_eq!( + turns[0].items, + vec![ThreadItem::CommandExecution { + id: "exec-1".to_string(), + command: "echo 'hello world'".to_string(), + cwd: test_path_buf("/tmp").abs().into(), + process_id: Some("pid-1".to_string()), + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::Completed, + command_actions: vec![CommandAction::Unknown { + command: "echo hello world".to_string(), + }], + aggregated_output: Some("hello world\n".to_string()), + exit_code: Some(0), + duration_ms: Some(12), + }] + ); + } + #[test] fn preserves_user_message_client_id_from_legacy_event() { let turn_id = "turn-1"; From 2445ed550ff8b562009f519b0d751f108ab76a78 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 26 Jun 2026 16:31:39 -0700 Subject: [PATCH 6/6] codex: fix CI failure on PR #30282 --- codex-rs/core/src/session/tests.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 46f7ca6e60c0..d6336acd57a8 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -7016,6 +7016,7 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() { dynamic_tools: Vec::new(), selected_capability_roots: Vec::new(), multi_agent_version: None, + history_mode: Default::default(), initial_window_id: Uuid::now_v7().to_string(), metadata: ThreadPersistenceMetadata { cwd: Some(config.cwd.to_path_buf()),