feat(core): zsh exec bridge (#12052)

zsh fork PR stack:
- https://github.com/openai/codex/pull/12051 
- https://github.com/openai/codex/pull/12052 👈 

### Summary
This PR introduces a feature-gated native shell runtime path that routes
shell execution through a patched zsh exec bridge, removing MCP-specific
behavior from the shell hot path while preserving existing
CommandExecution lifecycle semantics.

When shell_zsh_fork is enabled, shell commands run via patched zsh with
per-`execve` interception through EXEC_WRAPPER. Core receives wrapper
IPC requests over a Unix socket, applies existing approval policy, and
returns allow/deny before the subcommand executes.

### What’s included
**1) New zsh exec bridge runtime in core**
- Wrapper-mode entrypoint (maybe_run_zsh_exec_wrapper_mode) for
EXEC_WRAPPER invocations.
- Per-execution Unix-socket IPC handling for wrapper requests/responses.
- Approval callback integration using existing core approval
orchestration.
- Streaming stdout/stderr deltas to existing command output event
pipeline.
- Error handling for malformed IPC, denial/abort, and execution
failures.

**2) Session lifecycle integration**
SessionServices now owns a `ZshExecBridge`.
Session startup initializes bridge state; shutdown tears it down
cleanly.

**3) Shell runtime routing (feature-gated)**
When `shell_zsh_fork` is enabled:
- Build execution env/spec as usual.
- Add wrapper socket env wiring.
- Execute via `zsh_exec_bridge.execute_shell_request(...)` instead of
the regular shell path.
- Non-zsh-fork behavior remains unchanged.

**4) Config + feature wiring**
- Added `Feature::ShellZshFork` (under development).
- Added config support for `zsh_path` (optional absolute path to patched
zsh):
- `Config`, `ConfigToml`, `ConfigProfile`, overrides, and schema.
- Session startup validates that `zsh_path` exists/usable when zsh-fork
is enabled.
- Added startup test for missing `zsh_path` failure mode.

**5) Seatbelt/sandbox updates for wrapper IPC**
- Extended seatbelt policy generation to optionally allow outbound
connection to explicitly permitted Unix sockets.
- Wired sandboxing path to pass wrapper socket path through to seatbelt
policy generation.
- Added/updated seatbelt tests for explicit socket allow rule and
argument emission.

**6) Runtime entrypoint hooks**
- This allows the same binary to act as the zsh wrapper subprocess when
invoked via `EXEC_WRAPPER`.

**7) Tool selection behavior**
- ToolsConfig now prefers ShellCommand type when shell_zsh_fork is
enabled.
- Added test coverage for precedence with unified-exec enabled.

codex-rs/app-server-test-client/src/lib.rs

@@ -1,4 +1,5 @@
 use std::collections::VecDeque;
+use std::ffi::OsString;
 use std::fs;
 use std::fs::OpenOptions;
 use std::io::BufRead;
@@ -29,6 +30,7 @@ use codex_app_server_protocol::ClientRequest;
 use codex_app_server_protocol::CommandExecutionApprovalDecision;
 use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
 use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
+use codex_app_server_protocol::CommandExecutionStatus;
 use codex_app_server_protocol::DynamicToolSpec;
 use codex_app_server_protocol::FileChangeApprovalDecision;
 use codex_app_server_protocol::FileChangeRequestApprovalParams;
@@ -55,6 +57,7 @@ use codex_app_server_protocol::SendUserMessageParams;
 use codex_app_server_protocol::SendUserMessageResponse;
 use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ServerRequest;
+use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadListParams;
 use codex_app_server_protocol::ThreadListResponse;
 use codex_app_server_protocol::ThreadResumeParams;
@@ -78,6 +81,30 @@ use tungstenite::stream::MaybeTlsStream;
 use url::Url;
 use uuid::Uuid;
 
