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

mod common;

use std::collections::HashMap;
#[cfg(target_os = "linux")]
use std::os::unix::fs::PermissionsExt;
use std::sync::Arc;

use anyhow::Context;
use anyhow::Result;
use codex_exec_server::Environment;
use codex_exec_server::ExecBackend;
#[cfg(unix)]
use codex_exec_server::ExecEnvPolicy;
use codex_exec_server::ExecOutputStream;
use codex_exec_server::ExecParams;
use codex_exec_server::ExecProcess;
use codex_exec_server::ExecProcessEvent;
#[cfg(any(unix, windows))]
use codex_exec_server::FileSystemSandboxContext;
use codex_exec_server::ProcessId;
use codex_exec_server::ProcessSignal;
use codex_exec_server::ReadResponse;
#[cfg(unix)]
use codex_exec_server::ShellInfo;
#[cfg(unix)]
use codex_exec_server::ShellSnapshotRequest;
use codex_exec_server::StartedExecProcess;
use codex_exec_server::WriteStatus;
#[cfg(unix)]
use codex_network_proxy::NetworkProxyConfig;
#[cfg(unix)]
use codex_network_proxy::RemoteNetworkProxyConfig;
#[cfg(unix)]
use codex_network_proxy::RemoteNetworkProxyLaunchConfig;
#[cfg(unix)]
use codex_protocol::config_types::ShellEnvironmentPolicyInherit;
use codex_protocol::config_types::WindowsSandboxLevel;
#[cfg(unix)]
use codex_protocol::models::PermissionProfile;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemAccessMode;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemPath;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemSandboxEntry;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemSandboxPolicy;
#[cfg(unix)]
use codex_protocol::permissions::FileSystemSpecialPath;
#[cfg(unix)]
use codex_protocol::permissions::NetworkSandboxPolicy;
use codex_protocol::protocol::SandboxPolicy;
use codex_utils_path_uri::PathUri;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use test_case::test_case;
use tokio::sync::watch;
use tokio::time::Duration;
use tokio::time::sleep;
use tokio::time::timeout;

use common::DELAYED_OUTPUT_AFTER_EXIT_PARENT_ARG;
use common::current_test_binary_helper_paths;
use common::exec_server::ExecServerHarness;
use common::exec_server::exec_server;

struct ProcessContext {
    backend: Arc<dyn ExecBackend>,
    _server: Option<ExecServerHarness>,
}

#[derive(Debug, PartialEq, Eq)]
enum ProcessEventSnapshot {
    Output {
        seq: u64,
        stream: ExecOutputStream,
        text: String,
    },
    Exited {
        seq: u64,
        exit_code: i32,
    },
    Closed {
        seq: u64,
    },
}

async fn create_process_context(use_remote: bool) -> Result<ProcessContext> {
    if use_remote {
        let server = exec_server().await?;
        let environment = Environment::create_for_tests(Some(server.websocket_url().to_string()))?;
        Ok(ProcessContext {
            backend: environment.get_exec_backend(),
            _server: Some(server),
        })
    } else {
        let environment = Environment::create_for_tests(/*exec_server_url*/ None)?;
        Ok(ProcessContext {
            backend: environment.get_exec_backend(),
            _server: None,
        })
    }
}

