fix(exec-server): route remote network policy decisions

Co-authored-by: Codex noreply@openai.com

codex-rs/core/src/tools/network_approval.rs

@@ -51,6 +51,7 @@ pub(crate) enum NetworkApprovalMode {
 pub(crate) struct NetworkApprovalSpec {
     pub network: Option<NetworkProxy>,
     pub mode: NetworkApprovalMode,
+    pub execution_scoped_proxy: bool,
     pub trigger: GuardianNetworkAccessTrigger,
     pub command: String,
     pub environment_id: String,
@@ -857,6 +858,7 @@ pub(crate) async fn begin_network_approval(
     let NetworkApprovalSpec {
         network,
         mode,
+        execution_scoped_proxy,
         trigger,
         command,
         environment_id,
@@ -872,7 +874,8 @@ pub(crate) async fn begin_network_approval(
     }
 
     let registration_id = Uuid::new_v4().to_string();
-    let execution_proxy = if selected_sandbox == SandboxType::LinuxSeccomp {
+    let execution_proxy = if execution_scoped_proxy || selected_sandbox == SandboxType::LinuxSeccomp
+    {
         let attribution_token = Uuid::new_v4().to_string();
         network
             .for_execution(&environment_id, &registration_id, attribution_token)

codex-rs/core/src/tools/runtimes/shell.rs

@@ -223,6 +223,7 @@ impl ToolRuntime<ShellRequest, ExecToolCallOutput> for ShellRuntime {
         Some(NetworkApprovalSpec {
             network: Some(network.clone()),
             mode: NetworkApprovalMode::Immediate,
+            execution_scoped_proxy: false,
             trigger: GuardianNetworkAccessTrigger {
                 call_id: ctx.call_id.clone(),
                 tool_name: flat_tool_name(&ctx.tool_name).into_owned(),

codex-rs/core/src/tools/runtimes/unified_exec.rs

@@ -274,6 +274,7 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
         Some(NetworkApprovalSpec {
             network: Some(network.clone()),
             mode: NetworkApprovalMode::Deferred,
+            execution_scoped_proxy: req.turn_environment.environment.is_remote(),
             trigger: GuardianNetworkAccessTrigger {
                 call_id: ctx.call_id.clone(),
                 tool_name: flat_tool_name(&ctx.tool_name).into_owned(),
@@ -325,9 +326,10 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
         let env = exec_env_for_sandbox_permissions(&req.env, launch_sandbox_permissions);
         let (env, managed_network_context, network_proxy_launch) = match managed_network {
             Some(network) if environment_is_remote => {
-                let launch = network.remote_launch_config().await.map_err(|err| {
+                let mut launch = network.remote_launch_config().await.map_err(|err| {
                     ToolError::Codex(CodexErr::Io(io::Error::other(err.to_string())))
                 })?;
+                launch.proxy.request_policy_decisions = network.has_policy_decider();
                 (env, None, Some(launch))
             }
             Some(network) => {

codex-rs/core/src/unified_exec/process_manager.rs

@@ -923,13 +923,17 @@ impl UnifiedExecProcessManager {
         spawn_lifecycle: SpawnLifecycleHandle,
         environment: &codex_exec_server::Environment,
     ) -> Result<UnifiedExecProcess, ToolError> {
-        let mut request = if environment.is_remote() {
+        let remote = environment.is_remote();
+        let mut request = if remote {
             attempt.env_for_exec_server(command, options)
         } else {
             attempt.env_for(command, options, network, environment_id)
         }
         .map_err(ToolError::Codex)?;
         request.exec_server_network_proxy = network_proxy_launch;
+        if remote {
+            request.network = network.cloned();
+        }
         request.exec_server_env_config = exec_server_env_config;
         self.open_session_with_prepared_exec_env(
             process_id,
@@ -1053,11 +1057,21 @@ impl UnifiedExecProcessManager {
                 ));
             }
 
-            let started = environment
-                .get_exec_backend()
-                .start(exec_server_params_for_request(process_id, request, tty))
-                .await
-                .map_err(|err| UnifiedExecError::create_process(err.to_string()))?;
+            let backend = environment.get_exec_backend();
+            let params = exec_server_params_for_request(process_id, request, tty);
+            let started = match request
+                .network
+                .as_ref()
+                .and_then(NetworkProxy::remote_policy_decider)
+            {
+                Some(decider) => {
+                    backend
+                        .start_with_network_policy_decider(params, decider)
+                        .await
+                }
+                None => backend.start(params).await,
+            }
+            .map_err(|err| UnifiedExecError::create_process(err.to_string()))?;
             spawn_lifecycle.after_spawn();
             return UnifiedExecProcess::from_exec_server_started(started).await;
         }

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

@@ -232,6 +232,83 @@ async fn guardian_receives_exact_trigger_for_single_network_request() -> Result<
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn remote_guardian_receives_exact_trigger_for_network_request() -> Result<()> {
+    skip_if_target_windows!(Ok(()), "uses the POSIX/Python network fixture");
+    skip_if_host_windows!(Ok(()));
+    skip_if_no_network!(Ok(()));
+    skip_if_sandbox!(Ok(()));
+    skip_if_no_remote_env!(Ok(()));
+
+    let server = start_mock_server().await;
+    let test = managed_network_unified_exec_test(&server).await?;
+    let command = "python3 -c \"import urllib.request; urllib.request.build_opener(urllib.request.ProxyHandler()).open('http://1.1.1.1', timeout=10).read()\"".to_string();
+    mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| {
+            !is_guardian_request(request)
+                && request_body_contains(request, "run one remote network request")
+                && !request_body_contains(request, "exec-network-remote-guardian")
+        },
+        sse(vec![
+            ev_response_created("resp-network-remote"),
+            ev_function_call(
+                "exec-network-remote-guardian",
+                "exec_command",
+                &serde_json::to_string(&network_exec_args(&command))?,
+            ),
+            ev_completed("resp-network-remote"),
+        ]),
+    )
+    .await;
+    let guardian = mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| {
+            guardian_request_is_for(request, "exec-network-remote-guardian")
+        },
+        sse(vec![
+            ev_response_created("resp-network-remote-guardian"),
+            ev_assistant_message("msg-network-remote-guardian", r#"{"outcome":"deny"}"#),
+            ev_completed("resp-network-remote-guardian"),
+        ]),
+    )
+    .await;
+    mount_sse_once_match(
+        &server,
+        |request: &wiremock::Request| {
+            !is_guardian_request(request)
+                && request_body_contains(request, "exec-network-remote-guardian")
+        },
+        sse(vec![
+            ev_response_created("resp-network-remote-done"),
+            ev_assistant_message("msg-network-remote-done", "done"),
+            ev_completed("resp-network-remote-done"),
+        ]),
+    )
+    .await;
+    let remote = test.executor_environment().selection().clone();
+    assert_eq!(remote.environment_id, REMOTE_ENVIRONMENT_ID);
+
+    submit_managed_network_turn(
+        &test,
+        "run one remote network request",
+        vec![remote],
+        ApprovalsReviewer::AutoReview,
+        AskForApproval::OnRequest,
+    )
+    .await?;
+    tokio::time::timeout(Duration::from_secs(5), wait_for_turn_complete(&test))
+        .await
+        .context("remote Guardian denial should return before the command network timeout")?;
+
+    assert_eq!(
+        guardian_network_triggers(&[&guardian])?,
+        vec![("exec-network-remote-guardian".to_string(), command)]
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn approved_network_host_for_one_environment_still_prompts_in_another() -> Result<()> {
     skip_if_target_windows!(Ok(()), "uses the POSIX/Python network fixture");

codex-rs/exec-server-protocol/src/protocol.rs

@@ -5,6 +5,8 @@ use codex_file_system::FileSystemSandboxContext;
 pub use codex_file_system::WalkOptions;
 pub use codex_file_system::WalkOutcome;
 use codex_network_proxy::ManagedNetworkSandboxContext;
+use codex_network_proxy::NetworkDecision;
+use codex_network_proxy::NetworkPolicyRequest;
 use codex_network_proxy::RemoteNetworkProxyLaunchConfig;
 use codex_protocol::config_types::ShellEnvironmentPolicyInherit;
 use codex_shell_command::shell_detect::DetectedShell;
@@ -24,6 +26,8 @@ pub const EXEC_TERMINATE_METHOD: &str = "process/terminate";
 pub const EXEC_OUTPUT_DELTA_METHOD: &str = "process/output";
 pub const EXEC_EXITED_METHOD: &str = "process/exited";
 pub const EXEC_CLOSED_METHOD: &str = "process/closed";
+pub const NETWORK_POLICY_REQUEST_METHOD: &str = "network/policyRequest";
+pub const NETWORK_POLICY_DECISION_METHOD: &str = "network/policyDecision";
 pub const ENVIRONMENT_INFO_METHOD: &str = "environment/info";
 pub const FS_READ_FILE_METHOD: &str = "fs/readFile";
 pub const FS_OPEN_METHOD: &str = "fs/open";
@@ -166,6 +170,22 @@ pub struct ExecResponse {
     pub process_id: ProcessId,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NetworkPolicyRequestNotification {
+    pub request_id: String,
+    pub process_id: ProcessId,
+    pub request: NetworkPolicyRequest,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NetworkPolicyDecisionNotification {
+    pub request_id: String,
+    pub process_id: ProcessId,
+    pub decision: NetworkDecision,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ReadParams {

codex-rs/exec-server/src/client.rs

@@ -10,6 +10,8 @@ use std::time::Duration;
 
 use arc_swap::ArcSwap;
 use codex_exec_server_protocol::JSONRPCNotification;
+use codex_network_proxy::NetworkDecision;
+use codex_network_proxy::NetworkPolicyDecider;
 use futures::FutureExt;
 use futures::future::BoxFuture;
 use serde_json::Value;
@@ -91,6 +93,10 @@ use crate::protocol::INITIALIZE_METHOD;
 use crate::protocol::INITIALIZED_METHOD;
 use crate::protocol::InitializeParams;
 use crate::protocol::InitializeResponse;
+use crate::protocol::NETWORK_POLICY_DECISION_METHOD;
+use crate::protocol::NETWORK_POLICY_REQUEST_METHOD;
+use crate::protocol::NetworkPolicyDecisionNotification;
+use crate::protocol::NetworkPolicyRequestNotification;
 use crate::protocol::ProcessOutputChunk;
 use crate::protocol::ProcessSignal;
 use crate::protocol::ReadParams;
@@ -162,6 +168,7 @@ pub(crate) struct SessionState {
     ordered_events: StdMutex<OrderedSessionEvents>,
     recoverable: AtomicBool,
     next_write_id: AtomicU64,
+    network_policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
 }
 
 #[derive(Default)]
@@ -648,6 +655,24 @@ impl ExecServerClient {
     pub(crate) async fn start_process(
         &self,
         params: ExecParams,
+    ) -> Result<Session, ExecServerError> {
+        self.start_process_inner(params, /*network_policy_decider*/ None)
+            .await
+    }
+
+    pub(crate) async fn start_process_with_network_policy_decider(
+        &self,
+        params: ExecParams,
+        network_policy_decider: Arc<dyn NetworkPolicyDecider>,
+    ) -> Result<Session, ExecServerError> {
+        self.start_process_inner(params, Some(network_policy_decider))
+            .await
+    }
+
+    async fn start_process_inner(
+        &self,
+        params: ExecParams,
+        network_policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
     ) -> Result<Session, ExecServerError> {
         loop {
             let rpc_client = self.inner.rpc_client().await?;
@@ -656,7 +681,10 @@ impl ExecServerClient {
             }
 
             let process_id = params.process_id.clone();
-            let state = Arc::new(SessionState::new(/*recoverable*/ false));
+            let state = Arc::new(SessionState::new(
+                /*recoverable*/ false,
+                network_policy_decider.clone(),
+            ));
             if let Err(error) = self.inner.insert_session(&process_id, Arc::clone(&state)) {
                 self.inner.finish_process_start();
                 return Err(error);
@@ -710,7 +738,9 @@ impl ExecServerClient {
         &self,
         process_id: &ProcessId,
     ) -> Result<Session, ExecServerError> {
-        let state = Arc::new(SessionState::new(/*recoverable*/ true));
+        let state = Arc::new(SessionState::new(
+            /*recoverable*/ true, /*network_policy_decider*/ None,
+        ));
         self.inner.insert_session(process_id, Arc::clone(&state))?;
         Ok(Session {
             client: self.clone(),
@@ -835,7 +865,10 @@ impl From<RpcCallError> for ExecServerError {
 }
 
 impl SessionState {
-    fn new(recoverable: bool) -> Self {
+    fn new(
+        recoverable: bool,
+        network_policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
+    ) -> Self {
         let (wake_tx, _wake_rx) = watch::channel(0);
         Self {
             wake_tx,
@@ -846,6 +879,7 @@ impl SessionState {
             ordered_events: StdMutex::new(OrderedSessionEvents::default()),
             recoverable: AtomicBool::new(recoverable),
             next_write_id: AtomicU64::new(1),
+            network_policy_decider,
         }
     }
 
@@ -1196,6 +1230,37 @@ async fn handle_server_notification(
                 .handle_http_body_delta_notification(notification.params)
                 .await?;
         }
+        NETWORK_POLICY_REQUEST_METHOD => {
+            let params: NetworkPolicyRequestNotification =
+                serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
+            let decider = inner
+                .get_session(&params.process_id)
+                .and_then(|session| session.network_policy_decider.clone());
+            let inner = Arc::clone(inner);
+            tokio::spawn(async move {
+                let decision = match decider {
+                    Some(decider) => decider.decide(params.request).await,
+                    None => NetworkDecision::deny("not_allowed"),
+                };
+                let response = NetworkPolicyDecisionNotification {
+                    request_id: params.request_id,
+                    process_id: params.process_id,
+                    decision,
+                };
+                let Ok(rpc_client) = inner.rpc_client().await else {
+                    return;
+                };
+                if let Err(err) = rpc_client
+                    .notify(NETWORK_POLICY_DECISION_METHOD, &response)
+                    .await
+                {
+                    debug!(
+                        ?err,
+                        "failed to send network policy decision to exec-server"
+                    );
+                }
+            });
+        }
         other => {
             debug!("ignoring unknown exec-server notification: {other}");
         }

codex-rs/exec-server/src/client_recovery_tests.rs

@@ -59,7 +59,9 @@ fn recovery_does_not_retry_other_registry_conflicts() {
 
 #[tokio::test]
 async fn recovery_adds_sandbox_denial_to_pending_exit_event() {
-    let state = SessionState::new(/*recoverable*/ true);
+    let state = SessionState::new(
+        /*recoverable*/ true, /*network_policy_decider*/ None,
+    );
     assert!(!state.publish_ordered_event(ExecProcessEvent::Exited {
         seq: 2,
         exit_code: 1,

codex-rs/exec-server/src/lib.rs

@@ -12,6 +12,7 @@ mod fs_helper_main;
 mod fs_sandbox;
 mod local_file_system;
 mod local_process;
+mod network_policy_decisions;
 mod noise_channel;
 mod noise_relay;
 mod process;
@@ -133,6 +134,8 @@ pub use protocol::HttpRequestParams;
 pub use protocol::HttpRequestResponse;
 pub use protocol::InitializeParams;
 pub use protocol::InitializeResponse;
+pub use protocol::NetworkPolicyDecisionNotification;
+pub use protocol::NetworkPolicyRequestNotification;
 pub use protocol::ProcessOutputChunk;
 pub use protocol::ProcessSignal;
 pub use protocol::ReadParams;

codex-rs/exec-server/src/local_process.rs

@@ -34,6 +34,7 @@ use crate::ExecServerError;
 use crate::ExecServerRuntimePaths;
 use crate::ProcessId;
 use crate::StartedExecProcess;
+use crate::network_policy_decisions::NetworkPolicyDecisionRelay;
 use crate::process::ExecProcessEventLog;
 use crate::process_sandbox::prepare_exec_request;
 use crate::protocol::EXEC_CLOSED_METHOD;
@@ -44,6 +45,7 @@ use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecOutputStream;
 use crate::protocol::ExecParams;
 use crate::protocol::ExecResponse;
+use crate::protocol::NetworkPolicyDecisionNotification;
 use crate::protocol::ProcessOutputChunk;
 use crate::protocol::ProcessSignal;
 use crate::protocol::ReadParams;
@@ -140,9 +142,10 @@ enum ProcessEntry {
 }
 
 struct Inner {
-    notifications: std::sync::RwLock<Option<RpcNotificationSender>>,
+    notifications: Arc<std::sync::RwLock<Option<RpcNotificationSender>>>,
     processes: Mutex<HashMap<ProcessId, ProcessEntry>>,
     telemetry: ExecServerTelemetry,
+    network_policy_decisions: Arc<NetworkPolicyDecisionRelay>,
 }
 
 #[derive(Clone)]
@@ -195,15 +198,19 @@ impl LocalProcess {
     ) -> Self {
         Self {
             inner: Arc::new(Inner {
-                notifications: std::sync::RwLock::new(Some(notifications)),
+                notifications: Arc::new(std::sync::RwLock::new(Some(notifications))),
                 processes: Mutex::new(HashMap::new()),
                 telemetry,
+                network_policy_decisions: Arc::new(NetworkPolicyDecisionRelay::default()),
             }),
             runtime_paths,
         }
     }
 
     pub(crate) async fn shutdown(&self) {
+        self.inner
+            .network_policy_decisions
+            .fail_pending(/*process_id*/ None);
         let remaining = {
             let mut processes = self.inner.processes.lock().await;
             processes
@@ -223,12 +230,25 @@ impl LocalProcess {
     }
 
     pub(crate) fn set_notification_sender(&self, notifications: Option<RpcNotificationSender>) {
+        let disconnected = notifications.is_none();
         let mut notification_sender = self
             .inner
             .notifications
             .write()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
         *notification_sender = notifications;
+        if disconnected {
+            self.inner
+                .network_policy_decisions
+                .fail_pending(/*process_id*/ None);
+        }
+    }
+
+    pub(crate) fn resolve_network_policy_decision(
+        &self,
+        params: NetworkPolicyDecisionNotification,
+    ) -> Result<(), String> {
+        self.inner.network_policy_decisions.resolve(params)
     }
 
     async fn start_process(
@@ -236,8 +256,23 @@ impl LocalProcess {
         params: ExecParams,
     ) -> Result<(ExecResponse, watch::Sender<u64>, ExecProcessEventLog), JSONRPCErrorError> {
         let process_id = params.process_id.clone();
-        let prepared =
-            prepare_exec_request(&params, child_env(&params), self.runtime_paths.as_ref()).await?;
+        let network_policy_decider = params
+            .managed_network
+            .as_ref()
+            .and_then(|context| context.proxy_config.as_ref())
+            .is_some_and(|config| config.request_policy_decisions)
+            .then(|| {
+                self.inner
+                    .network_policy_decisions
+                    .decider(process_id.clone(), Arc::clone(&self.inner.notifications))
+            });
+        let prepared = prepare_exec_request(
+            &params,
+            child_env(&params),
+            self.runtime_paths.as_ref(),
+            network_policy_decider,
+        )
+        .await?;
         let (program, args) = prepared
             .command
             .split_first()
@@ -850,6 +885,9 @@ async fn watch_exit(
     {
         tracing::warn!("failed to shut down executor network proxy: {err}");
     }
+    inner
+        .network_policy_decisions
+        .fail_pending(Some(&process_id));
     if sandboxed {
         let _ = tokio::time::timeout(Duration::from_millis(20), output_notify.notified()).await;
     }

codex-rs/exec-server/src/network_policy_decisions.rs

@@ -0,0 +1,127 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::RwLock;
+
+use codex_network_proxy::NetworkDecision;
+use codex_network_proxy::NetworkPolicyDecider;
+use codex_network_proxy::NetworkPolicyRequest;
+use tokio::sync::oneshot;
+use uuid::Uuid;
+
+use crate::ProcessId;
+use crate::protocol::NETWORK_POLICY_REQUEST_METHOD;
+use crate::protocol::NetworkPolicyDecisionNotification;
+use crate::protocol::NetworkPolicyRequestNotification;
+use crate::rpc::RpcNotificationSender;
+
+#[derive(Default)]
+pub(crate) struct NetworkPolicyDecisionRelay {
+    pending: Mutex<HashMap<String, PendingNetworkPolicyDecision>>,
+}
+
+struct PendingNetworkPolicyDecision {
+    process_id: ProcessId,
+    response_tx: oneshot::Sender<NetworkDecision>,
+}
+
+impl NetworkPolicyDecisionRelay {
+    pub(crate) fn decider(
+        self: &Arc<Self>,
+        process_id: ProcessId,
+        notifications: Arc<RwLock<Option<RpcNotificationSender>>>,
+    ) -> Arc<dyn NetworkPolicyDecider> {
+        let relay = Arc::clone(self);
+        Arc::new(move |request: NetworkPolicyRequest| {
+            let relay = Arc::clone(&relay);
+            let process_id = process_id.clone();
+            let notifications = Arc::clone(&notifications);
+            async move { relay.request(process_id, request, notifications).await }
+        })
+    }
+
+    pub(crate) fn resolve(&self, params: NetworkPolicyDecisionNotification) -> Result<(), String> {
+        let mut pending_requests = self
+            .pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let Some(request) = pending_requests.get(&params.request_id) else {
+            return Ok(());
+        };
+        if request.process_id != params.process_id {
+            return Err("network policy decision process id does not match request".to_string());
+        }
+        let Some(pending) = pending_requests.remove(&params.request_id) else {
+            return Ok(());
+        };
+        drop(pending_requests);
+        let _ = pending.response_tx.send(params.decision);
+        Ok(())
+    }
+
+    pub(crate) fn fail_pending(&self, process_id: Option<&ProcessId>) {
+        let mut pending = self
+            .pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        match process_id {
+            Some(process_id) => pending.retain(|_, pending| pending.process_id != *process_id),
+            None => pending.clear(),
+        }
+    }
+
+    async fn request(
+        &self,
+        process_id: ProcessId,
+        request: NetworkPolicyRequest,
+        notifications: Arc<RwLock<Option<RpcNotificationSender>>>,
+    ) -> NetworkDecision {
+        let request_id = Uuid::new_v4().to_string();
+        let (response_tx, response_rx) = oneshot::channel();
+        self.pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(
+                request_id.clone(),
+                PendingNetworkPolicyDecision {
+                    process_id: process_id.clone(),
+                    response_tx,
+                },
+            );
+        let notification = NetworkPolicyRequestNotification {
+            request_id: request_id.clone(),
+            process_id,
+            request,
+        };
+        let Some(notifications) = notifications
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone()
+        else {
+            self.fail(&request_id);
+            return NetworkDecision::deny("not_allowed");
+        };
+        if notifications
+            .notify(NETWORK_POLICY_REQUEST_METHOD, &notification)
+            .await
+            .is_err()
+        {
+            self.fail(&request_id);
+            return NetworkDecision::deny("not_allowed");
+        }
+        response_rx
+            .await
+            .unwrap_or_else(|_| NetworkDecision::deny("not_allowed"))
+    }
+
+    fn fail(&self, request_id: &str) {
+        self.pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .remove(request_id);
+    }
+}
+
+#[cfg(test)]
+#[path = "network_policy_decisions_tests.rs"]
+mod tests;

codex-rs/exec-server/src/network_policy_decisions_tests.rs

@@ -0,0 +1,198 @@
+use std::sync::Arc;
+use std::sync::RwLock;
+use std::time::Duration;
+
+use codex_network_proxy::NetworkDecision;
+use codex_network_proxy::NetworkPolicyRequest;
+use codex_network_proxy::NetworkPolicyRequestArgs;
+use codex_network_proxy::NetworkProtocol;
+use pretty_assertions::assert_eq;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use tokio::time::timeout;
+
+use super::*;
+use crate::protocol::NETWORK_POLICY_REQUEST_METHOD;
+use crate::rpc::RpcServerOutboundMessage;
+
+const NOTIFICATION_CHANNEL_CAPACITY: usize = 8;
+
+fn network_policy_request(host: &str) -> NetworkPolicyRequest {
+    NetworkPolicyRequest::new(NetworkPolicyRequestArgs {
+        protocol: NetworkProtocol::HttpsConnect,
+        host: host.to_string(),
+        port: 443,
+        environment_id: None,
+        client_addr: None,
+        method: None,
+        command: None,
+        exec_policy_hint: None,
+    })
+}
+
+async fn next_network_policy_request(
+    outgoing_rx: &mut mpsc::Receiver<RpcServerOutboundMessage>,
+) -> NetworkPolicyRequestNotification {
+    let outbound = timeout(Duration::from_secs(1), outgoing_rx.recv())
+        .await
+        .expect("policy request notification should arrive")
+        .expect("policy request notification");
+    let RpcServerOutboundMessage::Notification(notification) = outbound else {
+        panic!("expected policy request notification");
+    };
+    assert_eq!(notification.method, NETWORK_POLICY_REQUEST_METHOD);
+    serde_json::from_value(
+        notification
+            .params
+            .expect("policy request notification params"),
+    )
+    .expect("deserialize policy request notification")
+}
+
+fn network_policy_relay() -> (
+    Arc<NetworkPolicyDecisionRelay>,
+    Arc<RwLock<Option<RpcNotificationSender>>>,
+    mpsc::Receiver<RpcServerOutboundMessage>,
+) {
+    let (outgoing_tx, outgoing_rx) = mpsc::channel(NOTIFICATION_CHANNEL_CAPACITY);
+    (
+        Arc::new(NetworkPolicyDecisionRelay::default()),
+        Arc::new(RwLock::new(Some(RpcNotificationSender::new(outgoing_tx)))),
+        outgoing_rx,
+    )
+}
+
+fn request_decision(
+    relay: &Arc<NetworkPolicyDecisionRelay>,
+    notifications: &Arc<RwLock<Option<RpcNotificationSender>>>,
+    process_id: ProcessId,
+    host: &str,
+) -> JoinHandle<NetworkDecision> {
+    let decider = relay.decider(process_id, Arc::clone(notifications));
+    let request = network_policy_request(host);
+    tokio::spawn(async move { decider.decide(request).await })
+}
+
+async fn await_decision(decision: JoinHandle<NetworkDecision>) -> NetworkDecision {
+    timeout(Duration::from_secs(1), decision)
+        .await
+        .expect("network policy decision should resolve")
+        .expect("network policy decision task")
+}
+
+#[tokio::test]
+async fn decisions_correlate_concurrent_requests() {
+    let (relay, notifications, mut outgoing_rx) = network_policy_relay();
+    let process_a = ProcessId::from("network-decision-a");
+    let process_b = ProcessId::from("network-decision-b");
+    let decision_a = request_decision(&relay, &notifications, process_a.clone(), "a.example.com");
+    let decision_b = request_decision(&relay, &notifications, process_b.clone(), "b.example.com");
+    let first = next_network_policy_request(&mut outgoing_rx).await;
+    let second = next_network_policy_request(&mut outgoing_rx).await;
+    let (request_a, request_b) = if first.process_id == process_a {
+        (first, second)
+    } else {
+        (second, first)
+    };
+    assert_eq!(request_a.request.host, "a.example.com");
+    assert_eq!(request_b.request.host, "b.example.com");
+
+    let error = relay
+        .resolve(NetworkPolicyDecisionNotification {
+            request_id: request_a.request_id.clone(),
+            process_id: process_b.clone(),
+            decision: NetworkDecision::Allow,
+        })
+        .expect_err("mismatched process must not resolve request");
+    assert_eq!(
+        error,
+        "network policy decision process id does not match request"
+    );
+
+    relay
+        .resolve(NetworkPolicyDecisionNotification {
+            request_id: request_b.request_id,
+            process_id: process_b,
+            decision: NetworkDecision::Allow,
+        })
+        .expect("resolve second network policy decision");
+    let denied = NetworkDecision::deny("not_allowed");
+    relay
+        .resolve(NetworkPolicyDecisionNotification {
+            request_id: request_a.request_id,
+            process_id: process_a,
+            decision: denied.clone(),
+        })
+        .expect("resolve first network policy decision");
+    assert_eq!(await_decision(decision_a).await, denied);
+    assert_eq!(await_decision(decision_b).await, NetworkDecision::Allow);
+}
+
+#[tokio::test]
+async fn decisions_fail_closed_on_disconnect() {
+    let (relay, notifications, mut outgoing_rx) = network_policy_relay();
+    let decision = request_decision(
+        &relay,
+        &notifications,
+        ProcessId::from("network-decision-disconnect"),
+        "disconnect.example.com",
+    );
+    next_network_policy_request(&mut outgoing_rx).await;
+
+    *notifications
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
+    relay.fail_pending(/*process_id*/ None);
+
+    assert_eq!(
+        await_decision(decision).await,
+        NetworkDecision::deny("not_allowed")
+    );
+}
+
+#[tokio::test]
+async fn process_exit_fails_only_its_decisions() {
+    let (relay, notifications, mut outgoing_rx) = network_policy_relay();
+    let process_a = ProcessId::from("network-decision-exit-a");
+    let process_b = ProcessId::from("network-decision-exit-b");
+    let decision_a = request_decision(
+        &relay,
+        &notifications,
+        process_a.clone(),
+        "exit-a.example.com",
+    );
+    let mut decision_b = request_decision(
+        &relay,
+        &notifications,
+        process_b.clone(),
+        "exit-b.example.com",
+    );
+    let first = next_network_policy_request(&mut outgoing_rx).await;
+    let second = next_network_policy_request(&mut outgoing_rx).await;
+    let request_b = if first.process_id == process_b {
+        first
+    } else {
+        second
+    };
+
+    relay.fail_pending(Some(&process_a));
+
+    assert_eq!(
+        await_decision(decision_a).await,
+        NetworkDecision::deny("not_allowed")
+    );
+    assert!(
+        timeout(Duration::from_millis(20), &mut decision_b)
+            .await
+            .is_err(),
+        "another process decision should remain pending"
+    );
+    relay
+        .resolve(NetworkPolicyDecisionNotification {
+            request_id: request_b.request_id,
+            process_id: process_b,
+            decision: NetworkDecision::Allow,
+        })
+        .expect("resolve remaining process decision");
+    assert_eq!(await_decision(decision_b).await, NetworkDecision::Allow);
+}

codex-rs/exec-server/src/process.rs

@@ -202,6 +202,14 @@ pub type ExecProcessFuture<'a, T> =
 
 pub trait ExecBackend: Send + Sync {
     fn start(&self, params: ExecParams) -> ExecBackendFuture<'_>;
+
+    fn start_with_network_policy_decider(
+        &self,
+        params: ExecParams,
+        _decider: Arc<dyn codex_network_proxy::NetworkPolicyDecider>,
+    ) -> ExecBackendFuture<'_> {
+        self.start(params)
+    }
 }
 
 pub type ExecBackendFuture<'a> =

codex-rs/exec-server/src/process_sandbox.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_network_proxy::CUSTOM_CA_ENV_KEYS;
 use codex_network_proxy::ManagedNetworkSandboxContext;
+use codex_network_proxy::NetworkPolicyDecider;
 use codex_network_proxy::NetworkProxy;
 use codex_network_proxy::NetworkProxyHandle;
 use codex_network_proxy::NetworkProxyState;
@@ -38,11 +39,13 @@ pub(crate) async fn prepare_exec_request(
     params: &ExecParams,
     env: HashMap<String, String>,
     runtime_paths: Option<&ExecServerRuntimePaths>,
+    network_policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
 ) -> Result<PreparedExecRequest, JSONRPCErrorError> {
     let (env, managed_network, network_proxy_handle) = prepare_managed_network(
         params.managed_network.as_ref(),
         params.network_proxy.as_ref(),
         env,
+        network_policy_decider,
     )
     .await?;
     let Some(sandbox_context) = params.sandbox.as_ref() else {
@@ -162,6 +165,7 @@ async fn prepare_managed_network(
     managed_network: Option<&ManagedNetworkSandboxContext>,
     network_proxy: Option<&RemoteNetworkProxyLaunchConfig>,
     env: HashMap<String, String>,
+    network_policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
 ) -> Result<
     (
         HashMap<String, String>,
@@ -175,8 +179,11 @@ async fn prepare_managed_network(
     };
     let state = NetworkProxyState::from_remote_launch_config(network_proxy)
         .map_err(|err| invalid_params(format!("invalid network proxy config: {err}")))?;
-    let proxy = NetworkProxy::builder()
-        .state(Arc::new(state))
+    let mut builder = NetworkProxy::builder().state(Arc::new(state));
+    if let Some(network_policy_decider) = network_policy_decider {
+        builder = builder.policy_decider_arc(network_policy_decider);
+    }
+    let proxy = builder
         .build()
         .await
         .map_err(|err| internal_error(format!("failed to build executor network proxy: {err}")))?;

codex-rs/exec-server/src/process_sandbox_tests.rs

@@ -58,9 +58,14 @@ async fn sandbox_request_wraps_native_argv_on_executor() {
         network_proxy: None,
     };
 
-    let prepared = prepare_exec_request(&params, HashMap::new(), Some(&runtime_paths))
-        .await
-        .expect("prepare sandboxed request");
+    let prepared = prepare_exec_request(
+        &params,
+        HashMap::new(),
+        Some(&runtime_paths),
+        /*network_policy_decider*/ None,
+    )
+    .await
+    .expect("prepare sandboxed request");
 
     assert_ne!(prepared.command, params.argv);
     assert_eq!(prepared.cwd, cwd);
@@ -124,9 +129,14 @@ async fn sandbox_request_allows_prepared_managed_proxy_port() {
         network_proxy: None,
     };
 
-    let prepared = prepare_exec_request(&params, HashMap::new(), Some(&runtime_paths))
-        .await
-        .expect("prepare managed-network sandbox request");
+    let prepared = prepare_exec_request(
+        &params,
+        HashMap::new(),
+        Some(&runtime_paths),
+        /*network_policy_decider*/ None,
+    )
+    .await
+    .expect("prepare managed-network sandbox request");
     let policy = prepared
         .command
         .windows(2)
@@ -159,9 +169,14 @@ async fn native_request_preserves_native_launch_fields() {
         network_proxy: None,
     };
 
-    let prepared = prepare_exec_request(&params, env.clone(), /*runtime_paths*/ None)
-        .await
-        .expect("prepare native request");
+    let prepared = prepare_exec_request(
+        &params,
+        env.clone(),
+        /*runtime_paths*/ None,
+        /*network_policy_decider*/ None,
+    )
+    .await
+    .expect("prepare native request");
 
     assert_eq!(prepared.command, params.argv);
     assert_eq!(prepared.cwd, cwd);
@@ -202,9 +217,11 @@ async fn remote_proxy_config_starts_executor_local_proxy() {
     let stale_proxy = "http://127.0.0.1:9".to_string();
     let env = HashMap::from([("HTTP_PROXY".to_string(), stale_proxy.clone())]);
 
-    let prepared = prepare_exec_request(&params, env, /*runtime_paths*/ None)
-        .await
-        .expect("prepare request with executor-local proxy");
+    let prepared = prepare_exec_request(
+        &params, env, /*runtime_paths*/ None, /*network_policy_decider*/ None,
+    )
+    .await
+    .expect("prepare request with executor-local proxy");
 
     let http_proxy = prepared.env.get("HTTP_PROXY").expect("HTTP proxy env");
     assert_ne!(http_proxy, &stale_proxy);

codex-rs/exec-server/src/remote_process.rs

@@ -15,6 +15,7 @@ use crate::protocol::ExecParams;
 use crate::protocol::ProcessSignal;
 use crate::protocol::ReadResponse;
 use crate::protocol::WriteResponse;
+use codex_network_proxy::NetworkPolicyDecider;
 
 #[derive(Clone)]
 pub(crate) struct RemoteProcess {
@@ -48,6 +49,22 @@ impl ExecBackend for RemoteProcess {
     fn start(&self, params: ExecParams) -> ExecBackendFuture<'_> {
         Box::pin(RemoteProcess::start(self, params))
     }
+
+    fn start_with_network_policy_decider(
+        &self,
+        params: ExecParams,
+        decider: Arc<dyn NetworkPolicyDecider>,
+    ) -> ExecBackendFuture<'_> {
+        Box::pin(async move {
+            let client = self.client.get().await?;
+            let session = client
+                .start_process_with_network_policy_decider(params, decider)
+                .await?;
+            Ok(StartedExecProcess {
+                process: Arc::new(RemoteExecProcess { session }),
+            })
+        })
+    }
 }
 
 impl RemoteExecProcess {

codex-rs/exec-server/src/server/handler.rs

@@ -44,6 +44,7 @@ use crate::protocol::FsWriteFileResponse;
 use crate::protocol::HttpRequestParams;
 use crate::protocol::InitializeParams;
 use crate::protocol::InitializeResponse;
+use crate::protocol::NetworkPolicyDecisionNotification;
 use crate::protocol::ReadParams;
 use crate::protocol::ReadResponse;
 use crate::protocol::SignalParams;
@@ -161,6 +162,16 @@ impl ExecServerHandler {
         session.process().exec(params).await
     }
 
+    pub(crate) fn resolve_network_policy_decision(
+        &self,
+        params: NetworkPolicyDecisionNotification,
+    ) -> Result<(), String> {
+        let session = self
+            .require_initialized_for("network policy decision")
+            .map_err(|error| error.message)?;
+        session.process().resolve_network_policy_decision(params)
+    }
+
     pub(crate) fn environment_info(&self) -> Result<EnvironmentInfo, JSONRPCErrorError> {
         self.require_initialized_for("environment info")?;
         Ok(EnvironmentInfo::local())

codex-rs/exec-server/src/server/process_handler.rs

@@ -4,6 +4,7 @@ use crate::ExecServerRuntimePaths;
 use crate::local_process::LocalProcess;
 use crate::protocol::ExecParams;
 use crate::protocol::ExecResponse;
+use crate::protocol::NetworkPolicyDecisionNotification;
 use crate::protocol::ReadParams;
 use crate::protocol::ReadResponse;
 use crate::protocol::SignalParams;
@@ -43,6 +44,13 @@ impl ProcessHandler {
         self.process.exec(params).await
     }
 
+    pub(crate) fn resolve_network_policy_decision(
+        &self,
+        params: NetworkPolicyDecisionNotification,
+    ) -> Result<(), String> {
+        self.process.resolve_network_policy_decision(params)
+    }
+
     pub(crate) async fn exec_read(
         &self,
         params: ReadParams,

codex-rs/exec-server/src/server/registry.rs

@@ -36,6 +36,8 @@ use crate::protocol::HttpRequestParams;
 use crate::protocol::INITIALIZE_METHOD;
 use crate::protocol::INITIALIZED_METHOD;
 use crate::protocol::InitializeParams;
+use crate::protocol::NETWORK_POLICY_DECISION_METHOD;
+use crate::protocol::NetworkPolicyDecisionNotification;
 use crate::protocol::ReadParams;
 use crate::protocol::SignalParams;
 use crate::protocol::TerminateParams;
@@ -57,6 +59,12 @@ pub(crate) fn build_router() -> RpcRouter<ExecServerHandler> {
             handler.initialize(params).await
         },
     );
+    router.notification(
+        NETWORK_POLICY_DECISION_METHOD,
+        |handler: Arc<ExecServerHandler>, params: NetworkPolicyDecisionNotification| async move {
+            handler.resolve_network_policy_decision(params)
+        },
+    );
     router.request_with_id(
         HTTP_REQUEST_METHOD,
         |handler: Arc<ExecServerHandler>, request_id, params: HttpRequestParams| async move {

codex-rs/network-proxy/src/network_policy.rs

@@ -19,7 +19,8 @@ const POLICY_REASON_ALLOW: &str = "allow";
 const DEFAULT_METHOD: &str = "none";
 const DEFAULT_CLIENT_ADDRESS: &str = "unknown";
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
 pub enum NetworkProtocol {
     Http,
     HttpsConnect,
@@ -74,7 +75,8 @@ impl NetworkDecisionSource {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
 pub struct NetworkPolicyRequest {
     pub protocol: NetworkProtocol,
     pub host: String,
@@ -124,7 +126,8 @@ impl NetworkPolicyRequest {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
+#[serde(tag = "type", rename_all = "snake_case")]
 pub enum NetworkDecision {
     Allow,
     Deny {

codex-rs/network-proxy/src/proxy.rs

@@ -680,6 +680,29 @@ impl NetworkProxy {
         })
     }
 
+    pub fn has_policy_decider(&self) -> bool {
+        self.policy_decider.is_some()
+    }
+
+    /// Returns the policy decider with this proxy's execution attribution attached.
+    pub fn remote_policy_decider(&self) -> Option<Arc<dyn NetworkPolicyDecider>> {
+        let decider = self.policy_decider.clone()?;
+        let environment_id = self
+            .execution_scope
+            .as_ref()
+            .map(|scope| scope.environment_id.clone());
+        let execution_id = self
+            .execution_scope
+            .as_ref()
+            .map(|scope| scope.execution_id.clone());
+        Some(Arc::new(move |mut request: crate::NetworkPolicyRequest| {
+            let decider = Arc::clone(&decider);
+            request.environment_id = request.environment_id.or_else(|| environment_id.clone());
+            request.execution_id = request.execution_id.or_else(|| execution_id.clone());
+            async move { decider.decide(request).await }
+        }))
+    }
+
     pub async fn add_allowed_domain(&self, host: &str) -> Result<()> {
         self.state.add_allowed_domain(host).await
     }
@@ -1118,6 +1141,7 @@ impl Drop for NetworkProxyHandle {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::NetworkDecision;
     use crate::config::NetworkProxySettings;
     use crate::state::network_proxy_state_for_policy;
     use pretty_assertions::assert_eq;
@@ -1125,6 +1149,52 @@ mod tests {
     use std::net::Ipv4Addr;
     use std::path::Path;
 
+    #[tokio::test]
+    async fn remote_policy_decider_attaches_execution_attribution() {
+        let captured = Arc::new(Mutex::new(None));
+        let captured_for_decider = Arc::clone(&captured);
+        let state = Arc::new(network_proxy_state_for_policy(
+            NetworkProxySettings::default(),
+        ));
+        let proxy = NetworkProxy::builder()
+            .state(state)
+            .policy_decider(move |request: crate::NetworkPolicyRequest| {
+                let captured = Arc::clone(&captured_for_decider);
+                async move {
+                    *captured
+                        .lock()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner) =
+                        Some((request.environment_id, request.execution_id));
+                    NetworkDecision::Allow
+                }
+            })
+            .build()
+            .await
+            .expect("build proxy");
+        let scoped = proxy
+            .for_execution("remote", "execution-1", "attribution-token".to_string())
+            .expect("scope proxy to execution");
+        let decider = scoped.remote_policy_decider().expect("remote decider");
+        let request = crate::NetworkPolicyRequest::new(crate::NetworkPolicyRequestArgs {
+            protocol: crate::NetworkProtocol::HttpsConnect,
+            host: "example.com".to_string(),
+            port: 443,
+            environment_id: None,
+            client_addr: None,
+            method: None,
+            command: None,
+            exec_policy_hint: None,
+        });
+
+        assert_eq!(decider.decide(request).await, NetworkDecision::Allow);
+        assert_eq!(
+            *captured
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+            Some((Some("remote".to_string()), Some("execution-1".to_string())))
+        );
+    }
+
     #[tokio::test]
     async fn managed_proxy_builder_uses_loopback_ports() {
         let http_listener = StdTcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();

codex-rs/network-proxy/src/remote_config.rs

@@ -66,6 +66,9 @@ pub struct RemoteNetworkProxyConfig {
     pub domains: Option<NetworkDomainPermissions>,
     pub unix_sockets: Option<NetworkUnixSocketPermissions>,
     pub allow_local_binding: bool,
+    /// Whether the executor may send policy misses back to the client for a decision.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub request_policy_decisions: bool,
 }
 
 impl RemoteNetworkProxyConfig {
@@ -88,6 +91,7 @@ impl RemoteNetworkProxyConfig {
             domains: settings.domains.clone(),
             unix_sockets: settings.unix_sockets.clone(),
             allow_local_binding: settings.allow_local_binding,
+            request_policy_decisions: false,
         })
     }
 

codex-rs/network-proxy/src/remote_config_tests.rs

@@ -71,3 +71,29 @@ fn launch_config_materializes_audit_and_execution_attribution() {
     assert_eq!(state.environment_id(), Some("remote"));
     assert_eq!(state.execution_id().as_deref(), Some("execution-1"));
 }
+
+#[test]
+fn policy_decision_callback_opt_in_is_backward_compatible() {
+    let mut remote =
+        RemoteNetworkProxyConfig::from_effective_config(&NetworkProxyConfig::default())
+            .expect("supported remote config");
+    remote.request_policy_decisions = true;
+
+    let mut value = serde_json::to_value(&remote).expect("serialize remote config");
+    assert_eq!(value["requestPolicyDecisions"], true);
+    assert_eq!(
+        serde_json::from_value::<RemoteNetworkProxyConfig>(value.clone())
+            .expect("deserialize callback-enabled config"),
+        remote
+    );
+
+    value
+        .as_object_mut()
+        .expect("remote config object")
+        .remove("requestPolicyDecisions");
+    assert!(
+        !serde_json::from_value::<RemoteNetworkProxyConfig>(value)
+            .expect("deserialize legacy remote config")
+            .request_policy_decisions
+    );
+}