codex/codex-rs/app-server/tests/suite/v2/remote_control.rs

use std::ffi::OsStr;
use std::ffi::OsString;
use std::io::ErrorKind;
use std::time::Duration;

use anyhow::Context;
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::DEFAULT_CLIENT_NAME;
use app_test_support::TestAppServer;
use app_test_support::to_response;
use app_test_support::write_chatgpt_auth;
use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
use codex_app_server::AppServerRuntimeOptions;
use codex_app_server::AppServerTransport;
use codex_app_server::AppServerWebsocketAuthSettings;
use codex_app_server::PluginStartupTasks;
use codex_app_server::RemoteControlStartupMode;
use codex_app_server::run_main_with_transport_options;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RemoteControlClient;
use codex_app_server_protocol::RemoteControlClientsListOrder;
use codex_app_server_protocol::RemoteControlClientsListParams;
use codex_app_server_protocol::RemoteControlClientsListResponse;
use codex_app_server_protocol::RemoteControlClientsRevokeParams;
use codex_app_server_protocol::RemoteControlClientsRevokeResponse;
use codex_app_server_protocol::RemoteControlConnectionStatus;
use codex_app_server_protocol::RemoteControlDisableResponse;
use codex_app_server_protocol::RemoteControlEnableResponse;
use codex_app_server_protocol::RemoteControlPairingStartParams;
use codex_app_server_protocol::RemoteControlPairingStartResponse;
use codex_app_server_protocol::RemoteControlPairingStatusParams;
use codex_app_server_protocol::RemoteControlPairingStatusResponse;
use codex_app_server_protocol::RemoteControlStatusChangedNotification;
use codex_app_server_protocol::RemoteControlStatusReadResponse;
use codex_app_server_protocol::RequestId;
use codex_arg0::Arg0DispatchPaths;
use codex_config::LoaderOverrides;
use codex_config::types::AuthCredentialsStoreMode;
use codex_protocol::protocol::SessionSource;
use codex_state::RemoteControlEnrollmentRecord;
use codex_state::StateRuntime;
use codex_utils_cli::CliConfigOverrides;
use pretty_assertions::assert_eq;
use serial_test::serial;
use tempfile::TempDir;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio::time::timeout;

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
const STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
const REMOTE_CONTROL_DISABLED_BY_REQUIREMENTS_MESSAGE: &str =
    "remote control is disabled by managed requirements";

struct EnvVarGuard {
    key: &'static str,
    original: Option<OsString>,
}

impl EnvVarGuard {
    fn set(key: &'static str, value: &OsStr) -> Self {
        let original = std::env::var_os(key);
        unsafe {
            std::env::set_var(key, value);
        }
        Self { key, original }
    }
}

impl Drop for EnvVarGuard {
    fn drop(&mut self) {
        unsafe {
            match &self.original {
                Some(value) => std::env::set_var(self.key, value),
                None => std::env::remove_var(self.key),
            }
        }
    }
}

async fn remote_control_preference(
    state_db: &StateRuntime,
    websocket_url: &str,
) -> Result<Option<bool>> {
    Ok(state_db
        .get_remote_control_enrollment(websocket_url, "account_id", Some(DEFAULT_CLIENT_NAME))
        .await?
        .context("enrollment should exist")?
        .remote_control_enabled)
}

async fn wait_for_response(mcp: &mut TestAppServer, request_id: i64) -> Result<JSONRPCResponse> {
    timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await?
}

async fn assert_remote_control_disabled_by_requirements(
    mcp: &mut TestAppServer,
    request_id: i64,
) -> Result<()> {
    let JSONRPCError { error, .. } = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(error.code, -32600);
    assert_eq!(
        error.message,
        REMOTE_CONTROL_DISABLED_BY_REQUIREMENTS_MESSAGE
    );
    Ok(())
}