#[cfg(unix)]
#[test_case(false, false, false, false, "bash"; "local_pipe")]
#[test_case(false, true, false, false, "bash"; "local_tty")]
#[test_case(true, false, false, false, "bash"; "remote_pipe")]
#[test_case(true, true, false, false, "bash"; "remote_tty")]
#[test_case(true, false, true, false, "bash"; "remote_sandbox")]
#[test_case(false, false, false, false, "sh"; "local_sh_pipe")]
#[test_case(false, false, false, true, "bash"; "local_bash_env")]
#[test_case(true, false, false, true, "bash"; "remote_bash_env")]
#[cfg_attr(
    target_os = "macos",
    test_case(false, false, false, false, "zsh"; "local_zsh_pipe")
)]
#[cfg_attr(
    target_os = "macos",
    test_case(false, false, false, true, "zsh"; "local_zshenv")
)]
#[cfg_attr(
    target_os = "macos",
    test_case(true, false, false, true, "zsh"; "remote_zshenv")
)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn shell_snapshot_v2_filters_profile_exports_and_stays_in_memory(
    use_remote: bool,
    tty: bool,
    use_sandbox: bool,
    automatic_startup: bool,
    shell_name: &str,
) -> Result<()> {
    if use_sandbox
        && let Some(warning) =
            codex_sandboxing::system_bwrap_warning(&PermissionProfile::read_only())
    {
        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" if automatic_startup => ("/bin/bash", ".bash-env"),
        "bash" => ("/bin/bash", ".bashrc"),
        "sh" => ("/bin/sh", ".snapshot-env"),
        "zsh" if automatic_startup => ("/bin/zsh", ".zshenv"),
        "zsh" => ("/bin/zsh", ".zshrc"),
        name => anyhow::bail!("unsupported test shell {name}"),
    };
    let profile_path = home.path().join(profile_name);
    let profile_path_entry = home.path().join("profile-bin");
    let runtime_path_entry = home.path().join("runtime-bin");
    let padding = if !use_remote && !tty && shell_name == "bash" {
        format!(
            "snapshot_padding() {{ printf '%s' '{}'; }}\n",
            "🦀".repeat(20_000)
        )
    } else {
        String::new()
    };
    let shadowed_builtins = if shell_name == "sh" {
        ""
    } else {
        "unset() { exit 41; }\nbuiltin() { :; }\n"
    };
    std::fs::write(
        &profile_path,
        format!(
            "printf x >> \"$HOME/captures\"\nexport PATH=\"$HOME/profile-bin:/usr/bin:/bin\"\nexport PROFILE_ALLOWED=profile\nexport PROFILE_SECRET=secret\nexport PROFILE_DENIED=denied\nprofile_helper() {{ printf helper; }}\n{shadowed_builtins}{padding}"
        ),
    )?;
    if shell_name == "zsh" && automatic_startup {
        std::fs::write(
            home.path().join(".zshrc"),
            "export PATH=\"$HOME/profile-bin:/usr/bin:/bin\"\n",
        )?;
    }
    let mut configured_environment = HashMap::from([(
        "HOME".to_string(),
        home.path().to_string_lossy().into_owned(),
    )]);
    if shell_name == "sh" {
        configured_environment.insert(
            "ENV".to_string(),
            profile_path.to_string_lossy().into_owned(),
        );
    }
    if shell_name == "bash" && automatic_startup {
        configured_environment.insert(
            "BASH_ENV".to_string(),
            profile_path.to_string_lossy().into_owned(),
        );
    }
    let policy = ExecEnvPolicy {
        inherit: ShellEnvironmentPolicyInherit::All,
        ignore_default_excludes: false,
        exclude: vec!["PROFILE_DENIED".to_string()],
        r#set: configured_environment,
        include_only: vec![
            "BASH_ENV".to_string(),
            "ENV".to_string(),
            "HOME".to_string(),
            "PATH".to_string(),
            "PROFILE_*".to_string(),
        ],
    };
    let (command_prefix, expected_prefix) = if shell_name == "sh" {
        ("", "")
    } else {
        ("profile_helper; ", "helper")
    };
    let command = format!(
        "export PATH='{}':\"$PATH\"; {command_prefix}printf '|%s|%s|%s|%s|%s|%s' \"$PROFILE_ALLOWED\" \"${{PROFILE_SECRET-missing}}\" \"${{PROFILE_DENIED-missing}}\" \"$PATH\" \"${{__CODEX_SHELL_SNAPSHOT_STATE_0-missing}}\" \"${{__CODEX_SHELL_SNAPSHOT_STATE_1-missing}}\"",
        runtime_path_entry.display(),
    );
    let expected_stdout = format!(
        "{expected_prefix}|profile|missing|missing|{}:{}:/usr/bin:/bin|missing|missing",
        runtime_path_entry.display(),
        profile_path_entry.display(),
    );

    for attempt in 0..2 {
        let started = context
            .backend
            .start(ExecParams {
                process_id: ProcessId::from(format!("snapshot-{attempt}")),
                argv: vec![shell_path.to_string(), "-lc".to_string(), command.clone()],
                cwd: cwd.clone(),
                env_policy: Some(policy.clone()),
                shell_snapshot: Some(ShellSnapshotRequest {
                    scope_id: "attachment-1".to_string(),
                    shell: ShellInfo {
                        name: shell_name.to_string(),
                        path: shell_path.to_string(),
                    },
                }),
                env: HashMap::new(),
                tty,
                pipe_stdin: false,
                arg0: None,
                sandbox: (use_sandbox && attempt == 0).then(|| {
                    FileSystemSandboxContext::from_permission_profile_with_cwd(
                        PermissionProfile::read_only(),
                        cwd.clone(),
                    )
                }),
                enforce_managed_network: false,
                managed_network: None,
                network_proxy: None,
            })
            .await?;
        let (stdout, stderr, status, closed) =
            collect_process_output_from_events(started.process).await?;
        assert_eq!(
            (stdout, stderr, status, closed),
            (expected_stdout.clone(), String::new(), Some(0), true,)
        );
    }

    assert_eq!(std::fs::read_to_string(home.path().join("captures"))?, "x");
    if let Some(server) = context._server {
        assert!(!server.codex_home().join("shell_snapshots").exists());
    }
    Ok(())
}

