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

use std::future::Future;
use std::io::ErrorKind;
use std::mem::swap;
use std::path::Path;
use std::path::PathBuf;
use std::process::Command;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::Duration;

use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use codex_config::CloudConfigBundleLoader;
use codex_core::CodexThread;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::resolve_installation_id;
use codex_core::shell::Shell;
use codex_core::shell::get_shell_by_model_provided_path;
use codex_core::thread_store_from_config;
use codex_exec_server::CreateDirectoryOptions;
use codex_exec_server::ExecutorFileSystem;
use codex_exec_server::RemoveOptions;
use codex_extension_api::ExtensionRegistry;
use codex_extension_api::empty_extension_registry;
use codex_login::CodexAuth;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::built_in_model_providers;
use codex_models_manager::bundled_models_response;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RealtimeConversationVersion as RealtimeWsVersion;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionConfiguredEvent;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::user_input::UserInput;
use codex_utils_absolute_path::AbsolutePathBuf;
use futures::future::BoxFuture;
use serde_json::Value;
use tempfile::TempDir;
use wiremock::MockServer;

use crate::PathBufExt;
use crate::TempDirExt;
use crate::get_remote_test_env;
use crate::load_default_config_for_test;
use crate::load_default_config_for_test_with_cloud_config_bundle;
use crate::responses::WebSocketTestServer;
use crate::responses::output_value_to_text;
use crate::responses::start_mock_server;
use crate::streaming_sse::StreamingSseServer;
use crate::wait_for_event_match;
use crate::wait_for_event_with_timeout;
use wiremock::Match;
use wiremock::matchers::path_regex;

type ConfigMutator = dyn FnOnce(&mut Config) + Send;
type PreBuildHook = dyn FnOnce(&Path) + Send + 'static;
type WorkspaceSetup = dyn FnOnce(AbsolutePathBuf, Arc<dyn ExecutorFileSystem>) -> BoxFuture<'static, Result<()>>
    + Send;
const TEST_MODEL_WITH_EXPERIMENTAL_TOOLS: &str = "test-gpt-5.1-codex";
const REMOTE_EXEC_SERVER_URL_ENV_VAR: &str = "CODEX_TEST_REMOTE_EXEC_SERVER_URL";
static REMOTE_TEST_INSTANCE_COUNTER: AtomicU64 = AtomicU64::new(0);
const SUBMIT_TURN_COMPLETE_TIMEOUT: Duration = Duration::from_secs(30);

pub fn local(cwd: AbsolutePathBuf) -> TurnEnvironmentSelection {
    TurnEnvironmentSelection {
        environment_id: codex_exec_server::LOCAL_ENVIRONMENT_ID.to_string(),
        cwd,
    }
}

pub fn local_selections(cwd: AbsolutePathBuf) -> TurnEnvironmentSelections {
    TurnEnvironmentSelections::new(cwd.clone(), vec![local(cwd)])
}

#[derive(Debug)]
pub struct TestEnv {
    environment: codex_exec_server::Environment,
    exec_server_url: Option<String>,
    cwd: AbsolutePathBuf,
    local_cwd_temp_dir: Option<Arc<TempDir>>,
    remote_container_name: Option<String>,
}

impl TestEnv {
    pub async fn local() -> Result<Self> {
        let local_cwd_temp_dir = Arc::new(TempDir::new()?);
        let cwd = local_cwd_temp_dir.abs();
        let environment =
            codex_exec_server::Environment::create_for_tests(/*exec_server_url*/ None)?;
        Ok(Self {
            environment,
            exec_server_url: None,
            cwd,
            local_cwd_temp_dir: Some(local_cwd_temp_dir),
            remote_container_name: None,
        })
    }

    pub fn cwd(&self) -> &AbsolutePathBuf {
        &self.cwd
    }

    pub fn environment(&self) -> &codex_exec_server::Environment {
        &self.environment
    }

    fn local_cwd_temp_dir(&self) -> Option<Arc<TempDir>> {
        self.local_cwd_temp_dir.clone()
    }
}

impl Drop for TestEnv {
    fn drop(&mut self) {
        if let Some(container_name) = &self.remote_container_name {
            let script = format!("rm -rf {}", self.cwd.as_path().display());
            let _ = docker_command_capture_stdout(["exec", container_name, "sh", "-lc", &script]);
        }
    }
}