#[tokio::test]
async fn managed_requirements_reject_all_remote_control_rpcs() -> Result<()> {
    let codex_home = TempDir::new()?;
    std::fs::write(
        codex_home.path().join("requirements.toml"),
        "allow_remote_control = false\n",
    )?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let notification = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_notification_message("remoteControl/status/changed"),
    )
    .await??;
    let status: RemoteControlStatusChangedNotification = serde_json::from_value(
        notification
            .params
            .context("remote-control status notification should include params")?,
    )?;
    assert_eq!(status.status, RemoteControlConnectionStatus::Disabled);
    assert_eq!(status.environment_id, None);

    let request_ids = [
        mcp.send_remote_control_enable_request().await?,
        mcp.send_remote_control_disable_request().await?,
        mcp.send_remote_control_status_read_request().await?,
        mcp.send_remote_control_pairing_start_request(RemoteControlPairingStartParams {
            manual_code: false,
        })
        .await?,
        mcp.send_remote_control_pairing_status_request(RemoteControlPairingStatusParams {
            pairing_code: Some("pairing-code".to_string()),
            manual_pairing_code: None,
        })
        .await?,
        mcp.send_remote_control_clients_list_request(RemoteControlClientsListParams {
            environment_id: "environment-id".to_string(),
            cursor: None,
            limit: None,
            order: None,
        })
        .await?,
        mcp.send_remote_control_clients_revoke_request(RemoteControlClientsRevokeParams {
            environment_id: "environment-id".to_string(),
            client_id: "client-id".to_string(),
        })
        .await?,
    ];

    for request_id in request_ids {
        assert_remote_control_disabled_by_requirements(&mut mcp, request_id).await?;
    }

    Ok(())
}

#[tokio::test]
async fn managed_requirements_allow_remote_control_true_does_not_enable_or_block_it() -> Result<()>
{
    let codex_home = TempDir::new()?;
    std::fs::write(
        codex_home.path().join("requirements.toml"),
        "allow_remote_control = true\n",
    )?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_status_read_request().await?;
    let response = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlStatusReadResponse = to_response(response)?;
    assert_eq!(received.status, RemoteControlConnectionStatus::Disabled);
    Ok(())
}

#[tokio::test]
#[serial]
async fn explicit_remote_control_startup_fails_when_disabled_by_requirements() -> Result<()> {
    let codex_home = TempDir::new()?;
    std::fs::write(
        codex_home.path().join("requirements.toml"),
        "allow_remote_control = false\n",
    )?;
    let managed_config_path = codex_home.path().join("managed_config.toml");
    let socket_path = codex_home.path().join("app-server.sock");
    let transport =
        AppServerTransport::from_listen_url(&format!("unix://{}", socket_path.display()))?;
    let _codex_home_guard = EnvVarGuard::set("CODEX_HOME", codex_home.path().as_os_str());

    let result = timeout(
        STARTUP_TIMEOUT,
        run_main_with_transport_options(
            Arg0DispatchPaths {
                codex_self_exe: Some(std::env::current_exe()?),
                codex_linux_sandbox_exe: None,
                main_execve_wrapper_exe: None,
            },
            CliConfigOverrides::default(),
            LoaderOverrides::with_managed_config_path_for_tests(managed_config_path),
            /*strict_config*/ false,
            /*default_analytics_enabled*/ false,
            transport,
            SessionSource::VSCode,
            AppServerWebsocketAuthSettings::default(),
            AppServerRuntimeOptions {
                plugin_startup_tasks: PluginStartupTasks::Skip,
                remote_control_startup_mode: RemoteControlStartupMode::EnabledEphemeral,
                install_shutdown_signal_handler: false,
            },
        ),
    )
    .await?;
    let err = result.expect_err("managed requirements should reject explicit remote control");
    assert_eq!(err.kind(), ErrorKind::InvalidInput);
    assert_eq!(
        err.to_string(),
        REMOTE_CONTROL_DISABLED_BY_REQUIREMENTS_MESSAGE
    );
    assert!(!socket_path.exists());
    Ok(())
}

