feat: flag exec events for user shell commands

Author: zhao-oai

codex-rs/core/src/exec.rs

@@ -6,6 +6,7 @@ use std::io;
 use std::path::Path;
 use std::path::PathBuf;
 use std::process::ExitStatus;
+use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -84,6 +85,45 @@ pub struct StdoutStream {
     pub tx_event: Sender<Event>,
 }
 
+type DeltaEventFn = dyn Fn(&str, ExecOutputStream, Vec<u8>) -> EventMsg + Send + Sync;
+
+#[derive(Clone)]
+pub struct DeltaEventBuilder {
+    inner: Arc<DeltaEventFn>,
+}
+
+impl DeltaEventBuilder {
+    pub fn exec_command() -> Self {
+        Self {
+            inner: Arc::new(|call_id, stream, chunk| {
+                EventMsg::ExecCommandOutputDelta(ExecCommandOutputDeltaEvent {
+                    call_id: call_id.to_string(),
+                    stream,
+                    chunk,
+                    is_user_shell_command: false,
+                })
+            }),
+        }
+    }
+
+    pub fn user_command() -> Self {
+        Self {
+            inner: Arc::new(|call_id, stream, chunk| {
+                EventMsg::ExecCommandOutputDelta(ExecCommandOutputDeltaEvent {
+                    call_id: call_id.to_string(),
+                    stream,
+                    chunk,
+                    is_user_shell_command: true,
+                })
+            }),
+        }
+    }
+
+    pub fn build(&self, call_id: &str, stream: ExecOutputStream, chunk: Vec<u8>) -> EventMsg {
+        (self.inner)(call_id, stream, chunk)
+    }
+}
+
 pub async fn process_exec_tool_call(
     params: ExecParams,
     sandbox_type: SandboxType,
@@ -138,6 +178,7 @@ pub(crate) async fn execute_exec_env(
     env: ExecEnv,
     sandbox_policy: &SandboxPolicy,
     stdout_stream: Option<StdoutStream>,
+    delta_event_builder: Option<DeltaEventBuilder>,
 ) -> Result<ExecToolCallOutput> {
     let ExecEnv {
         command,
@@ -161,7 +202,15 @@ pub(crate) async fn execute_exec_env(
     };
 
     let start = Instant::now();
-    let raw_output_result = exec(params, sandbox, sandbox_policy, stdout_stream).await;
+    let delta_event_builder = delta_event_builder.unwrap_or_else(DeltaEventBuilder::exec_command);
+    let raw_output_result = exec(
+        params,
+        sandbox,
+        sandbox_policy,
+        stdout_stream,
+        delta_event_builder.clone(),
+    )
+    .await;
     let duration = start.elapsed();
     finalize_exec_result(raw_output_result, sandbox, duration)
 }
@@ -434,6 +483,7 @@ async fn exec(
     sandbox: SandboxType,
     sandbox_policy: &SandboxPolicy,
     stdout_stream: Option<StdoutStream>,
+    delta_event_builder: DeltaEventBuilder,
 ) -> Result<RawExecToolCallOutput> {
     #[cfg(target_os = "windows")]
     if sandbox == SandboxType::WindowsRestrictedToken {
@@ -465,7 +515,7 @@ async fn exec(
         env,
     )
     .await?;
-    consume_truncated_output(child, timeout, stdout_stream).await
+    consume_truncated_output(child, timeout, stdout_stream, delta_event_builder).await
 }
 
 /// Consumes the output of a child process, truncating it so it is suitable for
@@ -474,6 +524,7 @@ async fn consume_truncated_output(
     mut child: Child,
     timeout: Duration,
     stdout_stream: Option<StdoutStream>,
+    delta_event_builder: DeltaEventBuilder,
 ) -> Result<RawExecToolCallOutput> {
     // Both stdout and stderr were configured with `Stdio::piped()`
     // above, therefore `take()` should normally return `Some`.  If it doesn't
@@ -497,12 +548,14 @@ async fn consume_truncated_output(
         stdout_stream.clone(),
         false,
         Some(agg_tx.clone()),
+        delta_event_builder.clone(),
     ));
     let stderr_handle = tokio::spawn(read_capped(
         BufReader::new(stderr_reader),
         stdout_stream.clone(),
         true,
         Some(agg_tx.clone()),
+        delta_event_builder.clone(),
     ));
 
     let (exit_status, timed_out) = tokio::select! {
@@ -554,6 +607,7 @@ async fn read_capped<R: AsyncRead + Unpin + Send + 'static>(
     stream: Option<StdoutStream>,
     is_stderr: bool,
     aggregate_tx: Option<Sender<Vec<u8>>>,
+    delta_event_builder: DeltaEventBuilder,
 ) -> io::Result<StreamOutput<Vec<u8>>> {
     let mut buf = Vec::with_capacity(AGGREGATE_BUFFER_INITIAL_CAPACITY);
     let mut tmp = [0u8; READ_CHUNK_SIZE];
@@ -571,15 +625,15 @@ async fn read_capped<R: AsyncRead + Unpin + Send + 'static>(
             && emitted_deltas < MAX_EXEC_OUTPUT_DELTAS_PER_CALL
         {
             let chunk = tmp[..n].to_vec();
-            let msg = EventMsg::ExecCommandOutputDelta(ExecCommandOutputDeltaEvent {
-                call_id: stream.call_id.clone(),
-                stream: if is_stderr {
+            let msg = delta_event_builder.build(
+                &stream.call_id,
+                if is_stderr {
                     ExecOutputStream::Stderr
                 } else {
                     ExecOutputStream::Stdout
                 },
                 chunk,
-            });
+            );
             let event = Event {
                 id: stream.sub_id.clone(),
                 msg,

codex-rs/core/src/sandboxing/mod.rs

@@ -165,5 +165,5 @@ pub async fn execute_env(
     policy: &SandboxPolicy,
     stdout_stream: Option<StdoutStream>,
 ) -> crate::error::Result<ExecToolCallOutput> {
-    execute_exec_env(env.clone(), policy, stdout_stream).await
+    execute_exec_env(env.clone(), policy, stdout_stream, None).await
 }

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

@@ -337,6 +337,7 @@ async fn emit_exec_end(
                 exit_code,
                 duration,
                 formatted_output,
+                is_user_shell_command: false,
             }),
         )
         .await;

codex-rs/core/src/tools/handlers/unified_exec.rs

@@ -157,6 +157,7 @@ impl ToolHandler for UnifiedExecHandler {
                 call_id: response.event_call_id.clone(),
                 stream: ExecOutputStream::Stdout,
                 chunk: response.output.as_bytes().to_vec(),
+                is_user_shell_command: false,
             };
             session
                 .send_event(turn.as_ref(), EventMsg::ExecCommandOutputDelta(delta))

codex-rs/exec/tests/event_processor_with_json_output.rs

@@ -656,6 +656,7 @@ fn exec_command_end_success_produces_completed_command_item() {
             exit_code: 0,
             duration: Duration::from_millis(5),
             formatted_output: String::new(),
+            is_user_shell_command: false,
         }),
     );
     let out_ok = ep.collect_thread_events(&end_ok);
@@ -716,6 +717,7 @@ fn exec_command_end_failure_produces_failed_command_item() {
             exit_code: 1,
             duration: Duration::from_millis(2),
             formatted_output: String::new(),
+            is_user_shell_command: false,
         }),
     );
     let out_fail = ep.collect_thread_events(&end_fail);
@@ -750,6 +752,7 @@ fn exec_command_end_without_begin_is_ignored() {
             exit_code: 0,
             duration: Duration::from_millis(1),
             formatted_output: String::new(),
+            is_user_shell_command: false,
         }),
     );
     let out = ep.collect_thread_events(&end_only);