#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(remote_exec_server)]
async fn shell_snapshot_v2_remote_managed_proxy_uses_prepared_execution_context() -> Result<()> {
    let context = create_process_context(/*use_remote*/ true).await?;
    let home = TempDir::new()?;
    let cwd = PathUri::from_host_native_path(home.path())?;
    std::fs::write(
        home.path().join(".bashrc"),
        "printf '%s\\n' \"$HTTP_PROXY\" >> \"$HOME/captures\"\ntest \"$CODEX_NETWORK_PROXY_ACTIVE\" = 1 || exit 41\nexport PROFILE_ALLOWED=profile\nprofile_helper() { printf helper; }\n",
    )?;
    let policy = ExecEnvPolicy {
        inherit: ShellEnvironmentPolicyInherit::All,
        ignore_default_excludes: false,
        exclude: Vec::new(),
        r#set: HashMap::from([(
            "HOME".to_string(),
            home.path().to_string_lossy().into_owned(),
        )]),
        include_only: vec![
            "HOME".to_string(),
            "PATH".to_string(),
            "PROFILE_*".to_string(),
        ],
    };
    let proxy_config = RemoteNetworkProxyConfig::from_effective_config(&NetworkProxyConfig {
        enabled: true,
        ..NetworkProxyConfig::default()
    })?;
    let mut proxy_addresses = Vec::new();

    for attempt in 0..2 {
        let started = context
            .backend
            .start(ExecParams {
                process_id: ProcessId::from(format!("managed-snapshot-{attempt}")),
                argv: vec![
                    "/bin/bash".to_string(),
                    "-lc".to_string(),
                    "profile_helper; printf '|%s|%s|%s' \"$PROFILE_ALLOWED\" \"$CODEX_NETWORK_PROXY_ACTIVE\" \"$HTTP_PROXY\"".to_string(),
                ],
                cwd: cwd.clone(),
                env_policy: Some(policy.clone()),
                shell_snapshot: Some(ShellSnapshotRequest {
                    scope_id: "managed-attachment".to_string(),
                    shell: ShellInfo {
                        name: "bash".to_string(),
                        path: "/bin/bash".to_string(),
                    },
                }),
                env: HashMap::new(),
                tty: false,
                pipe_stdin: false,
                arg0: None,
                sandbox: None,
                enforce_managed_network: true,
                managed_network: None,
                network_proxy: Some(
                    RemoteNetworkProxyLaunchConfig::new(proxy_config.clone()).for_execution(
                        "remote-environment".to_string(),
                        format!("managed-snapshot-{attempt}"),
                    ),
                ),
            })
            .await?;
        let (stdout, stderr, status, closed) =
            collect_process_output_from_events(started.process).await?;
        let proxy_address = stdout
            .strip_prefix("helper|profile|1|")
            .context("snapshot should restore profile functions and live proxy state")?;
        assert!(proxy_address.starts_with("http://127.0.0.1:"));
        assert_eq!((stderr, status, closed), (String::new(), Some(0), true));
        proxy_addresses.push(proxy_address.to_string());
    }

    assert_eq!(
        std::fs::read_to_string(home.path().join("captures"))?,
        format!("{}\n", proxy_addresses[0])
    );
    Ok(())
}

#[cfg(unix)]
#[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?;
    let home = TempDir::new()?;
    let cwd = PathUri::from_host_native_path(home.path())?;
    std::fs::write(
        home.path().join(".bashrc"),
        "printf x >> \"$HOME/captures\"\nexit 7\n",
    )?;
    let policy = ExecEnvPolicy {
        inherit: ShellEnvironmentPolicyInherit::All,
        ignore_default_excludes: false,
        exclude: Vec::new(),
        r#set: HashMap::from([(
            "HOME".to_string(),
            home.path().to_string_lossy().into_owned(),
        )]),
        include_only: vec!["HOME".to_string(), "PATH".to_string()],
    };
    let mut params = ExecParams {
        process_id: ProcessId::from("snapshot-first"),
        argv: vec![
            "/bin/bash".to_string(),
            "-lc".to_string(),
            "printf original".to_string(),
        ],
        cwd,
        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(),
            },
        }),
        env: HashMap::new(),
        tty: false,
        pipe_stdin: false,
        arg0: None,
        sandbox: None,
        enforce_managed_network: false,
        managed_network: None,
        network_proxy: None,
    };

    for attempt in 0..2 {
        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?;
        assert_eq!(
            fallback_output,
            ("original".to_string(), String::new(), Some(0), true)
        );
    }
    assert_eq!(std::fs::read_to_string(home.path().join("captures"))?, "x");
    Ok(())
}

#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_sandboxed_process_preserves_custom_arg0() -> Result<()> {
    if let Some(warning) = codex_sandboxing::system_bwrap_warning(&PermissionProfile::read_only()) {
        eprintln!("skipping bwrap test: {warning}");
        return Ok(());
    }

    let context = create_process_context(/*use_remote*/ true).await?;
    let workspace = TempDir::new()?;
    let outside_workspace = TempDir::new()?;
    let denied_file = outside_workspace.path().join("denied.txt");
    std::fs::write(&denied_file, b"denied")?;
    let cwd = PathUri::from_host_native_path(workspace.path())?;
    let policy = FileSystemSandboxPolicy::restricted(vec![
        FileSystemSandboxEntry {
            path: FileSystemPath::Special {
                value: FileSystemSpecialPath::Minimal,
            },
            access: FileSystemAccessMode::Read,
            missing_path_behavior: None,
        },
        FileSystemSandboxEntry {
            path: FileSystemPath::Special {
                value: FileSystemSpecialPath::project_roots(/*subpath*/ None),
            },
            access: FileSystemAccessMode::Read,
            missing_path_behavior: None,
        },
    ]);
    let sandbox = FileSystemSandboxContext::from_permission_profile_with_cwd(
        PermissionProfile::from_runtime_permissions(&policy, NetworkSandboxPolicy::Restricted),
        cwd.clone(),
    );
    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-custom-arg0"),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "printf '%s' \"$0\"; if /bin/cat \"$CODEX_TEST_DENIED_FILE\" >/dev/null 2>&1; then exit 42; fi"
                    .to_string(),
            ],
            cwd,
            shell_snapshot: None,
            env_policy: None,
            env: HashMap::from([
                ("PATH".to_string(), std::env::var("PATH")?),
                (
                    "CODEX_TEST_DENIED_FILE".to_string(),
                    denied_file.to_string_lossy().into_owned(),
                ),
            ]),
            tty: false,
            pipe_stdin: false,
            arg0: Some("custom-arg0".to_string()),
            sandbox: Some(sandbox),
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    let output = collect_process_output_from_events(session.process).await?;

    assert_eq!(
        output,
        ("custom-arg0".to_string(), String::new(), Some(0), true)
    );
    Ok(())
}

