codex-rs/app-server/src/message_processor.rs

use std::collections::HashSet;
use std::future::Future;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;

use crate::attestation::app_server_attestation_provider;
use crate::config_manager::ConfigManager;
use crate::connection_rpc_gate::ConnectionRpcGate;
use crate::current_time::app_server_time_provider;
use crate::error_code::internal_error;
use crate::error_code::invalid_params;
use crate::error_code::invalid_request;
use crate::extensions::ThreadExtensionDependencies;
use crate::extensions::app_server_extension_event_sink;
use crate::extensions::thread_extensions;
use crate::external_agent_migration::ExternalAgentConfigRequestProcessor;
use crate::external_agent_migration::ExternalAgentConfigRequestProcessorArgs;
use crate::fs_watch::FsWatchManager;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::ConnectionRequestId;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::RequestContext;
use crate::plugin_config_reload;
use crate::plugin_config_reload::PluginStartupConfig;
use crate::request_processors::AccountRequestProcessor;
use crate::request_processors::AppsRequestProcessor;
use crate::request_processors::CatalogRequestProcessor;
use crate::request_processors::CommandExecRequestProcessor;
use crate::request_processors::ConfigRequestProcessor;
use crate::request_processors::EnvironmentRequestProcessor;
use crate::request_processors::FeedbackRequestProcessor;
use crate::request_processors::FsRequestProcessor;
use crate::request_processors::GitRequestProcessor;
use crate::request_processors::InitializeRequestProcessor;
use crate::request_processors::MarketplaceRequestProcessor;
use crate::request_processors::McpEventStreamReady;
use crate::request_processors::McpEventStreams;
use crate::request_processors::McpRequestProcessor;
use crate::request_processors::PluginRequestProcessor;
use crate::request_processors::ProcessExecRequestProcessor;
use crate::request_processors::ProjectRequestProcessor;
use crate::request_processors::RemoteControlRequestProcessor;
use crate::request_processors::SearchRequestProcessor;
use crate::request_processors::ThreadGoalRequestProcessor;
use crate::request_processors::ThreadQueueRequestProcessor;
use crate::request_processors::ThreadRequestProcessor;
use crate::request_processors::ThreadResumeTarget;
use crate::request_processors::TurnRequestProcessor;
use crate::request_processors::WindowsSandboxRequestProcessor;
use crate::request_processors::read_server_diagnostics;
use crate::request_serialization::QueuedInitializedRequest;
use crate::request_serialization::RequestSerializationQueueKey;
use crate::request_serialization::RequestSerializationQueues;
use crate::skills_watcher::SkillsWatcher;
use crate::thread_state::ConnectionCapabilities;
use crate::thread_state::ThreadStateManager;
use crate::transport::AppServerTransport;
use crate::transport::RemoteControlHandle;
use crate::turn_cost_worker::TurnCostWorker;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppServerRpcTransport;
use codex_app_server_protocol::ClientNotification;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponsePayload;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::ExperimentalApi;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::UserVerificationCancelResponse;
use codex_app_server_protocol::experimental_required_message;
use codex_arg0::Arg0DispatchPaths;
use codex_code_mode::CodeModeSessionProvider;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::config::ThreadStoreConfig;
use codex_exec_server::EnvironmentManager;
use codex_extension_api::TurnStartAdmission;
use codex_feedback::CodexFeedback;
use codex_goal_extension::GoalService;
use codex_home::CodexHomeUserInstructionsProvider;
use codex_login::AuthManager;
use codex_protocol::ThreadId;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::W3cTraceContext;
use codex_queue_extension::QueuedItemService;
use codex_rollout::StateDbHandle;
use codex_state::log_db::LogDbLayer;
use codex_thread_store::LocalQueueStore;
use codex_thread_store::QueueStore;
use tokio::sync::Mutex;
use tokio::sync::Semaphore;
use tokio::sync::broadcast;
use tokio::sync::watch;
use tokio::time::Duration;
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;
use tracing::Instrument;

use crate::models_refresh_worker::ModelsRefreshWorker;
use crate::turn_admission::TurnAdmission;

const CONNECTION_RPC_DRAIN_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 30);

fn deserialize_client_request(request: JSONRPCRequest) -> Result<ClientRequest, JSONRPCErrorError> {
    reject_obsolete_request_fields(&request)?;

    ClientRequest::try_from(request)
        .map_err(|err| invalid_request(format!("Invalid request: {err}")))
}

fn reject_obsolete_request_fields(request: &JSONRPCRequest) -> Result<(), JSONRPCErrorError> {
    reject_removed_permission_profile(request)?;
    Ok(())
}

fn reject_removed_permission_profile(request: &JSONRPCRequest) -> Result<(), JSONRPCErrorError> {
    if matches!(
        request.method.as_str(),
        "thread/start" | "thread/resume" | "thread/fork" | "turn/start"
    ) && request
        .params
        .as_ref()
        .and_then(serde_json::Value::as_object)
        .is_some_and(|params| params.contains_key("permissionProfile"))
    {
        let method = request.method.as_str();
        return Err(invalid_params(format!(
            "`permissionProfile` is no longer supported for `{method}`; use `permissions` with a named profile id instead"
        )));
    }

    Ok(())
}

pub(crate) struct MessageProcessor {
    pub(crate) turn_admission: TurnAdmission,
    user_verification: Arc<crate::user_verification::Service>,
    outgoing: Arc<OutgoingMessageSender>,
    models_refresh_worker: ModelsRefreshWorker,
    turn_cost_worker: Option<TurnCostWorker>,
    skills_watcher: Arc<SkillsWatcher>,
    account_processor: AccountRequestProcessor,
    apps_processor: AppsRequestProcessor,
    catalog_processor: CatalogRequestProcessor,
    command_exec_processor: CommandExecRequestProcessor,
    process_exec_processor: ProcessExecRequestProcessor,
    config_processor: ConfigRequestProcessor,
    environment_processor: EnvironmentRequestProcessor,
    external_agent_config_processor: ExternalAgentConfigRequestProcessor,
    feedback_processor: FeedbackRequestProcessor,
    fs_processor: FsRequestProcessor,
    git_processor: GitRequestProcessor,
    initialize_processor: InitializeRequestProcessor,
    marketplace_processor: MarketplaceRequestProcessor,
    mcp_processor: McpRequestProcessor,
    plugin_processor: PluginRequestProcessor,
    project_processor: ProjectRequestProcessor,
    remote_control_processor: RemoteControlRequestProcessor,
    search_processor: SearchRequestProcessor,
    thread_goal_processor: ThreadGoalRequestProcessor,
    thread_queue_processor: ThreadQueueRequestProcessor,
    thread_processor: ThreadRequestProcessor,
    turn_processor: TurnRequestProcessor,
    windows_sandbox_processor: WindowsSandboxRequestProcessor,
    request_serialization_queues: RequestSerializationQueues,
}

#[derive(Debug)]
pub(crate) struct ConnectionSessionState {
    pub(crate) rpc_gate: Arc<ConnectionRpcGate>,
    pub(crate) origin: crate::transport::ConnectionOrigin,
    pub(crate) mcp_event_streams: McpEventStreams,
    initialized: OnceLock<InitializedConnectionSessionState>,
}

#[derive(Debug)]
pub(crate) struct InitializedConnectionSessionState {
    pub(crate) experimental_api_enabled: bool,
    pub(crate) opted_out_notification_methods: HashSet<String>,
    pub(crate) app_server_client_name: String,
    pub(crate) client_version: String,
    pub(crate) request_attestation: bool,
    pub(crate) client_mcp_extensions: ClientMcpExtensions,
}

