feat: end events on unified exec (#5551)

Author: jif-oai

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

@@ -200,10 +200,51 @@ impl ToolEmitter {
             ) => {
                 emit_patch_end(ctx, String::new(), (*message).to_string(), false).await;
             }
-            (Self::UnifiedExec { command, cwd, .. }, _) => {
-                // TODO(jif) add end and failures.
+            (Self::UnifiedExec { command, cwd, .. }, ToolEventStage::Begin) => {
                 emit_exec_command_begin(ctx, &[command.to_string()], cwd.as_path()).await;
             }
+            (Self::UnifiedExec { .. }, ToolEventStage::Success(output)) => {
+                emit_exec_end(
+                    ctx,
+                    output.stdout.text.clone(),
+                    output.stderr.text.clone(),
+                    output.aggregated_output.text.clone(),
+                    output.exit_code,
+                    output.duration,
+                    format_exec_output_str(&output),
+                )
+                .await;
+            }
+            (
+                Self::UnifiedExec { .. },
+                ToolEventStage::Failure(ToolEventFailure::Output(output)),
+            ) => {
+                emit_exec_end(
+                    ctx,
+                    output.stdout.text.clone(),
+                    output.stderr.text.clone(),
+                    output.aggregated_output.text.clone(),
+                    output.exit_code,
+                    output.duration,
+                    format_exec_output_str(&output),
+                )
+                .await;
+            }
+            (
+                Self::UnifiedExec { .. },
+                ToolEventStage::Failure(ToolEventFailure::Message(message)),
+            ) => {
+                emit_exec_end(
+                    ctx,
+                    String::new(),
+                    (*message).to_string(),
+                    (*message).to_string(),
+                    -1,
+                    Duration::ZERO,
+                    format_exec_output(&message),
+                )
+                .await;
+            }
         }
     }
 

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

@@ -5,6 +5,9 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use crate::function_tool::FunctionCallError;
+use crate::protocol::EventMsg;
+use crate::protocol::ExecCommandOutputDeltaEvent;
+use crate::protocol::ExecOutputStream;
 use crate::tools::context::ToolInvocation;
 use crate::tools::context::ToolOutput;
 use crate::tools::context::ToolPayload;
@@ -87,11 +90,7 @@ impl ToolHandler for UnifiedExecHandler {
         };
 
         let manager: &UnifiedExecSessionManager = &session.services.unified_exec_manager;
-        let context = UnifiedExecContext {
-            session: &session,
-            turn: turn.as_ref(),
-            call_id: &call_id,
-        };
+        let context = UnifiedExecContext::new(session.clone(), turn.clone(), call_id.clone());
 
         let response = match tool_name.as_str() {
             "exec_command" => {
@@ -101,8 +100,12 @@ impl ToolHandler for UnifiedExecHandler {
                     ))
                 })?;
 
-                let event_ctx =
-                    ToolEventCtx::new(context.session, context.turn, context.call_id, None);
+                let event_ctx = ToolEventCtx::new(
+                    context.session.as_ref(),
+                    context.turn.as_ref(),
+                    &context.call_id,
+                    None,
+                );
                 let emitter =
                     ToolEmitter::unified_exec(args.cmd.clone(), context.turn.cwd.clone(), true);
                 emitter.emit(event_ctx, ToolEventStage::Begin).await;
@@ -148,6 +151,18 @@ impl ToolHandler for UnifiedExecHandler {
             }
         };
 
+        // Emit a delta event with the chunk of output we just produced, if any.
+        if !response.output.is_empty() {
+            let delta = ExecCommandOutputDeltaEvent {
+                call_id: response.event_call_id.clone(),
+                stream: ExecOutputStream::Stdout,
+                chunk: response.output.as_bytes().to_vec(),
+            };
+            session
+                .send_event(turn.as_ref(), EventMsg::ExecCommandOutputDelta(delta))
+                .await;
+        }
+
         let content = serialize_response(&response).map_err(|err| {
             FunctionCallError::RespondToModel(format!(
                 "failed to serialize unified exec output: {err:?}"

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

@@ -22,6 +22,8 @@
 //! - `session_manager.rs`: orchestration (approvals, sandboxing, reuse) and request handling.
 
 use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
 use std::sync::atomic::AtomicI32;
 use std::time::Duration;
 
@@ -45,10 +47,20 @@ pub(crate) const MAX_YIELD_TIME_MS: u64 = 30_000;
 pub(crate) const DEFAULT_MAX_OUTPUT_TOKENS: usize = 10_000;
 pub(crate) const UNIFIED_EXEC_OUTPUT_MAX_BYTES: usize = 1024 * 1024; // 1 MiB
 
-pub(crate) struct UnifiedExecContext<'a> {
-    pub session: &'a Session,
-    pub turn: &'a TurnContext,
-    pub call_id: &'a str,
+pub(crate) struct UnifiedExecContext {
+    pub session: Arc<Session>,
+    pub turn: Arc<TurnContext>,
+    pub call_id: String,
+}
+
+impl UnifiedExecContext {
+    pub fn new(session: Arc<Session>, turn: Arc<TurnContext>, call_id: String) -> Self {
+        Self {
+            session,
+            turn,
+            call_id,
+        }
+    }
 }
 
 #[derive(Debug)]
@@ -70,6 +82,7 @@ pub(crate) struct WriteStdinRequest<'a> {
 
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct UnifiedExecResponse {
+    pub event_call_id: String,
     pub chunk_id: String,
     pub wall_time: Duration,
     pub output: String,
@@ -78,10 +91,20 @@ pub(crate) struct UnifiedExecResponse {
     pub original_token_count: Option<usize>,
 }
 
-#[derive(Debug, Default)]
+#[derive(Default)]
 pub(crate) struct UnifiedExecSessionManager {
     next_session_id: AtomicI32,
-    sessions: Mutex<HashMap<i32, session::UnifiedExecSession>>,
+    sessions: Mutex<HashMap<i32, SessionEntry>>,
+}
+
+struct SessionEntry {
+    session: session::UnifiedExecSession,
+    session_ref: Arc<Session>,
+    turn_ref: Arc<TurnContext>,
+    call_id: String,
+    command: String,
+    cwd: PathBuf,
+    started_at: tokio::time::Instant,
 }
 
 pub(crate) fn clamp_yield_time(yield_time_ms: Option<u64>) -> u64 {
@@ -163,11 +186,8 @@ mod tests {
         cmd: &str,
         yield_time_ms: Option<u64>,
     ) -> Result<UnifiedExecResponse, UnifiedExecError> {
-        let context = UnifiedExecContext {
-            session,
-            turn: turn.as_ref(),
-            call_id: "call",
-        };
+        let context =
+            UnifiedExecContext::new(Arc::clone(session), Arc::clone(turn), "call".to_string());
 
         session
             .services