async fn assert_exec_process_starts_and_exits(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-1"),
            argv: vec!["true".to_string()],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), "proc-1");
    let wake_rx = session.process.subscribe_wake();
    let (_, exit_code, closed) =
        collect_process_output_from_reads(session.process, wake_rx).await?;

    assert_eq!(exit_code, Some(0));
    assert!(closed);
    Ok(())
}

#[cfg(target_os = "linux")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_process_keeps_sandbox_helper_visible_with_restricted_reads() -> Result<()> {
    if let Some(warning) = codex_sandboxing::system_bwrap_warning(&PermissionProfile::read_only()) {
        eprintln!("skipping bwrap test: {warning}");
        return Ok(());
    }

    let context = create_process_context(/*use_remote*/ true).await?;
    let workspace = TempDir::new()?;
    let file = workspace.path().join("allowed.txt");
    std::fs::write(&file, b"allowed")?;
    let cwd = PathUri::from_host_native_path(workspace.path())?;
    let policy = FileSystemSandboxPolicy::restricted(vec![
        FileSystemSandboxEntry {
            path: FileSystemPath::Special {
                value: FileSystemSpecialPath::Minimal,
            },
            access: FileSystemAccessMode::Read,
            missing_path_behavior: None,
        },
        FileSystemSandboxEntry {
            path: FileSystemPath::Special {
                value: FileSystemSpecialPath::project_roots(/*subpath*/ None),
            },
            access: FileSystemAccessMode::Read,
            missing_path_behavior: None,
        },
    ]);
    let sandbox = FileSystemSandboxContext::from_permission_profile_with_cwd(
        PermissionProfile::from_runtime_permissions(&policy, NetworkSandboxPolicy::Restricted),
        cwd.clone(),
    );

    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-restricted-helper"),
            argv: vec!["/bin/cat".to_string(), file.to_string_lossy().into_owned()],
            cwd,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: HashMap::from([("PATH".to_string(), std::env::var("PATH")?)]),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: Some(sandbox),
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    let output = collect_process_output_from_events(session.process).await?;

    assert_eq!(
        output,
        ("allowed".to_string(), String::new(), Some(0), true)
    );
    Ok(())
}

#[cfg(target_os = "linux")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_tty_process_uses_configured_sandbox_helper_with_hostile_path() -> Result<()> {
    if let Some(warning) = codex_sandboxing::system_bwrap_warning(&PermissionProfile::read_only()) {
        eprintln!("skipping bwrap test: {warning}");
        return Ok(());
    }

    let context = create_process_context(/*use_remote*/ true).await?;
    let workspace = TempDir::new()?;
    let file = workspace.path().join("allowed.txt");
    std::fs::write(&file, b"allowed")?;
    let hostile_helper = workspace.path().join("codex-linux-sandbox");
    std::fs::write(&hostile_helper, b"#!/bin/sh\nprintf hostile")?;
    let mut permissions = std::fs::metadata(&hostile_helper)?.permissions();
    permissions.set_mode(0o755);
    std::fs::set_permissions(&hostile_helper, permissions)?;
    let path = std::env::var_os("PATH").context("PATH is not set")?;
    let hostile_path = std::env::join_paths(
        std::iter::once(workspace.path().to_path_buf()).chain(std::env::split_paths(&path)),
    )?;
    let cwd = PathUri::from_host_native_path(workspace.path())?;
    let policy = FileSystemSandboxPolicy::restricted(vec![
        FileSystemSandboxEntry {
            path: FileSystemPath::Special {
                value: FileSystemSpecialPath::Minimal,
            },
            access: FileSystemAccessMode::Read,
            missing_path_behavior: None,
        },
        FileSystemSandboxEntry {
            path: FileSystemPath::Special {
                value: FileSystemSpecialPath::project_roots(/*subpath*/ None),
            },
            access: FileSystemAccessMode::Read,
            missing_path_behavior: None,
        },
    ]);
    let sandbox = FileSystemSandboxContext::from_permission_profile_with_cwd(
        PermissionProfile::from_runtime_permissions(&policy, NetworkSandboxPolicy::Restricted),
        cwd.clone(),
    );

    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-hostile-helper-path"),
            argv: vec!["/bin/cat".to_string(), file.to_string_lossy().into_owned()],
            cwd,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: HashMap::from([(
                "PATH".to_string(),
                hostile_path.to_string_lossy().into_owned(),
            )]),
            tty: true,
            pipe_stdin: false,
            arg0: None,
            sandbox: Some(sandbox),
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    let output = collect_process_output_from_events(session.process).await?;

    assert_eq!(
        output,
        ("allowed".to_string(), String::new(), Some(0), true)
    );
    Ok(())
}

