Retry failed shell snapshot captures (#40447)

## Why

A failed shell snapshot capture was cached for the lifetime of its cache entry, so later commands continued using the original command even after a transient profile error was fixed.

## What changed

- Retry failed captures after a one-second backoff, up to three total attempts.
- Share each attempt across concurrent requests and preserve command fallback during failures and backoff.
- Keep the final failure cached after the retry budget is exhausted.

## Testing

Cover recovery and retry exhaustion for concurrent requests, local and remote execution, pipes and TTYs, and Bash and zsh where available.

GitOrigin-RevId: b432ef7c9eb35a3a0a703ac675b4db6954c43a80

codex-rs/exec-server/tests/exec_process.rs

@@ -351,13 +351,39 @@ async fn shell_snapshot_v2_remote_managed_proxy_uses_prepared_execution_context(
 }
 
 #[cfg(unix)]
+#[test_case(false, false, "bash", 1; "local_pipe_recovery")]
+#[test_case(false, true, "bash", 1; "local_tty_recovery")]
+#[test_case(true, false, "bash", 1; "remote_pipe_recovery")]
+#[test_case(true, true, "bash", 1; "remote_tty_recovery")]
+#[test_case(false, false, "bash", 3; "local_retry_budget_exhausted")]
+#[test_case(true, false, "bash", 3; "remote_retry_budget_exhausted")]
+#[cfg_attr(target_os = "macos", test_case(false, false, "zsh", 1; "local_zsh_recovery"))]
+#[cfg_attr(target_os = "macos", test_case(true, false, "zsh", 1; "remote_zsh_recovery"))]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn shell_snapshot_v2_capture_failure_falls_back_to_original_command() -> Result<()> {
-    let context = create_process_context(/*use_remote*/ false).await?;
+#[serial_test::serial(remote_exec_server)]
+async fn shell_snapshot_v2_capture_failure_falls_back_and_retries(
+    use_remote: bool,
+    tty: bool,
+    shell_name: &str,
+    failures_before_repair: usize,
+) -> Result<()> {
+    if use_remote
+        && let Some(warning) =
+            codex_sandboxing::system_bwrap_warning(&PermissionProfile::workspace_write())
+    {
+        eprintln!("skipping sandbox test: {warning}");
+        return Ok(());
+    }
+    let context = create_process_context(use_remote).await?;
     let home = TempDir::new()?;
     let cwd = PathUri::from_host_native_path(home.path())?;
+    let (shell_path, profile_name) = match shell_name {
+        "bash" => ("/bin/bash", ".bashrc"),
+        "zsh" => ("/bin/zsh", ".zshrc"),
+        name => anyhow::bail!("unsupported test shell {name}"),
+    };
     std::fs::write(
-        home.path().join(".bashrc"),
+        home.path().join(profile_name),
         "printf x >> \"$HOME/captures\"\nexit 7\n",
     )?;
     let policy = ExecEnvPolicy {
@@ -373,30 +399,35 @@ async fn shell_snapshot_v2_capture_failure_falls_back_to_original_command() -> R
     let mut params = ExecParams {
         process_id: ProcessId::from("snapshot-first"),
         argv: vec![
-            "/bin/bash".to_string(),
+            shell_path.to_string(),
             "-lc".to_string(),
-            "printf original".to_string(),
+            "if command -v profile_helper >/dev/null; then profile_helper; else printf original; fi".to_string(),
         ],
-        cwd,
+        cwd: cwd.clone(),
         env_policy: Some(policy),
         shell_snapshot: Some(ShellSnapshotRequest {
             scope_id: "attachment-1".to_string(),
             shell: ShellInfo {
-                name: "bash".to_string(),
-                path: "/bin/bash".to_string(),
+                name: shell_name.to_string(),
+                path: shell_path.to_string(),
             },
         }),
         env: HashMap::new(),
-        tty: false,
+        tty,
         pipe_stdin: false,
         arg0: None,
-        sandbox: None,
+        sandbox: use_remote.then(|| {
+            FileSystemSandboxContext::from_permission_profile_with_cwd(
+                PermissionProfile::workspace_write(),
+                cwd,
+            )
+        }),
         enforce_managed_network: false,
         managed_network: None,
         network_proxy: None,
     };
 
-    for attempt in 0..2 {
+    for attempt in 0..failures_before_repair {
         params.process_id = ProcessId::from(format!("snapshot-fallback-{attempt}"));
         let fallback = context.backend.start(params.clone()).await?;
         let fallback_output = collect_process_output_from_events(fallback.process).await?;
@@ -404,8 +435,36 @@ async fn shell_snapshot_v2_capture_failure_falls_back_to_original_command() -> R
             fallback_output,
             ("original".to_string(), String::new(), Some(0), true)
         );
+        // A real remote executor has its own clock; the unit test uses a
+        // paused clock to check requests made during the one-second backoff.
+        sleep(Duration::from_millis(1100)).await;
     }
-    assert_eq!(std::fs::read_to_string(home.path().join("captures"))?, "x");
+    assert_eq!(
+        std::fs::read_to_string(home.path().join("captures"))?,
+        "x".repeat(failures_before_repair)
+    );
+
+    std::fs::write(
+        home.path().join(profile_name),
+        "printf x >> \"$HOME/captures\"\nprofile_helper() { printf recovered; }\n",
+    )?;
+    let (expected_output, expected_captures) = if failures_before_repair == 3 {
+        ("original", "xxx")
+    } else {
+        ("recovered", "xx")
+    };
+    for attempt in 0..2 {
+        params.process_id = ProcessId::from(format!("snapshot-after-repair-{attempt}"));
+        let started = context.backend.start(params.clone()).await?;
+        assert_eq!(
+            collect_process_output_from_events(started.process).await?,
+            (expected_output.to_string(), String::new(), Some(0), true)
+        );
+    }
+    assert_eq!(
+        std::fs::read_to_string(home.path().join("captures"))?,
+        expected_captures
+    );
     Ok(())
 }