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::AtomicUsize;
use std::sync::atomic::Ordering;
use std::time::Duration;

use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use codex_analytics::AnalyticsEventsClient;
use codex_attachment_store::AttachmentStore;
use codex_config::CloudConfigBundleLoader;
use codex_core::CodexThread;
pub use codex_core::StartThreadOptions;
use codex_core::ThreadManager;
use codex_core::TimeProvider;
pub use codex_core::TurnInputRequest;
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::LoadInstructionsFuture;
use codex_extension_api::UserInstructionsProvider;
use codex_extension_api::empty_extension_registry;
use codex_features::Feature;
use codex_home::CodexHomeUserInstructionsProvider;
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_models_manager::manager::SharedModelsManager;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::Settings;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::mcp::OPENAI_FORM_EXTENSION_ID;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::openai_models::TruncationPolicyConfig;
use codex_protocol::openai_models::WebSearchToolType;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EnvironmentConfigState;
use codex_protocol::protocol::EventMsg;
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::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::user_input::UserInput;
use codex_thread_store::ThreadStore;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path_uri::LegacyAppPathString;
use codex_utils_path_uri::PathConvention;
use codex_utils_path_uri::PathUri;
use futures::future::BoxFuture;
use serde_json::Value;
use tempfile::TempDir;
use wiremock::MockServer;

use crate::TempDirExt;
use crate::TestEnvironment;
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::test_environment;
use crate::wait_for_event;
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 struct RecordingUserInstructionsProvider {
    inner: Arc<dyn UserInstructionsProvider>,
    load_count: AtomicUsize,
}

impl RecordingUserInstructionsProvider {
    pub fn new(inner: Arc<dyn UserInstructionsProvider>) -> Self {
        Self {
            inner,
            load_count: AtomicUsize::new(0),
        }
    }

    pub fn load_count(&self) -> usize {
        self.load_count.load(Ordering::SeqCst)
    }
}

impl UserInstructionsProvider for RecordingUserInstructionsProvider {
    fn load_user_instructions(&self) -> LoadInstructionsFuture<'_> {
        self.load_count.fetch_add(1, Ordering::SeqCst);
        self.inner.load_user_instructions()
    }
}

pub fn local(cwd: AbsolutePathBuf) -> TurnEnvironmentSelection {
    TurnEnvironmentSelection {
        environment_id: codex_exec_server::LOCAL_ENVIRONMENT_ID.to_string(),
        cwd: PathUri::from_abs_path(&cwd),
        workspace_roots: vec![PathUri::from_abs_path(&cwd)],
        config: EnvironmentConfigState::FromThread,
    }
}

/// Converts the host-shaped /C:/... cwd projection used by Wine tests back
/// into the selected executor's Windows URI.
pub fn executor_path_uri(path: impl AsRef<Path>) -> Result<PathUri> {
    let path = path.as_ref();
    if matches!(test_environment(), TestEnvironment::WineExec)
        && let Some(path) = path.to_str()
        && matches!(path.as_bytes(), [b'/', drive, b':', b'/' | b'\\', ..] if drive.is_ascii_alphabetic())
    {
        return LegacyAppPathString::from_string(path[1..].to_string())
            .to_path_uri(PathConvention::Windows)
            .map_err(Into::into);
    }
    Ok(PathUri::from_host_native_path(path)?)
}

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,
    selection: TurnEnvironmentSelection,
    local_cwd_temp_dir: Option<Arc<TempDir>>,
    remote_container_name: Option<String>,
}

impl TestEnv {
    pub async fn local() -> Result<Self> {
        Self::local_with_exec_server_url(/*exec_server_url*/ None).await
    }