#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_process_preserves_empty_workspace_roots() -> Result<()> {
    if let Some(warning) = codex_sandboxing::system_bwrap_warning(&PermissionProfile::read_only()) {
        eprintln!("skipping bwrap test: {warning}");
        return Ok(());
    }

    let context = create_process_context(/*use_remote*/ true).await?;
    let tmp = TempDir::new()?;
    let file = tmp.path().join("excluded.txt");
    std::fs::write(&file, b"excluded")?;
    let cwd = PathUri::from_host_native_path(tmp.path())?;
    let policy = FileSystemSandboxPolicy::restricted(vec![FileSystemSandboxEntry {
        path: FileSystemPath::Special {
            value: FileSystemSpecialPath::project_roots(/*subpath*/ None),
        },
        access: FileSystemAccessMode::Read,
        missing_path_behavior: None,
    }]);
    let mut sandbox = FileSystemSandboxContext::from_permission_profile_with_cwd(
        PermissionProfile::from_runtime_permissions(&policy, NetworkSandboxPolicy::Restricted),
        cwd.clone(),
    );
    sandbox.workspace_roots.clear();

    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-empty-workspace-roots"),
            argv: vec!["/bin/cat".to_string(), file.to_string_lossy().into_owned()],
            cwd,
            shell_snapshot: None,
            env_policy: None,
            env: HashMap::new(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: Some(sandbox),
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    let (stdout, _stderr, exit_code, closed) =
        collect_process_output_from_events(session.process).await?;

    assert!(!stdout.contains("excluded"), "unexpected stdout: {stdout}");
    assert_ne!(exit_code, Some(0));
    assert!(closed);
    Ok(())
}

async fn read_process_until_change(
    session: Arc<dyn ExecProcess>,
    wake_rx: &mut watch::Receiver<u64>,
    after_seq: Option<u64>,
) -> Result<ReadResponse> {
    let response = session
        .read(after_seq, /*max_bytes*/ None, /*wait_ms*/ Some(0))
        .await?;
    if !response.chunks.is_empty() || response.closed || response.failure.is_some() {
        return Ok(response);
    }

    timeout(Duration::from_secs(2), wake_rx.changed()).await??;
    session
        .read(after_seq, /*max_bytes*/ None, /*wait_ms*/ Some(0))
        .await
        .map_err(Into::into)
}

async fn collect_process_output_from_reads(
    session: Arc<dyn ExecProcess>,
    mut wake_rx: watch::Receiver<u64>,
) -> Result<(String, Option<i32>, bool)> {
    let mut output = String::new();
    let mut exit_code = None;
    let mut after_seq = None;
    loop {
        let response =
            read_process_until_change(Arc::clone(&session), &mut wake_rx, after_seq).await?;
        if let Some(message) = response.failure {
            anyhow::bail!("process failed before closed state: {message}");
        }
        for chunk in response.chunks {
            output.push_str(&String::from_utf8_lossy(&chunk.chunk.into_inner()));
            after_seq = Some(chunk.seq);
        }
        if response.exited {
            exit_code = response.exit_code;
        }
        if response.closed {
            break;
        }
        after_seq = response.next_seq.checked_sub(1).or(after_seq);
    }
    drop(session);
    Ok((output, exit_code, true))
}

async fn collect_process_output_from_events(
    session: Arc<dyn ExecProcess>,
) -> Result<(String, String, Option<i32>, bool)> {
    let mut events = session.subscribe_events();
    let mut stdout = String::new();
    let mut stderr = String::new();
    let mut exit_code = None;
    loop {
        match timeout(Duration::from_secs(2), events.recv()).await?? {
            ExecProcessEvent::Output(chunk) => match chunk.stream {
                ExecOutputStream::Stdout | ExecOutputStream::Pty => {
                    stdout.push_str(&String::from_utf8_lossy(&chunk.chunk.into_inner()));
                }
                ExecOutputStream::Stderr => {
                    stderr.push_str(&String::from_utf8_lossy(&chunk.chunk.into_inner()));
                }
            },
            ExecProcessEvent::Exited {
                seq: _,
                exit_code: code,
                ..
            } => {
                exit_code = Some(code);
            }
            ExecProcessEvent::Closed { seq: _ } => {
                drop(session);
                return Ok((stdout, stderr, exit_code, true));
            }
            ExecProcessEvent::Failed(message) => {
                anyhow::bail!("process failed before closed state: {message}");
            }
        }
    }
}

async fn collect_process_event_snapshots(
    session: Arc<dyn ExecProcess>,
) -> Result<Vec<ProcessEventSnapshot>> {
    let mut events = session.subscribe_events();
    let mut snapshots = Vec::new();
    loop {
        let snapshot = match timeout(Duration::from_secs(2), events.recv()).await?? {
            ExecProcessEvent::Output(chunk) => ProcessEventSnapshot::Output {
                seq: chunk.seq,
                stream: chunk.stream,
                text: String::from_utf8_lossy(&chunk.chunk.into_inner()).into_owned(),
            },
            ExecProcessEvent::Exited { seq, exit_code, .. } => {
                ProcessEventSnapshot::Exited { seq, exit_code }
            }
            ExecProcessEvent::Closed { seq } => ProcessEventSnapshot::Closed { seq },
            ExecProcessEvent::Failed(message) => {
                anyhow::bail!("process failed before closed state: {message}");
            }
        };
        let closed = matches!(snapshot, ProcessEventSnapshot::Closed { .. });
        snapshots.push(snapshot);
        if closed {
            drop(session);
            return Ok(snapshots);
        }
    }
}