impl ConnectionSessionState {
    pub(crate) fn new(origin: crate::transport::ConnectionOrigin) -> Self {
        Self {
            origin,
            rpc_gate: Arc::new(ConnectionRpcGate::new()),
            mcp_event_streams: McpEventStreams::default(),
            initialized: OnceLock::new(),
        }
    }

    pub(crate) fn initialized(&self) -> bool {
        self.initialized.get().is_some()
    }

    pub(crate) fn experimental_api_enabled(&self) -> bool {
        self.initialized
            .get()
            .is_some_and(|session| session.experimental_api_enabled)
    }

    pub(crate) fn opted_out_notification_methods(&self) -> HashSet<String> {
        self.initialized
            .get()
            .map(|session| session.opted_out_notification_methods.clone())
            .unwrap_or_default()
    }

    pub(crate) fn app_server_client_name(&self) -> Option<&str> {
        self.initialized
            .get()
            .map(|session| session.app_server_client_name.as_str())
    }

    pub(crate) fn client_version(&self) -> Option<&str> {
        self.initialized
            .get()
            .map(|session| session.client_version.as_str())
    }

    pub(crate) fn request_attestation(&self) -> bool {
        self.initialized
            .get()
            .is_some_and(|session| session.request_attestation)
    }

    pub(crate) fn client_mcp_extensions(&self) -> ClientMcpExtensions {
        self.initialized
            .get()
            .map(|session| session.client_mcp_extensions.clone())
            .unwrap_or_default()
    }
    pub(crate) fn initialize(&self, session: InitializedConnectionSessionState) -> Result<(), ()> {
        self.initialized.set(session).map_err(|_| ())
    }
}

pub(crate) struct MessageProcessorArgs {
    pub(crate) outgoing: Arc<OutgoingMessageSender>,
    pub(crate) analytics_events_client: AnalyticsEventsClient,
    pub(crate) arg0_paths: Arg0DispatchPaths,
    pub(crate) config: Arc<Config>,
    pub(crate) config_manager: ConfigManager,
    pub(crate) environment_manager: Arc<EnvironmentManager>,
    pub(crate) feedback: CodexFeedback,
    pub(crate) log_db: Option<LogDbLayer>,
    pub(crate) state_db: Option<StateDbHandle>,
    pub(crate) config_warnings: Vec<ConfigWarningNotification>,
    pub(crate) session_source: SessionSource,
    pub(crate) auth_manager: Arc<AuthManager>,
    pub(crate) user_verification: Arc<crate::user_verification::Service>,
    pub(crate) installation_id: String,
    pub(crate) code_mode_session_provider: Option<Arc<dyn CodeModeSessionProvider>>,
    pub(crate) rpc_transport: AppServerRpcTransport,
    pub(crate) remote_control_handle: Option<RemoteControlHandle>,
    /// `None` skips startup tasks; otherwise preserve the initial config-loading path.
    pub(crate) plugin_startup_tasks: Option<PluginStartupConfig>,
}