#[tokio::test]
async fn listen_off_honors_persisted_remote_control_enable() -> Result<()> {
    let codex_home = TempDir::new()?;
    let listener = configured_remote_control_listener(codex_home.path()).await?;
    let websocket_url = format!(
        "ws://{}/backend-api/wham/remote/control/server",
        listener.local_addr()?
    );
    let state_db =
        StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?;
    state_db
        .upsert_remote_control_enrollment(&RemoteControlEnrollmentRecord {
            websocket_url,
            account_id: "account_id".to_string(),
            app_server_client_name: None,
            server_id: "server-id".to_string(),
            environment_id: "environment-id".to_string(),
            server_name: "server-name".to_string(),
            remote_control_enabled: Some(true),
        })
        .await?;

    let _app_server = TestAppServer::new_with_args(codex_home.path(), &["--listen", "off"]).await?;
    let request = timeout(STARTUP_TIMEOUT, read_http_request(&listener)).await??;
    assert!(
        request
            .request_line
            .starts_with("GET /backend-api/wham/remote/control/server ")
            || request
                .request_line
                .starts_with("POST /backend-api/wham/remote/control/server/refresh ")
    );
    Ok(())
}

#[tokio::test]
async fn listen_off_ignores_persisted_enable_when_disabled_by_requirements() -> Result<()> {
    let codex_home = TempDir::new()?;
    let listener = configured_remote_control_listener(codex_home.path()).await?;
    std::fs::write(
        codex_home.path().join("requirements.toml"),
        "allow_remote_control = false\n",
    )?;
    let websocket_url = format!(
        "ws://{}/backend-api/wham/remote/control/server",
        listener.local_addr()?
    );
    let state_db =
        StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?;
    state_db
        .upsert_remote_control_enrollment(&RemoteControlEnrollmentRecord {
            websocket_url: websocket_url.clone(),
            account_id: "account_id".to_string(),
            app_server_client_name: None,
            server_id: "server-id".to_string(),
            environment_id: "environment-id".to_string(),
            server_name: "server-name".to_string(),
            remote_control_enabled: Some(true),
        })
        .await?;

    let mut app_server =
        TestAppServer::new_with_args(codex_home.path(), &["--listen", "off"]).await?;
    let status = timeout(STARTUP_TIMEOUT, app_server.wait_for_exit()).await??;
    assert!(!status.success());
    timeout(Duration::from_millis(100), listener.accept())
        .await
        .expect_err("managed requirements should prevent a remote-control connection");
    assert_eq!(
        state_db
            .get_remote_control_enrollment(
                &websocket_url,
                "account_id",
                /*app_server_client_name*/ None
            )
            .await?
            .context("enrollment should remain persisted")?
            .remote_control_enabled,
        Some(true)
    );
    Ok(())
}

#[tokio::test]
async fn listen_off_exits_without_persisted_remote_control_enable() -> Result<()> {
    for persisted_preference in [None, Some(false)] {
        let codex_home = TempDir::new()?;
        let listener = configured_remote_control_listener(codex_home.path()).await?;
        if let Some(remote_control_enabled) = persisted_preference {
            let websocket_url = format!(
                "ws://{}/backend-api/wham/remote/control/server",
                listener.local_addr()?
            );
            let state_db =
                StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string())
                    .await?;
            state_db
                .upsert_remote_control_enrollment(&RemoteControlEnrollmentRecord {
                    websocket_url,
                    account_id: "account_id".to_string(),
                    app_server_client_name: None,
                    server_id: "server-id".to_string(),
                    environment_id: "environment-id".to_string(),
                    server_name: "server-name".to_string(),
                    remote_control_enabled: Some(remote_control_enabled),
                })
                .await?;
        }

        let mut app_server =
            TestAppServer::new_with_args(codex_home.path(), &["--listen", "off"]).await?;
        let status = timeout(STARTUP_TIMEOUT, app_server.wait_for_exit()).await??;
        assert!(!status.success());
    }
    Ok(())
}

