Emit command lifecycle events for unified exec launch failures (#47529)

## Why

When unified exec cannot create a process, command lifecycle events are missing, leaving clients without a command failure notification.

## What changed

Emit start and failure events for terminal process-creation errors in both regular and zsh-fork execution. Report no process ID, an exit code of `-1`, zero duration, and the launch diagnostic, while retaining plugin attribution. Once event publication starts, let it finish even if the caller is cancelled.

Preserve approval rejection handling and retryable sandbox denials.

## Testing

Extend app-server tests to cover pipe and PTY launch failures, plugin attribution on failed launches, and zsh-fork launch failures after approval. Assert failure metadata and check for duplicate lifecycle notifications while retaining approval-decline coverage.

GitOrigin-RevId: 26ed191fc69b903ea4710dd1a6c3b031e8caa4eb

CHANGELOG.md

@@ -1 +1,5 @@
 The changelog can be found on the [releases page](https://github.com/openai/codex/releases).
+
+## Unreleased
+
+- Fix missing command lifecycle events when unified exec cannot create a process, preserving approval rejection and sandbox retry behavior. Thanks @Marvinthebored ([#44557](https://github.com/openai/codex/issues/44557)).

codex-rs/app-server/tests/suite/v2/turn_start.rs

@@ -5495,9 +5495,13 @@ async fn run_turn_start_file_change_approval_rejection_v2(
     Ok(())
 }
 
+#[cfg_attr(not(windows), test_case(None; "started"))]
+#[test_case(Some(json!({"cmd": "echo unreachable", "workdir": "missing-work-directory", "tty": false})); "pipe_launch_failure")]
+#[test_case(Some(json!({"cmd": "echo unreachable\u{0}", "tty": true, "login": false})); "pty_launch_failure")]
 #[tokio::test]
-#[cfg_attr(windows, ignore = "process id reporting differs on Windows")]
-async fn command_execution_notifications_include_process_id() -> Result<()> {
+async fn command_execution_notifications_include_process_id(
+    launch_failure_args: Option<Value>,
+) -> Result<()> {
     // TODO(anp): Add target-Windows process-id expectations for remote executors.
     skip_if_wine_exec!(
         Ok(()),
@@ -5505,8 +5509,18 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
     );
     skip_if_no_network!(Ok(()));
 
+    let launch_failed = launch_failure_args.is_some();
+    let command = if let Some(args) = launch_failure_args {
+        responses::sse(vec![
+            responses::ev_response_created("launch"),
+            responses::ev_function_call("uexec-1", "exec_command", &args.to_string()),
+            responses::ev_completed("launch"),
+        ])
+    } else {
+        create_exec_command_sse_response("uexec-1")?
+    };
     let responses = vec![
-        create_exec_command_sse_response("uexec-1")?,
+        command,
         create_final_assistant_message_sse_response("done")?,
     ];
     let server = create_mock_responses_server_sequence(responses).await;
@@ -5514,6 +5528,8 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
     MockResponsesConfig::new(&server.uri())
         .with_sandbox_mode("danger-full-access")
         .enable_feature(Feature::UnifiedExec)
+        .disable_feature(Feature::ShellZshFork)
+        .disable_feature(Feature::ShellSnapshot)
         .write(codex_home.path())?;
 
     let mut mcp = TestAppServer::builder()
@@ -5564,7 +5580,7 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
     };
     assert_eq!(id, "uexec-1");
     assert_eq!(status, CommandExecutionStatus::InProgress);
-    let started_process_id = started_process_id.expect("process id should be present");
+    assert_eq!(started_process_id.is_none(), launch_failed);
 
     let completed_command = timeout(DEFAULT_READ_TIMEOUT, async {
         loop {
@@ -5581,6 +5597,8 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
         process_id: completed_process_id,
         status: completed_status,
         exit_code,
+        duration_ms,
+        aggregated_output,
         ..
     } = completed_command
     else {
@@ -5594,15 +5612,22 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
         ),
         "unexpected command execution status: {completed_status:?}"
     );
-    if completed_status == CommandExecutionStatus::Completed {
+    if launch_failed {
+        assert_eq!(
+            (completed_status, exit_code, duration_ms),
+            (CommandExecutionStatus::Failed, Some(-1), Some(0))
+        );
+        assert!(
+            aggregated_output
+                .context("launch diagnostic")?
+                .starts_with("Failed to create unified exec process:")
+        );
+    } else if completed_status == CommandExecutionStatus::Completed {
         assert_eq!(exit_code, Some(0));
     } else {
         assert!(exit_code.is_some(), "expected exit_code for failed command");
     }
-    assert_eq!(
-        completed_process_id.as_deref(),
-        Some(started_process_id.as_str())
-    );
+    assert_eq!(completed_process_id, started_process_id);
 
     timeout(
         DEFAULT_READ_TIMEOUT,
@@ -5610,16 +5635,30 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
     )
     .await??;
 
+    for method in mcp.pending_notification_methods() {
+        let notification = mcp.read_stream_until_notification_message(&method).await?;
+        assert_ne!(
+            notification.params.context("notification params")?["item"]["id"],
+            "uexec-1"
+        );
+    }
+
     Ok(())
 }
 
 #[cfg_attr(windows, ignore = "plugin attribution fixture is Unix-only")]
+#[test_case(CommandExecutionStatus::Completed; "completed")]
+#[test_case(CommandExecutionStatus::Failed; "launch_failure")]
 #[tokio::test]
-async fn command_execution_notifications_include_trusted_plugin_id() -> Result<()> {
+async fn command_execution_notifications_include_trusted_plugin_id(
+    expected_status: CommandExecutionStatus,
+) -> Result<()> {
     skip_if_no_network!(Ok(()));
     skip_if_wine_exec!(Ok(()), "plugin attribution fixture is Unix-only");
 
     let codex_home = TempDir::new()?;
+    let missing_cwd = codex_home.path().join("missing-work-directory");
+    let launch_failed = expected_status == CommandExecutionStatus::Failed;
     let curated_sha = "0123456789abcdef0123456789abcdef01234567";
     let plugin_root = codex_home
         .path()
@@ -5661,7 +5700,7 @@ async fn command_execution_notifications_include_trusted_plugin_id() -> Result<(
                 "/bin/sh".to_string(),
                 script_path.to_string_lossy().into_owned(),
             ],
-            /*workdir*/ None,
+            launch_failed.then_some(missing_cwd.as_path()),
             /*timeout_ms*/ None,
             "plugin-command",
         )?,
@@ -5672,7 +5711,10 @@ async fn command_execution_notifications_include_trusted_plugin_id() -> Result<(
         .with_approval_policy("on-request")
         .with_sandbox_mode("danger-full-access")
         .enable_feature(Feature::Plugins)
+        .enable_feature(Feature::UnifiedExec)
         .disable_feature(Feature::RemotePlugin)
+        .disable_feature(Feature::ShellZshFork)
+        .disable_feature(Feature::ShellSnapshot)
         .with_extra_config("[plugins.\"google-calendar@openai-api-curated\"]\nenabled = true")
         .write(codex_home.path())?;
 
@@ -5733,7 +5775,7 @@ async fn command_execution_notifications_include_trusted_plugin_id() -> Result<(
         if method == "item/started" {
             assert_eq!(status, CommandExecutionStatus::InProgress);
         } else {
-            assert_eq!(status, CommandExecutionStatus::Completed);
+            assert_eq!(status, expected_status);
         }
     }
 

codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs

@@ -38,6 +38,7 @@ use std::collections::BTreeMap;
 use std::path::Path;
 use std::path::PathBuf;
 use tempfile::TempDir;
+use test_case::test_case;
 use tokio::time::timeout;
 
 #[cfg(windows)]
@@ -169,8 +170,12 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
     Ok(())
 }
 
+#[test_case(CommandExecutionApprovalDecision::Decline; "declined")]
+#[test_case(CommandExecutionApprovalDecision::Accept; "launch_failure")]
 #[tokio::test]
-async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
+async fn turn_start_shell_zsh_fork_exec_approval_v2(
+    decision: CommandExecutionApprovalDecision,
+) -> Result<()> {
     // TODO(anp): Remove after zsh-fork fixtures can run in the selected remote environment.
     skip_if_remote!(
         Ok(()),
@@ -190,6 +195,8 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
     };
     eprintln!("using zsh path for zsh-fork test: {}", zsh_path.display());
 
+    let launch_failed = matches!(decision, CommandExecutionApprovalDecision::Accept);
+    let missing_cwd = workspace.join("missing-work-directory");
     let responses = vec![
         create_escalated_command_execution_sse_response(
             vec![
@@ -197,7 +204,7 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
                 "-c".to_string(),
                 "print(42)".to_string(),
             ],
-            /*workdir*/ None,
+            launch_failed.then_some(missing_cwd.as_path()),
             Some(5000),
             "call-zsh-fork-decline",
         )?,
@@ -209,6 +216,7 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
         &server.uri(),
         "on-request",
         &BTreeMap::from([
+            (Feature::UnifiedExec, true),
             (Feature::ShellZshFork, true),
             (Feature::ShellSnapshot, false),
         ]),
@@ -253,9 +261,7 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
 
     mcp.send_response(
         request_id,
-        serde_json::to_value(CommandExecutionRequestApprovalResponse {
-            decision: CommandExecutionApprovalDecision::Decline,
-        })?,
+        serde_json::to_value(CommandExecutionRequestApprovalResponse { decision })?,
     )
     .await?;
 
@@ -280,6 +286,8 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
         id,
         status,
         exit_code,
+        process_id,
+        duration_ms,
         aggregated_output,
         ..
     } = completed_command_execution
@@ -287,9 +295,24 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
         unreachable!("loop ensures we break on command execution items");
     };
     assert_eq!(id, "call-zsh-fork-decline");
-    assert_eq!(status, CommandExecutionStatus::Declined);
-    assert!(exit_code.is_none());
-    assert!(aggregated_output.is_none());
+    assert_eq!(process_id, None);
+    if launch_failed {
+        assert_eq!(
+            (status, exit_code, duration_ms),
+            (CommandExecutionStatus::Failed, Some(-1), Some(0))
+        );
+        assert!(
+            aggregated_output
+                .expect("launch diagnostic")
+                .starts_with("Failed to create unified exec process:")
+        );
+    } else {
+        assert_eq!(
+            (status, exit_code, duration_ms),
+            (CommandExecutionStatus::Declined, None, None)
+        );
+        assert!(aggregated_output.is_none());
+    }
 
     timeout(
         DEFAULT_READ_TIMEOUT,
@@ -297,6 +320,17 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> {
     )
     .await??;
 
+    let mut command_events = Vec::new();
+    for method in mcp.pending_notification_methods() {
+        let notification = mcp.read_stream_until_notification_message(&method).await?;
+        if notification.params.expect("notification params")["item"]["id"]
+            == "call-zsh-fork-decline"
+        {
+            command_events.push(method);
+        }
+    }
+    assert_eq!(command_events, ["item/started"]);
+
     Ok(())
 }
 

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

@@ -61,6 +61,10 @@ use std::path::PathBuf;
 use std::time::Duration;
 use tokio_util::sync::CancellationToken;
 
+mod launch;
+
+use launch::with_launch_failure_events;
+
 // Allow 5s for Guardian cleanup and 5s for controller processing after review.
 const REMOTE_NETWORK_POLICY_DECISION_MARGIN: Duration = Duration::from_secs(10);
 
@@ -653,7 +657,7 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecAttempt> for UnifiedExecRunt
                                 .to_string(),
                         ));
                     }
-                    let mut process = self
+                    let process = self
                         .manager
                         .open_session_with_prepared_exec_env(
                             req.process_id,
@@ -674,7 +678,8 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecAttempt> for UnifiedExecRunt
                                 }))
                             }
                             other => ToolError::Rejected(other.to_string()),
-                        })?;
+                        });
+                    let mut process = with_launch_failure_events(process, req, ctx).await?;
                     process._shell_snapshot = shell_snapshot;
                     return Ok(UnifiedExecAttempt {
                         process,
@@ -703,7 +708,7 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecAttempt> for UnifiedExecRunt
             error @ ToolError::Codex(_) => error,
         })?;
         let options = unified_exec_options(attempt.network_denial_cancellation_token.clone());