pub async fn test_env() -> Result<TestEnv> {
    match get_remote_test_env() {
        Some(remote_env) => {
            let websocket_url = remote_exec_server_url()?;
            let environment =
                codex_exec_server::Environment::create_for_tests(Some(websocket_url.clone()))?;
            let cwd = remote_aware_cwd_path();
            environment
                .get_filesystem()
                .create_directory(
                    &cwd,
                    CreateDirectoryOptions { recursive: true },
                    /*sandbox*/ None,
                )
                .await?;
            Ok(TestEnv {
                environment,
                exec_server_url: Some(websocket_url),
                cwd,
                local_cwd_temp_dir: None,
                remote_container_name: Some(remote_env.container_name),
            })
        }
        None => TestEnv::local().await,
    }
}

fn remote_aware_cwd_path() -> AbsolutePathBuf {
    PathBuf::from(format!(
        "/tmp/codex-core-test-cwd-{}",
        remote_test_instance_id()
    ))
    .abs()
}

fn remote_exec_server_url() -> Result<String> {
    let listen_url = std::env::var(REMOTE_EXEC_SERVER_URL_ENV_VAR).with_context(|| {
        format!("{REMOTE_EXEC_SERVER_URL_ENV_VAR} must be set for remote tests")
    })?;
    let listen_url = listen_url.trim();
    if listen_url.is_empty() {
        return Err(anyhow!(
            "{REMOTE_EXEC_SERVER_URL_ENV_VAR} must not be empty"
        ));
    }
    Ok(listen_url.to_string())
}

fn remote_test_instance_id() -> String {
    let instance = REMOTE_TEST_INSTANCE_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("{}-{instance}", std::process::id())
}

fn docker_command_capture_stdout<const N: usize>(args: [&str; N]) -> Result<String> {
    let output = Command::new("docker")
        .args(args)
        .output()
        .with_context(|| format!("run docker {args:?}"))?;
    if !output.status.success() {
        return Err(anyhow!(
            "docker {:?} failed: stdout={} stderr={}",
            args,
            String::from_utf8_lossy(&output.stdout).trim(),
            String::from_utf8_lossy(&output.stderr).trim()
        ));
    }
    String::from_utf8(output.stdout).context("docker stdout must be utf-8")
}

/// Non-default apply_patch model output shapes used by compatibility tests.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum ApplyPatchModelOutput {
    ShellCommandViaHeredoc,
}

/// A collection of different ways the model can output an apply_patch call
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum ShellModelOutput {
    ShellCommand,
    // UnifiedExec has its own set of tests
}

/// Returns the permission fields required by test thread-settings overrides.
pub fn turn_permission_fields(
    permission_profile: PermissionProfile,
    cwd: &Path,
) -> (SandboxPolicy, Option<PermissionProfile>) {
    let sandbox_policy = permission_profile
        .to_legacy_sandbox_policy(cwd)
        .unwrap_or_else(|_| SandboxPolicy::new_read_only_policy());
    (sandbox_policy, Some(permission_profile))
}

pub struct TestCodexBuilder {
    config_mutators: Vec<Box<ConfigMutator>>,
    auth: CodexAuth,
    pre_build_hooks: Vec<Box<PreBuildHook>>,
    workspace_setups: Vec<Box<WorkspaceSetup>>,
    home: Option<Arc<TempDir>>,
    cloud_config_bundle: Option<CloudConfigBundleLoader>,
    user_shell_override: Option<Shell>,
    exec_server_url: Option<String>,
    extensions: Arc<ExtensionRegistry<Config>>,
}

impl TestCodexBuilder {
    pub fn with_config<T>(mut self, mutator: T) -> Self
    where
        T: FnOnce(&mut Config) + Send + 'static,
    {
        self.config_mutators.push(Box::new(mutator));
        self
    }

    pub fn with_auth(mut self, auth: CodexAuth) -> Self {
        self.auth = auth;
        self
    }

    pub fn with_model(self, model: &str) -> Self {
        let new_model = model.to_string();
        self.with_config(move |config| {
            config.model = Some(new_model);
        })
    }

    pub fn with_model_info_override<T>(self, model: &str, override_model_info: T) -> Self
    where
        T: FnOnce(&mut ModelInfo) + Send + 'static,
    {
        let model = model.to_string();
        self.with_config(move |config| {
            let model_catalog = config.model_catalog.get_or_insert_with(|| {
                bundled_models_response()
                    .unwrap_or_else(|err| panic!("bundled models.json should parse: {err}"))
            });
            let model_info = model_catalog
                .models
                .iter_mut()
                .find(|model_info| model_info.slug == model)
                .unwrap_or_else(|| panic!("{model} should exist in the configured model catalog"));
            override_model_info(model_info);
            config.model = Some(model);
        })
    }