#[tokio::test]
async fn remote_control_disable_returns_disabled_status() -> Result<()> {
    let codex_home = TempDir::new()?;
    let _listener = configured_remote_control_listener(codex_home.path()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_disable_request().await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlDisableResponse = to_response(response)?;

    assert_eq!(received.status, RemoteControlConnectionStatus::Disabled);
    assert!(!received.server_name.is_empty());
    assert_eq!(received.environment_id, None);
    assert!(!received.installation_id.is_empty());
    Ok(())
}

#[tokio::test]
async fn remote_control_status_read_returns_disabled_status() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_status_read_request().await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlStatusReadResponse = to_response(response)?;

    assert_eq!(received.status, RemoteControlConnectionStatus::Disabled);
    assert!(!received.server_name.is_empty());
    assert_eq!(received.environment_id, None);
    assert!(!received.installation_id.is_empty());
    Ok(())
}

#[tokio::test]
async fn remote_control_enable_returns_connecting_status() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = BlockingRemoteControlBackend::start(codex_home.path()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_enable_request().await?;
    assert_eq!(
        timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??,
        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
    );
    timeout(
        Duration::from_millis(100),
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await
    .expect_err("enable response should wait for enrollment");
    backend.complete_enrollment()?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlEnableResponse = to_response(response)?;

    assert_eq!(received.status, RemoteControlConnectionStatus::Connecting);
    assert!(!received.server_name.is_empty());
    assert_eq!(received.environment_id.as_deref(), Some("environment-id"));
    assert!(!received.installation_id.is_empty());
    Ok(())
}

#[tokio::test]
async fn disable_waits_for_in_flight_durable_enable() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = BlockingRemoteControlBackend::start(codex_home.path()).await?;
    let websocket_url = backend.websocket_url().to_string();
    let state_db =
        StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    mcp.send_remote_control_enable_request().await?;
    timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??;
    let disable_request_id = mcp.send_remote_control_disable_request().await?;
    timeout(
        Duration::from_millis(100),
        mcp.read_stream_until_response_message(RequestId::Integer(disable_request_id)),
    )
    .await
    .expect_err("disable response should wait for the in-flight enable");

    backend.complete_enrollment()?;
    let response = wait_for_response(&mut mcp, disable_request_id).await?;
    let received: RemoteControlDisableResponse = to_response(response)?;
    assert_eq!(received.status, RemoteControlConnectionStatus::Disabled);
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(false)
    );
    Ok(())
}

#[tokio::test]
async fn rpc_updates_durable_preference_but_ephemeral_does_not() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = BlockingRemoteControlBackend::start(codex_home.path()).await?;
    let websocket_url = backend.websocket_url().to_string();
    let state_db =
        StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?;

    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_enable_request().await?;
    assert_eq!(
        timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??,
        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
    );
    backend.complete_enrollment()?;
    wait_for_response(&mut mcp, request_id).await?;
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(true)
    );

    let request_id = mcp.send_remote_control_ephemeral_disable_request().await?;
    wait_for_response(&mut mcp, request_id).await?;
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(true)
    );

    let request_id = mcp.send_remote_control_disable_request().await?;
    wait_for_response(&mut mcp, request_id).await?;
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(false)
    );

    let request_id = mcp.send_remote_control_enable_request().await?;
    wait_for_response(&mut mcp, request_id).await?;
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(true)
    );

    let request_id = mcp.send_remote_control_disable_request().await?;
    wait_for_response(&mut mcp, request_id).await?;
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(false)
    );

    let request_id = mcp.send_remote_control_ephemeral_enable_request().await?;
    wait_for_response(&mut mcp, request_id).await?;
    assert_eq!(
        remote_control_preference(&state_db, &websocket_url).await?,
        Some(false)
    );

    Ok(())
}