codex-rs/protocol/src/protocol.rs

@@ -1236,6 +1236,10 @@ pub struct ExecCommandEndEvent {
     pub duration: Duration,
     /// Formatted output from the command, as seen by the model.
     pub formatted_output: String,
+    /// True when this exec was initiated directly by the user (e.g. bang command),
+    /// not by the agent/model. Defaults to false for backwards compatibility.
+    #[serde(default)]
+    pub is_user_shell_command: bool,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
@@ -1265,6 +1269,10 @@ pub struct ExecCommandOutputDeltaEvent {
     #[schemars(with = "String")]
     #[ts(type = "string")]
     pub chunk: Vec<u8>,
+    /// True when this exec was initiated directly by the user (e.g. bang command),
+    /// not by the agent/model. Defaults to false for backwards compatibility.
+    #[serde(default)]
+    pub is_user_shell_command: bool,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
@@ -1546,10 +1554,11 @@ mod tests {
             call_id: "call21".to_string(),
             stream: ExecOutputStream::Stdout,
             chunk: vec![1, 2, 3, 4, 5],
+            is_user_shell_command: false,
         };
         let serialized = serde_json::to_string(&event)?;
         assert_eq!(
-            r#"{"call_id":"call21","stream":"stdout","chunk":"AQIDBAU="}"#,
+            r#"{"call_id":"call21","stream":"stdout","chunk":"AQIDBAU=","is_user_shell_command":false}"#,
             serialized,
         );
 

codex-rs/tui/src/chatwidget.rs

@@ -785,11 +785,23 @@ impl ChatWidget {
         self.request_redraw();
     }
 
-    pub(crate) fn handle_exec_end_now(&mut self, ev: ExecCommandEndEvent) {
-        let running = self.running_commands.remove(&ev.call_id);
+    fn handle_command_end_internal(
+        &mut self,
+        call_id: String,
+        aggregated_output: String,
+        formatted_output: String,
+        exit_code: i32,
+        duration: std::time::Duration,
+        default_is_user_shell_command: bool,
+    ) {
+        let running = self.running_commands.remove(&call_id);
         let (command, parsed, is_user_shell_command) = match running {
             Some(rc) => (rc.command, rc.parsed_cmd, rc.is_user_shell_command),
-            None => (vec![ev.call_id.clone()], Vec::new(), false),
+            None => (
+                vec![call_id.clone()],
+                Vec::new(),
+                default_is_user_shell_command,
+            ),
         };
 
         let needs_new = self
@@ -800,7 +812,7 @@ impl ChatWidget {
         if needs_new {
             self.flush_active_cell();
             self.active_cell = Some(Box::new(new_active_exec_command(
-                ev.call_id.clone(),
+                call_id.clone(),
                 command,
                 parsed,
                 is_user_shell_command,
@@ -813,20 +825,40 @@ impl ChatWidget {
             .and_then(|c| c.as_any_mut().downcast_mut::<ExecCell>())
         {
             cell.complete_call(
-                &ev.call_id,
+                &call_id,
                 CommandOutput {
-                    exit_code: ev.exit_code,
-                    formatted_output: ev.formatted_output.clone(),
-                    aggregated_output: ev.aggregated_output.clone(),
+                    exit_code,
+                    formatted_output,
+                    aggregated_output,
                 },
-                ev.duration,
+                duration,
             );
             if cell.should_flush() {
                 self.flush_active_cell();
             }
         }
     }
 
+    pub(crate) fn handle_exec_end_now(&mut self, ev: ExecCommandEndEvent) {
+        let ExecCommandEndEvent {
+            call_id,
+            aggregated_output,
+            formatted_output,
+            exit_code,
+            duration,
+            is_user_shell_command,
+            ..
+        } = ev;
+        self.handle_command_end_internal(
+            call_id,
+            aggregated_output,
+            formatted_output,
+            exit_code,
+            duration,
+            is_user_shell_command,
+        );
+    }
+
     pub(crate) fn handle_patch_apply_end_now(
         &mut self,
         event: codex_core::protocol::PatchApplyEndEvent,
@@ -875,42 +907,58 @@ impl ChatWidget {
         });
     }
 
-    pub(crate) fn handle_exec_begin_now(&mut self, ev: ExecCommandBeginEvent) {
-        // Ensure the status indicator is visible while the command runs.
+    fn handle_command_begin_internal(
+        &mut self,
+        call_id: String,
+        command: Vec<String>,
+        parsed_cmd: Vec<ParsedCommand>,
+        is_user_shell_command: bool,
+    ) {
         self.running_commands.insert(
-            ev.call_id.clone(),
+            call_id.clone(),
             RunningCommand {
-                command: ev.command.clone(),
-                parsed_cmd: ev.parsed_cmd.clone(),
-                is_user_shell_command: ev.is_user_shell_command,
+                command: command.clone(),
+                parsed_cmd: parsed_cmd.clone(),
+                is_user_shell_command,
             },
         );
         if let Some(cell) = self
             .active_cell
             .as_mut()
             .and_then(|c| c.as_any_mut().downcast_mut::<ExecCell>())
             && let Some(new_exec) = cell.with_added_call(
-                ev.call_id.clone(),
-                ev.command.clone(),
-                ev.parsed_cmd.clone(),
-                ev.is_user_shell_command,
+                call_id.clone(),
+                command.clone(),
+                parsed_cmd.clone(),
+                is_user_shell_command,
             )
         {
             *cell = new_exec;
         } else {
             self.flush_active_cell();
 
             self.active_cell = Some(Box::new(new_active_exec_command(
-                ev.call_id.clone(),
-                ev.command.clone(),
-                ev.parsed_cmd,
-                ev.is_user_shell_command,
+                call_id,
+                command,
+                parsed_cmd,
+                is_user_shell_command,
             )));
         }
 
         self.request_redraw();
     }
 
+    pub(crate) fn handle_exec_begin_now(&mut self, ev: ExecCommandBeginEvent) {
+        let ExecCommandBeginEvent {
+            call_id,
+            command,
+            parsed_cmd,
+            is_user_shell_command,
+            ..
+        } = ev;
+        self.handle_command_begin_internal(call_id, command, parsed_cmd, is_user_shell_command);
+    }
+
     pub(crate) fn handle_mcp_begin_now(&mut self, ev: McpToolCallBeginEvent) {
         self.flush_answer_stream_with_separator();
         self.flush_active_cell();

codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__binary_size_ideal_response.snap

@@ -26,7 +26,7 @@ expression: "lines[start_idx..].join(\"\\n\")"
   through crates for heavy dependencies in Cargo.toml, including cli, core,
   exec, linux-sandbox, tui, login, ollama, and mcp.
 
-• Ran for d in ansi-escape apply-patch arg0 cli common core exec execpolicy
+• Ran: for d in ansi-escape apply-patch arg0 cli common core exec execpolicy
   │ file-search linux-sandbox login mcp-client mcp-server mcp-types ollama
   │ tui; do echo "--- $d/Cargo.toml"; sed -n '1,200p' $d/Cargo.toml; echo;
   │ … +1 lines

codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__interrupt_exec_marks_failed.snap

@@ -2,5 +2,5 @@
 source: tui/src/chatwidget/tests.rs
 expression: exec_blob
 ---
-• Ran sleep 1
+• Ran: sleep 1
   └ (no output)