    /// Builds a host-local test environment, optionally using the provided
    /// exec-server URL instead of the normal implicit local executor.
    pub async fn local_with_exec_server_url(exec_server_url: Option<String>) -> Result<Self> {
        let local_cwd_temp_dir = Arc::new(TempDir::new()?);
        let cwd = local_cwd_temp_dir.abs();
        let selection = match exec_server_url {
            Some(_) => TurnEnvironmentSelection {
                environment_id: codex_exec_server::REMOTE_ENVIRONMENT_ID.to_string(),
                cwd: PathUri::from_abs_path(&cwd),
                workspace_roots: vec![PathUri::from_abs_path(&cwd)],
                config: EnvironmentConfigState::FromThread,
            },
            None => local(cwd.clone()),
        };
        let environment =
            codex_exec_server::Environment::create_for_tests(exec_server_url.clone())?;
        Ok(Self {
            environment,
            exec_server_url,
            cwd,
            selection,
            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
    }

    pub fn exec_server_url(&self) -> Option<&str> {
        self.exec_server_url.as_deref()
    }

    /// Returns the environment and target-native cwd selected by the test harness.
    pub fn selection(&self) -> &TurnEnvironmentSelection {
        &self.selection
    }

    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 test_environment() {
        remote_env @ (TestEnvironment::Docker { .. } | TestEnvironment::WineExec) => {
            let websocket_url = remote_exec_server_url()?;
            let environment =
                codex_exec_server::Environment::create_for_tests(Some(websocket_url.clone()))?;
            let cwd = remote_env
                .remote_cwd(&remote_test_instance_id())?
                .context("remote test environment should define a cwd")?;
            let cwd_uri = cwd.to_path_uri(remote_env.path_convention())?;
            environment
                .get_filesystem()
                .create_directory(
                    &cwd_uri,
                    CreateDirectoryOptions {
                        recursive: true,
                        follow_symlinks: true,
                    },
                    /*sandbox*/ None,
                )
                .await?;
            let selection = TurnEnvironmentSelection {
                environment_id: codex_exec_server::REMOTE_ENVIRONMENT_ID.to_string(),
                cwd: cwd_uri.clone(),
                workspace_roots: vec![cwd_uri.clone()],
                config: EnvironmentConfigState::FromThread,
            };
            let cwd = if remote_env == TestEnvironment::WineExec {
                // TODO(anp): Convert `Config::cwd` to `LegacyAppPathString` and remove this
                // compatibility projection.
                // `Config::cwd` still requires `AbsolutePathBuf`. Preserve the test harness's
                // Linux-absolute `/C:/...` compatibility spelling so converting it back to a
                // `PathUri` recovers the remote Windows convention. Production conversions stay
                // strict: `PathUri::to_abs_path` intentionally rejects foreign paths.
                let path = cwd_uri.to_url().to_file_path().map_err(|()| {
                    anyhow!("remote test cwd URI cannot be projected onto the host: {cwd_uri}")
                })?;
                AbsolutePathBuf::try_from(path)?
            } else {
                cwd_uri.to_abs_path()?
            };
            Ok(TestEnv {
                environment,
                exec_server_url: Some(websocket_url),
                cwd,
                selection,
                local_cwd_temp_dir: None,
                remote_container_name: remote_env.docker_container_name().map(str::to_owned),
            })
        }
        TestEnvironment::Local => TestEnv::local().await,
    }
}

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 {
    ExecCommandViaHeredoc,
}

/// 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,
    analytics_events_client: Option<AnalyticsEventsClient>,
    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>>,
    user_instructions_provider: Option<Arc<dyn UserInstructionsProvider>>,
    supports_openai_form_elicitation: bool,
    external_time_provider: Option<Arc<dyn TimeProvider>>,
    code_mode_host_program: Option<PathBuf>,
    history_mode: Option<ThreadHistoryMode>,
    models_manager: Option<SharedModelsManager>,
    thread_store: Option<Arc<dyn ThreadStore>>,
    image_store: Arc<dyn AttachmentStore>,
}

impl TestCodexBuilder {
    pub fn with_thread_store(mut self, thread_store: Arc<dyn ThreadStore>) -> Self {
        self.thread_store = Some(thread_store);
        self
    }

    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_analytics_events_client(
        mut self,
        analytics_events_client: AnalyticsEventsClient,
    ) -> Self {
        self.analytics_events_client = Some(analytics_events_client);
        self
    }

    pub fn with_models_manager(mut self, models_manager: SharedModelsManager) -> Self {
        self.models_manager = Some(models_manager);
        self
    }