    pub fn with_pre_build_hook<F>(mut self, hook: F) -> Self
    where
        F: FnOnce(&Path) + Send + 'static,
    {
        self.pre_build_hooks.push(Box::new(hook));
        self
    }

    pub fn with_workspace_setup<F, Fut>(mut self, setup: F) -> Self
    where
        F: FnOnce(AbsolutePathBuf, Arc<dyn ExecutorFileSystem>) -> Fut + Send + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        self.workspace_setups
            .push(Box::new(move |cwd, fs| Box::pin(setup(cwd, fs))));
        self
    }

    pub fn with_home(mut self, home: Arc<TempDir>) -> Self {
        self.home = Some(home);
        self
    }

    pub fn with_cloud_config_bundle(
        mut self,
        cloud_config_bundle: CloudConfigBundleLoader,
    ) -> Self {
        self.cloud_config_bundle = Some(cloud_config_bundle);
        self
    }

    pub fn with_user_shell(mut self, user_shell: Shell) -> Self {
        self.user_shell_override = Some(user_shell);
        self
    }

    pub fn with_exec_server_url(mut self, exec_server_url: impl Into<String>) -> Self {
        self.exec_server_url = Some(exec_server_url.into());
        self
    }

    pub fn with_extensions(mut self, extensions: Arc<ExtensionRegistry<Config>>) -> Self {
        self.extensions = extensions;
        self
    }

    pub fn with_windows_cmd_shell(self) -> Self {
        if cfg!(windows) {
            self.with_user_shell(get_shell_by_model_provided_path(&PathBuf::from("cmd.exe")))
        } else {
            self
        }
    }

    pub async fn build(&mut self, server: &wiremock::MockServer) -> anyhow::Result<TestCodex> {
        let home = match self.home.clone() {
            Some(home) => home,
            None => Arc::new(TempDir::new()?),
        };
        let base_url = format!("{}/v1", server.uri());
        let test_env = TestEnv::local().await?;
        Box::pin(self.build_with_home_and_base_url(
            base_url, home, /*resume_from*/ None, test_env,
            /*include_local_environment*/ false,
        ))
        .await
    }

    pub async fn build_with_remote_env(
        &mut self,
        server: &wiremock::MockServer,
    ) -> anyhow::Result<TestCodex> {
        let home = match self.home.clone() {
            Some(home) => home,
            None => Arc::new(TempDir::new()?),
        };
        let base_url = format!("{}/v1", server.uri());
        let test_env = test_env().await?;
        Box::pin(self.build_with_home_and_base_url(
            base_url, home, /*resume_from*/ None, test_env,
            /*include_local_environment*/ false,
        ))
        .await
    }

    pub async fn build_with_remote_and_local_env(
        &mut self,
        server: &wiremock::MockServer,
    ) -> anyhow::Result<TestCodex> {
        let home = match self.home.clone() {
            Some(home) => home,
            None => Arc::new(TempDir::new()?),
        };
        let base_url = format!("{}/v1", server.uri());
        let test_env = test_env().await?;
        Box::pin(self.build_with_home_and_base_url(
            base_url, home, /*resume_from*/ None, test_env,
            /*include_local_environment*/ true,
        ))
        .await
    }

    pub async fn build_with_streaming_server(
        &mut self,
        server: &StreamingSseServer,
    ) -> anyhow::Result<TestCodex> {
        let base_url = server.uri();
        let home = match self.home.clone() {
            Some(home) => home,
            None => Arc::new(TempDir::new()?),
        };
        let test_env = TestEnv::local().await?;
        Box::pin(self.build_with_home_and_base_url(
            format!("{base_url}/v1"),
            home,
            /*resume_from*/ None,
            test_env,
            /*include_local_environment*/ false,
        ))
        .await
    }

    pub async fn build_with_websocket_server(
        &mut self,
        server: &WebSocketTestServer,
    ) -> anyhow::Result<TestCodex> {
        let base_url = format!("{}/v1", server.uri());
        let home = match self.home.clone() {
            Some(home) => home,
            None => Arc::new(TempDir::new()?),
        };
        let base_url_clone = base_url.clone();
        self.config_mutators.push(Box::new(move |config| {
            config.model_provider.base_url = Some(base_url_clone);
            config.model_provider.supports_websockets = true;
            config.experimental_realtime_ws_model = Some("realtime-test-model".to_string());
            config.realtime.version = RealtimeWsVersion::V1;
        }));
        let test_env = TestEnv::local().await?;
        Box::pin(self.build_with_home_and_base_url(
            base_url, home, /*resume_from*/ None, test_env,
            /*include_local_environment*/ false,
        ))
        .await
    }

    pub async fn resume(
        &mut self,
        server: &wiremock::MockServer,
        home: Arc<TempDir>,
        rollout_path: PathBuf,
    ) -> anyhow::Result<TestCodex> {
        let base_url = format!("{}/v1", server.uri());
        let test_env = TestEnv::local().await?;
        Box::pin(self.build_with_home_and_base_url(
            base_url,
            home,
            Some(rollout_path),
            test_env,
            /*include_local_environment*/ false,
        ))
        .await
    }

    async fn build_with_home_and_base_url(
        &mut self,
        base_url: String,
        home: Arc<TempDir>,
        resume_from: Option<PathBuf>,
        test_env: TestEnv,
        include_local_environment: bool,
    ) -> anyhow::Result<TestCodex> {
        let (config, fallback_cwd) = self
            .prepare_config(base_url, &home, test_env.cwd().clone())
            .await?;
        let exec_server_url = self
            .exec_server_url
            .clone()
            .or_else(|| test_env.exec_server_url.clone());
        #[cfg(target_os = "linux")]
        let codex_linux_sandbox_exe = Some(
            crate::find_codex_linux_sandbox_exe()
                .context("should find binary for codex-linux-sandbox")?,
        );
        #[cfg(not(target_os = "linux"))]
        let codex_linux_sandbox_exe = None;
        let local_runtime_paths = codex_exec_server::ExecServerRuntimePaths::new(
            std::env::current_exe()?,
            codex_linux_sandbox_exe,
        )?;
        let environment_manager = Arc::new(if include_local_environment {
            codex_exec_server::EnvironmentManager::create_for_tests_with_local(
                exec_server_url,
                local_runtime_paths,
            )
            .await
        } else {
            codex_exec_server::EnvironmentManager::create_for_tests(
                exec_server_url,
                Some(local_runtime_paths),
            )
            .await
        });
        let file_system = test_env.environment().get_filesystem();
        let mut workspace_setups = vec![];
        swap(&mut self.workspace_setups, &mut workspace_setups);
        for setup in workspace_setups {
            setup(config.cwd.clone(), Arc::clone(&file_system)).await?;
        }
        let cwd = test_env.local_cwd_temp_dir().unwrap_or(fallback_cwd);
        Box::pin(self.build_from_config(
            config,
            cwd,
            home,
            resume_from,
            test_env,
            environment_manager,
        ))
        .await
    }

    async fn build_from_config(
        &mut self,
        config: Config,
        cwd: Arc<TempDir>,
        home: Arc<TempDir>,
        resume_from: Option<PathBuf>,
        test_env: TestEnv,
        environment_manager: Arc<codex_exec_server::EnvironmentManager>,
    ) -> anyhow::Result<TestCodex> {
        let auth = self.auth.clone();
        let state_db = codex_core::init_state_db(&config).await;
        let thread_store = thread_store_from_config(&config, state_db.clone());
        let installation_id = resolve_installation_id(&config.codex_home).await?;
        let thread_manager = ThreadManager::new(
            &config,
            codex_core::test_support::auth_manager_from_auth(auth.clone()),
            SessionSource::Exec,
            Arc::clone(&environment_manager),
            Arc::clone(&self.extensions),
            /*analytics_events_client*/ None,
            thread_store,
            state_db.clone(),
            installation_id,
            /*attestation_provider*/ None,
        );
        let thread_manager = if std::env::var_os("CODEX_CODE_MODE_HOST_PATH").is_some() {
            thread_manager
        } else {
            thread_manager.with_code_mode_session_provider(Arc::new(
                codex_code_mode::InProcessCodeModeSessionProvider,
            ))
        };
        let thread_manager = Arc::new(thread_manager);
        let user_shell_override = self.user_shell_override.clone();

        let new_conversation = match (resume_from, user_shell_override) {
            (Some(path), Some(user_shell_override)) => {
                let auth_manager = codex_core::test_support::auth_manager_from_auth(auth);
                Box::pin(
                    codex_core::test_support::resume_thread_from_rollout_with_user_shell_override(
                        thread_manager.as_ref(),
                        config.clone(),
                        path,
                        auth_manager,
                        user_shell_override,
                    ),
                )
                .await?
            }
            (Some(path), None) => {
                let auth_manager = codex_core::test_support::auth_manager_from_auth(auth);
                Box::pin(thread_manager.resume_thread_from_rollout(
                    config.clone(),
                    path,
                    auth_manager,
                    /*parent_trace*/ None,
                ))
                .await?
            }
            (None, Some(user_shell_override)) => {
                Box::pin(
                    codex_core::test_support::start_thread_with_user_shell_override(
                        thread_manager.as_ref(),
                        config.clone(),
                        user_shell_override,
                    ),
                )
                .await?
            }
            (None, None) => Box::pin(thread_manager.start_thread(config.clone())).await?,
        };

        Ok(TestCodex {
            home,
            cwd,
            config,
            codex: new_conversation.thread,
            session_configured: new_conversation.session_configured,
            thread_manager,
            _test_env: test_env,
        })
    }

    async fn prepare_config(
        &mut self,
        base_url: String,
        home: &TempDir,
        cwd_override: AbsolutePathBuf,
    ) -> anyhow::Result<(Config, Arc<TempDir>)> {
        let model_provider = ModelProviderInfo {
            base_url: Some(base_url),
            // Most core tests use SSE-only mock servers, so keep websocket transport off unless
            // a test explicitly opts into websocket coverage.
            supports_websockets: false,
            ..built_in_model_providers(/*openai_base_url*/ None)["openai"].clone()
        };
        let cwd = Arc::new(TempDir::new()?);
        for hook in self.pre_build_hooks.drain(..) {
            hook(home.path());
        }
        let mut config = if let Some(cloud_config_bundle) = self.cloud_config_bundle.take() {
            load_default_config_for_test_with_cloud_config_bundle(home, cloud_config_bundle).await
        } else {
            load_default_config_for_test(home).await
        };
        config.cwd = cwd_override;
        config.model_provider = model_provider;
        if let Ok(path) = codex_utils_cargo_bin::cargo_bin("codex") {
            config.codex_self_exe = Some(path);
        } else if let Ok(path) = codex_utils_cargo_bin::cargo_bin("codex-exec") {
            // `codex-exec` also supports `--codex-run-as-apply-patch`, so use it
            // when the multitool binary is not available in test builds.
            config.codex_self_exe = Some(path);
        } else if let Ok(exe) = std::env::current_exe()
            && let Some(bin_dir) = exe.parent().and_then(|parent| parent.parent())
        {
            let codex = bin_dir.join("codex");
            let codex_exec = bin_dir.join("codex-exec");
            if codex.is_file() {
                config.codex_self_exe = Some(codex);
            } else if codex_exec.is_file() {
                config.codex_self_exe = Some(codex_exec);
            }
        }

        let mut mutators = vec![];
        swap(&mut self.config_mutators, &mut mutators);
        for mutator in mutators {
            mutator(&mut config);
        }
        ensure_test_model_catalog(&mut config)?;

        Ok((config, cwd))
    }
}