async fn assert_exec_process_streams_output(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-stream".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "sleep 0.05; printf 'session output\\n'".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
    assert_eq!(output, "session output\n");
    assert_eq!(exit_code, Some(0));
    assert!(closed);
    Ok(())
}

async fn assert_exec_process_pushes_events(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-events".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "printf 'event output\\n'; sleep 0.1; printf 'event err\\n' >&2; sleep 0.1; exit 7".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    let StartedExecProcess { process, .. } = session;
    let actual = collect_process_event_snapshots(process).await?;
    assert_eq!(
        actual,
        vec![
            ProcessEventSnapshot::Output {
                seq: 1,
                stream: ExecOutputStream::Stdout,
                text: "event output\n".to_string(),
            },
            ProcessEventSnapshot::Output {
                seq: 2,
                stream: ExecOutputStream::Stderr,
                text: "event err\n".to_string(),
            },
            ProcessEventSnapshot::Exited {
                seq: 3,
                exit_code: 7,
            },
            ProcessEventSnapshot::Closed { seq: 4 },
        ]
    );
    Ok(())
}

async fn assert_exec_process_replays_events_after_close(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-events-late".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "printf 'late one\\n'; printf 'late two\\n'".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let read_result = collect_process_output_from_reads(Arc::clone(&process), wake_rx).await?;
    assert_eq!(
        read_result,
        ("late one\nlate two\n".to_string(), Some(0), true)
    );

    let event_result = collect_process_output_from_events(process).await?;
    assert_eq!(
        event_result,
        (
            "late one\nlate two\n".to_string(),
            String::new(),
            Some(0),
            true
        )
    );
    Ok(())
}

async fn assert_exec_process_retains_output_after_exit_until_streams_close(
    use_remote: bool,
) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let (helper_binary, _) = current_test_binary_helper_paths()?;
    let release_dir = TempDir::new()?;
    let release_path = release_dir.path().join("release-delayed-output");
    let process_id = "proc-output-after-exit".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                helper_binary.to_string_lossy().into_owned(),
                DELAYED_OUTPUT_AFTER_EXIT_PARENT_ARG.to_string(),
                release_path.to_string_lossy().into_owned(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    let StartedExecProcess { process, .. } = session;

    let exit_response = timeout(
        Duration::from_secs(2),
        process.read(
            /*after_seq*/ None,
            /*max_bytes*/ None,
            /*wait_ms*/ Some(2_000),
        ),
    )
    .await??;
    assert!(
        exit_response.chunks.is_empty(),
        "parent should exit before child writes delayed output"
    );
    assert_eq!(exit_response.exit_code, Some(0));
    assert!(!exit_response.closed);
    let exit_seq = exit_response
        .next_seq
        .checked_sub(1)
        .context("exit response should advance next_seq")?;
    std::fs::write(&release_path, b"go")?;

    let late_response = timeout(
        Duration::from_secs(2),
        process.read(
            /*after_seq*/ Some(exit_seq),
            /*max_bytes*/ None,
            /*wait_ms*/ Some(2_000),
        ),
    )
    .await??;
    let mut late_output = String::new();
    for chunk in late_response.chunks {
        assert_eq!(chunk.stream, ExecOutputStream::Stdout);
        late_output.push_str(&String::from_utf8_lossy(&chunk.chunk.into_inner()));
    }
    assert_eq!(late_output, "late output after exit\n");

    let wake_rx = process.subscribe_wake();
    let actual = collect_process_output_from_reads(process, wake_rx).await?;
    assert_eq!(
        actual,
        ("late output after exit\n".to_string(), Some(0), true)
    );
    Ok(())
}

async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-stdin".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                // Use `/bin/sh` instead of Python so this stdin round-trip test
                // stays portable across Bazel and non-macOS runners where
                // `/usr/bin/python3` is not guaranteed to exist.
                "/bin/sh".to_string(),
                "-c".to_string(),
                "IFS= read line; printf 'from-stdin:%s\\n' \"$line\"".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: true,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    tokio::time::sleep(Duration::from_millis(200)).await;
    session.process.write(b"hello\n".to_vec()).await?;
    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;

    assert!(
        output.contains("from-stdin:hello"),
        "unexpected output: {output:?}"
    );
    assert_eq!(exit_code, Some(0));
    assert!(closed);
    Ok(())
}

async fn assert_exec_process_write_then_read_without_tty(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-stdin-pipe".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "IFS= read line; printf 'from-stdin:%s\\n' \"$line\"".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: true,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    tokio::time::sleep(Duration::from_millis(200)).await;
    let write_response = session.process.write(b"hello\n".to_vec()).await?;
    assert_eq!(write_response.status, WriteStatus::Accepted);
    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let actual = collect_process_output_from_reads(process, wake_rx).await?;

    assert_eq!(actual, ("from-stdin:hello\n".to_string(), Some(0), true));
    Ok(())
}

