chore: sanbox extraction

codex-rs/app-server/tests/common/mcp_process.rs

@@ -1,3 +1,4 @@
+use std::collections::VecDeque;
 use std::path::Path;
 use std::process::Stdio;
 use std::sync::atomic::AtomicI64;
@@ -47,6 +48,7 @@ pub struct McpProcess {
     process: Child,
     stdin: ChildStdin,
     stdout: BufReader<ChildStdout>,
+    pending_user_messages: VecDeque<JSONRPCNotification>,
 }
 
 impl McpProcess {
@@ -117,6 +119,7 @@ impl McpProcess {
             process,
             stdin,
             stdout,
+            pending_user_messages: VecDeque::new(),
         })
     }
 
@@ -375,8 +378,9 @@ impl McpProcess {
             let message = self.read_jsonrpc_message().await?;
 
             match message {
-                JSONRPCMessage::Notification(_) => {
-                    eprintln!("notification: {message:?}");
+                JSONRPCMessage::Notification(notification) => {
+                    eprintln!("notification: {notification:?}");
+                    self.enqueue_user_message(notification);
                 }
                 JSONRPCMessage::Request(jsonrpc_request) => {
                     return jsonrpc_request.try_into().with_context(
@@ -402,8 +406,9 @@ impl McpProcess {
         loop {
             let message = self.read_jsonrpc_message().await?;
             match message {
-                JSONRPCMessage::Notification(_) => {
-                    eprintln!("notification: {message:?}");
+                JSONRPCMessage::Notification(notification) => {
+                    eprintln!("notification: {notification:?}");
+                    self.enqueue_user_message(notification);
                 }
                 JSONRPCMessage::Request(_) => {
                     anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
@@ -427,8 +432,9 @@ impl McpProcess {
         loop {
             let message = self.read_jsonrpc_message().await?;
             match message {
-                JSONRPCMessage::Notification(_) => {
-                    eprintln!("notification: {message:?}");
+                JSONRPCMessage::Notification(notification) => {
+                    eprintln!("notification: {notification:?}");
+                    self.enqueue_user_message(notification);
                 }
                 JSONRPCMessage::Request(_) => {
                     anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
@@ -451,13 +457,18 @@ impl McpProcess {
     ) -> anyhow::Result<JSONRPCNotification> {
         eprintln!("in read_stream_until_notification_message({method})");
 
+        if let Some(notification) = self.take_pending_notification_by_method(method) {
+            return Ok(notification);
+        }
+
         loop {
             let message = self.read_jsonrpc_message().await?;
             match message {
                 JSONRPCMessage::Notification(notification) => {
                     if notification.method == method {
                         return Ok(notification);
                     }
+                    self.enqueue_user_message(notification);
                 }
                 JSONRPCMessage::Request(_) => {
                     anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
@@ -471,4 +482,21 @@ impl McpProcess {
             }
         }
     }
+
+    fn take_pending_notification_by_method(&mut self, method: &str) -> Option<JSONRPCNotification> {
+        if let Some(pos) = self
+            .pending_user_messages
+            .iter()
+            .position(|notification| notification.method == method)
+        {
+            return self.pending_user_messages.remove(pos);
+        }
+        None
+    }
+
+    fn enqueue_user_message(&mut self, notification: JSONRPCNotification) {
+        if notification.method == "codex/event/user_message" {
+            self.pending_user_messages.push_back(notification);
+        }
+    }
 }

codex-rs/app-server/tests/suite/codex_message_processor_flow.rs

@@ -8,6 +8,7 @@ use app_test_support::to_response;
 use codex_app_server_protocol::AddConversationListenerParams;
 use codex_app_server_protocol::AddConversationSubscriptionResponse;
 use codex_app_server_protocol::ExecCommandApprovalParams;
+use codex_app_server_protocol::InputItem;
 use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::NewConversationParams;
@@ -25,6 +26,10 @@ use codex_core::protocol::SandboxPolicy;
 use codex_core::protocol_config_types::ReasoningEffort;
 use codex_core::protocol_config_types::ReasoningSummary;
 use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
+use codex_protocol::config_types::SandboxMode;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::InputMessageKind;
 use pretty_assertions::assert_eq;
 use std::env;
 use tempfile::TempDir;
@@ -367,6 +372,234 @@ async fn test_send_user_turn_changes_approval_policy_behavior() {
 }
 
 // Helper: minimal config.toml pointing at mock provider.
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_send_user_turn_updates_sandbox_and_cwd_between_turns() {
+    if env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
+        println!(
+            "Skipping test because it cannot execute when network is disabled in a Codex sandbox."
+        );
+        return;
+    }
+
+    let tmp = TempDir::new().expect("tmp dir");
+    let codex_home = tmp.path().join("codex_home");
+    std::fs::create_dir(&codex_home).expect("create codex home dir");
+    let workspace_root = tmp.path().join("workspace");
+    std::fs::create_dir(&workspace_root).expect("create workspace root");
+    let first_cwd = workspace_root.join("turn1");
+    let second_cwd = workspace_root.join("turn2");
+    std::fs::create_dir(&first_cwd).expect("create first cwd");
+    std::fs::create_dir(&second_cwd).expect("create second cwd");
+
+    let responses = vec![
+        create_shell_sse_response(
+            vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "echo first turn".to_string(),
+            ],
+            None,
+            Some(5000),
+            "call-first",
+        )
+        .expect("create first shell response"),
+        create_final_assistant_message_sse_response("done first")
+            .expect("create first final assistant message"),
+        create_shell_sse_response(
+            vec![
+                "bash".to_string(),
+                "-lc".to_string(),
+                "echo second turn".to_string(),
+            ],
+            None,
+            Some(5000),
+            "call-second",
+        )
+        .expect("create second shell response"),
+        create_final_assistant_message_sse_response("done second")
+            .expect("create second final assistant message"),
+    ];
+    let server = create_mock_chat_completions_server(responses).await;
+    create_config_toml(&codex_home, &server.uri()).expect("write config");
+
+    let mut mcp = McpProcess::new(&codex_home)
+        .await
+        .expect("spawn mcp process");
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize())
+        .await
+        .expect("init timeout")
+        .expect("init failed");
+
+    let new_conv_id = mcp
+        .send_new_conversation_request(NewConversationParams {
+            cwd: Some(first_cwd.to_string_lossy().into_owned()),
+            approval_policy: Some(AskForApproval::Never),
+            sandbox: Some(SandboxMode::WorkspaceWrite),
+            ..Default::default()
+        })
+        .await
+        .expect("send newConversation");
+    let new_conv_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(new_conv_id)),
+    )
+    .await
+    .expect("newConversation timeout")
+    .expect("newConversation resp");
+    let NewConversationResponse {
+        conversation_id,
+        model,
+        ..
+    } = to_response::<NewConversationResponse>(new_conv_resp)
+        .expect("deserialize newConversation response");
+
+    let add_listener_id = mcp
+        .send_add_conversation_listener_request(AddConversationListenerParams { conversation_id })
+        .await
+        .expect("send addConversationListener");
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(add_listener_id)),
+    )
+    .await
+    .expect("addConversationListener timeout")
+    .expect("addConversationListener resp");
+
+    let first_turn_id = mcp
+        .send_send_user_turn_request(SendUserTurnParams {
+            conversation_id,
+            items: vec![InputItem::Text {
+                text: "first turn".to_string(),
+            }],
+            cwd: first_cwd.clone(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::WorkspaceWrite {
+                writable_roots: vec![first_cwd.clone()],
+                network_access: false,
+                exclude_tmpdir_env_var: false,
+                exclude_slash_tmp: false,
+            },
+            model: model.clone(),
+            effort: Some(ReasoningEffort::Medium),
+            summary: ReasoningSummary::Auto,
+        })
+        .await
+        .expect("send first sendUserTurn");
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(first_turn_id)),
+    )
+    .await
+    .expect("sendUserTurn 1 timeout")
+    .expect("sendUserTurn 1 resp");
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("codex/event/task_complete"),
+    )
+    .await
+    .expect("task_complete 1 timeout")
+    .expect("task_complete 1 notification");
+
+    let second_turn_id = mcp
+        .send_send_user_turn_request(SendUserTurnParams {
+            conversation_id,
+            items: vec![InputItem::Text {
+                text: "second turn".to_string(),
+            }],
+            cwd: second_cwd.clone(),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::DangerFullAccess,
+            model: model.clone(),
+            effort: Some(ReasoningEffort::Medium),
+            summary: ReasoningSummary::Auto,
+        })
+        .await
+        .expect("send second sendUserTurn");
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)),
+    )
+    .await
+    .expect("sendUserTurn 2 timeout")
+    .expect("sendUserTurn 2 resp");
+
+    let mut env_message: Option<String> = None;
+    let second_cwd_str = second_cwd.to_string_lossy().into_owned();
+    for _ in 0..10 {
+        let notification = timeout(
+            DEFAULT_READ_TIMEOUT,
+            mcp.read_stream_until_notification_message("codex/event/user_message"),
+        )
+        .await
+        .expect("user_message timeout")
+        .expect("user_message notification");
+        let params = notification
+            .params
+            .clone()
+            .expect("user_message should include params");
+        let event: Event = serde_json::from_value(params).expect("deserialize user_message event");
+        if let EventMsg::UserMessage(user) = event.msg
+            && matches!(user.kind, Some(InputMessageKind::EnvironmentContext))
+            && user.message.contains(&second_cwd_str)
+        {
+            env_message = Some(user.message);
+            break;
+        }
+    }
+    let env_message = env_message.expect("expected environment context update");
+    assert!(
+        env_message.contains("<sandbox_mode>danger-full-access</sandbox_mode>"),
+        "env context should reflect new sandbox mode: {env_message}"
+    );
+    assert!(
+        env_message.contains("<network_access>enabled</network_access>"),
+        "env context should enable network access for danger-full-access policy: {env_message}"
+    );
+    assert!(
+        env_message.contains(&second_cwd_str),
+        "env context should include updated cwd: {env_message}"
+    );
+
+    let exec_begin_notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("codex/event/exec_command_begin"),
+    )
+    .await
+    .expect("exec_command_begin timeout")
+    .expect("exec_command_begin notification");
+    let params = exec_begin_notification
+        .params
+        .clone()
+        .expect("exec_command_begin params");
+    let event: Event = serde_json::from_value(params).expect("deserialize exec begin event");
+    let exec_begin = match event.msg {
+        EventMsg::ExecCommandBegin(exec_begin) => exec_begin,
+        other => panic!("expected ExecCommandBegin event, got {other:?}"),
+    };
+    assert_eq!(
+        exec_begin.cwd, second_cwd,
+        "exec turn should run from updated cwd"
+    );
+    assert_eq!(
+        exec_begin.command,
+        vec![
+            "bash".to_string(),
+            "-lc".to_string(),
+            "echo second turn".to_string()
+        ],
+        "exec turn should run expected command"
+    );
+
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("codex/event/task_complete"),
+    )
+    .await
+    .expect("task_complete 2 timeout")
+    .expect("task_complete 2 notification");
+}
+
 fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
     let config_toml = codex_home.join("config.toml");
     std::fs::write(

codex-rs/core/src/apply_patch.rs

@@ -27,6 +27,7 @@ pub(crate) enum InternalApplyPatchInvocation {
     DelegateToExec(ApplyPatchExec),
 }
 
+#[derive(Debug)]
 pub(crate) struct ApplyPatchExec {
     pub(crate) action: ApplyPatchAction,
     pub(crate) user_explicitly_approved_this_action: bool,
@@ -109,3 +110,28 @@ pub(crate) fn convert_apply_patch_to_protocol(
     }
     result
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+
+    use tempfile::tempdir;
+
+    #[test]
+    fn convert_apply_patch_maps_add_variant() {
+        let tmp = tempdir().expect("tmp");
+        let p = tmp.path().join("a.txt");
+        // Create an action with a single Add change
+        let action = ApplyPatchAction::new_add_for_test(&p, "hello".to_string());
+
+        let got = convert_apply_patch_to_protocol(&action);
+
+        assert_eq!(
+            got.get(&p),
+            Some(&FileChange::Add {
+                content: "hello".to_string()
+            })
+        );
+    }
+}