impl MessageProcessor {
    /// Create a new `MessageProcessor`, retaining a handle to the outgoing
    /// `Sender` so handlers can enqueue messages to be written to stdout.
    pub(crate) fn new(args: MessageProcessorArgs) -> Self {
        let MessageProcessorArgs {
            outgoing,
            analytics_events_client,
            arg0_paths,
            config,
            config_manager,
            environment_manager,
            feedback,
            log_db,
            state_db,
            config_warnings,
            session_source,
            auth_manager,
            user_verification,
            installation_id,
            code_mode_session_provider,
            rpc_transport,
            remote_control_handle,
            plugin_startup_tasks,
        } = args;
        let thread_state_manager = ThreadStateManager::new();
        outgoing.watch_user_verification_auth(Arc::clone(&auth_manager));
        // The thread store is intentionally process-scoped. Config reloads can
        // affect per-thread behavior, but they must not move newly started,
        // resumed, or forked threads to a different persistence backend/root.
        let thread_store = codex_core::thread_store_from_config(config.as_ref(), state_db.clone());
        // Queue persistence requires SQLite, so in-memory thread stores and
        // app servers without a state database do not have a queue backend.
        let queue_store: Option<Arc<dyn QueueStore>> = match &config.experimental_thread_store {
            ThreadStoreConfig::Local => state_db.as_ref().map(|state_db| {
                Arc::new(LocalQueueStore::new(Arc::clone(state_db))) as Arc<dyn QueueStore>
            }),
            ThreadStoreConfig::InMemory { .. } => None,
        };
        let environment_manager_for_requests = Arc::clone(&environment_manager);
        let environment_manager_for_extensions = Arc::clone(&environment_manager);
        let restriction_product = session_source.restriction_product();
        let executor_skill_provider: Arc<dyn codex_skills_extension::SkillProvider> = Arc::new(
            codex_skills_extension::ExecutorSkillProvider::new_with_restriction_product(
                Arc::clone(&environment_manager_for_extensions),
                restriction_product,
            ),
        );
        let goal_service = Arc::new(GoalService::new());
        let turn_admission = TurnAdmission::default();
        let turn_start_admission: Arc<dyn TurnStartAdmission> = Arc::new(turn_admission.clone());
        let extension_event_sink =
            app_server_extension_event_sink(outgoing.clone(), thread_state_manager.clone());
        let mut queue_service = None;
        let thread_manager = Arc::new_cyclic(|thread_manager| {
            queue_service = queue_store.map(|queue| {
                Arc::new(QueuedItemService::new(
                    queue,
                    thread_manager.clone(),
                    Arc::clone(&extension_event_sink),
                ))
            });
            let manager = ThreadManager::new(
                config.as_ref(),
                auth_manager.clone(),
                codex_core::build_models_manager(config.as_ref(), auth_manager.clone()),
                codex_core::CodexAppsToolsCache::default(),
                session_source,
                environment_manager,
                thread_extensions(ThreadExtensionDependencies {
                    event_sink: Arc::clone(&extension_event_sink),
                    auth_manager: auth_manager.clone(),
                    state_db: state_db.clone(),
                    analytics_events_client: analytics_events_client.clone(),
                    thread_manager: thread_manager.clone(),
                    goal_service: Arc::clone(&goal_service),
                    environment_manager: Arc::clone(&environment_manager_for_extensions),
                    executor_skill_provider: Arc::clone(&executor_skill_provider),
                    git_attribution_base_url: config.chatgpt_base_url.clone(),
                    http_client_factory: config.http_client_factory(),
                    queue_service: queue_service.clone(),
                    turn_start_admission: Some(Arc::clone(&turn_start_admission)),
                }),
                Arc::new(CodexHomeUserInstructionsProvider::new(
                    config.codex_home.clone(),
                )),
                Some(analytics_events_client.clone()),
                codex_core::passthrough_image_store(),
                Arc::clone(&thread_store),
                codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
                installation_id,
                Some(app_server_attestation_provider(
                    outgoing.clone(),
                    thread_state_manager.clone(),
                )),
                Some(app_server_time_provider(
                    outgoing.clone(),
                    thread_state_manager.clone(),
                )),
            );
            match code_mode_session_provider {
                Some(provider) => manager.with_code_mode_session_provider(provider),
                None => manager,
            }
        });
        let models_manager = thread_manager.get_models_manager();
        let models_refresh_worker =
            crate::models_refresh_worker::spawn(&models_manager, config.http_client_factory());
        let turn_cost_worker =
            TurnCostWorker::spawn(Arc::clone(&config), Arc::clone(&auth_manager));
        thread_manager
            .plugins_manager()
            .set_analytics_events_client(analytics_events_client.clone());
        let skills_watcher = SkillsWatcher::new(
            thread_manager.skills_service(),
            &config.codex_home,
            outgoing.clone(),
        );

        let pending_thread_unloads = Arc::new(Mutex::new(HashSet::new()));
        let thread_watch_manager =
            crate::thread_status::ThreadWatchManager::new_with_outgoing(outgoing.clone());
        let thread_list_state_permit = Arc::new(Semaphore::new(/*permits*/ 1));
        let app_list_shutdown_token = CancellationToken::new();
        let request_serialization_queues = RequestSerializationQueues::default();
        let config_processor = ConfigRequestProcessor::new(
            outgoing.clone(),
            config_manager.clone(),
            thread_manager.clone(),
            analytics_events_client.clone(),
        );
        let on_effective_plugins_changed =
            crate::effective_plugin_change::effective_plugins_changed_callback(
                auth_manager.clone(),
                Arc::clone(&thread_manager),
                config_manager.clone(),
                config_processor.clone(),
                request_serialization_queues.clone(),
            );
        let account_processor = AccountRequestProcessor::new(
            auth_manager.clone(),
            Arc::clone(&thread_manager),
            outgoing.clone(),
            Arc::clone(&config),
            config_manager.clone(),
        );
        let apps_processor = AppsRequestProcessor::new(
            auth_manager.clone(),
            Arc::clone(&thread_manager),
            outgoing.clone(),
            config_manager.clone(),
            app_list_shutdown_token,
        );
        let catalog_processor = CatalogRequestProcessor::new(
            outgoing.clone(),
            Arc::clone(&skills_watcher),
            Arc::clone(&thread_manager),
            Arc::clone(&config),
            config_manager.clone(),
        );
        let command_exec_processor = CommandExecRequestProcessor::new(
            arg0_paths.clone(),
            Arc::clone(&config),
            outgoing.clone(),
            config_manager.clone(),
            Arc::clone(&environment_manager_for_requests),
        );
        let process_exec_processor = ProcessExecRequestProcessor::new(
            outgoing.clone(),
            Arc::clone(&environment_manager_for_requests),
        );
        let feedback_processor = FeedbackRequestProcessor::new(
            auth_manager.clone(),
            Arc::clone(&thread_manager),
            Arc::clone(&config),
            feedback,
            log_db.clone(),
            state_db.clone(),
        );
        let git_processor = GitRequestProcessor::new();
        let initialize_processor = InitializeRequestProcessor::new(
            outgoing.clone(),
            analytics_events_client.clone(),
            Arc::clone(&config),
            config_warnings.clone(),
            rpc_transport,
            Arc::clone(&user_verification),
        );
        let marketplace_processor = MarketplaceRequestProcessor::new(
            Arc::clone(&config),
            config_manager.clone(),
            Arc::clone(&thread_manager),
        );
        let mcp_processor = McpRequestProcessor::new(
            auth_manager.clone(),
            Arc::clone(&thread_manager),
            thread_state_manager.clone(),
            outgoing.clone(),
            config_manager.clone(),
        );
        let plugin_processor = PluginRequestProcessor::new(
            auth_manager.clone(),
            Arc::clone(&thread_manager),
            outgoing.clone(),
            analytics_events_client.clone(),
            config_manager.clone(),
            on_effective_plugins_changed,
        );
        let remote_control_processor = RemoteControlRequestProcessor::new(remote_control_handle);
        let search_processor = SearchRequestProcessor::new(outgoing.clone());
        let thread_goal_processor = ThreadGoalRequestProcessor::new(
            Arc::clone(&thread_manager),
            outgoing.clone(),
            Arc::clone(&config),
            thread_state_manager.clone(),
            state_db.clone(),
            Arc::clone(&goal_service),
            config_manager.clone(),
        );
        let thread_queue_processor = ThreadQueueRequestProcessor::new(
            Arc::clone(&thread_manager),
            Arc::clone(&thread_store),
            outgoing.clone(),
            config_manager.clone(),
            queue_service,
        );
        let project_processor = ProjectRequestProcessor::new(
            Arc::clone(&thread_store),
            outgoing.clone(),
            Arc::clone(&thread_list_state_permit),
        );
        let thread_processor = ThreadRequestProcessor::new(
            auth_manager.clone(),
            Arc::clone(&thread_manager),
            outgoing.clone(),
            arg0_paths.clone(),
            Arc::clone(&config),
            config_manager.clone(),
            Arc::clone(&thread_store),
            Arc::clone(&pending_thread_unloads),
            thread_state_manager.clone(),
            thread_watch_manager.clone(),
            Arc::clone(&thread_list_state_permit),
            thread_goal_processor.clone(),
            state_db.clone(),
            log_db,
            Arc::clone(&skills_watcher),
            turn_cost_worker.as_ref().map(TurnCostWorker::handle),
            config_warnings,
        );
        let turn_processor = TurnRequestProcessor::new(
            auth_manager,
            Arc::clone(&thread_manager),
            outgoing.clone(),
            analytics_events_client.clone(),
            arg0_paths.clone(),
            Arc::clone(&config),
            config_manager.clone(),
            pending_thread_unloads,
            thread_state_manager,
            thread_watch_manager,
            Arc::clone(&skills_watcher),
            turn_cost_worker.as_ref().map(TurnCostWorker::handle),
        );
        if let Some(startup_config) = plugin_startup_tasks {
            // Keep plugin startup warmups aligned at app-server startup.
            let reload_config = match startup_config {
                PluginStartupConfig::Current => {
                    plugin_config_reload::for_cwd(config_manager.clone(), config.cwd.clone())
                }
                PluginStartupConfig::Defaults => {
                    plugin_config_reload::defaults(config_manager.clone())
                }
            };
            let on_effective_plugins_changed =
                plugin_processor.effective_plugins_changed_callback();
            thread_manager
                .plugins_manager()
                .maybe_start_plugin_startup_tasks_for_config(
                    &config.plugins_config_input(),
                    reload_config,
                    Some(on_effective_plugins_changed),
                );
        }
        let external_agent_config_processor =
            ExternalAgentConfigRequestProcessor::new(ExternalAgentConfigRequestProcessorArgs {
                outgoing: outgoing.clone(),
                thread_manager: Arc::clone(&thread_manager),
                thread_store: Arc::clone(&thread_store),
                config_manager: config_manager.clone(),
                config_processor: config_processor.clone(),
                state_db,
                analytics_events_client,
                arg0_paths,
                codex_home: config.codex_home.to_path_buf(),
            });
        let environment_processor =
            EnvironmentRequestProcessor::new(thread_manager.environment_manager());
        let fs_processor = FsRequestProcessor::new(
            Arc::clone(&environment_manager_for_requests),
            FsWatchManager::new(outgoing.clone()),
        );
        let windows_sandbox_processor = WindowsSandboxRequestProcessor::new(
            outgoing.clone(),
            Arc::clone(&config),
            config_manager,
        );

        Self {
            turn_admission,
            user_verification,
            outgoing,
            models_refresh_worker,
            turn_cost_worker,
            skills_watcher,
            account_processor,
            apps_processor,
            catalog_processor,
            command_exec_processor,
            process_exec_processor,
            config_processor,
            environment_processor,
            external_agent_config_processor,
            feedback_processor,
            fs_processor,
            git_processor,
            initialize_processor,
            marketplace_processor,
            mcp_processor,
            plugin_processor,
            project_processor,
            remote_control_processor,
            search_processor,
            thread_goal_processor,
            thread_queue_processor,
            thread_processor,
            turn_processor,
            windows_sandbox_processor,
            request_serialization_queues,
        }
    }

    pub(crate) fn clear_runtime_references(&self) {
        self.account_processor.clear_external_auth();
        self.apps_processor.shutdown();
        self.models_refresh_worker.shutdown();
        self.skills_watcher.shutdown();
    }