async fn assert_remote_windows_sandbox_process_write() -> Result<()> {
    let context = create_process_context(/*use_remote*/ true).await?;
    let workspace = TempDir::new()?;
    let blocked_file = workspace.path().join("blocked.txt");
    let cwd = PathUri::from_host_native_path(workspace.path())?;
    let mut sandbox = FileSystemSandboxContext::from_legacy_sandbox_policy(
        SandboxPolicy::new_read_only_policy(),
        cwd.clone(),
    )?;
    sandbox.windows_sandbox_level = WindowsSandboxLevel::RestrictedToken;

    let session = match context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-windows-sandbox-stdin"),
            argv: vec![
                r"C:\Windows\System32\cmd.exe".to_string(),
                "/D".to_string(),
                "/V:ON".to_string(),
                "/S".to_string(),
                "/C".to_string(),
                format!(
                    "set /P line= & echo blocked > \"{}\" & echo from-stdin:!line!",
                    blocked_file.display()
                ),
            ],
            cwd,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: true,
            arg0: None,
            sandbox: Some(sandbox),
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await
    {
        Ok(session) => session,
        Err(err) => return Err(err.into()),
    };

    let write_response = session.process.write(b"hello\n".to_vec()).await?;
    assert_eq!(write_response.status, WriteStatus::Accepted);
    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;

    assert!(
        output.contains("from-stdin:hello"),
        "unexpected output: {output:?}"
    );
    assert_eq!(exit_code, Some(0));
    assert!(closed);
    assert!(!blocked_file.exists());
    Ok(())
}

async fn assert_exec_process_rejects_write_without_pipe_stdin(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-stdin-closed".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "sleep 0.3; if IFS= read -r line; then printf 'read:%s\\n' \"$line\"; else printf 'eof\\n'; fi".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    let write_response = session.process.write(b"ignored\n".to_vec()).await?;
    assert_eq!(write_response.status, WriteStatus::StdinClosed);
    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;

    assert_eq!(output, "eof\n");
    assert_eq!(exit_code, Some(0));
    assert!(closed);
    Ok(())
}