    pub fn with_image_store(mut self, image_store: Arc<dyn AttachmentStore>) -> Self {
        self.image_store = image_store;
        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_history_mode(mut self, history_mode: ThreadHistoryMode) -> Self {
        self.history_mode = Some(history_mode);
        self
    }

    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().expect("test model catalog should parse")
            });
            if !model_catalog
                .models
                .iter()
                .any(|candidate| candidate.slug == model)
            {
                let mut fixture = bundled_models_response()
                    .expect("bundled model catalog should parse")
                    .models
                    .into_iter()
                    .find(|candidate| candidate.slug == "gpt-5.5")
                    .expect("missing bundled model gpt-5.5");
                fixture.slug = model.clone();
                fixture.display_name = model.clone();
                model_catalog.models.push(fixture);
            }
            let model_info = model_catalog
                .models
                .iter_mut()
                .find(|model_info| model_info.slug == model)
                .expect("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_user_instructions_provider(
        mut self,
        provider: Arc<dyn UserInstructionsProvider>,
    ) -> Self {
        self.user_instructions_provider = Some(provider);
        self
    }

    pub fn with_openai_form_elicitation(mut self) -> Self {
        self.supports_openai_form_elicitation = true;
        self
    }

    pub fn with_external_time_provider(mut self, provider: Arc<dyn TimeProvider>) -> Self {
        self.external_time_provider = Some(provider);
        self
    }

    pub fn with_code_mode_host_program(mut self, host_program: PathBuf) -> Self {
        self.code_mode_host_program = Some(host_program);
        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
    }

    /// Builds a test runtime using the execution environment selected by the test process.
    ///
    /// With no remote test configuration, or with `CODEX_TEST_ENVIRONMENT=local`, this uses a
    /// temporary local environment just like [`Self::build`]. `CODEX_TEST_ENVIRONMENT=docker` or
    /// `CODEX_TEST_ENVIRONMENT=wine-exec` selects the remote exec server configured by
    /// `CODEX_TEST_REMOTE_EXEC_SERVER_URL`; the legacy `CODEX_TEST_REMOTE_ENV` Docker-container
    /// configuration does the same. Only the automatically selected environment is registered.
    /// Use [`Self::build_with_remote_and_local_env`] when a remote test also needs the local
    /// environment to be selectable explicitly.
    pub async fn build_with_auto_env(
        &mut self,
        server: &wiremock::MockServer,
    ) -> anyhow::Result<TestCodex> {
        let test_env = test_env().await?;
        self.build_with_environment(server, test_env).await
    }