    pub(crate) async fn process_request(
        self: &Arc<Self>,
        connection_id: ConnectionId,
        request: JSONRPCRequest,
        transport: &AppServerTransport,
        session: Arc<ConnectionSessionState>,
    ) {
        let request_method = request.method.as_str();
        tracing::trace!(
            ?connection_id,
            request_id = ?request.id,
            "app-server request: {request_method}"
        );
        let request_id = ConnectionRequestId {
            connection_id,
            request_id: request.id.clone(),
        };
        let request_span =
            crate::app_server_tracing::request_span(&request, transport, connection_id, &session);
        let request_trace = request.trace.as_ref().map(|trace| W3cTraceContext {
            traceparent: trace.traceparent.clone(),
            tracestate: trace.tracestate.clone(),
        });
        let request_context = RequestContext::new(
            request_id.clone(),
            request_method,
            request_span,
            request_trace,
        );
        Self::run_request_with_context(
            Arc::clone(&self.outgoing),
            request_context.clone(),
            async {
                let codex_request = deserialize_client_request(request);
                let result = match codex_request {
                    Ok(codex_request) => {
                        // Websocket callers finalize outbound readiness in lib.rs after mirroring
                        // session state into outbound state and sending initialize notifications to
                        // this specific connection. Passing `None` avoids marking the connection
                        // ready too early from inside the shared request handler.
                        self.handle_client_request(
                            request_id.clone(),
                            codex_request,
                            Arc::clone(&session),
                            /*outbound_initialized*/ None,
                            request_context.clone(),
                        )
                        .await
                    }
                    Err(error) => Err(error),
                };
                if let Err(error) = result {
                    self.outgoing.send_error(request_id.clone(), error).await;
                }
            },
        )
        .await;
    }

    /// Handles a typed request path used by in-process embedders.
    ///
    /// This bypasses JSON request deserialization but keeps identical request
    /// semantics by delegating to `handle_client_request`.
    pub(crate) async fn process_client_request(
        self: &Arc<Self>,
        connection_id: ConnectionId,
        request: ClientRequest,
        session: Arc<ConnectionSessionState>,
        outbound_initialized: &AtomicBool,
        cancellation: tokio_util::sync::CancellationToken,
    ) {
        let request_id = ConnectionRequestId {
            connection_id,
            request_id: request.id().clone(),
        };
        let request_span =
            crate::app_server_tracing::typed_request_span(&request, connection_id, &session);
        let mut request_context = RequestContext::new(
            request_id.clone(),
            request.method_name(),
            request_span,
            /*parent_trace*/ None,
        );
        request_context.cancellation = cancellation;
        tracing::trace!(
            ?connection_id,
            request_id = ?request_id.request_id,
            "app-server typed request"
        );
        Self::run_request_with_context(
            Arc::clone(&self.outgoing),
            request_context.clone(),
            async {
                // In-process clients do not have the websocket transport loop that performs
                // post-initialize bookkeeping, so they still finalize outbound readiness in
                // the shared request handler.
                let result = self
                    .handle_client_request(
                        request_id.clone(),
                        request,
                        Arc::clone(&session),
                        Some(outbound_initialized),
                        request_context.clone(),
                    )
                    .await;
                if let Err(error) = result {
                    self.outgoing.send_error(request_id.clone(), error).await;
                }
            },
        )
        .await;
    }

    pub(crate) async fn process_notification(&self, notification: JSONRPCNotification) {
        // Currently, we do not expect to receive any notifications from the
        // client, so we just log them.
        tracing::info!("<- notification: {:?}", notification);
    }

    /// Handles typed notifications from in-process clients.
    pub(crate) async fn process_client_notification(&self, notification: ClientNotification) {
        // Currently, we do not expect to receive any typed notifications from
        // in-process clients, so we just log them.
        tracing::info!("<- typed notification: {:?}", notification);
    }

    async fn run_request_with_context<F>(
        outgoing: Arc<OutgoingMessageSender>,
        request_context: RequestContext,
        request_fut: F,
    ) where
        F: Future<Output = ()>,
    {
        outgoing
            .register_request_context(request_context.clone())
            .await;
        request_fut.instrument(request_context.span()).await;
    }

    pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> {
        self.thread_processor.thread_created_receiver()
    }

    pub(crate) async fn daemon_recovery_snapshot(
        &self,
    ) -> codex_app_server_transport::daemon_recovery::RecoverySnapshot {
        self.thread_processor.daemon_recovery_snapshot().await
    }

    pub(crate) async fn restore_daemon_threads(
        &self,
        mut snapshot: codex_app_server_transport::daemon_recovery::RecoverySnapshot,
    ) {
        for thread_id in snapshot.loaded {
            let Ok(_permit) = self.turn_admission.admit() else {
                break;
            };
            if let Err(err) = self
                .thread_processor
                .thread_resume(
                    ThreadResumeTarget::DaemonRecovery(snapshot.interrupted.remove(&thread_id)),
                    codex_app_server_protocol::ThreadResumeParams {
                        thread_id: thread_id.clone(),
                        exclude_turns: true,
                        ..Default::default()
                    },
                    /*app_server_client_name*/ None,
                    /*app_server_client_version*/ None,
                    ClientMcpExtensions::default(),
                )
                .await
            {
                tracing::warn!(code = err.code, "failed to restore saved daemon thread");
            }
        }
    }

    pub(crate) async fn send_initialize_notifications_to_connection(
        &self,
        connection_id: ConnectionId,
    ) {
        self.initialize_processor
            .send_initialize_notifications_to_connection(connection_id)
            .await;
    }

    pub(crate) async fn connection_initialized(
        &self,
        connection_id: ConnectionId,
        request_attestation: bool,
    ) {
        self.account_processor
            .notify_workspace_routing_to_connection(connection_id);
        self.thread_processor
            .connection_initialized(
                connection_id,
                ConnectionCapabilities {
                    request_attestation,
                },
            )
            .await;
    }

    pub(crate) async fn send_initialize_notifications(&self) {
        self.initialize_processor
            .send_initialize_notifications()
            .await;
    }

    pub(crate) async fn try_attach_thread_listener(
        &self,
        thread_id: ThreadId,
        connection_ids: Vec<ConnectionId>,
    ) {
        self.thread_processor
            .try_attach_thread_listener(thread_id, connection_ids)
            .await;
    }

    pub(crate) async fn drain_background_tasks(&self) {
        self.models_refresh_worker.shutdown();
        if let Some(worker) = &self.turn_cost_worker {
            worker.shutdown();
        }
        self.thread_processor.drain_background_tasks().await;
    }

    pub(crate) async fn cancel_active_login(&self) {
        self.account_processor.cancel_active_login().await;
    }

    pub(crate) async fn clear_all_thread_listeners(&self) {
        self.thread_processor.clear_all_thread_listeners().await;
    }

    pub(crate) async fn shutdown_threads(&self) {
        self.thread_processor.shutdown_threads().await;
    }

    pub(crate) async fn connection_closed(
        &self,
        connection_id: ConnectionId,
        session_state: &ConnectionSessionState,
    ) {
        session_state.rpc_gate.close().await;
        self.request_serialization_queues.discard_closed().await;
        self.outgoing
            .disconnect_user_verification_connection(connection_id)
            .await;
        session_state.mcp_event_streams.clear().await;
        if timeout(
            CONNECTION_RPC_DRAIN_TIMEOUT,
            session_state.rpc_gate.shutdown(),
        )
        .await
        .is_err()
        {
            tracing::warn!(
                ?connection_id,
                timeout_seconds = CONNECTION_RPC_DRAIN_TIMEOUT.as_secs(),
                "timed out waiting for connection RPCs to drain"
            );
        }
        self.outgoing.connection_closed(connection_id).await;
        self.fs_processor.connection_closed(connection_id).await;
        self.command_exec_processor
            .connection_closed(connection_id)
            .await;
        self.process_exec_processor
            .connection_closed(connection_id)
            .await;
        self.thread_processor.connection_closed(connection_id).await;
    }