async fn assert_exec_process_signal_interrupts_process(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let process_id = "proc-signal".to_string();
    let session = context
        .backend
        .start(ExecParams {
            process_id: process_id.clone().into(),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "trap 'printf \"signal:2\\n\"; exit 7' INT; printf 'ready\\n'; while :; do :; done".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;
    assert_eq!(session.process.process_id().as_str(), process_id);

    let StartedExecProcess { process, .. } = session;
    let mut wake_rx = process.subscribe_wake();
    let mut ready_output = String::new();
    let mut after_seq = None;
    loop {
        let response =
            read_process_until_change(Arc::clone(&process), &mut wake_rx, after_seq).await?;
        for chunk in response.chunks {
            ready_output.push_str(&String::from_utf8_lossy(&chunk.chunk.into_inner()));
            after_seq = Some(chunk.seq);
        }
        if ready_output.contains("ready\n") {
            break;
        }
        if response.closed {
            anyhow::bail!("process closed before readiness marker: {ready_output:?}");
        }
        after_seq = response.next_seq.checked_sub(1).or(after_seq);
    }

    process.signal(ProcessSignal::Interrupt).await?;
    let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;

    assert!(
        output.contains("signal:2"),
        "expected signal handler output, got {output:?}"
    );
    assert_eq!(exit_code, Some(7));
    assert!(closed);
    Ok(())
}

async fn assert_exec_process_signal_terminates_on_windows(use_remote: bool) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-windows-signal"),
            argv: vec![
                "cmd".to_string(),
                "/C".to_string(),
                "echo ready && ping -n 30 127.0.0.1 >NUL".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;

    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    process.signal(ProcessSignal::Interrupt).await?;
    let (_output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;

    assert_eq!(exit_code, Some(1));
    assert!(closed);
    Ok(())
}

async fn assert_exec_process_preserves_queued_events_before_subscribe(
    use_remote: bool,
) -> Result<()> {
    let context = create_process_context(use_remote).await?;
    let session = context
        .backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-queued"),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "printf 'queued output\\n'".to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: Default::default(),
            tty: false,
            pipe_stdin: false,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;

    tokio::time::sleep(Duration::from_millis(200)).await;

    let StartedExecProcess { process, .. } = session;
    let wake_rx = process.subscribe_wake();
    let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
    assert_eq!(output, "queued output\n");
    assert_eq!(exit_code, Some(0));
    assert!(closed);
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn remote_exec_process_recovers_after_transport_disconnect() -> Result<()> {
    let server = exec_server().await?;
    let mut proxy = server.disconnectable_websocket_proxy().await?;
    let environment = Environment::create_for_tests(Some(proxy.websocket_url().to_string()))?;
    let backend = environment.get_exec_backend();
    let temp_dir = TempDir::new()?;
    let gate_path = temp_dir.path().join("release-output");
    let emitted_path = temp_dir.path().join("output-emitted");
    let session = backend
        .start(ExecParams {
            process_id: ProcessId::from("proc-recover"),
            argv: vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                concat!(
                    "printf 'ready:%s\\n' \"$$\"; ",
                    "while [ ! -f \"$GATE\" ]; do /bin/sleep 0.01; done; ",
                    "printf 'during:%s\\n' \"$$\"; ",
                    ": > \"$EMITTED\"; ",
                    "IFS= read -r line; ",
                    "printf 'after:%s:%s\\n' \"$$\" \"$line\"; ",
                    "exit 7",
                )
                .to_string(),
            ],
            cwd: PathUri::from_host_native_path(std::env::current_dir()?)?,
            shell_snapshot: None,
            env_policy: /*env_policy*/ None,
            env: HashMap::from([
                (
                    "GATE".to_string(),
                    gate_path.to_string_lossy().into_owned(),
                ),
                (
                    "EMITTED".to_string(),
                    emitted_path.to_string_lossy().into_owned(),
                ),
            ]),
            tty: false,
            pipe_stdin: true,
            arg0: None,
            sandbox: None,
            enforce_managed_network: false,
            managed_network: None,
            network_proxy: None,
        })
        .await?;

    let process = Arc::clone(&session.process);
    let mut events = process.subscribe_events();
    let mut output = Vec::new();
    let mut last_seq = 0;
    while !output.ends_with(b"\n") {
        match timeout(Duration::from_secs(5), events.recv()).await?? {
            ExecProcessEvent::Output(chunk) => {
                assert_eq!(chunk.seq, last_seq + 1);
                last_seq = chunk.seq;
                output.extend_from_slice(&chunk.chunk.into_inner());
            }
            event => anyhow::bail!("expected ready output before disconnect, got {event:?}"),
        }
    }
    let ready = String::from_utf8(output.clone())?;
    let pid = ready
        .strip_prefix("ready:")
        .and_then(|line| line.strip_suffix('\n'))
        .context("ready output should contain the process id")?
        .to_string();

    proxy.pause_and_disconnect().await?;
    tokio::fs::write(&gate_path, b"").await?;
    timeout(Duration::from_secs(5), async {
        while tokio::fs::metadata(&emitted_path).await.is_err() {
            sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .context("process did not emit output while disconnected")?;

    let process_for_read = Arc::clone(&process);
    let mut pending_read = tokio::spawn(async move {
        process_for_read
            .read(
                /*after_seq*/ Some(last_seq),
                /*max_bytes*/ None,
                /*wait_ms*/ Some(0),
            )
            .await
    });
    assert!(
        timeout(Duration::from_millis(200), &mut pending_read)
            .await
            .is_err(),
        "process reads should wait while recovery is in progress"
    );
    proxy.resume()?;

    let recovered_read = timeout(Duration::from_secs(5), pending_read)
        .await
        .context("timed out waiting for a read after recovery")??;
    let recovered_read = recovered_read?;
    assert_eq!(recovered_read.failure, None);
    let recovered_output = recovered_read
        .chunks
        .into_iter()
        .flat_map(|chunk| chunk.chunk.into_inner())
        .collect::<Vec<_>>();
    assert_eq!(
        String::from_utf8(recovered_output)?,
        format!("during:{pid}\n")
    );

    let write = timeout(Duration::from_secs(5), process.write(b"hello\n".to_vec()))
        .await
        .context("timed out waiting for a write after recovery")??;
    assert_eq!(write.status, WriteStatus::Accepted);

    let mut saw_exit = false;
    loop {
        match timeout(Duration::from_secs(5), events.recv()).await?? {
            ExecProcessEvent::Output(chunk) => {
                assert_eq!(chunk.seq, last_seq + 1);
                last_seq = chunk.seq;
                output.extend_from_slice(&chunk.chunk.into_inner());
            }
            ExecProcessEvent::Exited { seq, exit_code, .. } => {
                assert_eq!(seq, last_seq + 1);
                assert_eq!(exit_code, 7);
                last_seq = seq;
                saw_exit = true;
            }
            ExecProcessEvent::Closed { seq } => {
                assert!(saw_exit, "closed must be delivered after exit");
                assert_eq!(seq, last_seq + 1);
                break;
            }
            ExecProcessEvent::Failed(message) => {
                anyhow::bail!("process recovery failed: {message}");
            }
        }
    }
    assert_eq!(
        String::from_utf8(output)?,
        format!("ready:{pid}\nduring:{pid}\nafter:{pid}:hello\n")
    );

    Ok(())
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_starts_and_exits(use_remote: bool) -> Result<()> {
    assert_exec_process_starts_and_exits(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_streams_output(use_remote: bool) -> Result<()> {
    assert_exec_process_streams_output(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_pushes_events(use_remote: bool) -> Result<()> {
    assert_exec_process_pushes_events(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_replays_events_after_close(use_remote: bool) -> Result<()> {
    assert_exec_process_replays_events_after_close(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_retains_output_after_exit_until_streams_close(
    use_remote: bool,
) -> Result<()> {
    assert_exec_process_retains_output_after_exit_until_streams_close(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_write_then_read(use_remote: bool) -> Result<()> {
    assert_exec_process_write_then_read(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_write_then_read_without_tty(use_remote: bool) -> Result<()> {
    assert_exec_process_write_then_read_without_tty(use_remote).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[cfg_attr(not(windows), ignore = "Windows-only exec-server sandbox process test")]
#[serial_test::serial(remote_exec_server)]
async fn remote_windows_sandbox_process_accepts_process_write() -> Result<()> {
    assert_remote_windows_sandbox_process_write().await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_rejects_write_without_pipe_stdin(use_remote: bool) -> Result<()> {
    assert_exec_process_rejects_write_without_pipe_stdin(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_signal_interrupts_process(use_remote: bool) -> Result<()> {
    assert_exec_process_signal_interrupts_process(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(windows), ignore = "Windows-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_signal_terminates_on_windows(use_remote: bool) -> Result<()> {
    assert_exec_process_signal_terminates_on_windows(use_remote).await
}

#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_preserves_queued_events_before_subscribe(use_remote: bool) -> Result<()> {
    assert_exec_process_preserves_queued_events_before_subscribe(use_remote).await
}