+const NOTIFICATIONS_TO_OPT_OUT: &[&str] = &[
+    // Legacy codex/event (v1-style) deltas.
+    "codex/event/agent_message_content_delta",
+    "codex/event/agent_message_delta",
+    "codex/event/agent_reasoning_delta",
+    "codex/event/reasoning_content_delta",
+    "codex/event/reasoning_raw_content_delta",
+    "codex/event/exec_command_output_delta",
+    // Other legacy events.
+    "codex/event/exec_approval_request",
+    "codex/event/exec_command_begin",
+    "codex/event/exec_command_end",
+    "codex/event/exec_output",
+    "codex/event/item_started",
+    "codex/event/item_completed",
+    // v2 item deltas.
+    "item/agentMessage/delta",
+    "item/plan/delta",
+    "item/commandExecution/outputDelta",
+    "item/fileChange/outputDelta",
+    "item/reasoning/summaryTextDelta",
+    "item/reasoning/textDelta",
+];
+
 /// Minimal launcher that initializes the Codex app-server and logs the handshake.
 #[derive(Parser)]
 #[command(author = "Codex", version, about = "Bootstrap Codex app-server", long_about = None)]
@@ -180,6 +207,18 @@ enum CliCommand {
         /// Follow-up user message for the second turn.
         follow_up_message: String,
     },
+    /// Trigger zsh-fork multi-subcommand approvals and assert expected approval behavior.
+    #[command(name = "trigger-zsh-fork-multi-cmd-approval")]
+    TriggerZshForkMultiCmdApproval {
+        /// Optional prompt; defaults to an explicit `/usr/bin/true && /usr/bin/true` command.
+        user_message: Option<String>,
+        /// Minimum number of command-approval callbacks expected in the turn.
+        #[arg(long, default_value_t = 2)]
+        min_approvals: usize,
+        /// One-based approval index to abort (e.g. --abort-on 2 aborts the second approval).
+        #[arg(long)]
+        abort_on: Option<usize>,
+    },
     /// Trigger the ChatGPT login flow and wait for completion.
     TestLogin,
     /// Fetch the current account rate limits from the Codex app-server.
@@ -265,6 +304,21 @@ pub fn run() -> Result<()> {
                 &dynamic_tools,
             )
         }
+        CliCommand::TriggerZshForkMultiCmdApproval {
+            user_message,
+            min_approvals,
+            abort_on,
+        } => {
+            let endpoint = resolve_endpoint(codex_bin, url)?;
+            trigger_zsh_fork_multi_cmd_approval(
+                &endpoint,
+                &config_overrides,
+                user_message,
+                min_approvals,
+                abort_on,
+                &dynamic_tools,
+            )
+        }
         CliCommand::TestLogin => {
             ensure_dynamic_tools_unused(&dynamic_tools, "test-login")?;
             let endpoint = resolve_endpoint(codex_bin, url)?;
@@ -470,6 +524,101 @@ fn send_message_v2_endpoint(
     )
 }
 