fn ensure_test_model_catalog(config: &mut Config) -> Result<()> {
    if config.model.as_deref() != Some(TEST_MODEL_WITH_EXPERIMENTAL_TOOLS)
        || config.model_catalog.is_some()
    {
        return Ok(());
    }

    let bundled_models = bundled_models_response()
        .unwrap_or_else(|err| panic!("bundled models.json should parse: {err}"));
    let mut model = bundled_models
        .models
        .iter()
        .find(|candidate| candidate.slug == "gpt-5.2")
        .cloned()
        .unwrap_or_else(|| panic!("missing bundled model gpt-5.2"));
    model.slug = TEST_MODEL_WITH_EXPERIMENTAL_TOOLS.to_string();
    model.display_name = TEST_MODEL_WITH_EXPERIMENTAL_TOOLS.to_string();
    model.experimental_supported_tools = vec!["test_sync_tool".to_string()];
    config.model_catalog = Some(ModelsResponse {
        models: vec![model],
    });
    Ok(())
}

pub struct TestCodex {
    pub home: Arc<TempDir>,
    pub cwd: Arc<TempDir>,
    pub codex: Arc<CodexThread>,
    pub session_configured: SessionConfiguredEvent,
    pub config: Config,
    pub thread_manager: Arc<ThreadManager>,
    _test_env: TestEnv,
}

