fix: prepare ExecPolicy in exec-server for execpolicy2 cutover (#6888)

This PR introduces an extra layer of abstraction to prepare us for the
migration to execpolicy2:

- introduces a new trait, `EscalationPolicy`, whose `determine_action()`
method is responsible for producing the `EscalateAction`
- the existing `ExecPolicy` typedef is changed to return an intermediate
`ExecPolicyOutcome` instead of `EscalateAction`
- the default implementation of `EscalationPolicy`,
`McpEscalationPolicy`, composes `ExecPolicy`
- the `ExecPolicyOutcome` includes `codex_execpolicy2::Decision`, which
has a `Prompt` variant
- when `McpEscalationPolicy` gets `Decision::Prompt` back from
`ExecPolicy`, it prompts the user via an MCP elicitation and maps the
result into an `ElicitationAction`
- now that the end user can reply to an elicitation with `Decline` or
`Cancel`, we introduce a new variant, `EscalateAction::Deny`, which the
client handles by returning exit code `1` without running anything

Note the way the elicitation is created is still not quite right, but I
will fix that once we have things running end-to-end for real in a
follow-up PR.

codex-rs/exec-server/src/posix/escalate_server.rs

@@ -1,8 +1,8 @@
 use std::collections::HashMap;
 use std::os::fd::AsRawFd;
-use std::path::Path;
 use std::path::PathBuf;
 use std::process::Stdio;
+use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context as _;
@@ -21,25 +21,24 @@ use crate::posix::escalate_protocol::EscalateRequest;
 use crate::posix::escalate_protocol::EscalateResponse;
 use crate::posix::escalate_protocol::SuperExecMessage;
 use crate::posix::escalate_protocol::SuperExecResult;
+use crate::posix::escalation_policy::EscalationPolicy;
 use crate::posix::socket::AsyncDatagramSocket;
 use crate::posix::socket::AsyncSocket;
 
-/// This is the policy which decides how to handle an exec() call.
-///
-/// `file` is the absolute, canonical path to the executable to run, i.e. the first arg to exec.
-/// `argv` is the argv, including the program name (`argv[0]`).
-/// `workdir` is the absolute, canonical path to the working directory in which to execute the
-/// command.
-pub(crate) type ExecPolicy = fn(file: &Path, argv: &[String], workdir: &Path) -> EscalateAction;
-
 pub(crate) struct EscalateServer {
     bash_path: PathBuf,
-    policy: ExecPolicy,
+    policy: Arc<dyn EscalationPolicy>,
 }
 
 impl EscalateServer {
-    pub fn new(bash_path: PathBuf, policy: ExecPolicy) -> Self {
-        Self { bash_path, policy }
+    pub fn new<P>(bash_path: PathBuf, policy: P) -> Self
+    where
+        P: EscalationPolicy + Send + Sync + 'static,
+    {
+        Self {
+            bash_path,
+            policy: Arc::new(policy),
+        }
     }
 
     pub async fn exec(
@@ -53,7 +52,7 @@ impl EscalateServer {
         let client_socket = escalate_client.into_inner();
         client_socket.set_cloexec(false)?;
 
-        let escalate_task = tokio::spawn(escalate_task(escalate_server, self.policy));
+        let escalate_task = tokio::spawn(escalate_task(escalate_server, self.policy.clone()));
         let mut env = env.clone();
         env.insert(
             ESCALATE_SOCKET_ENV_VAR.to_string(),
@@ -96,7 +95,10 @@ impl EscalateServer {
     }
 }
 
-async fn escalate_task(socket: AsyncDatagramSocket, policy: ExecPolicy) -> anyhow::Result<()> {
+async fn escalate_task(
+    socket: AsyncDatagramSocket,
+    policy: Arc<dyn EscalationPolicy>,
+) -> anyhow::Result<()> {
     loop {
         let (_, mut fds) = socket.receive_with_fds().await?;
         if fds.len() != 1 {
@@ -104,6 +106,7 @@ async fn escalate_task(socket: AsyncDatagramSocket, policy: ExecPolicy) -> anyho
             continue;
         }
         let stream_socket = AsyncSocket::from_fd(fds.remove(0))?;
+        let policy = policy.clone();
         tokio::spawn(async move {
             if let Err(err) = handle_escalate_session_with_policy(stream_socket, policy).await {
                 tracing::error!("escalate session failed: {err:?}");
@@ -122,7 +125,7 @@ pub(crate) struct ExecResult {
 
 async fn handle_escalate_session_with_policy(
     socket: AsyncSocket,
-    policy: ExecPolicy,
+    policy: Arc<dyn EscalationPolicy>,
 ) -> anyhow::Result<()> {
     let EscalateRequest {
         file,
@@ -132,8 +135,12 @@ async fn handle_escalate_session_with_policy(
     } = socket.receive::<EscalateRequest>().await?;
     let file = PathBuf::from(&file).absolutize()?.into_owned();
     let workdir = PathBuf::from(&workdir).absolutize()?.into_owned();
-    let action = policy(file.as_path(), &argv, &workdir);
+    let action = policy
+        .determine_action(file.as_path(), &argv, &workdir)
+        .await?;
+
     tracing::debug!("decided {action:?} for {file:?} {argv:?} {workdir:?}");
+
     match action {
         EscalateAction::Run => {
             socket
@@ -195,6 +202,13 @@ async fn handle_escalate_session_with_policy(
                 })
                 .await?;
         }
+        EscalateAction::Deny { reason } => {
+            socket
+                .send(EscalateResponse {
+                    action: EscalateAction::Deny { reason },
+                })
+                .await?;
+        }
     }
     Ok(())
 }
@@ -204,14 +218,33 @@ mod tests {
     use super::*;
     use pretty_assertions::assert_eq;
     use std::collections::HashMap;
+    use std::path::Path;
     use std::path::PathBuf;
 
+    struct DeterministicEscalationPolicy {
+        action: EscalateAction,
+    }
+
+    #[async_trait::async_trait]
+    impl EscalationPolicy for DeterministicEscalationPolicy {
+        async fn determine_action(
+            &self,
+            _file: &Path,
+            _argv: &[String],
+            _workdir: &Path,
+        ) -> Result<EscalateAction, rmcp::ErrorData> {
+            Ok(self.action.clone())
+        }
+    }
+
     #[tokio::test]
     async fn handle_escalate_session_respects_run_in_sandbox_decision() -> anyhow::Result<()> {
         let (server, client) = AsyncSocket::pair()?;
         let server_task = tokio::spawn(handle_escalate_session_with_policy(
             server,
-            |_file, _argv, _workdir| EscalateAction::Run,
+            Arc::new(DeterministicEscalationPolicy {
+                action: EscalateAction::Run,
+            }),
         ));
 
         client
@@ -238,7 +271,9 @@ mod tests {
         let (server, client) = AsyncSocket::pair()?;
         let server_task = tokio::spawn(handle_escalate_session_with_policy(
             server,
-            |_file, _argv, _workdir| EscalateAction::Escalate,
+            Arc::new(DeterministicEscalationPolicy {
+                action: EscalateAction::Escalate,
+            }),
         ));
 
         client