+fn trigger_zsh_fork_multi_cmd_approval(
+    endpoint: &Endpoint,
+    config_overrides: &[String],
+    user_message: Option<String>,
+    min_approvals: usize,
+    abort_on: Option<usize>,
+    dynamic_tools: &Option<Vec<DynamicToolSpec>>,
+) -> Result<()> {
+    if let Some(abort_on) = abort_on
+        && abort_on == 0
+    {
+        bail!("--abort-on must be >= 1 when provided");
+    }
+
+    let default_prompt = "Run this exact command using shell command execution without rewriting or splitting it: /usr/bin/true && /usr/bin/true";
+    let message = user_message.unwrap_or_else(|| default_prompt.to_string());
+
+    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    let initialize = client.initialize()?;
+    println!("< initialize response: {initialize:?}");
+
+    let thread_response = client.thread_start(ThreadStartParams {
+        dynamic_tools: dynamic_tools.clone(),
+        ..Default::default()
+    })?;
+    println!("< thread/start response: {thread_response:?}");
+
+    client.command_approval_behavior = match abort_on {
+        Some(index) => CommandApprovalBehavior::AbortOn(index),
+        None => CommandApprovalBehavior::AlwaysAccept,
+    };
+    client.command_approval_count = 0;
+    client.command_approval_item_ids.clear();
+    client.command_execution_statuses.clear();
+    client.last_turn_status = None;
+
+    let mut turn_params = TurnStartParams {
+        thread_id: thread_response.thread.id.clone(),
+        input: vec![V2UserInput::Text {
+            text: message,
+            text_elements: Vec::new(),
+        }],
+        ..Default::default()
+    };
+    turn_params.approval_policy = Some(AskForApproval::OnRequest);
+    turn_params.sandbox_policy = Some(SandboxPolicy::ReadOnly {
+        access: ReadOnlyAccess::FullAccess,
+    });
+
+    let turn_response = client.turn_start(turn_params)?;
+    println!("< turn/start response: {turn_response:?}");
+    client.stream_turn(&thread_response.thread.id, &turn_response.turn.id)?;
+
+    if client.command_approval_count < min_approvals {
+        bail!(
+            "expected at least {min_approvals} command approvals, got {}",
+            client.command_approval_count
+        );
+    }
+    let mut approvals_per_item = std::collections::BTreeMap::new();
+    for item_id in &client.command_approval_item_ids {
+        *approvals_per_item.entry(item_id.clone()).or_insert(0usize) += 1;
+    }
+    let max_approvals_for_one_item = approvals_per_item.values().copied().max().unwrap_or(0);
+    if max_approvals_for_one_item < min_approvals {
+        bail!(
+            "expected at least {min_approvals} approvals for one command item, got max {max_approvals_for_one_item} with map {approvals_per_item:?}"
+        );
+    }
+
+    let last_command_status = client.command_execution_statuses.last();
+    if abort_on.is_none() {
+        if last_command_status != Some(&CommandExecutionStatus::Completed) {
+            bail!("expected completed command execution, got {last_command_status:?}");
+        }
+        if client.last_turn_status != Some(TurnStatus::Completed) {
+            bail!(
+                "expected completed turn in all-accept flow, got {:?}",
+                client.last_turn_status
+            );
+        }
+    } else if last_command_status == Some(&CommandExecutionStatus::Completed) {
+        bail!(
+            "expected non-completed command execution in mixed approval/decline flow, got {last_command_status:?}"
+        );
+    }
+
+    println!(
+        "[zsh-fork multi-approval summary] approvals={}, approvals_per_item={approvals_per_item:?}, command_statuses={:?}, turn_status={:?}",
+        client.command_approval_count, client.command_execution_statuses, client.last_turn_status
+    );
+
+    Ok(())
+}
+
 fn resume_message_v2(
     endpoint: &Endpoint,
     config_overrides: &[String],
@@ -791,6 +940,17 @@ enum ClientTransport {
 struct CodexClient {
     transport: ClientTransport,
     pending_notifications: VecDeque<JSONRPCNotification>,
+    command_approval_behavior: CommandApprovalBehavior,
+    command_approval_count: usize,
+    command_approval_item_ids: Vec<String>,
+    command_execution_statuses: Vec<CommandExecutionStatus>,
+    last_turn_status: Option<TurnStatus>,
+}
+
+#[derive(Debug, Clone, Copy)]
+enum CommandApprovalBehavior {
+    AlwaysAccept,
+    AbortOn(usize),
 }
 
 impl CodexClient {
@@ -804,6 +964,14 @@ impl CodexClient {
     fn spawn_stdio(codex_bin: &Path, config_overrides: &[String]) -> Result<Self> {
         let codex_bin_display = codex_bin.display();
         let mut cmd = Command::new(codex_bin);
+        if let Some(codex_bin_parent) = codex_bin.parent() {
+            let mut path = OsString::from(codex_bin_parent.as_os_str());
+            if let Some(existing_path) = std::env::var_os("PATH") {
+                path.push(":");
+                path.push(existing_path);
+            }
+            cmd.env("PATH", path);
+        }
         for override_kv in config_overrides {
             cmd.arg("--config").arg(override_kv);
         }
@@ -831,6 +999,11 @@ impl CodexClient {
                 stdout: BufReader::new(stdout),
             },
             pending_notifications: VecDeque::new(),
+            command_approval_behavior: CommandApprovalBehavior::AlwaysAccept,
+            command_approval_count: 0,
+            command_approval_item_ids: Vec::new(),
+            command_execution_statuses: Vec::new(),
+            last_turn_status: None,
         })
     }
 
@@ -847,6 +1020,11 @@ impl CodexClient {
                 socket: Box::new(socket),
             },
             pending_notifications: VecDeque::new(),
+            command_approval_behavior: CommandApprovalBehavior::AlwaysAccept,
+            command_approval_count: 0,
+            command_approval_item_ids: Vec::new(),
+            command_execution_statuses: Vec::new(),
+            last_turn_status: None,
         })
     }
 