impl TestCodex {
    pub fn cwd_path(&self) -> &Path {
        self.cwd.path()
    }

    pub fn codex_home_path(&self) -> &Path {
        self.config.codex_home.as_path()
    }

    pub fn workspace_path(&self, rel: impl AsRef<Path>) -> PathBuf {
        self.cwd_path().join(rel)
    }

    pub fn executor_environment(&self) -> &TestEnv {
        &self._test_env
    }

    pub fn fs(&self) -> Arc<dyn ExecutorFileSystem> {
        self._test_env.environment().get_filesystem()
    }

    pub async fn submit_turn(&self, prompt: &str) -> Result<()> {
        self.submit_turn_with_permission_profile(prompt, PermissionProfile::Disabled)
            .await
    }

    pub async fn submit_turn_with_permission_profile(
        &self,
        prompt: &str,
        permission_profile: PermissionProfile,
    ) -> Result<()> {
        self.submit_turn_with_approval_and_permission_profile(
            prompt,
            AskForApproval::Never,
            permission_profile,
        )
        .await
    }

    pub async fn submit_turn_with_policy(
        &self,
        prompt: &str,
        sandbox_policy: SandboxPolicy,
    ) -> Result<()> {
        self.submit_turn_with_policies(prompt, AskForApproval::Never, sandbox_policy)
            .await
    }