-        let mut process = self
+        let process = self
             .manager
             .open_session_with_exec_env(
                 req.process_id,
@@ -721,7 +726,8 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecAttempt> for UnifiedExecRunt
                 Box::new(NoopSpawnLifecycle),
                 req.turn_environment.environment.as_ref(),
             )
-            .await?;
+            .await;
+        let mut process = with_launch_failure_events(process, req, ctx).await?;
         process._shell_snapshot = shell_snapshot;
         Ok(UnifiedExecAttempt {
             process,

codex-rs/core/src/tools/runtimes/unified_exec/launch.rs

@@ -0,0 +1,75 @@
+//! Reports terminal process-creation failures as command attempts without a process handle.
+//! Approval rejections retain their own lifecycle, and sandbox denials remain retryable.
+//! Once started, failure event publication survives caller cancellation.
+
+use super::UnifiedExecRequest;
+use crate::tools::events::ToolEmitter;
+use crate::tools::events::ToolEventCtx;
+use crate::tools::events::ToolEventFailure;
+use crate::tools::events::ToolEventStage;
+use crate::tools::sandboxing::ToolCtx;
+use crate::tools::sandboxing::ToolError;
+use crate::unified_exec::UnifiedExecProcess;
+use codex_protocol::protocol::ExecCommandSource;
+use std::sync::Arc;
+use tracing::Instrument;
+
+pub(super) async fn with_launch_failure_events(
+    result: Result<UnifiedExecProcess, ToolError>,
+    req: &UnifiedExecRequest,
+    ctx: &ToolCtx,
+) -> Result<UnifiedExecProcess, ToolError> {
+    let Err(ToolError::Rejected(message)) = &result else {
+        return result;
+    };
+    let plugin_attribution = if req.turn_environment.environment.is_remote() {
+        let file_system = req
+            .turn_environment
+            .environment
+            .get_filesystem_without_reconnect();
+        ctx.step_context
+            .turn
+            .plugin_attribution_for_executor_command(&req.command, &req.cwd, file_system.as_ref())
+            .await
+    } else {
+        req.cwd.to_abs_path().ok().and_then(|cwd| {
+            ctx.step_context
+                .turn
+                .plugin_attribution_for_command(&req.command, &cwd)
+        })
+    };
+    let emitter = ToolEmitter::unified_exec(
+        &req.command,
+        req.cwd.clone(),
+        ExecCommandSource::UnifiedExecStartup,
+        /*process_id*/ None,
+        plugin_attribution,
+    );
+    let session = Arc::clone(&ctx.session);
+    let step_context = Arc::clone(&ctx.step_context);
+    let call_id = ctx.call_id.clone();
+    let message = message.clone();
+    // Dropping the join handle on cancellation leaves both lifecycle events running.
+    let publish = async move {
+        let model_context = step_context.model_context();
+        let mut event_ctx = ToolEventCtx::new(
+            session.as_ref(),
+            step_context.turn.as_ref(),
+            &step_context.settings.model_info,
+            &call_id,
+            /*turn_diff_tracker*/ None,
+        );
+        event_ctx.model_context = Some(&model_context);
+        emitter.emit(event_ctx, ToolEventStage::Begin).await;
+        emitter
+            .emit(
+                event_ctx,
+                ToolEventStage::Failure(ToolEventFailure::Message(message)),
+            )
+            .await;
+    };
+    if let Err(err) = tokio::spawn(publish.in_current_span()).await {
+        tracing::warn!(%err, "failed to publish unified exec launch failure");
+    }
+    result
+}