@@ -862,7 +1040,12 @@ impl CodexClient {
                 },
                 capabilities: Some(InitializeCapabilities {
                     experimental_api: true,
-                    opt_out_notification_methods: None,
+                    opt_out_notification_methods: Some(
+                        NOTIFICATIONS_TO_OPT_OUT
+                            .iter()
+                            .map(|method| (*method).to_string())
+                            .collect(),
+                    ),
                 }),
             },
         };
@@ -1121,10 +1304,14 @@ impl CodexClient {
                     println!("\n< item started: {:?}", payload.item);
                 }
                 ServerNotification::ItemCompleted(payload) => {
+                    if let ThreadItem::CommandExecution { status, .. } = payload.item.clone() {
+                        self.command_execution_statuses.push(status);
+                    }
                     println!("< item completed: {:?}", payload.item);
                 }
                 ServerNotification::TurnCompleted(payload) => {
                     if payload.turn.id == turn_id {
+                        self.last_turn_status = Some(payload.turn.status.clone());
                         println!("\n< turn/completed notification: {:?}", payload.turn.status);
                         if payload.turn.status == TurnStatus::Failed
                             && let Some(error) = payload.turn.error
@@ -1314,6 +1501,8 @@ impl CodexClient {
             "\n< commandExecution approval requested for thread {thread_id}, turn {turn_id}, item {item_id}, approval {}",
             approval_id.as_deref().unwrap_or("<none>")
         );
+        self.command_approval_count += 1;
+        self.command_approval_item_ids.push(item_id.clone());
         if let Some(reason) = reason.as_deref() {
             println!("< reason: {reason}");
         }
@@ -1332,11 +1521,21 @@ impl CodexClient {
             println!("< proposed execpolicy amendment: {execpolicy_amendment:?}");
         }
 
+        let decision = match self.command_approval_behavior {
+            CommandApprovalBehavior::AlwaysAccept => CommandExecutionApprovalDecision::Accept,
+            CommandApprovalBehavior::AbortOn(index) if self.command_approval_count == index => {
+                CommandExecutionApprovalDecision::Cancel
+            }
+            CommandApprovalBehavior::AbortOn(_) => CommandExecutionApprovalDecision::Accept,
+        };
         let response = CommandExecutionRequestApprovalResponse {
-            decision: CommandExecutionApprovalDecision::Accept,
+            decision: decision.clone(),
         };
         self.send_server_request_response(request_id, &response)?;
-        println!("< approved commandExecution request for item {item_id}");
+        println!(
+            "< commandExecution decision for approval #{} on item {item_id}: {:?}",
+            self.command_approval_count, decision
+        );
         Ok(())
     }