    pub async fn submit_turn_with_service_tier(
        &self,
        prompt: &str,
        service_tier: Option<&str>,
    ) -> Result<()> {
        self.submit_turn_with_permission_profile_context(
            prompt,
            AskForApproval::Never,
            PermissionProfile::Disabled,
            Some(service_tier.map(str::to_string)),
            /*environments*/ None,
        )
        .await
    }

    pub async fn submit_turn_with_policies(
        &self,
        prompt: &str,
        approval_policy: AskForApproval,
        sandbox_policy: SandboxPolicy,
    ) -> Result<()> {
        let permission_profile = PermissionProfile::from_legacy_sandbox_policy_for_cwd(
            &sandbox_policy,
            self.config.cwd.as_path(),
        );
        self.submit_turn_with_context(
            prompt,
            approval_policy,
            permission_profile,
            /*service_tier*/ None,
            /*environments*/ None,
        )
        .await
    }

    pub async fn submit_turn_with_approval_and_permission_profile(
        &self,
        prompt: &str,
        approval_policy: AskForApproval,
        permission_profile: PermissionProfile,
    ) -> Result<()> {
        self.submit_turn_with_permission_profile_context(
            prompt,
            approval_policy,
            permission_profile,
            /*service_tier*/ None,
            /*environments*/ None,
        )
        .await
    }

    pub async fn submit_turn_with_environments(
        &self,
        prompt: &str,
        environments: Option<Vec<TurnEnvironmentSelection>>,
    ) -> Result<()> {
        self.submit_turn_with_permission_profile_context(
            prompt,
            AskForApproval::Never,
            PermissionProfile::Disabled,
            /*service_tier*/ None,
            environments,
        )
        .await
    }

    async fn submit_turn_with_permission_profile_context(
        &self,
        prompt: &str,
        approval_policy: AskForApproval,
        permission_profile: PermissionProfile,
        service_tier: Option<Option<String>>,
        environments: Option<Vec<TurnEnvironmentSelection>>,
    ) -> Result<()> {
        self.submit_turn_with_context(
            prompt,
            approval_policy,
            permission_profile,
            service_tier,
            environments,
        )
        .await
    }

