feat(core): emit canonical command execution items (#31297)

This PR depends on [#31296](https://github.com/openai/codex/pull/31296)
for the canonical-to-legacy event mappings.

## Description

This PR makes command execution emit canonical
`TurnItem::CommandExecution` lifecycle from both the shell tool path and
user `/shell` commands.

App-server v2 consumes the canonical command items directly and ignores
the mapped `ExecCommandBegin` / `ExecCommandEnd` compatibility events,
so clients still receive one command item lifecycle.

`UnifiedExecInteraction` stays on the legacy path because
`TerminalInteraction` is still the v2 surface for stdin and poll events.
Emitting a command item there would render the same wait twice.

## Why

This is the first live producer migration after the compatibility
mappings in #31296. Keeping command execution separate makes the unified
exec exception reviewable without mixing in dynamic tools or multi-agent
behavior.

## What changed

- Emit canonical command execution items from shell tool events and user
shell commands.
- Preserve the existing unified exec interaction carveout.
- Move app-server command deduplication and completion bookkeeping onto
canonical item events.
- Update unified exec coverage to assert the completed command item.

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -90,6 +90,7 @@ use codex_core::ThreadManager;
 use codex_core::review_format::format_review_findings_block;
 use codex_core::review_prompts;
 use codex_protocol::ThreadId;
+use codex_protocol::items::TurnItem as CoreTurnItem;
 use codex_protocol::items::parse_hook_prompt_message;
 use codex_protocol::models::AdditionalPermissionProfile as CoreAdditionalPermissionProfile;
 use codex_protocol::plan_tool::UpdatePlanArgs;
@@ -1027,10 +1028,37 @@ pub(crate) async fn apply_bespoke_event_handling(
                 .send_server_notification(ServerNotification::ItemCompleted(completed))
                 .await;
         }
-        msg @ (EventMsg::ItemStarted(_)
-        | EventMsg::ItemCompleted(_)
-        | EventMsg::PatchApplyUpdated(_)
-        | EventMsg::TerminalInteraction(_)) => {
+        EventMsg::ItemStarted(event) => {
+            let should_emit = match &event.item {
+                // Approval and guardian flows can emit the command start notification before core
+                // emits the canonical item. Reuse the same set to suppress that duplicate.
+                CoreTurnItem::CommandExecution(item) => thread_state
+                    .lock()
+                    .await
+                    .turn_summary
+                    .command_execution_started
+                    .insert(item.id.clone()),
+                _ => true,
+            };
+            if should_emit {
+                let notification = item_event_to_server_notification(
+                    EventMsg::ItemStarted(event),
+                    &conversation_id.to_string(),
+                    &event_turn_id,
+                );
+                outgoing.send_server_notification(notification).await;
+            }
+        }
+        EventMsg::ItemCompleted(event) => {
+            apply_canonical_item_completed_side_effects(&thread_state, &event.item).await;
+            let notification = item_event_to_server_notification(
+                EventMsg::ItemCompleted(event),
+                &conversation_id.to_string(),
+                &event_turn_id,
+            );
+            outgoing.send_server_notification(notification).await;
+        }
+        msg @ (EventMsg::PatchApplyUpdated(_) | EventMsg::TerminalInteraction(_)) => {
             let notification = item_event_to_server_notification(
                 msg,
                 &conversation_id.to_string(),
@@ -1106,32 +1134,10 @@ pub(crate) async fn apply_bespoke_event_handling(
             // Core still fans out these deprecated events for legacy clients;
             // v2 clients receive the canonical FileChange item instead.
         }
-        EventMsg::ExecCommandBegin(exec_command_begin_event) => {
-            if matches!(
-                exec_command_begin_event.source,
-                codex_protocol::protocol::ExecCommandSource::UnifiedExecInteraction
-            ) {
-                // TerminalInteraction is the v2 surface for unified exec
-                // stdin/poll events. Suppress the legacy CommandExecution
-                // item so clients do not render the same wait twice.
-                return;
-            }
-            let item_id = exec_command_begin_event.call_id.clone();
-            let first_start = {
-                let mut state = thread_state.lock().await;
-                state
-                    .turn_summary
-                    .command_execution_started
-                    .insert(item_id.clone())
-            };
-            if first_start {
-                let notification = item_event_to_server_notification(
-                    EventMsg::ExecCommandBegin(exec_command_begin_event),
-                    &conversation_id.to_string(),
-                    &event_turn_id,
-                );
-                outgoing.send_server_notification(notification).await;
-            }
+        EventMsg::ExecCommandBegin(_) | EventMsg::ExecCommandEnd(_) => {
+            // Deprecated command-execution events are still fanned out for raw-event and rollout
+            // compatibility consumers. App-server v2 receives the canonical CommandExecution
+            // item lifecycle instead.
         }
         EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event) => {
             let notification = item_event_to_server_notification(
@@ -1141,31 +1147,6 @@ pub(crate) async fn apply_bespoke_event_handling(
             );
             outgoing.send_server_notification(notification).await;
         }
-        EventMsg::ExecCommandEnd(exec_command_end_event) => {
-            let call_id = exec_command_end_event.call_id.clone();
-            {
-                let mut state = thread_state.lock().await;
-                state
-                    .turn_summary
-                    .command_execution_started
-                    .remove(&call_id);
-            }
-            if matches!(
-                exec_command_end_event.source,
-                codex_protocol::protocol::ExecCommandSource::UnifiedExecInteraction
-            ) {
-                // The paired begin event is suppressed above; keep the
-                // completion out of v2 as well so no orphan legacy item is
-                // emitted for unified exec interactions.
-                return;
-            }
-            let notification = item_event_to_server_notification(
-                EventMsg::ExecCommandEnd(exec_command_end_event),
-                &conversation_id.to_string(),
-                &event_turn_id,
-            );
-            outgoing.send_server_notification(notification).await;
-        }
         // If this is a TurnAborted, reply to any pending interrupt requests.
         EventMsg::TurnAborted(turn_aborted_event) => {
             // All per-thread requests are bound to a turn, so abort them.
@@ -1369,6 +1350,20 @@ async fn emit_turn_completed_with_status(
         .await;
 }
 
+async fn apply_canonical_item_completed_side_effects(
+    thread_state: &Arc<Mutex<ThreadState>>,
+    item: &CoreTurnItem,
+) {
+    if let CoreTurnItem::CommandExecution(item) = item {
+        thread_state
+            .lock()
+            .await
+            .turn_summary
+            .command_execution_started
+            .remove(&item.id);
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn start_command_execution_item(
     conversation_id: &ThreadId,

codex-rs/core/src/tasks/user_shell.rs

@@ -25,16 +25,15 @@ use crate::tools::runtimes::RuntimePathPrepends;
 use crate::tools::runtimes::apply_package_path_prepend;
 use crate::tools::runtimes::maybe_wrap_shell_lc_with_snapshot;
 use crate::tools::runtimes::strip_managed_proxy_env;
-use crate::turn_timing::now_unix_timestamp_ms;
 use crate::user_shell_command::user_shell_command_record_item;
 use codex_protocol::exec_output::ExecToolCallOutput;
 use codex_protocol::exec_output::StreamOutput;
+use codex_protocol::items::CommandExecutionItem;
+use codex_protocol::items::CommandExecutionStatus;
+use codex_protocol::items::TurnItem;
 use codex_protocol::protocol::ErrorEvent;
 use codex_protocol::protocol::EventMsg;
-use codex_protocol::protocol::ExecCommandBeginEvent;
-use codex_protocol::protocol::ExecCommandEndEvent;
 use codex_protocol::protocol::ExecCommandSource;
-use codex_protocol::protocol::ExecCommandStatus;
 use codex_protocol::protocol::TurnStartedEvent;
 use codex_sandboxing::SandboxType;
 use codex_shell_command::parse_command::parse_command;
@@ -181,18 +180,23 @@ pub(crate) async fn execute_user_shell_command(
 
     let parsed_cmd = parse_command(&display_command);
     session
-        .send_event(
+        .emit_turn_item_started(
             turn_context.as_ref(),
-            EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
-                call_id: call_id.clone(),
+            &TurnItem::CommandExecution(CommandExecutionItem {
+                id: call_id.clone(),
                 process_id: None,
-                turn_id: turn_context.sub_id.clone(),
-                started_at_ms: now_unix_timestamp_ms(),
                 command: display_command.clone(),
                 cwd: cwd.clone().into(),
                 parsed_cmd: parsed_cmd.clone(),
                 source: ExecCommandSource::UserShell,
                 interaction_input: None,
+                status: CommandExecutionStatus::InProgress,
+                stdout: None,
+                stderr: None,
+                aggregated_output: None,
+                exit_code: None,
+                duration: None,
+                formatted_output: None,
             }),
         )
         .await;
@@ -259,57 +263,53 @@ pub(crate) async fn execute_user_shell_command(
             )
             .await;
             session
-                .send_event(
+                .emit_turn_item_completed(
                     turn_context.as_ref(),
-                    EventMsg::ExecCommandEnd(ExecCommandEndEvent {
-                        call_id,
+                    TurnItem::CommandExecution(CommandExecutionItem {
+                        id: call_id,
                         process_id: None,
-                        turn_id: turn_context.sub_id.clone(),
-                        completed_at_ms: now_unix_timestamp_ms(),
                         command: display_command.clone(),
                         cwd: cwd.clone().into(),
                         parsed_cmd: parsed_cmd.clone(),
                         source: ExecCommandSource::UserShell,
                         interaction_input: None,
-                        stdout: String::new(),
-                        stderr: aborted_message.clone(),
-                        aggregated_output: aborted_message.clone(),
-                        exit_code: -1,
-                        duration: Duration::ZERO,
-                        formatted_output: aborted_message,
-                        status: ExecCommandStatus::Failed,
+                        status: CommandExecutionStatus::Failed,
+                        stdout: Some(String::new()),
+                        stderr: Some(aborted_message.clone()),
+                        aggregated_output: Some(aborted_message.clone()),
+                        exit_code: Some(-1),
+                        duration: Some(Duration::ZERO),
+                        formatted_output: Some(aborted_message),
                     }),
                 )
                 .await;
         }
         Ok(Ok(output)) => {
             session
-                .send_event(
+                .emit_turn_item_completed(
                     turn_context.as_ref(),
-                    EventMsg::ExecCommandEnd(ExecCommandEndEvent {
-                        call_id: call_id.clone(),
+                    TurnItem::CommandExecution(CommandExecutionItem {
+                        id: call_id.clone(),
                         process_id: None,
-                        turn_id: turn_context.sub_id.clone(),
-                        completed_at_ms: now_unix_timestamp_ms(),
                         command: display_command.clone(),
                         cwd: cwd.clone().into(),
                         parsed_cmd: parsed_cmd.clone(),
                         source: ExecCommandSource::UserShell,
                         interaction_input: None,
-                        stdout: output.stdout.text.clone(),
-                        stderr: output.stderr.text.clone(),
-                        aggregated_output: output.aggregated_output.text.clone(),
-                        exit_code: output.exit_code,
-                        duration: output.duration,
-                        formatted_output: format_exec_output_str(
+                        status: if output.exit_code == 0 {
+                            CommandExecutionStatus::Completed
+                        } else {
+                            CommandExecutionStatus::Failed
+                        },
+                        stdout: Some(output.stdout.text.clone()),
+                        stderr: Some(output.stderr.text.clone()),
+                        aggregated_output: Some(output.aggregated_output.text.clone()),
+                        exit_code: Some(output.exit_code),
+                        duration: Some(output.duration),
+                        formatted_output: Some(format_exec_output_str(
                             &output,
                             turn_context.model_info.truncation_policy.into(),
-                        ),
-                        status: if output.exit_code == 0 {
-                            ExecCommandStatus::Completed
-                        } else {
-                            ExecCommandStatus::Failed
-                        },
+                        )),
                     }),
                 )
                 .await;
@@ -329,28 +329,26 @@ pub(crate) async fn execute_user_shell_command(
                 timed_out: false,
             };
             session
-                .send_event(
+                .emit_turn_item_completed(
                     turn_context.as_ref(),
-                    EventMsg::ExecCommandEnd(ExecCommandEndEvent {
-                        call_id,
+                    TurnItem::CommandExecution(CommandExecutionItem {
+                        id: call_id,
                         process_id: None,
-                        turn_id: turn_context.sub_id.clone(),
-                        completed_at_ms: now_unix_timestamp_ms(),
                         command: display_command,
                         cwd: cwd.into(),
                         parsed_cmd,
                         source: ExecCommandSource::UserShell,
                         interaction_input: None,
-                        stdout: exec_output.stdout.text.clone(),
-                        stderr: exec_output.stderr.text.clone(),
-                        aggregated_output: exec_output.aggregated_output.text.clone(),
-                        exit_code: exec_output.exit_code,
-                        duration: exec_output.duration,
-                        formatted_output: format_exec_output_str(
+                        status: CommandExecutionStatus::Failed,
+                        stdout: Some(exec_output.stdout.text.clone()),
+                        stderr: Some(exec_output.stderr.text.clone()),
+                        aggregated_output: Some(exec_output.aggregated_output.text.clone()),
+                        exit_code: Some(exec_output.exit_code),
+                        duration: Some(exec_output.duration),
+                        formatted_output: Some(format_exec_output_str(
                             &exec_output,
                             turn_context.model_info.truncation_policy.into(),
-                        ),
-                        status: ExecCommandStatus::Failed,
+                        )),
                     }),
                 )
                 .await;

codex-rs/core/src/tools/events.rs

@@ -8,6 +8,8 @@ use codex_apply_patch::AppliedPatchDelta;
 use codex_protocol::error::CodexErr;
 use codex_protocol::error::SandboxErr;
 use codex_protocol::exec_output::ExecToolCallOutput;
+use codex_protocol::items::CommandExecutionItem;
+use codex_protocol::items::CommandExecutionStatus;
 use codex_protocol::items::FileChangeItem;
 use codex_protocol::items::TurnItem;
 use codex_protocol::parse_command::ParsedCommand;
@@ -102,19 +104,43 @@ pub(crate) async fn emit_exec_command_begin(
     interaction_input: Option<String>,
     process_id: Option<&str>,
 ) {
+    if matches!(source, ExecCommandSource::UnifiedExecInteraction) {
+        ctx.session
+            .send_event(
+                ctx.turn,
+                EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
+                    call_id: ctx.call_id.to_string(),
+                    process_id: process_id.map(str::to_owned),
+                    turn_id: ctx.turn.sub_id.clone(),
+                    started_at_ms: now_unix_timestamp_ms(),
+                    command: command.to_vec(),
+                    cwd: cwd.clone(),
+                    parsed_cmd: parsed_cmd.to_vec(),
+                    source,
+                    interaction_input,
+                }),
+            )
+            .await;
+        return;
+    }
     ctx.session
-        .send_event(
+        .emit_turn_item_started(
             ctx.turn,
-            EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
-                call_id: ctx.call_id.to_string(),
+            &TurnItem::CommandExecution(CommandExecutionItem {
+                id: ctx.call_id.to_string(),
                 process_id: process_id.map(str::to_owned),
-                turn_id: ctx.turn.sub_id.clone(),
-                started_at_ms: now_unix_timestamp_ms(),
                 command: command.to_vec(),
                 cwd: cwd.clone(),
                 parsed_cmd: parsed_cmd.to_vec(),
                 source,
                 interaction_input,
+                status: CommandExecutionStatus::InProgress,
+                stdout: None,
+                stderr: None,
+                aggregated_output: None,
+                exit_code: None,
+                duration: None,
+                formatted_output: None,
             }),
         )
         .await;
@@ -542,26 +568,50 @@ async fn emit_exec_end(
     exec_input: ExecCommandInput<'_>,
     exec_result: ExecCommandResult,
 ) {
+    if matches!(exec_input.source, ExecCommandSource::UnifiedExecInteraction) {
+        ctx.session
+            .send_event(
+                ctx.turn,
+                EventMsg::ExecCommandEnd(ExecCommandEndEvent {
+                    call_id: ctx.call_id.to_string(),
+                    process_id: exec_input.process_id.map(str::to_owned),
+                    turn_id: ctx.turn.sub_id.clone(),
+                    completed_at_ms: now_unix_timestamp_ms(),
+                    command: exec_input.command.to_vec(),
+                    cwd: exec_input.cwd.clone(),
+                    parsed_cmd: exec_input.parsed_cmd.to_vec(),
+                    source: exec_input.source,
+                    interaction_input: exec_input.interaction_input.map(str::to_owned),
+                    stdout: exec_result.stdout,
+                    stderr: exec_result.stderr,
+                    aggregated_output: exec_result.aggregated_output,
+                    exit_code: exec_result.exit_code,
+                    duration: exec_result.duration,
+                    formatted_output: exec_result.formatted_output,
+                    status: exec_result.status,
+                }),
+            )
+            .await;
+        return;
+    }
     ctx.session
-        .send_event(
+        .emit_turn_item_completed(
             ctx.turn,
-            EventMsg::ExecCommandEnd(ExecCommandEndEvent {
-                call_id: ctx.call_id.to_string(),
+            TurnItem::CommandExecution(CommandExecutionItem {
+                id: ctx.call_id.to_string(),
                 process_id: exec_input.process_id.map(str::to_owned),
-                turn_id: ctx.turn.sub_id.clone(),
-                completed_at_ms: now_unix_timestamp_ms(),
                 command: exec_input.command.to_vec(),
                 cwd: exec_input.cwd.clone(),
                 parsed_cmd: exec_input.parsed_cmd.to_vec(),
                 source: exec_input.source,
                 interaction_input: exec_input.interaction_input.map(str::to_owned),
-                stdout: exec_result.stdout,
-                stderr: exec_result.stderr,
-                aggregated_output: exec_result.aggregated_output,
-                exit_code: exec_result.exit_code,
-                duration: exec_result.duration,
-                formatted_output: exec_result.formatted_output,
-                status: exec_result.status,
+                status: exec_result.status.into(),
+                stdout: Some(exec_result.stdout),
+                stderr: Some(exec_result.stderr),
+                aggregated_output: Some(exec_result.aggregated_output),
+                exit_code: Some(exec_result.exit_code),
+                duration: Some(exec_result.duration),
+                formatted_output: Some(exec_result.formatted_output),
             }),
         )
         .await;

codex-rs/core/src/unified_exec/process_manager.rs

@@ -573,9 +573,8 @@ impl UnifiedExecProcessManager {
                 }
             }
         } else {
-            // Short‑lived command: emit ExecCommandEnd immediately using the
-            // same helper as the background watcher, so all end events share
-            // one implementation.
+            // Short-lived command: emit the completed command item immediately
+            // using the same helper as the background watcher.
             let finish_result = finish_deferred_network_approval_after_process_exit_for_session(
                 Some(&context.session),
                 deferred_network_approval.take(),

codex-rs/core/src/unified_exec/process_manager_tests.rs

@@ -315,21 +315,24 @@ async fn failed_initial_end_for_unstored_process_uses_fallback_output() {
 
     let event = tokio::time::timeout(Duration::from_secs(1), rx_event.recv())
         .await
-        .expect("timed out waiting for failed exec end event")
+        .expect("timed out waiting for failed command execution item")
         .expect("event channel closed");
-    let codex_protocol::protocol::EventMsg::ExecCommandEnd(end_event) = event.msg else {
-        panic!("expected ExecCommandEnd event");
+    let codex_protocol::protocol::EventMsg::ItemCompleted(completed_event) = event.msg else {
+        panic!("expected ItemCompleted event");
     };
-    assert_eq!(end_event.call_id, "call-unified-denied");
+    let codex_protocol::items::TurnItem::CommandExecution(item) = completed_event.item else {
+        panic!("expected CommandExecution item");
+    };
+    assert_eq!(item.id, "call-unified-denied");
     assert_eq!(
-        end_event.status,
-        codex_protocol::protocol::ExecCommandStatus::Failed
+        item.status,
+        codex_protocol::items::CommandExecutionStatus::Failed
     );
-    assert_eq!(end_event.exit_code, -1);
-    assert_eq!(end_event.process_id.as_deref(), Some("123"));
+    assert_eq!(item.exit_code, Some(-1));
+    assert_eq!(item.process_id.as_deref(), Some("123"));
     assert_eq!(
-        end_event.aggregated_output,
-        "PRE_DENIAL_MARKER\nNetwork access denied"
+        item.aggregated_output.as_deref(),
+        Some("PRE_DENIAL_MARKER\nNetwork access denied")
     );
 }