    pub(crate) fn subscribe_running_assistant_turn_count(&self) -> watch::Receiver<usize> {
        self.thread_processor
            .subscribe_running_assistant_turn_count()
    }

    /// Handle a standalone JSON-RPC response originating from the peer.
    pub(crate) async fn process_response(
        &self,
        connection_id: ConnectionId,
        response: JSONRPCResponse,
    ) {
        let JSONRPCResponse { id, result, .. } = response;
        self.outgoing
            .notify_client_response(connection_id, id, result)
            .await
    }

    /// Handle an error object received from the peer.
    pub(crate) async fn process_error(&self, connection_id: ConnectionId, err: JSONRPCError) {
        self.outgoing
            .notify_client_error(connection_id, err.id, err.error)
            .await;
    }

    async fn handle_client_request(
        self: &Arc<Self>,
        connection_request_id: ConnectionRequestId,
        codex_request: ClientRequest,
        session: Arc<ConnectionSessionState>,
        // `Some(...)` means the caller wants initialize to immediately mark the
        // connection outbound-ready. Websocket JSON-RPC calls pass `None` so
        // lib.rs can deliver connection-scoped initialize notifications first.
        outbound_initialized: Option<&AtomicBool>,
        request_context: RequestContext,
    ) -> Result<(), JSONRPCErrorError> {
        let connection_id = connection_request_id.connection_id;
        if let ClientRequest::Initialize { request_id, params } = codex_request {
            let connection_initialized = self
                .initialize_processor
                .initialize(
                    connection_id,
                    request_id,
                    params,
                    &session,
                    outbound_initialized,
                )
                .await?;
            if connection_initialized {
                self.connection_initialized(connection_id, session.request_attestation())
                    .await;
            }
            return Ok(());
        }

        self.dispatch_initialized_client_request(
            connection_request_id,
            codex_request,
            session,
            request_context,
        )
        .await
    }

    async fn dispatch_initialized_client_request(
        self: &Arc<Self>,
        connection_request_id: ConnectionRequestId,
        codex_request: ClientRequest,
        session: Arc<ConnectionSessionState>,
        request_context: RequestContext,
    ) -> Result<(), JSONRPCErrorError> {
        if !session.initialized() {
            return Err(invalid_request("Not initialized"));
        }

        if let Some(reason) = codex_request.experimental_reason()
            && !session.experimental_api_enabled()
        {
            return Err(invalid_request(experimental_required_message(reason)));
        }
        let connection_id = connection_request_id.connection_id;
        self.initialize_processor.track_initialized_request(
            connection_id,
            connection_request_id.request_id.clone(),
            &codex_request,
        );

        let (turn_admission, recheck_turn_admission) = match &codex_request {
            ClientRequest::ThreadStart { .. }
            | ClientRequest::ThreadFork { .. }
            | ClientRequest::ThreadResume { .. }
            | ClientRequest::ThreadRevert { .. }
            | ClientRequest::ThreadSettingsUpdate { .. }
            | ClientRequest::TurnSettingsUpdate { .. }
            | ClientRequest::ThreadDelete { .. }
            | ClientRequest::ThreadArchive { .. } => (Some(self.turn_admission.admit()?), false),
            ClientRequest::TurnStart { .. }
            | ClientRequest::TurnSteer { .. }
            | ClientRequest::ReviewStart { .. }
            | ClientRequest::ThreadCompactStart { .. }
            | ClientRequest::ThreadShellCommand { .. }
            | ClientRequest::ThreadQueueStart { .. }
            | ClientRequest::ThreadRealtimeStart { .. } => {
                (Some(self.turn_admission.admit()?), true)
            }
            _ => (None, false),
        };

        let event_stream_ready = match &codex_request {
            ClientRequest::McpServerEventStreamStart { params, .. } => Some(
                session
                    .mcp_event_streams
                    .start(connection_id, params.clone(), self.mcp_processor.clone())
                    .await?,
            ),
            _ => None,
        };
        let serialization_scope = codex_request.serialization_scope();
        let error_request_id = connection_request_id.clone();
        let rpc_gate = Arc::clone(&session.rpc_gate);
        let processor = Arc::clone(self);
        let span = request_context.span();
        let request = QueuedInitializedRequest::new(
            rpc_gate,
            async move {
                let _turn_admission = turn_admission;
                // Runtime changes already admitted before drain finish normally. Turn work
                // still waiting in serialization must observe the newly closed gate.
                if recheck_turn_admission && let Err(error) = processor.turn_admission.admit() {
                    processor.outgoing.send_error(error_request_id, error).await;
                    return;
                }
                let processor_for_request = Arc::clone(&processor);
                let result = processor_for_request
                    .handle_initialized_client_request(
                        connection_request_id,
                        codex_request,
                        request_context,
                        session,
                        event_stream_ready,
                    )
                    .await;
                if let Err(error) = result {
                    processor.outgoing.send_error(error_request_id, error).await;
                }
            }
            .instrument(span),
        );

        if let Some(scope) = serialization_scope {
            let (key, access) = RequestSerializationQueueKey::from_scope(connection_id, scope);
            self.request_serialization_queues
                .enqueue(key, access, request)
                .await;
        } else {
            tokio::spawn(async move {
                request.run().await;
            });
        }
        Ok(())
    }