#[tokio::test]
async fn remote_control_status_read_returns_connecting_status_after_enable() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = BlockingRemoteControlBackend::start(codex_home.path()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_enable_request().await?;
    let enroll_request = timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??;
    assert_eq!(
        enroll_request,
        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
    );
    backend.complete_enrollment()?;
    let _: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;

    let request_id = mcp.send_remote_control_status_read_request().await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlStatusReadResponse = to_response(response)?;

    assert_eq!(received.status, RemoteControlConnectionStatus::Connecting);
    assert!(!received.server_name.is_empty());
    assert_eq!(received.environment_id.as_deref(), Some("environment-id"));
    assert!(!received.installation_id.is_empty());
    Ok(())
}

#[tokio::test]
async fn remote_control_pairing_start_returns_pairing_artifacts() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = PairingRemoteControlBackend::start(codex_home.path()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp.send_remote_control_enable_request().await?;
    let _: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(
        timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??,
        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
    );
    timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_matching_notification(
            "remoteControl/status/changed enrolled",
            |notification| {
                notification.method == "remoteControl/status/changed"
                    && notification
                        .params
                        .as_ref()
                        .and_then(|params| params.get("environmentId"))
                        .and_then(serde_json::Value::as_str)
                        == Some("environment-id")
            },
        ),
    )
    .await??;

    let request_id = mcp
        .send_remote_control_pairing_start_request(RemoteControlPairingStartParams {
            manual_code: true,
        })
        .await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(response.result.get("serverId"), None);
    let received: RemoteControlPairingStartResponse = to_response(response)?;

    assert_eq!(
        received,
        RemoteControlPairingStartResponse {
            pairing_code: "pairing-code".to_string(),
            manual_pairing_code: Some("ABCD-EFGH".to_string()),
            environment_id: "environment-id".to_string(),
            expires_at: 33_336_362_096,
        }
    );

    let request_id = mcp
        .send_remote_control_pairing_status_request(RemoteControlPairingStatusParams {
            pairing_code: Some("pairing-code".to_string()),
            manual_pairing_code: None,
        })
        .await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(response.result.get("serverId"), None);
    let received: RemoteControlPairingStatusResponse = to_response(response)?;

    assert_eq!(
        received,
        RemoteControlPairingStatusResponse { claimed: true }
    );

    let request_id = mcp
        .send_remote_control_pairing_status_request(RemoteControlPairingStatusParams {
            pairing_code: None,
            manual_pairing_code: Some("ABCD-EFGH".to_string()),
        })
        .await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(response.result.get("serverId"), None);
    let received: RemoteControlPairingStatusResponse = to_response(response)?;

    assert_eq!(
        received,
        RemoteControlPairingStatusResponse { claimed: true }
    );
    Ok(())
}

#[tokio::test]
async fn pairing_start_works_after_ephemeral_enable() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = PairingRemoteControlBackend::start(codex_home.path()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
    let request_id = mcp.send_remote_control_ephemeral_enable_request().await?;
    wait_for_response(&mut mcp, request_id).await?;

    let request_id = mcp
        .send_remote_control_pairing_start_request(RemoteControlPairingStartParams {
            manual_code: true,
        })
        .await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(
        timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??,
        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
    );
    assert_eq!(response.result.get("serverId"), None);
    let received: RemoteControlPairingStartResponse = to_response(response)?;

    assert_eq!(
        received,
        RemoteControlPairingStartResponse {
            pairing_code: "pairing-code".to_string(),
            manual_pairing_code: Some("ABCD-EFGH".to_string()),
            environment_id: "environment-id".to_string(),
            expires_at: 33_336_362_096,
        }
    );
    Ok(())
}