    async fn submit_turn_with_context(
        &self,
        prompt: &str,
        approval_policy: AskForApproval,
        permission_profile: PermissionProfile,
        service_tier: Option<Option<String>>,
        environments: Option<Vec<TurnEnvironmentSelection>>,
    ) -> Result<()> {
        let (sandbox_policy, permission_profile) =
            turn_permission_fields(permission_profile, self.config.cwd.as_path());
        let session_model = self.session_configured.model.clone();
        let turn_environment_selections = environments.map(|environments| {
            TurnEnvironmentSelections::new(self.config.cwd.clone(), environments)
        });
        self.codex
            .submit(Op::UserInput {
                items: vec![UserInput::Text {
                    text: prompt.into(),
                    text_elements: Vec::new(),
                }],
                final_output_json_schema: None,
                responsesapi_client_metadata: None,
                additional_context: Default::default(),
                thread_settings: codex_protocol::protocol::ThreadSettingsOverrides {
                    environments: turn_environment_selections,
                    approval_policy: Some(approval_policy),
                    sandbox_policy: Some(sandbox_policy),
                    permission_profile,
                    service_tier,
                    collaboration_mode: Some(codex_protocol::config_types::CollaborationMode {
                        mode: codex_protocol::config_types::ModeKind::Default,
                        settings: codex_protocol::config_types::Settings {
                            model: session_model,
                            reasoning_effort: None,
                            developer_instructions: None,
                        },
                    }),
                    ..Default::default()
                },
            })
            .await?;

        let turn_id = wait_for_event_match(&self.codex, |event| match event {
            EventMsg::TurnStarted(event) => Some(event.turn_id.clone()),
            _ => None,
        })
        .await;
        wait_for_event_with_timeout(
            &self.codex,
            |event| match event {
                EventMsg::TurnComplete(event) => event.turn_id == turn_id,
                _ => false,
            },
            SUBMIT_TURN_COMPLETE_TIMEOUT,
        )
        .await;
        Ok(())
    }
}

pub struct TestCodexHarness {
    server: MockServer,
    test: TestCodex,
}

impl TestCodexHarness {
    pub async fn new() -> Result<Self> {
        Self::with_builder(test_codex()).await
    }

    pub async fn with_config(mutator: impl FnOnce(&mut Config) + Send + 'static) -> Result<Self> {
        Self::with_builder(test_codex().with_config(mutator)).await
    }

    pub async fn with_builder(mut builder: TestCodexBuilder) -> Result<Self> {
        let server = start_mock_server().await;
        let test = builder.build(&server).await?;
        Ok(Self { server, test })
    }

    pub async fn with_remote_env_builder(mut builder: TestCodexBuilder) -> Result<Self> {
        let server = start_mock_server().await;
        let test = builder.build_with_remote_env(&server).await?;
        Ok(Self { server, test })
    }

    pub fn server(&self) -> &MockServer {
        &self.server
    }

    pub fn test(&self) -> &TestCodex {
        &self.test
    }

    pub fn cwd(&self) -> &Path {
        self.test.config.cwd.as_path()
    }

    pub fn cwd_abs(&self) -> AbsolutePathBuf {
        self.test.config.cwd.clone()
    }

    pub fn path(&self, rel: impl AsRef<Path>) -> PathBuf {
        self.path_abs(rel).into_path_buf()
    }

    pub fn path_abs(&self, rel: impl AsRef<Path>) -> AbsolutePathBuf {
        self.test.config.cwd.join(rel)
    }

    pub async fn write_file(
        &self,
        rel: impl AsRef<Path>,
        contents: impl AsRef<[u8]>,
    ) -> Result<()> {
        let abs_path = self.path_abs(rel);
        if let Some(parent) = abs_path.parent() {
            self.test
                .fs()
                .create_directory(
                    &parent,
                    CreateDirectoryOptions { recursive: true },
                    /*sandbox*/ None,
                )
                .await?;
        }
        self.test
            .fs()
            .write_file(&abs_path, contents.as_ref().to_vec(), /*sandbox*/ None)
            .await?;
        Ok(())
    }

    pub async fn read_file_text(&self, rel: impl AsRef<Path>) -> Result<String> {
        Ok(self
            .test
            .fs()
            .read_file_text(&self.path_abs(rel), /*sandbox*/ None)
            .await?)
    }

    pub async fn create_dir_all(&self, rel: impl AsRef<Path>) -> Result<()> {
        self.test
            .fs()
            .create_directory(
                &self.path_abs(rel),
                CreateDirectoryOptions { recursive: true },
                /*sandbox*/ None,
            )
            .await?;
        Ok(())
    }

    pub async fn path_exists(&self, rel: impl AsRef<Path>) -> Result<bool> {
        self.abs_path_exists(&self.path_abs(rel)).await
    }

    pub async fn remove_abs_path(&self, path: &AbsolutePathBuf) -> Result<()> {
        self.test
            .fs()
            .remove(
                path,
                RemoveOptions {
                    recursive: false,
                    force: true,
                },
                /*sandbox*/ None,
            )
            .await?;
        Ok(())
    }

    pub async fn abs_path_exists(&self, path: &AbsolutePathBuf) -> Result<bool> {
        match self.test.fs().get_metadata(path, /*sandbox*/ None).await {
            Ok(_) => Ok(true),
            Err(err) if err.kind() == ErrorKind::NotFound => Ok(false),
            Err(err) => Err(err.into()),
        }
    }

