refactor: route zsh-fork through unified exec

codex-rs/core/tests/common/process.rs

@@ -1,6 +1,7 @@
 use anyhow::Context;
 use std::fs;
 use std::path::Path;
+use std::process::Stdio;
 use std::time::Duration;
 
 pub async fn wait_for_pid_file(path: &Path) -> anyhow::Result<String> {
@@ -24,6 +25,7 @@ pub async fn wait_for_pid_file(path: &Path) -> anyhow::Result<String> {
 pub fn process_is_alive(pid: &str) -> anyhow::Result<bool> {
     let status = std::process::Command::new("kill")
         .args(["-0", pid])
+        .stderr(Stdio::null())
         .status()
         .context("failed to probe process liveness with kill -0")?;
     Ok(status.success())

codex-rs/core/tests/common/zsh_fork.rs

@@ -18,6 +18,10 @@ pub struct ZshForkRuntime {
 }
 
 impl ZshForkRuntime {
+    pub fn zsh_path(&self) -> &Path {
+        &self.zsh_path
+    }
+
     fn apply_to_config(
         &self,
         config: &mut Config,
@@ -91,6 +95,29 @@ where
     builder.build(server).await
 }
 
+pub async fn build_unified_exec_zsh_fork_test<F>(
+    server: &wiremock::MockServer,
+    runtime: ZshForkRuntime,
+    approval_policy: AskForApproval,
+    sandbox_policy: SandboxPolicy,
+    pre_build_hook: F,
+) -> Result<TestCodex>
+where
+    F: FnOnce(&Path) + Send + 'static,
+{
+    let mut builder = test_codex()
+        .with_pre_build_hook(pre_build_hook)
+        .with_config(move |config| {
+            runtime.apply_to_config(config, approval_policy, sandbox_policy);
+            config.use_experimental_unified_exec_tool = true;
+            config
+                .features
+                .enable(Feature::UnifiedExec)
+                .expect("test config should allow feature update");
+        });
+    builder.build(server).await
+}
+
 fn find_test_zsh_path() -> Result<Option<PathBuf>> {
     let repo_root = codex_utils_cargo_bin::repo_root()?;
     let dotslash_zsh = repo_root.join("codex-rs/app-server/tests/suite/zsh");

codex-rs/core/tests/suite/approvals.rs

@@ -35,6 +35,7 @@ use core_test_support::test_codex::TestCodex;
 use core_test_support::test_codex::test_codex;
 use core_test_support::wait_for_event;
 use core_test_support::wait_for_event_with_timeout;
+use core_test_support::zsh_fork::build_unified_exec_zsh_fork_test;
 use core_test_support::zsh_fork::build_zsh_fork_test;
 use core_test_support::zsh_fork::restrictive_workspace_write_policy;
 use core_test_support::zsh_fork::zsh_fork_runtime;
@@ -1985,6 +1986,158 @@ async fn approving_execpolicy_amendment_persists_policy_and_skips_future_prompts
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[cfg(unix)]
+async fn unified_exec_zsh_fork_execpolicy_amendment_skips_later_subcommands() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let Some(runtime) = zsh_fork_runtime("unified exec zsh-fork execpolicy amendment test")? else {
+        return Ok(());
+    };
+
+    let approval_policy = AskForApproval::UnlessTrusted;
+    let sandbox_policy = SandboxPolicy::new_read_only_policy();
+    let server = start_mock_server().await;
+    let test = build_unified_exec_zsh_fork_test(
+        &server,
+        runtime,
+        approval_policy,
+        sandbox_policy.clone(),
+        |_| {},
+    )
+    .await?;
+    let allow_prefix_path = test.cwd.path().join("allow-prefix-zsh-fork.txt");
+    let _ = fs::remove_file(&allow_prefix_path);
+
+    let call_id = "allow-prefix-zsh-fork";
+    let command = "touch allow-prefix-zsh-fork.txt && touch allow-prefix-zsh-fork.txt";
+    let event = exec_command_event(
+        call_id,
+        command,
+        Some(1_000),
+        SandboxPermissions::UseDefault,
+        None,
+    )?;
+    let _ = mount_sse_once(
+        &server,
+        sse(vec![
+            ev_response_created("resp-zsh-fork-allow-prefix-1"),
+            event,
+            ev_completed("resp-zsh-fork-allow-prefix-1"),
+        ]),
+    )
+    .await;
+    let results = mount_sse_once(
+        &server,
+        sse(vec![
+            ev_assistant_message("msg-zsh-fork-allow-prefix-1", "done"),
+            ev_completed("resp-zsh-fork-allow-prefix-2"),
+        ]),
+    )
+    .await;
+
+    submit_turn(
+        &test,
+        "allow-prefix-zsh-fork",
+        approval_policy,
+        sandbox_policy,
+    )
+    .await?;
+
+    let expected_execpolicy_amendment = ExecPolicyAmendment::new(vec![
+        "touch".to_string(),
+        "allow-prefix-zsh-fork.txt".to_string(),
+    ]);
+    let mut saw_parent_approval = false;
+    let mut saw_subcommand_approval = false;
+    loop {
+        let event = wait_for_event(&test.codex, |event| {
+            matches!(
+                event,
+                EventMsg::ExecApprovalRequest(_) | EventMsg::TurnComplete(_)
+            )
+        })
+        .await;
+
+        match event {
+            EventMsg::TurnComplete(_) => break,
+            EventMsg::ExecApprovalRequest(approval) => {
+                let command_parts = approval.command.clone();
+                let last_arg = command_parts.last().map(String::as_str).unwrap_or_default();
+                if last_arg == command {
+                    assert!(
+                        !saw_parent_approval,
+                        "unexpected duplicate parent approval: {command_parts:?}"
+                    );
+                    saw_parent_approval = true;
+                    test.codex
+                        .submit(Op::ExecApproval {
+                            id: approval.effective_approval_id(),
+                            turn_id: None,
+                            decision: ReviewDecision::Approved,
+                        })
+                        .await?;
+                    continue;
+                }
+
+                let is_touch_subcommand = command_parts
+                    .iter()
+                    .any(|part| part == "allow-prefix-zsh-fork.txt")
+                    && command_parts
+                        .first()
+                        .is_some_and(|part| part.ends_with("/touch") || part == "touch");
+                if is_touch_subcommand {
+                    assert!(
+                        !saw_subcommand_approval,
+                        "execpolicy amendment should suppress later matching subcommand approvals: {command_parts:?}"
+                    );
+                    saw_subcommand_approval = true;
+                    assert_eq!(
+                        approval.proposed_execpolicy_amendment,
+                        Some(expected_execpolicy_amendment.clone())
+                    );
+                    test.codex
+                        .submit(Op::ExecApproval {
+                            id: approval.effective_approval_id(),
+                            turn_id: None,
+                            decision: ReviewDecision::ApprovedExecpolicyAmendment {
+                                proposed_execpolicy_amendment: expected_execpolicy_amendment
+                                    .clone(),
+                            },
+                        })
+                        .await?;
+                    continue;
+                }
+
+                test.codex
+                    .submit(Op::ExecApproval {
+                        id: approval.effective_approval_id(),
+                        turn_id: None,
+                        decision: ReviewDecision::Approved,
+                    })
+                    .await?;
+            }
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+
+    assert!(saw_parent_approval, "expected parent unified-exec approval");
+    assert!(
+        saw_subcommand_approval,
+        "expected at least one intercepted touch approval"
+    );
+
+    let result = parse_result(&results.single_request().function_call_output(call_id));
+    assert_eq!(result.exit_code.unwrap_or(0), 0);
+    assert!(
+        allow_prefix_path.exists(),
+        "expected touch command to complete after approving the first intercepted subcommand; output: {}",
+        result.stdout
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 #[cfg(unix)]
 async fn matched_prefix_rule_runs_unsandboxed_under_zsh_fork() -> Result<()> {

codex-rs/core/tests/suite/skill_approval.rs

@@ -20,6 +20,7 @@ use core_test_support::skip_if_no_network;
 use core_test_support::test_codex::TestCodex;
 use core_test_support::wait_for_event;
 use core_test_support::wait_for_event_match;
+use core_test_support::zsh_fork::build_unified_exec_zsh_fork_test;
 use core_test_support::zsh_fork::build_zsh_fork_test;
 use core_test_support::zsh_fork::restrictive_workspace_write_policy;
 use core_test_support::zsh_fork::zsh_fork_runtime;
@@ -50,6 +51,13 @@ fn shell_command_arguments(command: &str) -> Result<String> {
     }))?)
 }
 
+fn exec_command_arguments(command: &str) -> Result<String> {
+    Ok(serde_json::to_string(&json!({
+        "cmd": command,
+        "yield_time_ms": 500,
+    }))?)
+}
+
 async fn submit_turn_with_policies(
     test: &TestCodex,
     prompt: &str,
@@ -89,6 +97,38 @@ echo 'zsh-fork-stderr' >&2
     )
 }
 
+#[cfg(unix)]
+fn write_repo_skill_with_shell_script_contents(
+    repo_root: &Path,
+    name: &str,
+    script_name: &str,
+    script_contents: &str,
+) -> Result<PathBuf> {
+    use std::os::unix::fs::PermissionsExt;
+
+    let skill_dir = repo_root.join(".agents").join("skills").join(name);
+    let scripts_dir = skill_dir.join("scripts");
+    fs::create_dir_all(&scripts_dir)?;
+    fs::write(repo_root.join(".git"), "gitdir: here")?;
+    fs::write(
+        skill_dir.join("SKILL.md"),
+        format!(
+            r#"---
+name: {name}
+description: {name} skill
+---
+"#
+        ),
+    )?;
+
+    let script_path = scripts_dir.join(script_name);
+    fs::write(&script_path, script_contents)?;
+    let mut permissions = fs::metadata(&script_path)?.permissions();
+    permissions.set_mode(0o755);
+    fs::set_permissions(&script_path, permissions)?;
+    Ok(script_path)
+}
+
 #[cfg(unix)]
 fn write_skill_with_shell_script_contents(
     home: &Path,
@@ -541,6 +581,168 @@ permissions:
     Ok(())
 }
 
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn unified_exec_zsh_fork_prompts_for_skill_script_execution() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let Some(runtime) = zsh_fork_runtime("unified exec zsh-fork skill prompt test")? else {
+        return Ok(());
+    };
+
+    let server = start_mock_server().await;
+    let tool_call_id = "uexec-zsh-fork-skill-call";
+    let test = build_unified_exec_zsh_fork_test(
+        &server,
+        runtime,
+        AskForApproval::OnRequest,
+        SandboxPolicy::new_workspace_write_policy(),
+        |home| {
+            write_skill_with_shell_script(home, "mbolin-test-skill", "hello-mbolin.sh").unwrap();
+            write_skill_metadata(
+                home,
+                "mbolin-test-skill",
+                r#"
+permissions:
+  file_system:
+    read:
+      - "./data"
+    write:
+      - "./output"
+"#,
+            )
+            .unwrap();
+        },
+    )
+    .await?;
+
+    let (script_path_str, command) = skill_script_command(&test, "hello-mbolin.sh")?;
+    let arguments = exec_command_arguments(&command)?;
+    let mocks =
+        mount_function_call_agent_response(&server, tool_call_id, &arguments, "exec_command").await;
+
+    submit_turn_with_policies(
+        &test,
+        "use $mbolin-test-skill",
+        AskForApproval::OnRequest,
+        SandboxPolicy::new_workspace_write_policy(),
+    )
+    .await?;
+
+    let approval = wait_for_exec_approval_request(&test)
+        .await
+        .expect("expected exec approval request before completion");
+    assert_eq!(approval.call_id, tool_call_id);
+    assert_eq!(approval.command, vec![script_path_str.clone()]);
+    assert_eq!(
+        approval.available_decisions,
+        Some(vec![
+            ReviewDecision::Approved,
+            ReviewDecision::ApprovedForSession,
+            ReviewDecision::Abort,
+        ])
+    );
+    assert_eq!(
+        approval.additional_permissions,
+        Some(PermissionProfile {
+            file_system: Some(FileSystemPermissions {
+                read: Some(vec![absolute_path(
+                    &test.codex_home_path().join("skills/mbolin-test-skill/data"),
+                )]),
+                write: Some(vec![absolute_path(
+                    &test
+                        .codex_home_path()
+                        .join("skills/mbolin-test-skill/output"),
+                )]),
+            }),
+            ..Default::default()
+        })
+    );
+
+    test.codex
+        .submit(Op::ExecApproval {
+            id: approval.effective_approval_id(),
+            turn_id: None,
+            decision: ReviewDecision::Denied,
+        })
+        .await?;
+
+    wait_for_turn_complete(&test).await;
+
+    let call_output = mocks
+        .completion
+        .single_request()
+        .function_call_output(tool_call_id);
+    let output = call_output["output"].as_str().unwrap_or_default();
+    assert!(
+        output.contains("Execution denied: User denied execution"),
+        "expected rejection marker in function_call_output: {output:?}"
+    );
+
+    Ok(())
+}
+
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn unified_exec_zsh_fork_keeps_skill_loading_pinned_to_turn_cwd() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let Some(runtime) = zsh_fork_runtime("unified exec zsh-fork turn cwd skill test")? else {
+        return Ok(());
+    };
+
+    let server = start_mock_server().await;
+    let tool_call_id = "uexec-zsh-fork-repo-skill-call";
+    let test = build_unified_exec_zsh_fork_test(
+        &server,
+        runtime,
+        AskForApproval::OnRequest,
+        SandboxPolicy::new_workspace_write_policy(),
+        |_| {},
+    )
+    .await?;
+
+    let repo_root = test.cwd_path().join("repo");
+    let script_path = write_repo_skill_with_shell_script_contents(
+        &repo_root,
+        "repo-skill",
+        "repo-skill.sh",
+        "#!/bin/sh\necho 'repo-skill-output'\n",
+    )?;
+    let script_path_quoted = shlex::try_join([script_path.to_string_lossy().as_ref()])?;
+    let repo_root_quoted = shlex::try_join([repo_root.to_string_lossy().as_ref()])?;
+    let command = format!("cd {repo_root_quoted} && {script_path_quoted}");
+    let arguments = exec_command_arguments(&command)?;
+    let mocks =
+        mount_function_call_agent_response(&server, tool_call_id, &arguments, "exec_command").await;
+
+    submit_turn_with_policies(
+        &test,
+        "run the repo skill after changing directories",
+        AskForApproval::OnRequest,
+        SandboxPolicy::new_workspace_write_policy(),
+    )
+    .await?;
+
+    let approval = wait_for_exec_approval_request(&test).await;
+    assert!(
+        approval.is_none(),
+        "changing directories inside unified exec should not load repo-local skills from the shell cwd",
+    );
+
+    let call_output = mocks
+        .completion
+        .single_request()
+        .function_call_output(tool_call_id);
+    let output = call_output["output"].as_str().unwrap_or_default();
+    assert!(
+        output.contains("repo-skill-output"),
+        "expected repo skill script to run without skill-specific approval when only the shell cwd changes: {output:?}"
+    );
+
+    Ok(())
+}
+
 /// Permissionless skills should inherit the turn sandbox without prompting.
 #[cfg(unix)]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

codex-rs/core/tests/suite/unified_exec.rs

@@ -1,6 +1,8 @@
 use std::collections::HashMap;
 use std::ffi::OsStr;
 use std::fs;
+#[cfg(unix)]
+use std::path::PathBuf;
 use std::sync::OnceLock;
 
 use anyhow::Context;
@@ -32,6 +34,10 @@ use core_test_support::test_codex::test_codex;
 use core_test_support::wait_for_event;
 use core_test_support::wait_for_event_match;
 use core_test_support::wait_for_event_with_timeout;
+#[cfg(unix)]
+use core_test_support::zsh_fork::build_unified_exec_zsh_fork_test;
+#[cfg(unix)]
+use core_test_support::zsh_fork::zsh_fork_runtime;
 use pretty_assertions::assert_eq;
 use regex_lite::Regex;
 use serde_json::Value;
@@ -155,6 +161,27 @@ fn collect_tool_outputs(bodies: &[Value]) -> Result<HashMap<String, ParsedUnifie
     Ok(outputs)
 }
 
+#[cfg(unix)]
+fn process_text_binary_path(pid: &str) -> Result<PathBuf> {
+    let output = std::process::Command::new("lsof")
+        .args(["-Fn", "-a", "-p", pid, "-d", "txt"])
+        .output()
+        .with_context(|| format!("failed to inspect process {pid} executable mapping with lsof"))?;
+    if !output.status.success() {
+        return Err(anyhow::anyhow!(
+            "lsof failed for pid {pid} with status {:?}",
+            output.status.code()
+        ));
+    }
+
+    let stdout = String::from_utf8(output.stdout).context("lsof output was not UTF-8")?;
+    let path = stdout
+        .lines()
+        .find_map(|line| line.strip_prefix('n'))
+        .ok_or_else(|| anyhow::anyhow!("lsof did not report a text binary path for pid {pid}"))?;
+    Ok(PathBuf::from(path))
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn unified_exec_intercepts_apply_patch_exec_command() -> Result<()> {
     skip_if_no_network!(Ok(()));
@@ -1323,7 +1350,6 @@ async fn exec_command_reports_chunk_and_exit_metadata() -> Result<()> {
         .into_iter()
         .map(|request| request.body_json())
         .collect::<Vec<_>>();
-
     let outputs = collect_tool_outputs(&bodies)?;
     let metadata = outputs
         .get(call_id)
@@ -1445,7 +1471,6 @@ async fn unified_exec_defaults_to_pipe() -> Result<()> {
         .into_iter()
         .map(|request| request.body_json())
         .collect::<Vec<_>>();
-
     let outputs = collect_tool_outputs(&bodies)?;
     let output = outputs
         .get(call_id)
@@ -1539,7 +1564,6 @@ async fn unified_exec_can_enable_tty() -> Result<()> {
         .into_iter()
         .map(|request| request.body_json())
         .collect::<Vec<_>>();
-
     let outputs = collect_tool_outputs(&bodies)?;
     let output = outputs
         .get(call_id)
@@ -1624,7 +1648,6 @@ async fn unified_exec_respects_early_exit_notifications() -> Result<()> {
         .into_iter()
         .map(|request| request.body_json())
         .collect::<Vec<_>>();
-
     let outputs = collect_tool_outputs(&bodies)?;
     let output = outputs
         .get(call_id)
@@ -1823,6 +1846,350 @@ async fn write_stdin_returns_exit_metadata_and_clears_session() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[cfg(unix)]
+async fn unified_exec_zsh_fork_keeps_python_repl_attached_to_zsh_session() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let Some(runtime) = zsh_fork_runtime("unified exec zsh-fork tty session test")? else {
+        return Ok(());
+    };
+    let configured_zsh_path =
+        fs::canonicalize(runtime.zsh_path()).unwrap_or_else(|_| runtime.zsh_path().to_path_buf());
+
+    let python = match which("python3") {
+        Ok(path) => path,
+        Err(_) => {
+            eprintln!("python3 not found in PATH, skipping zsh-fork python repl test.");
+            return Ok(());
+        }
+    };
+
+    let server = start_mock_server().await;
+    let test = build_unified_exec_zsh_fork_test(
+        &server,
+        runtime,
+        AskForApproval::Never,
+        SandboxPolicy::new_workspace_write_policy(),
+        |_| {},
+    )
+    .await?;
+
+    let start_call_id = "uexec-zsh-fork-python-start";
+    let send_call_id = "uexec-zsh-fork-python-pid";
+    let exit_call_id = "uexec-zsh-fork-python-exit";
+
+    let start_command = format!("{}; :", python.display());
+    let start_args = serde_json::json!({
+        "cmd": start_command,
+        "yield_time_ms": 500,
+        "tty": true,
+    });
+    let send_args = serde_json::json!({
+        "chars": "import os; print('CODEX_PY_PID=' + str(os.getpid()))\r\n",
+        "session_id": 1000,
+        "yield_time_ms": 500,
+    });
+    let exit_args = serde_json::json!({
+        "chars": "import sys; sys.exit(0)\r\n",
+        "session_id": 1000,
+        "yield_time_ms": 500,
+    });
+
+    let responses = vec![
+        sse(vec![
+            ev_response_created("resp-1"),
+            ev_function_call(
+                start_call_id,
+                "exec_command",
+                &serde_json::to_string(&start_args)?,
+            ),
+            ev_completed("resp-1"),
+        ]),
+        sse(vec![
+            ev_response_created("resp-2"),
+            ev_function_call(
+                send_call_id,
+                "write_stdin",
+                &serde_json::to_string(&send_args)?,
+            ),
+            ev_completed("resp-2"),
+        ]),
+        sse(vec![
+            ev_assistant_message("msg-1", "python is running"),
+            ev_completed("resp-3"),
+        ]),
+        sse(vec![
+            ev_response_created("resp-4"),
+            ev_function_call(
+                exit_call_id,
+                "write_stdin",
+                &serde_json::to_string(&exit_args)?,
+            ),
+            ev_completed("resp-4"),
+        ]),
+        sse(vec![
+            ev_assistant_message("msg-2", "all done"),
+            ev_completed("resp-5"),
+        ]),
+    ];
+    let request_log = mount_sse_sequence(&server, responses).await;
+
+    test.codex
+        .submit(Op::UserTurn {
+            items: vec![UserInput::Text {
+                text: "test unified exec zsh-fork tty behavior".into(),
+                text_elements: Vec::new(),
+            }],
+            final_output_json_schema: None,
+            cwd: test.cwd_path().to_path_buf(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::new_workspace_write_policy(),
+            model: test.session_configured.model.clone(),
+            effort: None,
+            summary: None,
+            service_tier: None,
+            collaboration_mode: None,
+            personality: None,
+        })
+        .await?;
+
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+
+    let requests = request_log.requests();
+    assert!(!requests.is_empty(), "expected at least one POST request");
+    let bodies = requests
+        .into_iter()
+        .map(|request| request.body_json())
+        .collect::<Vec<_>>();
+    let outputs = collect_tool_outputs(&bodies)?;
+
+    let start_output = outputs
+        .get(start_call_id)
+        .expect("missing start output for exec_command");
+    let process_id = start_output
+        .process_id
+        .clone()
+        .expect("expected process id from exec_command");
+    assert!(
+        start_output.exit_code.is_none(),
+        "initial exec_command should leave the PTY session running"
+    );
+
+    let send_output = outputs
+        .get(send_call_id)
+        .expect("missing write_stdin output");
+    let normalized = send_output.output.replace("\r\n", "\n");
+    let python_pid = Regex::new(r"CODEX_PY_PID=(\d+)")
+        .expect("valid python pid marker regex")
+        .captures(&normalized)
+        .and_then(|captures| captures.get(1))
+        .map(|value| value.as_str().to_string())
+        .with_context(|| format!("missing python pid in output {normalized:?}"))?;
+    assert!(
+        process_is_alive(&python_pid)?,
+        "python process should still be alive after printing its pid, got output {normalized:?}"
+    );
+    assert_eq!(send_output.process_id.as_deref(), Some(process_id.as_str()));
+    assert!(
+        send_output.exit_code.is_none(),
+        "write_stdin should not report an exit code while the process is still running"
+    );
+
+    let zsh_pid = std::process::Command::new("ps")
+        .args(["-o", "ppid=", "-p", &python_pid])
+        .output()
+        .context("failed to look up python parent pid")?;
+    let zsh_pid = String::from_utf8(zsh_pid.stdout)
+        .context("python parent pid output is not UTF-8")?
+        .trim()
+        .to_string();
+    assert!(
+        !zsh_pid.is_empty(),
+        "expected python parent pid to identify the zsh session"
+    );
+    assert!(
+        process_is_alive(&zsh_pid)?,
+        "expected zsh parent process {zsh_pid} to still be alive"
+    );
+
+    let zsh_command = std::process::Command::new("ps")
+        .args(["-o", "command=", "-p", &zsh_pid])
+        .output()
+        .context("failed to look up zsh parent command")?;
+    let zsh_command =
+        String::from_utf8(zsh_command.stdout).context("zsh parent command output is not UTF-8")?;
+    assert!(
+        zsh_command.contains("zsh"),
+        "expected python parent command to be zsh, got {zsh_command:?}"
+    );
+    let zsh_text_binary = process_text_binary_path(&zsh_pid)?;
+    let zsh_text_binary = fs::canonicalize(&zsh_text_binary).unwrap_or(zsh_text_binary);
+    assert_eq!(
+        zsh_text_binary, configured_zsh_path,
+        "python parent shell should run with configured zsh-fork binary, got {:?} ({zsh_command:?})",
+        zsh_text_binary,
+    );
+
+    test.codex
+        .submit(Op::UserTurn {
+            items: vec![UserInput::Text {
+                text: "shut down the python repl".into(),
+                text_elements: Vec::new(),
+            }],
+            final_output_json_schema: None,
+            cwd: test.cwd_path().to_path_buf(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::new_workspace_write_policy(),
+            model: test.session_configured.model.clone(),
+            effort: None,
+            summary: None,
+            service_tier: None,
+            collaboration_mode: None,
+            personality: None,
+        })
+        .await?;
+
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+
+    let requests = request_log.requests();
+    assert!(!requests.is_empty(), "expected at least one POST request");
+    let bodies = requests
+        .into_iter()
+        .map(|request| request.body_json())
+        .collect::<Vec<_>>();
+    let outputs = collect_tool_outputs(&bodies)?;
+    let exit_output = outputs
+        .get(exit_call_id)
+        .expect("missing exit output after requesting python shutdown");
+    assert!(
+        exit_output.exit_code.is_none() || exit_output.exit_code == Some(0),
+        "exit request should either leave cleanup to the background watcher or report success directly, got {exit_output:?}"
+    );
+    wait_for_process_exit(&python_pid).await?;
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[cfg(unix)]
+async fn unified_exec_zsh_fork_rewrites_nested_zsh_exec_to_configured_binary() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let Some(runtime) = zsh_fork_runtime("unified exec zsh-fork nested zsh rewrite test")? else {
+        return Ok(());
+    };
+    let configured_zsh_path =
+        fs::canonicalize(runtime.zsh_path()).unwrap_or_else(|_| runtime.zsh_path().to_path_buf());
+    let host_zsh = match which("zsh") {
+        Ok(path) => path,
+        Err(_) => {
+            eprintln!("zsh not found in PATH, skipping nested zsh rewrite test.");
+            return Ok(());
+        }
+    };
+
+    let server = start_mock_server().await;
+    let test = build_unified_exec_zsh_fork_test(
+        &server,
+        runtime,
+        AskForApproval::Never,
+        SandboxPolicy::new_workspace_write_policy(),
+        |_| {},
+    )
+    .await?;
+
+    let start_call_id = "uexec-zsh-fork-nested-start";
+    let nested_command = format!(
+        "exec {} -lc 'echo CODEX_NESTED_ZSH_PID=$$; sleep 3; :'",
+        host_zsh.display(),
+    );
+    let start_args = serde_json::json!({
+        "cmd": nested_command,
+        "yield_time_ms": 500,
+        "tty": true,
+    });
+
+    let responses = vec![
+        sse(vec![
+            ev_response_created("resp-nested-1"),
+            ev_function_call(
+                start_call_id,
+                "exec_command",
+                &serde_json::to_string(&start_args)?,
+            ),
+            ev_completed("resp-nested-1"),
+        ]),
+        sse(vec![
+            ev_assistant_message("msg-nested-1", "done"),
+            ev_completed("resp-nested-2"),
+        ]),
+    ];
+    let request_log = mount_sse_sequence(&server, responses).await;
+
+    test.codex
+        .submit(Op::UserTurn {
+            items: vec![UserInput::Text {
+                text: "test nested zsh rewrite behavior".into(),
+                text_elements: Vec::new(),
+            }],
+            final_output_json_schema: None,
+            cwd: test.cwd_path().to_path_buf(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::new_workspace_write_policy(),
+            model: test.session_configured.model.clone(),
+            effort: None,
+            summary: None,
+            service_tier: None,
+            collaboration_mode: None,
+            personality: None,
+        })
+        .await?;
+
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+
+    let requests = request_log.requests();
+    let bodies = requests
+        .into_iter()
+        .map(|request| request.body_json())
+        .collect::<Vec<_>>();
+    let outputs = collect_tool_outputs(&bodies)?;
+
+    let start_output = outputs
+        .get(start_call_id)
+        .expect("missing start output for nested zsh exec_command");
+    let normalized = start_output.output.replace("\r\n", "\n");
+    let nested_zsh_pid = Regex::new(r"CODEX_NESTED_ZSH_PID=(\d+)")
+        .expect("valid nested zsh pid regex")
+        .captures(&normalized)
+        .and_then(|captures| captures.get(1))
+        .map(|value| value.as_str().to_string())
+        .with_context(|| format!("missing nested zsh pid marker in output {normalized:?}"))?;
+    assert!(
+        process_is_alive(&nested_zsh_pid)?,
+        "nested zsh process should be running before release, got output {normalized:?}"
+    );
+
+    let nested_text_binary = process_text_binary_path(&nested_zsh_pid)?;
+    let nested_text_binary = fs::canonicalize(&nested_text_binary).unwrap_or(nested_text_binary);
+    assert_eq!(
+        nested_text_binary, configured_zsh_path,
+        "nested zsh exec should be rewritten to configured zsh-fork binary, got {:?}",
+        nested_text_binary,
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn unified_exec_emits_end_event_when_session_dies_via_stdin() -> Result<()> {
     skip_if_no_network!(Ok(()));