    async fn handle_initialized_client_request(
        self: Arc<Self>,
        connection_request_id: ConnectionRequestId,
        codex_request: ClientRequest,
        request_context: RequestContext,
        session: Arc<ConnectionSessionState>,
        event_stream_ready: Option<McpEventStreamReady>,
    ) -> Result<(), JSONRPCErrorError> {
        let connection_id = connection_request_id.connection_id;
        let app_server_client_name = session.app_server_client_name().map(str::to_string);
        let client_version = session.client_version().map(str::to_string);
        let client_mcp_extensions = session.client_mcp_extensions();
        let request_id = ConnectionRequestId {
            connection_id,
            request_id: codex_request.id().clone(),
        };
        let result: Result<Option<ClientResponsePayload>, JSONRPCErrorError> = match codex_request {
            ClientRequest::Initialize { .. } => {
                panic!("Initialize should be handled before initialized request dispatch");
            }
            ClientRequest::UserVerificationCancel { params, .. } => {
                self.outgoing
                    .cancel_user_verification_request(&ConnectionRequestId {
                        connection_id,
                        request_id: params.request_id,
                    })
                    .await;
                Ok(Some(UserVerificationCancelResponse {}.into()))
            }
            request @ (ClientRequest::UserVerificationStatus { .. }
            | ClientRequest::UserVerificationEnroll { .. }
            | ClientRequest::UserVerificationDelete { .. }
            | ClientRequest::UserVerificationVerify { .. }) => {
                return Box::pin(async {
                    // Keep verification temporaries out of unrelated RPCs' stack frames.
                    let operation = match request {
                        ClientRequest::UserVerificationStatus { .. } => {
                            crate::user_verification::Operation::Status
                        }
                        ClientRequest::UserVerificationEnroll { .. } => {
                            crate::user_verification::Operation::Enroll
                        }
                        ClientRequest::UserVerificationDelete { .. } => {
                            crate::user_verification::Operation::Delete
                        }
                        ClientRequest::UserVerificationVerify { params, .. } => {
                            crate::user_verification::Operation::Verify(params)
                        }
                        _ => unreachable!("matched a user-verification request"),
                    };
                    let response = self
                        .user_verification
                        .handle(
                            operation,
                            Arc::clone(&session.rpc_gate),
                            request_context.cancellation.clone(),
                            session.origin,
                        )
                        .await?;
                    let (payload, check) = response.into_parts();
                    self.outgoing
                        .send_response_as_checked(request_id.clone(), payload, check)
                        .await;
                    Ok(())
                })
                .await;
            }
            ClientRequest::ServerDiagnostics { .. } => Ok(Some(read_server_diagnostics().into())),
            ClientRequest::ConfigRead { params, .. } => self
                .config_processor
                .read(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::WindowsSandboxReadiness { .. } => {
                self.windows_sandbox_processor
                    .windows_sandbox_readiness(&request_id)
                    .await
            }
            ClientRequest::ExternalAgentConfigDetect { params, .. } => self
                .external_agent_config_processor
                .detect(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ExternalAgentConfigImport { params, .. } => self
                .external_agent_config_processor
                .import(request_id.clone(), params)
                .await
                .map(|()| None),
            ClientRequest::ExternalAgentConfigImportHistoryRecord { params, .. } => self
                .external_agent_config_processor
                .record_import_history(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ExternalAgentConfigImportHistoriesRead { .. } => self
                .external_agent_config_processor
                .read_import_histories()
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ConfigValueWrite { params, .. } => {
                self.config_processor.value_write(params).await.map(Some)
            }
            ClientRequest::ConfigBatchWrite { params, .. } => {
                self.config_processor.batch_write(params).await.map(Some)
            }
            ClientRequest::ExperimentalFeatureEnablementSet { params, .. } => {
                self.config_processor
                    .experimental_feature_enablement_set(request_id.clone(), params)
                    .await
            }
            ClientRequest::RemoteControlEnable { params, .. } => self
                .remote_control_processor
                .enable(
                    params.is_some_and(|params| params.ephemeral),
                    app_server_client_name.as_deref(),
                )
                .await
                .map(|response| Some(response.into())),
            ClientRequest::RemoteControlDisable { params, .. } => self
                .remote_control_processor
                .disable(
                    params.is_some_and(|params| params.ephemeral),
                    app_server_client_name.as_deref(),
                )
                .await
                .map(|response| Some(response.into())),
            ClientRequest::RemoteControlStatusRead { .. } => self
                .remote_control_processor
                .status_read()
                .map(|response| Some(response.into())),
            ClientRequest::RemoteControlPairingStart { params, .. } => self
                .remote_control_processor
                .pairing_start(params, app_server_client_name.as_deref())
                .await
                .map(|response| Some(response.into())),
            ClientRequest::RemoteControlPairingStatus { params, .. } => self
                .remote_control_processor
                .pairing_status(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::RemoteControlClientsList { params, .. } => self
                .remote_control_processor
                .clients_list(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::RemoteControlClientsRevoke { params, .. } => self
                .remote_control_processor
                .clients_revoke(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ConfigRequirementsRead { params: _, .. } => self
                .config_processor
                .config_requirements_read()
                .await
                .map(|response| Some(response.into())),
            ClientRequest::EnvironmentAdd { params, .. } => {
                self.environment_processor.environment_add(params).await
            }
            ClientRequest::EnvironmentInfo { params, .. } => {
                self.environment_processor.environment_info(params).await
            }
            ClientRequest::EnvironmentStatus { params, .. } => {
                self.environment_processor.environment_status(params).await
            }
            ClientRequest::FsReadFile { params, .. } => self
                .fs_processor
                .read_file(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsWriteFile { params, .. } => self
                .fs_processor
                .write_file(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsCreateDirectory { params, .. } => self
                .fs_processor
                .create_directory(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsGetMetadata { params, .. } => self
                .fs_processor
                .get_metadata(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsReadDirectory { params, .. } => self
                .fs_processor
                .read_directory(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsRemove { params, .. } => self
                .fs_processor
                .remove(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsCopy { params, .. } => self
                .fs_processor
                .copy(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsWatch { params, .. } => self
                .fs_processor
                .watch(connection_id, params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FsUnwatch { params, .. } => self
                .fs_processor
                .unwatch(connection_id, params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ModelProviderCapabilitiesRead { params: _, .. } => self
                .config_processor
                .model_provider_capabilities_read()
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadStart { params, .. } => {
                self.thread_processor
                    .thread_start(
                        request_id.clone(),
                        params,
                        app_server_client_name.clone(),
                        client_version.clone(),
                        client_mcp_extensions.clone(),
                        request_context,
                    )
                    .await
            }
            ClientRequest::ThreadUnsubscribe { params, .. } => {
                let thread_id = params.thread_id.clone();
                let response = self
                    .thread_processor
                    .thread_unsubscribe(&request_id, params)
                    .await?;
                if let Ok(thread_id) = ThreadId::from_string(&thread_id) {
                    session.mcp_event_streams.stop_thread(thread_id).await;
                }
                Ok(response)
            }
            ClientRequest::ThreadResume { params, .. } => {
                self.thread_processor
                    .thread_resume(
                        ThreadResumeTarget::Client(request_id.clone()),
                        params,
                        app_server_client_name.clone(),
                        client_version.clone(),
                        client_mcp_extensions.clone(),
                    )
                    .await
            }
            ClientRequest::ThreadFork { params, .. } => {
                self.thread_processor
                    .thread_fork(
                        request_id.clone(),
                        params,
                        app_server_client_name.clone(),
                        client_version.clone(),
                        client_mcp_extensions.clone(),
                    )
                    .await
            }
            ClientRequest::ThreadArchive { params, .. } => {
                self.thread_processor
                    .thread_archive(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadDelete { params, .. } => {
                self.thread_processor
                    .thread_delete(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadIncrementElicitation { params, .. } => {
                self.thread_processor
                    .thread_increment_elicitation(params)
                    .await
            }
            ClientRequest::ThreadDecrementElicitation { params, .. } => {
                self.thread_processor
                    .thread_decrement_elicitation(params)
                    .await
            }
            ClientRequest::ThreadSetName { params, .. } => {
                self.thread_processor
                    .thread_set_name(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadGoalSet { params, .. } => {
                self.thread_goal_processor
                    .thread_goal_set(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadGoalGet { params, .. } => {
                self.thread_goal_processor.thread_goal_get(params).await
            }
            ClientRequest::ThreadGoalClear { params, .. } => {
                self.thread_goal_processor
                    .thread_goal_clear(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadQueueAdd { params, .. } => self
                .thread_queue_processor
                .add(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadQueueList { params, .. } => self
                .thread_queue_processor
                .list(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadQueueUpdate { params, .. } => self
                .thread_queue_processor
                .update(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadQueueDelete { params, .. } => self
                .thread_queue_processor
                .delete(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadQueueReorder { params, .. } => self
                .thread_queue_processor
                .reorder(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadQueueStart { params, .. } => self
                .thread_queue_processor
                .start(&request_id, params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::ThreadMetadataUpdate { params, .. } => {
                self.thread_processor.thread_metadata_update(params).await
            }
            ClientRequest::ThreadAttachmentAdd { params, .. } => {
                self.thread_processor
                    .thread_attachment_add(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadAttachmentList { params, .. } => {
                self.thread_processor.thread_attachment_list(params).await
            }
            ClientRequest::ThreadAttachmentRemove { params, .. } => {
                self.thread_processor
                    .thread_attachment_remove(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadSectionMove { params, .. } => {
                self.thread_processor.thread_section_move(params).await
            }
            ClientRequest::ThreadSectionList { params, .. } => {
                self.thread_processor.thread_section_list(params).await
            }
            ClientRequest::ThreadSectionCreate { params, .. } => {
                self.thread_processor.thread_section_create(params).await
            }
            ClientRequest::ThreadSectionUpdate { params, .. } => {
                self.thread_processor.thread_section_update(params).await
            }
            ClientRequest::ThreadSectionDelete { params, .. } => {
                self.thread_processor.thread_section_delete(params).await
            }
            ClientRequest::ThreadSettingsUpdate { params, .. } => {
                self.turn_processor
                    .thread_settings_update(&request_id, params)
                    .await
            }
            ClientRequest::ThreadMemoryModeSet { params, .. } => {
                self.thread_processor.thread_memory_mode_set(params).await
            }
            ClientRequest::MemoryStatus { params, .. } => {
                self.thread_processor.memory_status(params).await
            }
            ClientRequest::MemoryReset { .. } => self.thread_processor.memory_reset().await,
            ClientRequest::ThreadUnarchive { params, .. } => {
                self.thread_processor
                    .thread_unarchive(request_id.clone(), params)
                    .await
            }
            ClientRequest::ThreadCompactStart { params, .. } => {
                self.thread_processor
                    .thread_compact_start(&request_id, params)
                    .await
            }
            ClientRequest::ThreadBackgroundTerminalsClean { params, .. } => {
                self.thread_processor
                    .thread_background_terminals_clean(&request_id, params)
                    .await
            }
            ClientRequest::ThreadBackgroundTerminalsList { params, .. } => {
                self.thread_processor
                    .thread_background_terminals_list(params)
                    .await
            }
            ClientRequest::ThreadBackgroundTerminalsTerminate { params, .. } => {
                self.thread_processor
                    .thread_background_terminals_terminate(params)
                    .await
            }
            ClientRequest::ThreadRevert { params, .. } => {
                self.thread_processor
                    .thread_revert(
                        request_id.clone(),
                        params,
                        app_server_client_name.clone(),
                        client_version.clone(),
                    )
                    .await
            }
            ClientRequest::ThreadList { params, .. } => {
                self.thread_processor.thread_list(params).await
            }
            ClientRequest::ProjectList { params, .. } => {
                self.project_processor.project_list(params).await
            }
            ClientRequest::ProjectRead { params, .. } => {
                self.project_processor.project_read(params).await
            }
            ClientRequest::ProjectCreate { params, .. } => {
                self.project_processor.project_create(params).await
            }
            ClientRequest::ProjectImport { params, .. } => {
                self.project_processor.project_import(params).await
            }
            ClientRequest::ProjectUpdate { params, .. } => {
                self.project_processor.project_update(params).await
            }
            ClientRequest::ProjectMove { params, .. } => {
                self.project_processor.project_move(params).await
            }
            ClientRequest::ProjectDelete { params, .. } => {
                self.project_processor.project_delete(params).await
            }
            ClientRequest::ThreadSearch { params, .. } => {
                self.thread_processor.thread_search(params).await
            }
            ClientRequest::ThreadSearchOccurrences { params, .. } => {
                self.thread_processor
                    .thread_search_occurrences(params)
                    .await
            }
            ClientRequest::ThreadLoadedList { params, .. } => {
                self.thread_processor.thread_loaded_list(params).await
            }
            ClientRequest::ThreadRead { params, .. } => {
                self.thread_processor.thread_read(&request_id, params).await
            }
            ClientRequest::ThreadTurnsList { params, .. } => {
                self.thread_processor.thread_turns_list(params).await
            }
            ClientRequest::ThreadItemsList { params, .. } => {
                self.thread_processor.thread_items_list(params).await
            }
            ClientRequest::ThreadShellCommand { params, .. } => {
                self.thread_processor
                    .thread_shell_command(&request_id, params)
                    .await
            }
            ClientRequest::ThreadApproveGuardianDeniedAction { params, .. } => {
                self.thread_processor
                    .thread_approve_guardian_denied_action(&request_id, params)
                    .await
            }
            ClientRequest::GetConversationSummary { params, .. } => {
                self.thread_processor.conversation_summary(params).await
            }
            ClientRequest::SkillsList { params, .. } => {
                self.catalog_processor.skills_list(params).await
            }
            ClientRequest::SkillsExtraRootsSet { params, .. } => {
                self.catalog_processor.skills_extra_roots_set(params).await
            }
            ClientRequest::HooksList { params, .. } => {
                self.catalog_processor.hooks_list(params).await
            }
            ClientRequest::MarketplaceAdd { params, .. } => {
                self.marketplace_processor.marketplace_add(params).await
            }
            ClientRequest::MarketplaceRemove { params, .. } => {
                self.marketplace_processor.marketplace_remove(params).await
            }
            ClientRequest::MarketplaceUpgrade { params, .. } => {
                self.marketplace_processor.marketplace_upgrade(params).await
            }
            ClientRequest::PluginList { params, .. } => {
                self.plugin_processor.plugin_list(params).await
            }
            ClientRequest::PluginSearch { params, .. } => {
                self.plugin_processor.plugin_search(params).await
            }
            ClientRequest::PluginInstalled { params, .. } => {
                self.plugin_processor.plugin_installed(params).await
            }
            ClientRequest::PluginReconcile { params, .. } => {
                self.plugin_processor
                    .plugin_reconcile(
                        params,
                        self.config_processor.clone(),
                        &self.request_serialization_queues,
                    )
                    .await
            }
            ClientRequest::PluginRead { params, .. } => {
                self.plugin_processor.plugin_read(params).await
            }
            ClientRequest::PluginSkillRead { params, .. } => {
                self.plugin_processor.plugin_skill_read(params).await
            }
            ClientRequest::PluginShareSave { params, .. } => {
                self.plugin_processor.plugin_share_save(params).await
            }
            ClientRequest::PluginShareUpdateTargets { params, .. } => {
                self.plugin_processor
                    .plugin_share_update_targets(params)
                    .await
            }
            ClientRequest::PluginShareList { params, .. } => {
                self.plugin_processor.plugin_share_list(params).await
            }
            ClientRequest::PluginShareCheckout { params, .. } => {
                self.plugin_processor.plugin_share_checkout(params).await
            }
            ClientRequest::PluginShareDelete { params, .. } => {
                self.plugin_processor.plugin_share_delete(params).await
            }
            ClientRequest::AppsRead { params, .. } => self.apps_processor.apps_read(params).await,
            ClientRequest::AppsList { params, .. } => {
                self.apps_processor.apps_list(&request_id, params).await
            }
            ClientRequest::AppsInstalled { params, .. } => self
                .apps_processor
                .apps_installed(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::SkillsConfigWrite { params, .. } => {
                self.catalog_processor.skills_config_write(params).await
            }
            ClientRequest::PluginInstall { params, .. } => {
                self.plugin_processor.plugin_install(params).await
            }
            ClientRequest::PluginUninstall { params, .. } => {
                self.plugin_processor.plugin_uninstall(params).await
            }
            ClientRequest::ModelList { params, .. } => {
                self.catalog_processor.model_list(params).await
            }
            ClientRequest::ExperimentalFeatureList { params, .. } => {
                self.catalog_processor
                    .experimental_feature_list(params)
                    .await
            }
            ClientRequest::PermissionProfileList { params, .. } => {
                self.catalog_processor.permission_profile_list(params).await
            }
            ClientRequest::CollaborationModeList { params, .. } => {
                self.catalog_processor.collaboration_mode_list(params).await
            }
            ClientRequest::MockExperimentalMethod { params, .. } => {
                self.catalog_processor
                    .mock_experimental_method(params)
                    .await
            }
            ClientRequest::TurnStart { params, .. } => {
                self.turn_processor
                    .turn_start(
                        request_id.clone(),
                        params,
                        app_server_client_name.clone(),
                        client_version.clone(),
                    )
                    .await
            }
            ClientRequest::ThreadInjectItems { params, .. } => {
                self.turn_processor
                    .thread_inject_items(&request_id, params)
                    .await
            }
            ClientRequest::TurnSteer { params, .. } => {
                self.turn_processor.turn_steer(&request_id, params).await
            }
            ClientRequest::TurnSettingsUpdate { params, .. } => {
                self.turn_processor
                    .turn_settings_update(&request_id, params)
                    .await
            }
            ClientRequest::TurnInterrupt { params, .. } => {
                self.turn_processor
                    .turn_interrupt(&request_id, params)
                    .await
            }
            ClientRequest::ThreadRealtimeStart { params, .. } => {
                self.turn_processor
                    .thread_realtime_start(&request_id, params)
                    .await
            }
            ClientRequest::ThreadRealtimeAppendAudio { params, .. } => {
                self.turn_processor
                    .thread_realtime_append_audio(&request_id, params)
                    .await
            }
            ClientRequest::ThreadRealtimeAppendText { params, .. } => {
                self.turn_processor
                    .thread_realtime_append_text(&request_id, params)
                    .await
            }
            ClientRequest::ThreadRealtimeAppendSpeech { params, .. } => {
                self.turn_processor
                    .thread_realtime_append_speech(&request_id, params)
                    .await
            }
            ClientRequest::ThreadRealtimeStop { params, .. } => {
                self.turn_processor
                    .thread_realtime_stop(&request_id, params)
                    .await
            }
            ClientRequest::ThreadTimelineList { params, .. } => {
                self.thread_processor.thread_timeline_list(params).await
            }
            ClientRequest::ThreadRealtimeListVoices { params: _, .. } => {
                self.turn_processor.thread_realtime_list_voices().await
            }
            ClientRequest::ReviewStart { params, .. } => {
                self.turn_processor.review_start(&request_id, params).await
            }
            ClientRequest::McpServerOauthLogin { params, .. } => {
                self.mcp_processor.mcp_server_oauth_login(params).await
            }
            ClientRequest::McpServerRefresh { params, .. } => {
                self.mcp_processor.mcp_server_refresh(params).await
            }
            ClientRequest::McpServerStatusList { params, .. } => {
                self.mcp_processor
                    .mcp_server_status_list(&request_id, params)
                    .await
            }
            ClientRequest::McpResourceRead { params, .. } => {
                self.mcp_processor
                    .mcp_resource_read(&request_id, params)
                    .await
            }
            ClientRequest::McpServerEventStreamStart { params, .. } => {
                let ready = event_stream_ready.ok_or_else(|| {
                    internal_error("MCP event subscription was not reserved before startup")
                })?;
                session
                    .mcp_event_streams
                    .wait_for_activation(&params.subscription_id, ready)
                    .await?;
                Ok(Some(
                    codex_app_server_protocol::McpServerEventStreamStartResponse {}.into(),
                ))
            }
            ClientRequest::McpServerEventStreamStop { params, .. } => {
                session
                    .mcp_event_streams
                    .stop(&params.subscription_id)
                    .await;
                Ok(Some(
                    codex_app_server_protocol::McpServerEventStreamStopResponse {}.into(),
                ))
            }
            ClientRequest::McpServerToolCall { params, .. } => {
                self.mcp_processor
                    .mcp_server_tool_call(&request_id, params)
                    .await
            }
            ClientRequest::WindowsSandboxSetupStart { params, .. } => {
                self.windows_sandbox_processor
                    .windows_sandbox_setup_start(&request_id, params)
                    .await
            }
            ClientRequest::LoginAccount { params, .. } => {
                self.account_processor
                    .login_account(request_id.clone(), params)
                    .await
            }
            ClientRequest::BedrockDiscover { params, .. } => {
                self.account_processor.bedrock_discover(params).await
            }
            ClientRequest::BedrockSetup { params, .. } => {
                self.account_processor.bedrock_setup(params).await
            }
            ClientRequest::LogoutAccount { .. } => {
                self.account_processor
                    .logout_account(request_id.clone())
                    .await
            }
            ClientRequest::CancelLoginAccount { params, .. } => {
                self.account_processor.cancel_login_account(params).await
            }
            ClientRequest::GetAccount { params, .. } => {
                self.account_processor.get_account(params).await
            }
            ClientRequest::GetAuthStatus { params, .. } => {
                self.account_processor.get_auth_status(params).await
            }
            ClientRequest::GetAccountRateLimits { params, .. } => {
                self.account_processor.get_account_rate_limits(params).await
            }
            ClientRequest::ConsumeAccountRateLimitResetCredit { params, .. } => {
                self.account_processor
                    .consume_account_rate_limit_reset_credit(params)
                    .await
            }
            ClientRequest::GetAccountTokenUsage { params, .. } => {
                self.account_processor.get_account_token_usage(params).await
            }
            ClientRequest::GetWorkspaceMessages { .. } => {
                self.account_processor.get_workspace_messages().await
            }
            ClientRequest::SendAddCreditsNudgeEmail { params, .. } => {
                self.account_processor
                    .send_add_credits_nudge_email(params)
                    .await
            }
            ClientRequest::GitDiffToRemote { params, .. } => {
                self.git_processor.git_diff_to_remote(params).await
            }
            ClientRequest::FuzzyFileSearch { params, .. } => self
                .search_processor
                .fuzzy_file_search(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FuzzyFileSearchSessionStart { params, .. } => self
                .search_processor
                .fuzzy_file_search_session_start_response(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FuzzyFileSearchSessionUpdate { params, .. } => self
                .search_processor
                .fuzzy_file_search_session_update_response(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::FuzzyFileSearchSessionStop { params, .. } => self
                .search_processor
                .fuzzy_file_search_session_stop(params)
                .await
                .map(|response| Some(response.into())),
            ClientRequest::OneOffCommandExec { params, .. } => {
                self.command_exec_processor
                    .one_off_command_exec(&request_id, params)
                    .await
            }
            ClientRequest::CommandExecWrite { params, .. } => {
                self.command_exec_processor
                    .command_exec_write(request_id.clone(), params)
                    .await
            }
            ClientRequest::CommandExecResize { params, .. } => {
                self.command_exec_processor
                    .command_exec_resize(request_id.clone(), params)
                    .await
            }
            ClientRequest::CommandExecTerminate { params, .. } => {
                self.command_exec_processor
                    .command_exec_terminate(request_id.clone(), params)
                    .await
            }
            ClientRequest::ProcessSpawn { params, .. } => self
                .process_exec_processor
                .process_spawn(request_id.clone(), params)
                .await
                .map(|()| None),
            ClientRequest::ProcessWriteStdin { params, .. } => {
                self.process_exec_processor
                    .process_write_stdin(request_id.clone(), params)
                    .await
            }
            ClientRequest::ProcessKill { params, .. } => {
                self.process_exec_processor
                    .process_kill(request_id.clone(), params)
                    .await
            }
            ClientRequest::ProcessResizePty { params, .. } => {
                self.process_exec_processor
                    .process_resize_pty(request_id.clone(), params)
                    .await
            }
            ClientRequest::FeedbackUpload { params, .. } => {
                self.feedback_processor.feedback_upload(params).await
            }
        };

        match result {
            Ok(Some(response)) => {
                self.outgoing
                    .send_response_as(request_id.clone(), response)
                    .await;
            }
            Ok(None) => {}
            Err(error) => {
                self.outgoing.send_error(request_id.clone(), error).await;
            }
        }
        Ok(())
    }
}

#[cfg(test)]
#[path = "message_processor_tracing_tests.rs"]
mod message_processor_tracing_tests;