    pub async fn submit(&self, prompt: &str) -> Result<()> {
        // Box the submit-and-wait path so callers do not inline the full turn
        // future into their own async state.
        Box::pin(self.test.submit_turn(prompt)).await
    }

    pub async fn submit_with_policy(
        &self,
        prompt: &str,
        sandbox_policy: SandboxPolicy,
    ) -> Result<()> {
        self.test
            .submit_turn_with_policy(prompt, sandbox_policy)
            .await
    }

    pub async fn submit_with_permission_profile(
        &self,
        prompt: &str,
        permission_profile: PermissionProfile,
    ) -> Result<()> {
        self.test
            .submit_turn_with_permission_profile(prompt, permission_profile)
            .await
    }

    pub async fn request_bodies(&self) -> Vec<Value> {
        let path_matcher = path_regex(".*/responses$");
        self.server
            .received_requests()
            .await
            .expect("mock server should not fail")
            .into_iter()
            .filter(|req| path_matcher.matches(req))
            .map(|req| {
                req.body_json::<Value>()
                    .expect("request body to be valid JSON")
            })
            .collect()
    }

    pub async fn function_call_output_value(&self, call_id: &str) -> Value {
        let bodies = self.request_bodies().await;
        function_call_output(&bodies, call_id).clone()
    }

    pub async fn function_call_stdout(&self, call_id: &str) -> String {
        self.function_call_output_value(call_id)
            .await
            .get("output")
            .and_then(Value::as_str)
            .expect("output string")
            .to_string()
    }

    pub async fn custom_tool_call_output(&self, call_id: &str) -> String {
        let bodies = self.request_bodies().await;
        custom_tool_call_output_text(&bodies, call_id)
    }

    pub async fn apply_patch_output(&self, call_id: &str) -> String {
        self.custom_tool_call_output(call_id).await
    }
}

fn custom_tool_call_output<'a>(bodies: &'a [Value], call_id: &str) -> &'a Value {
    for body in bodies {
        if let Some(items) = body.get("input").and_then(Value::as_array) {
            for item in items {
                if item.get("type").and_then(Value::as_str) == Some("custom_tool_call_output")
                    && item.get("call_id").and_then(Value::as_str) == Some(call_id)
                {
                    return item;
                }
            }
        }
    }
    panic!("custom_tool_call_output {call_id} not found");
}

fn custom_tool_call_output_text(bodies: &[Value], call_id: &str) -> String {
    let output = custom_tool_call_output(bodies, call_id)
        .get("output")
        .unwrap_or_else(|| panic!("custom_tool_call_output {call_id} missing output"));
    output_value_to_text(output)
        .unwrap_or_else(|| panic!("custom_tool_call_output {call_id} missing text output"))
}

fn function_call_output<'a>(bodies: &'a [Value], call_id: &str) -> &'a Value {
    for body in bodies {
        if let Some(items) = body.get("input").and_then(Value::as_array) {
            for item in items {
                if item.get("type").and_then(Value::as_str) == Some("function_call_output")
                    && item.get("call_id").and_then(Value::as_str) == Some(call_id)
                {
                    return item;
                }
            }
        }
    }
    panic!("function_call_output {call_id} not found");
}

pub fn test_codex() -> TestCodexBuilder {
    TestCodexBuilder {
        config_mutators: vec![],
        auth: CodexAuth::from_api_key("dummy"),
        pre_build_hooks: vec![],
        workspace_setups: vec![],
        home: None,
        cloud_config_bundle: None,
        user_shell_override: None,
        exec_server_url: None,
        extensions: empty_extension_registry(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use pretty_assertions::assert_eq;
    use serde_json::json;

    #[test]
    fn custom_tool_call_output_text_returns_output_text() {
        let bodies = vec![json!({
            "input": [{
                "type": "custom_tool_call_output",
                "call_id": "call-1",
                "output": "hello"
            }]
        })];

        assert_eq!(custom_tool_call_output_text(&bodies, "call-1"), "hello");
    }

    #[test]
    #[should_panic(expected = "custom_tool_call_output call-2 missing output")]
    fn custom_tool_call_output_text_panics_when_output_is_missing() {
        let bodies = vec![json!({
            "input": [{
                "type": "custom_tool_call_output",
                "call_id": "call-2"
            }]
        })];

        let _ = custom_tool_call_output_text(&bodies, "call-2");
    }
}