#[tokio::test]
async fn remote_control_client_management_works_while_disabled() -> Result<()> {
    let codex_home = TempDir::new()?;
    let mut backend = ClientManagementRemoteControlBackend::start(codex_home.path()).await?;
    let mut mcp = TestAppServer::new(codex_home.path()).await?;
    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;

    let request_id = mcp
        .send_remote_control_clients_list_request(RemoteControlClientsListParams {
            environment_id: "environment-id".to_string(),
            cursor: Some("cursor-id".to_string()),
            limit: Some(10),
            order: Some(RemoteControlClientsListOrder::Desc),
        })
        .await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlClientsListResponse = to_response(response)?;
    assert_eq!(
        received,
        RemoteControlClientsListResponse {
            data: vec![RemoteControlClient {
                client_id: "client-id".to_string(),
                display_name: Some("Anton Phone".to_string()),
                device_type: Some("phone".to_string()),
                platform: Some("ios".to_string()),
                os_version: Some("19.0".to_string()),
                device_model: Some("iPhone".to_string()),
                app_version: Some("1.2.3".to_string()),
                last_seen_at: Some(1_772_694_000),
            }],
            next_cursor: Some("next-cursor".to_string()),
        }
    );

    let request_id = mcp
        .send_remote_control_clients_revoke_request(RemoteControlClientsRevokeParams {
            environment_id: "environment-id".to_string(),
            client_id: "client-id".to_string(),
        })
        .await?;
    let response: JSONRPCResponse = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
    )
    .await??;
    let received: RemoteControlClientsRevokeResponse = to_response(response)?;
    assert_eq!(received, RemoteControlClientsRevokeResponse {});
    assert_eq!(
        timeout(DEFAULT_TIMEOUT, backend.wait_for_requests()).await??,
        vec![
            "GET /backend-api/wham/remote/control/environments/environment-id/clients?cursor=cursor-id&limit=10&order=desc HTTP/1.1".to_string(),
            "DELETE /backend-api/wham/remote/control/environments/environment-id/clients/client-id HTTP/1.1".to_string(),
        ]
    );
    Ok(())
}

struct BlockingRemoteControlBackend {
    enroll_request_rx: Option<oneshot::Receiver<Result<String>>>,
    enroll_response_tx: Option<oneshot::Sender<()>>,
    websocket_url: String,
    server_task: JoinHandle<()>,
}

struct ClientManagementRemoteControlBackend {
    requests_rx: Option<oneshot::Receiver<Result<Vec<String>>>>,
    server_task: JoinHandle<()>,
}

impl ClientManagementRemoteControlBackend {
    async fn start(codex_home: &std::path::Path) -> Result<Self> {
        let listener = configured_remote_control_listener(codex_home).await?;
        let (requests_tx, requests_rx) = oneshot::channel();
        let server_task = tokio::spawn(async move {
            let result = async {
                let list_request = read_http_request(&listener).await?;
                let list_request_line = list_request.request_line;
                respond_with_json(
                    list_request.reader.into_inner(),
                    serde_json::json!({
                        "items": [{
                            "client_id": "client-id",
                            "account_user_id": "user-id",
                            "enrollment_status": "enrolled_device_key",
                            "display_name": "Anton Phone",
                            "device_type": "phone",
                            "platform": "ios",
                            "os_version": "19.0",
                            "device_model": "iPhone",
                            "app_version": "1.2.3",
                            "last_seen_at": "2026-03-05T07:00:00Z",
                            "last_seen_city": "San Francisco",
                        }],
                        "cursor": "next-cursor",
                    }),
                )
                .await?;

                let revoke_request = read_http_request(&listener).await?;
                let revoke_request_line = revoke_request.request_line;
                respond_with_status(revoke_request.reader.into_inner(), "204 No Content", "")
                    .await?;

                Ok(vec![list_request_line, revoke_request_line])
            }
            .await;
            let _ = requests_tx.send(result);
        });
        Ok(Self {
            requests_rx: Some(requests_rx),
            server_task,
        })
    }

    async fn wait_for_requests(&mut self) -> Result<Vec<String>> {
        self.requests_rx
            .take()
            .context("requests should only be awaited once")?
            .await?
    }
}