    /// Builds a test runtime using an explicitly selected execution environment.
    pub async fn build_with_environment(
        &mut self,
        server: &wiremock::MockServer,
        test_env: TestEnv,
    ) -> anyhow::Result<TestCodex> {
        let home = match self.home.clone() {
            Some(home) => home,
            None => Arc::new(TempDir::new()?),
        };
        let base_url = format!("{}/v1", server.uri());
        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()?),
        };
        self.config_mutators.push(Box::new(move |config| {
            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
    }

    pub async fn restart(
        &mut self,
        server: &MockServer,
        previous: &TestCodex,
    ) -> Result<TestCodex> {
        let rollout_path = previous
            .session_configured
            .rollout_path
            .clone()
            .context("rollout path")?;
        previous.codex.shutdown_and_wait().await?;
        self.resume(server, Arc::clone(&previous.home), rollout_path)
            .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>,
        mut 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 = self
            .thread_store
            .clone()
            .unwrap_or_else(|| thread_store_from_config(&config, state_db.clone()));
        let installation_id = resolve_installation_id(&config.codex_home).await?;
        let user_instructions_provider =
            self.user_instructions_provider.clone().unwrap_or_else(|| {
                Arc::new(CodexHomeUserInstructionsProvider::new(
                    config.codex_home.clone(),
                ))
            });
        let auth_manager = codex_core::test_support::auth_manager_from_auth_with_home(
            auth.clone(),
            config.codex_home.to_path_buf(),
        );
        let models_manager = self
            .models_manager
            .clone()
            .unwrap_or_else(|| codex_core::build_models_manager(&config, auth_manager.clone()));
        let code_mode_host_program = self
            .code_mode_host_program
            .take()
            .or_else(|| codex_utils_cargo_bin::cargo_bin("codex-code-mode-host").ok());
        let thread_manager = Arc::new_cyclic(|manager| {
            let mut extensions = self.extensions.to_builder();
            codex_guardian_v2::install_reviewer(&mut extensions, manager.clone());
            let thread_manager = ThreadManager::new(
                &config,
                auth_manager.clone(),
                models_manager,
                codex_core::CodexAppsToolsCache::default(),
                SessionSource::Exec,
                Arc::clone(&environment_manager),
                Arc::new(extensions.build()),
                user_instructions_provider,
                self.analytics_events_client.clone(),
                Arc::clone(&self.image_store),
                Arc::clone(&thread_store),
                codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
                installation_id,
                /*attestation_provider*/ None,
                /*external_time_provider*/ self.external_time_provider.clone(),
            );
            if config.features.enabled(Feature::CodeModeHost)
                && let Some(code_mode_host_program) = code_mode_host_program
            {
                codex_core::test_support::with_code_mode_host_program(
                    thread_manager,
                    code_mode_host_program,
                    &config,
                )
            } else {
                thread_manager
            }
        });
        let user_shell_override = self.user_shell_override.clone();
        let client_mcp_extensions = || {
            ClientMcpExtensions::new(
                self.supports_openai_form_elicitation
                    .then(|| (OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({}))),
            )
        };

        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_with_home(
                    auth,
                    config.codex_home.to_path_buf(),
                );
                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,
                        self.supports_openai_form_elicitation,
                    ),
                )
                .await?
            }
            (Some(path), None) => {
                let auth_manager = codex_core::test_support::auth_manager_from_auth_with_home(
                    auth,
                    config.codex_home.to_path_buf(),
                );
                Box::pin(thread_manager.resume_thread_from_rollout(
                    config.clone(),
                    path,
                    auth_manager,
                    /*parent_trace*/ None,
                    client_mcp_extensions(),
                ))
                .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,
                        self.supports_openai_form_elicitation,
                    ),
                )
                .await?
            }
            (None, None) => {
                let environments = if test_env.selection().cwd.infer_path_convention()
                    == Some(PathConvention::Windows)
                    && PathUri::from_abs_path(&config.cwd) != test_env.selection().cwd
                {
                    let cwd = executor_path_uri(&config.cwd)?;
                    let mut selection = test_env.selection().clone();
                    selection.cwd = cwd.clone();
                    selection.workspace_roots = vec![cwd];
                    test_env.selection = selection.clone();
                    Some(vec![selection])
                } else {
                    None
                };
                Box::pin(thread_manager.start_thread(StartThreadOptions {
                    history_mode: self.history_mode,
                    client_mcp_extensions: client_mcp_extensions(),
                    environments,
                    ..StartThreadOptions::new(config.clone())
                }))
                .await?
            }
        };

        Ok(TestCodex {
            home,
            cwd,
            config,
            codex: new_conversation.thread,
            session_configured: new_conversation.session_configured,
            thread_manager,
            thread_store,
            _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
        };
        // Keep generic tests stable when the bundled catalog default changes. Tests that need a
        // specific model can still override this with a config mutator.
        config.model = Some("gpt-5.5".to_string());
        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().expect("test model catalog should parse");
    let mut model = bundled_models
        .models
        .iter()
        .find(|candidate| candidate.slug == "gpt-5.5")
        .cloned()
        .expect("missing bundled model gpt-5.5");
    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()];
    model.truncation_policy = TruncationPolicyConfig::bytes(/*limit*/ 10_000);
    model.default_reasoning_summary = ReasoningSummary::Auto;
    model.comp_hash = None;
    model.service_tiers.clear();
    model.additional_speed_tiers.clear();
    model.web_search_tool_type = WebSearchToolType::Text;
    model.supports_image_detail_original = false;
    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>,
    pub thread_store: Arc<dyn ThreadStore>,
    _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 workspace_path_uri(&self, rel: impl AsRef<Path>) -> Result<PathUri> {
        let rel = rel
            .as_ref()
            .to_str()
            .context("test workspace path must be UTF-8")?;
        Ok(self.executor_environment().selection().cwd.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
    }

    /// Submits a text turn without changing the current thread settings.
    pub async fn submit_text_turn(&self, prompt: &str) -> Result<()> {
        self.codex
            .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
                text: prompt.into(),
                text_elements: Vec::new(),
            }]))
            .await?;

        wait_for_event(&self.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
        Ok(())
    }

    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
            .start_or_steer_turn(
                TurnInputRequest::user_input(vec![UserInput::Text {
                    text: prompt.into(),
                    text_elements: Vec::new(),
                }])
                .with_thread_settings(ThreadSettingsOverrides {
                    environments: turn_environment_selections,
                    approval_policy: Some(approval_policy),
                    sandbox_policy: Some(sandbox_policy),
                    permission_profile,
                    service_tier,
                    collaboration_mode: Some(CollaborationMode {
                        mode: ModeKind::Default,
                        settings: 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 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_auto_env_builder(mut builder: TestCodexBuilder) -> Result<Self> {
        let server = start_mock_server().await;
        let test = builder.build_with_auto_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 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 path_uri = self.test.workspace_path_uri(rel)?;
        if let Some(parent_uri) = path_uri.parent() {
            self.test
                .fs()
                .create_directory(
                    &parent_uri,
                    CreateDirectoryOptions {
                        recursive: true,
                        follow_symlinks: true,
                    },
                    /*sandbox*/ None,
                )
                .await?;
        }
        self.test
            .fs()
            .write_file(
                &path_uri,
                contents.as_ref().to_vec(),
                Default::default(),
                /*sandbox*/ None,
            )
            .await?;
        Ok(())
    }

    pub async fn read_file_text(&self, rel: impl AsRef<Path>) -> Result<String> {
        let path_uri = self.test.workspace_path_uri(rel)?;
        Ok(self
            .test
            .fs()
            .read_file_text(&path_uri, Default::default(), /*sandbox*/ None)
            .await?)
    }

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

    pub async fn path_exists(&self, rel: impl AsRef<Path>) -> Result<bool> {
        self.path_uri_exists(&self.test.workspace_path_uri(rel)?)
            .await
    }

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

    pub async fn abs_path_exists(&self, path: &AbsolutePathBuf) -> Result<bool> {
        let path_uri = PathUri::from_abs_path(path);
        self.path_uri_exists(&path_uri).await
    }

    async fn path_uri_exists(&self, path_uri: &PathUri) -> Result<bool> {
        match self
            .test
            .fs()
            .get_metadata(path_uri, Default::default(), /*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_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 {
    let missing_output = format!("custom_tool_call_output {call_id} not found");
    bodies
        .iter()
        .filter_map(|body| body.get("input").and_then(Value::as_array))
        .flatten()
        .find(|item| {
            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)
        })
        .expect(&missing_output)
}

fn custom_tool_call_output_text(bodies: &[Value], call_id: &str) -> String {
    let missing_output = format!("custom_tool_call_output {call_id} missing output");
    let output = custom_tool_call_output(bodies, call_id)
        .get("output")
        .expect(&missing_output);
    output_value_to_text(output).expect("custom tool call output missing text output")
}

fn function_call_output<'a>(bodies: &'a [Value], call_id: &str) -> &'a Value {
    let missing_output = format!("function_call_output {call_id} not found");
    bodies
        .iter()
        .filter_map(|body| body.get("input").and_then(Value::as_array))
        .flatten()
        .find(|item| {
            item.get("type").and_then(Value::as_str) == Some("function_call_output")
                && item.get("call_id").and_then(Value::as_str) == Some(call_id)
        })
        .expect(&missing_output)
}

pub fn test_codex() -> TestCodexBuilder {
    TestCodexBuilder {
        config_mutators: vec![Box::new(|config| {
            config
                .features
                .disable(Feature::Apps)
                .expect("test config should allow Apps override");
            // Snapshot tests opt in explicitly; avoid spawning login shells for every test.
            config
                .features
                .disable(Feature::ShellSnapshot)
                .expect("test config should allow ShellSnapshot override");
        })],
        auth: CodexAuth::from_api_key("dummy"),
        analytics_events_client: None,
        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(),
        user_instructions_provider: None,
        supports_openai_form_elicitation: false,
        external_time_provider: None,
        code_mode_host_program: None,
        history_mode: None,
        models_manager: None,
        thread_store: None,
        image_store: codex_core::passthrough_image_store(),
    }
}

#[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");
    }
}