impl BlockingRemoteControlBackend {
    async fn start(codex_home: &std::path::Path) -> Result<Self> {
        let listener = configured_remote_control_listener(codex_home).await?;
        let websocket_url = format!(
            "ws://{}/backend-api/wham/remote/control/server",
            listener.local_addr()?
        );

        let (enroll_request_tx, enroll_request_rx) = oneshot::channel();
        let (enroll_response_tx, enroll_response_rx) = oneshot::channel();
        let server_task = tokio::spawn(async move {
            match read_enroll_request(&listener).await {
                Ok((request_line, reader)) => {
                    let _ = enroll_request_tx.send(Ok(request_line));
                    if enroll_response_rx.await.is_err() {
                        return;
                    }
                    if respond_with_json(
                        reader.into_inner(),
                        serde_json::json!({
                            "server_id": "server-id",
                            "environment_id": "environment-id",
                            "remote_control_token": "remote-control-token",
                            "expires_at": "3026-05-22T12:34:56Z",
                        }),
                    )
                    .await
                    .is_err()
                    {
                        return;
                    }
                    let Ok(request) = read_http_request(&listener).await else {
                        return;
                    };
                    if !request
                        .request_line
                        .starts_with("GET /backend-api/wham/remote/control/server ")
                    {
                        return;
                    }
                    std::future::pending::<()>().await;
                }
                Err(err) => {
                    let _ = enroll_request_tx.send(Err(err));
                }
            }
        });

        Ok(Self {
            enroll_request_rx: Some(enroll_request_rx),
            enroll_response_tx: Some(enroll_response_tx),
            websocket_url,
            server_task,
        })
    }

    async fn wait_for_enroll_request(&mut self) -> Result<String> {
        let rx = self
            .enroll_request_rx
            .take()
            .context("enroll request should only be awaited once")?;
        rx.await?
    }

    fn complete_enrollment(&mut self) -> Result<()> {
        self.enroll_response_tx
            .take()
            .context("enrollment should only complete once")?
            .send(())
            .map_err(|()| anyhow::anyhow!("enrollment response receiver dropped"))
    }

    fn websocket_url(&self) -> &str {
        &self.websocket_url
    }
}

struct PairingRemoteControlBackend {
    enroll_request_rx: Option<oneshot::Receiver<Result<String>>>,
    server_task: JoinHandle<()>,
}

impl PairingRemoteControlBackend {
    async fn start(codex_home: &std::path::Path) -> Result<Self> {
        let listener = configured_remote_control_listener(codex_home).await?;
        let (enroll_request_tx, enroll_request_rx) = oneshot::channel();
        let server_task = tokio::spawn(async move {
            let mut enroll_request_tx = Some(enroll_request_tx);
            let result = async {
                let enroll_request = read_http_request(&listener).await?;
                if let Some(enroll_request_tx) = enroll_request_tx.take() {
                    let _ = enroll_request_tx.send(Ok(enroll_request.request_line.clone()));
                }
                respond_with_json(
                    enroll_request.reader.into_inner(),
                    serde_json::json!({
                        "server_id": "server-id",
                        "environment_id": "environment-id",
                        "remote_control_token": "remote-control-token",
                        "expires_at": "3026-05-22T12:34:56Z",
                    }),
                )
                .await?;

                let request_after_enroll = read_http_request(&listener).await?;
                let pair_http_request = if request_after_enroll.request_line.starts_with("GET ") {
                    read_http_request(&listener).await?
                } else {
                    request_after_enroll
                };
                respond_with_json(
                    pair_http_request.reader.into_inner(),
                    serde_json::json!({
                        "pairing_code": "pairing-code",
                        "manual_pairing_code": "ABCD-EFGH",
                        "server_id": "server-id",
                        "environment_id": "environment-id",
                        "expires_at": "3026-05-22T12:34:56Z",
                    }),
                )
                .await?;
                for expected_body in [
                    serde_json::json!({ "pairing_code": "pairing-code" }),
                    serde_json::json!({ "manual_pairing_code": "ABCD-EFGH" }),
                ] {
                    let status_http_request = read_http_request(&listener).await?;
                    assert_eq!(
                        status_http_request.request_line,
                        "POST /backend-api/wham/remote/control/server/pair/status HTTP/1.1"
                    );
                    assert_eq!(
                        serde_json::from_str::<serde_json::Value>(&status_http_request.body)?,
                        expected_body
                    );
                    respond_with_json(
                        status_http_request.reader.into_inner(),
                        serde_json::json!({ "claimed": true }),
                    )
                    .await?;
                }
                std::future::pending::<()>().await;
                Ok::<(), anyhow::Error>(())
            }
            .await;

            if let Err(err) = result {
                let err = err.to_string();
                if let Some(enroll_request_tx) = enroll_request_tx {
                    let _ = enroll_request_tx.send(Err(anyhow::anyhow!(err)));
                }
            }
        });

        Ok(Self {
            enroll_request_rx: Some(enroll_request_rx),
            server_task,
        })
    }

    async fn wait_for_enroll_request(&mut self) -> Result<String> {
        self.enroll_request_rx
            .take()
            .context("enroll request should only be awaited once")?
            .await?
    }
}

impl Drop for PairingRemoteControlBackend {
    fn drop(&mut self) {
        self.server_task.abort();
    }
}

impl Drop for BlockingRemoteControlBackend {
    fn drop(&mut self) {
        self.server_task.abort();
    }
}

impl Drop for ClientManagementRemoteControlBackend {
    fn drop(&mut self) {
        self.server_task.abort();
    }
}

struct HttpRequest {
    request_line: String,
    body: String,
    reader: BufReader<TcpStream>,
}

async fn configured_remote_control_listener(codex_home: &std::path::Path) -> Result<TcpListener> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let remote_control_url = format!("http://{}/backend-api/", listener.local_addr()?);
    write_mock_responses_config_toml_with_chatgpt_base_url(
        codex_home,
        &remote_control_url,
        &remote_control_url,
    )?;
    write_chatgpt_auth(
        codex_home,
        ChatGptAuthFixture::new("chatgpt-token")
            .account_id("account_id")
            .chatgpt_account_id("account_id"),
        AuthCredentialsStoreMode::File,
    )?;
    Ok(listener)
}

async fn read_enroll_request(listener: &TcpListener) -> Result<(String, BufReader<TcpStream>)> {
    let request = read_http_request(listener).await?;
    Ok((request.request_line, request.reader))
}

async fn read_http_request(listener: &TcpListener) -> Result<HttpRequest> {
    loop {
        let (stream, _) = listener.accept().await?;
        let mut reader = BufReader::new(stream);

        let mut request_line = String::new();
        reader.read_line(&mut request_line).await?;
        let mut content_length = 0;
        loop {
            let mut line = String::new();
            reader.read_line(&mut line).await?;
            if line == "\r\n" {
                break;
            }
            if let Some(value) = line
                .trim_end()
                .strip_prefix("content-length:")
                .or_else(|| line.trim_end().strip_prefix("Content-Length:"))
            {
                content_length = value.trim().parse::<usize>()?;
            }
        }
        let mut body = vec![0; content_length];
        if content_length > 0 {
            reader.read_exact(&mut body).await?;
        }

        let request_line = request_line.trim_end().to_string();
        if request_line.starts_with("GET ") && request_line.contains("/v1/models?") {
            respond_with_json(reader.into_inner(), serde_json::json!({ "models": [] })).await?;
            continue;
        }

        return Ok(HttpRequest {
            request_line,
            body: String::from_utf8(body)?,
            reader,
        });
    }
}

async fn respond_with_json(stream: TcpStream, body: serde_json::Value) -> Result<()> {
    let body = body.to_string();
    let mut stream = stream;
    stream
        .write_all(
            format!(
                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                body.len()
            )
            .as_bytes(),
        )
        .await?;
    Ok(())
}

async fn respond_with_status(mut stream: TcpStream, status: &str, body: &str) -> Result<()> {
    stream
        .write_all(
            format!(
                "HTTP/1.1 {status}\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                body.len()
            )
            .as_bytes(),
        )
        .await?;
    Ok(())
}