From 2a020f1a0a7845f5921eb26a8a7c456db88bc9c9 Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Sun, 26 Apr 2026 15:10:35 -0700 Subject: [PATCH] Lift app-server JSON-RPC error handling to request boundary (#19484) ## Why App-server request handling had a lot of repeated JSON-RPC error construction and one-off `send_error`/`return` branches. This made small handlers noisy and pushed error response details into leaf code that otherwise only needed to validate input or call the underlying API. ## What Changed - Added shared JSON-RPC error constructors in `codex-rs/app-server/src/error_code.rs`. - Lifted straightforward request result emission into `codex-rs/app-server/src/message_processor.rs` so response/error dispatch happens at the request boundary. - Reused the result helpers across command exec, config, filesystem, device-key, external-agent config, fs-watch, and outgoing-message paths. - Removed leaf wrapper handlers where the method body was only forwarding to a response helper. - Returned request validation errors upward in the simple cases instead of sending an error locally and immediately returning. ## Verification - `cargo test -p codex-app-server --lib command_exec::tests` - `cargo test -p codex-app-server --lib outgoing_message::tests` - `cargo test -p codex-app-server --lib in_process::tests` - `cargo test -p codex-app-server --test all v2::fs` - `cargo test -p codex-app-server --test all v2::config_rpc` - `cargo test -p codex-app-server --test all v2::external_agent_config` - `cargo test -p codex-app-server --test all v2::initialize` - `just fix -p codex-app-server` - `git diff --check` Note: full `cargo test -p codex-app-server` was attempted and stopped in `message_processor::tracing_tests::turn_start_jsonrpc_span_parents_core_turn_spans` with a stack overflow after unrelated tests had already passed. --- codex-rs/app-server/src/command_exec.rs | 49 +- codex-rs/app-server/src/config_api.rs | 41 +- codex-rs/app-server/src/device_key_api.rs | 17 +- codex-rs/app-server/src/error_code.rs | 22 + .../src/external_agent_config_api.rs | 17 +- codex-rs/app-server/src/fs_api.rs | 18 +- codex-rs/app-server/src/fs_watch.rs | 2 +- codex-rs/app-server/src/message_processor.rs | 602 ++++++------------ codex-rs/app-server/src/outgoing_message.rs | 27 +- 9 files changed, 270 insertions(+), 525 deletions(-) diff --git a/codex-rs/app-server/src/command_exec.rs b/codex-rs/app-server/src/command_exec.rs index ab8618996..18077a994 100644 --- a/codex-rs/app-server/src/command_exec.rs +++ b/codex-rs/app-server/src/command_exec.rs @@ -34,9 +34,9 @@ use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::sync::watch; -use crate::error_code::INTERNAL_ERROR_CODE; -use crate::error_code::INVALID_PARAMS_ERROR_CODE; -use crate::error_code::INVALID_REQUEST_ERROR_CODE; +use crate::error_code::internal_error; +use crate::error_code::invalid_params; +use crate::error_code::invalid_request; use crate::outgoing_message::ConnectionId; use crate::outgoing_message::ConnectionRequestId; use crate::outgoing_message::OutgoingMessageSender; @@ -158,7 +158,7 @@ impl CommandExecManager { } = params; if process_id.is_none() && (tty || stream_stdin || stream_stdout_stderr) { return Err(invalid_request( - "command/exec tty or streaming requires a client-supplied processId".to_string(), + "command/exec tty or streaming requires a client-supplied processId", )); } let process_id = process_id.map_or_else( @@ -178,12 +178,12 @@ impl CommandExecManager { if matches!(exec_request.sandbox, SandboxType::WindowsRestrictedToken) { if tty || stream_stdin || stream_stdout_stderr { return Err(invalid_request( - "streaming command/exec is not supported with windows sandbox".to_string(), + "streaming command/exec is not supported with windows sandbox", )); } if output_bytes_cap != Some(DEFAULT_OUTPUT_BYTES_CAP) { return Err(invalid_request( - "custom outputBytesCap is not supported with windows sandbox".to_string(), + "custom outputBytesCap is not supported with windows sandbox", )); } if let InternalProcessId::Client(_) = &process_id { @@ -249,7 +249,7 @@ impl CommandExecManager { let sessions = Arc::clone(&self.sessions); let (program, args) = command .split_first() - .ok_or_else(|| invalid_request("command must not be empty".to_string()))?; + .ok_or_else(|| invalid_request("command must not be empty"))?; { let mut sessions = self.sessions.lock().await; if sessions.contains_key(&process_key) { @@ -312,7 +312,7 @@ impl CommandExecManager { ) -> Result { if params.delta_base64.is_none() && !params.close_stdin { return Err(invalid_params( - "command/exec/write requires deltaBase64 or closeStdin".to_string(), + "command/exec/write requires deltaBase64 or closeStdin", )); } @@ -421,7 +421,7 @@ impl CommandExecManager { }; let CommandExecSession::Active { control_tx } = session else { return Err(invalid_request( - "command/exec/write, command/exec/terminate, and command/exec/resize are not supported for windows sandbox processes".to_string(), + "command/exec/write, command/exec/terminate, and command/exec/resize are not supported for windows sandbox processes", )); }; let (response_tx, response_rx) = oneshot::channel(); @@ -635,7 +635,7 @@ async fn handle_process_write( ) -> Result<(), JSONRPCErrorError> { if !stream_stdin { return Err(invalid_request( - "stdin streaming is not enabled for this command/exec".to_string(), + "stdin streaming is not enabled for this command/exec", )); } if !delta.is_empty() { @@ -643,7 +643,7 @@ async fn handle_process_write( .writer_sender() .send(delta) .await - .map_err(|_| invalid_request("stdin is already closed".to_string()))?; + .map_err(|_| invalid_request("stdin is already closed"))?; } if close_stdin { session.close_stdin(); @@ -665,7 +665,7 @@ pub(crate) fn terminal_size_from_protocol( ) -> Result { if size.rows == 0 || size.cols == 0 { return Err(invalid_params( - "command/exec size rows and cols must be greater than 0".to_string(), + "command/exec size rows and cols must be greater than 0", )); } Ok(TerminalSize { @@ -681,34 +681,11 @@ fn command_no_longer_running_error(process_id: &InternalProcessId) -> JSONRPCErr )) } -fn invalid_request(message: String) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - } -} - -fn invalid_params(message: String) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_PARAMS_ERROR_CODE, - message, - data: None, - } -} - -fn internal_error(message: String) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message, - data: None, - } -} - #[cfg(test)] mod tests { use std::collections::HashMap; + use crate::error_code::INVALID_REQUEST_ERROR_CODE; use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::models::PermissionProfile; use codex_utils_absolute_path::AbsolutePathBuf; diff --git a/codex-rs/app-server/src/config_api.rs b/codex-rs/app-server/src/config_api.rs index ce0ea3406..355b41543 100644 --- a/codex-rs/app-server/src/config_api.rs +++ b/codex-rs/app-server/src/config_api.rs @@ -1,7 +1,8 @@ use crate::config_manager::ConfigManager; use crate::config_manager_service::ConfigManagerError; -use crate::error_code::INTERNAL_ERROR_CODE; use crate::error_code::INVALID_REQUEST_ERROR_CODE; +use crate::error_code::internal_error; +use crate::error_code::invalid_request; use async_trait::async_trait; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ConfigBatchWriteParams; @@ -99,10 +100,10 @@ impl ConfigApi { self.config_manager .load_latest_config(fallback_cwd) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to resolve feature override precedence: {err}"), - data: None, + .map_err(|err| { + internal_error(format!( + "failed to resolve feature override precedence: {err}" + )) }) } @@ -197,14 +198,10 @@ impl ConfigApi { continue; } - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "unsupported feature enablement `{key}`: currently supported features are {}", - SUPPORTED_EXPERIMENTAL_FEATURE_ENABLEMENT.join(", ") - ), - data: None, - }); + return Err(invalid_request(format!( + "unsupported feature enablement `{key}`: currently supported features are {}", + SUPPORTED_EXPERIMENTAL_FEATURE_ENABLEMENT.join(", ") + ))); } let message = if let Some(feature) = feature_for_key(key) { @@ -215,11 +212,7 @@ impl ConfigApi { } else { format!("invalid feature enablement `{key}`") }; - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }); + return Err(invalid_request(message)); } if enablement.is_empty() { @@ -232,11 +225,7 @@ impl ConfigApi { .iter() .map(|(name, enabled)| (name.clone(), *enabled)), ) - .map_err(|_| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "failed to update feature enablement".to_string(), - data: None, - })?; + .map_err(|_| internal_error("failed to update feature enablement"))?; self.load_latest_config(/*fallback_cwd*/ None).await?; self.user_config_reloader.reload_user_config().await; @@ -468,11 +457,7 @@ fn map_error(err: ConfigManagerError) -> JSONRPCErrorError { return config_write_error(code, err.to_string()); } - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: err.to_string(), - data: None, - } + internal_error(err.to_string()) } fn config_write_error(code: ConfigWriteErrorCode, message: impl Into) -> JSONRPCErrorError { diff --git a/codex-rs/app-server/src/device_key_api.rs b/codex-rs/app-server/src/device_key_api.rs index dbbc32f1c..b3d31426d 100644 --- a/codex-rs/app-server/src/device_key_api.rs +++ b/codex-rs/app-server/src/device_key_api.rs @@ -1,5 +1,5 @@ -use crate::error_code::INTERNAL_ERROR_CODE; -use crate::error_code::INVALID_REQUEST_ERROR_CODE; +use crate::error_code::internal_error; +use crate::error_code::invalid_request; use async_trait::async_trait; use base64::Engine; use base64::engine::general_purpose::STANDARD; @@ -302,16 +302,13 @@ fn protection_class_from_store( } fn map_device_key_error(error: DeviceKeyError) -> JSONRPCErrorError { - let code = match error { + match &error { DeviceKeyError::DegradedProtectionNotAllowed { .. } | DeviceKeyError::HardwareBackedKeysUnavailable | DeviceKeyError::KeyNotFound - | DeviceKeyError::InvalidPayload(_) => INVALID_REQUEST_ERROR_CODE, - DeviceKeyError::Platform(_) | DeviceKeyError::Crypto(_) => INTERNAL_ERROR_CODE, - }; - JSONRPCErrorError { - code, - message: error.to_string(), - data: None, + | DeviceKeyError::InvalidPayload(_) => invalid_request(error.to_string()), + DeviceKeyError::Platform(_) | DeviceKeyError::Crypto(_) => { + internal_error(error.to_string()) + } } } diff --git a/codex-rs/app-server/src/error_code.rs b/codex-rs/app-server/src/error_code.rs index 924a7086a..0054d2988 100644 --- a/codex-rs/app-server/src/error_code.rs +++ b/codex-rs/app-server/src/error_code.rs @@ -1,5 +1,27 @@ +use codex_app_server_protocol::JSONRPCErrorError; + pub(crate) const INVALID_REQUEST_ERROR_CODE: i64 = -32600; pub const INVALID_PARAMS_ERROR_CODE: i64 = -32602; pub(crate) const INTERNAL_ERROR_CODE: i64 = -32603; pub(crate) const OVERLOADED_ERROR_CODE: i64 = -32001; pub const INPUT_TOO_LARGE_ERROR_CODE: &str = "input_too_large"; + +pub(crate) fn invalid_request(message: impl Into) -> JSONRPCErrorError { + error(INVALID_REQUEST_ERROR_CODE, message) +} + +pub(crate) fn invalid_params(message: impl Into) -> JSONRPCErrorError { + error(INVALID_PARAMS_ERROR_CODE, message) +} + +pub(crate) fn internal_error(message: impl Into) -> JSONRPCErrorError { + error(INTERNAL_ERROR_CODE, message) +} + +fn error(code: i64, message: impl Into) -> JSONRPCErrorError { + JSONRPCErrorError { + code, + message: message.into(), + data: None, + } +} diff --git a/codex-rs/app-server/src/external_agent_config_api.rs b/codex-rs/app-server/src/external_agent_config_api.rs index 0741ad5bd..34ad572ca 100644 --- a/codex-rs/app-server/src/external_agent_config_api.rs +++ b/codex-rs/app-server/src/external_agent_config_api.rs @@ -3,7 +3,7 @@ use crate::config::external_agent_config::ExternalAgentConfigMigrationItem as Co use crate::config::external_agent_config::ExternalAgentConfigMigrationItemType as CoreMigrationItemType; use crate::config::external_agent_config::ExternalAgentConfigService; use crate::config::external_agent_config::PendingPluginImport; -use crate::error_code::INTERNAL_ERROR_CODE; +use crate::error_code::internal_error; use codex_app_server_protocol::ExternalAgentConfigDetectParams; use codex_app_server_protocol::ExternalAgentConfigDetectResponse; use codex_app_server_protocol::ExternalAgentConfigImportParams; @@ -12,7 +12,6 @@ use codex_app_server_protocol::ExternalAgentConfigMigrationItemType; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::MigrationDetails; use codex_app_server_protocol::PluginsMigration; -use std::io; use std::path::PathBuf; #[derive(Clone)] @@ -38,7 +37,7 @@ impl ExternalAgentConfigApi { cwds: params.cwds, }) .await - .map_err(map_io_error)?; + .map_err(|err| internal_error(err.to_string()))?; Ok(ExternalAgentConfigDetectResponse { items: items @@ -125,7 +124,7 @@ impl ExternalAgentConfigApi { .collect(), ) .await - .map_err(map_io_error) + .map_err(|err| internal_error(err.to_string())) } pub(crate) async fn complete_pending_plugin_import( @@ -139,14 +138,6 @@ impl ExternalAgentConfigApi { ) .await .map(|_| ()) - .map_err(map_io_error) - } -} - -fn map_io_error(err: io::Error) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: err.to_string(), - data: None, + .map_err(|err| internal_error(err.to_string())) } } diff --git a/codex-rs/app-server/src/fs_api.rs b/codex-rs/app-server/src/fs_api.rs index 93b4f21c2..203b053e5 100644 --- a/codex-rs/app-server/src/fs_api.rs +++ b/codex-rs/app-server/src/fs_api.rs @@ -1,5 +1,5 @@ -use crate::error_code::INTERNAL_ERROR_CODE; -use crate::error_code::INVALID_REQUEST_ERROR_CODE; +use crate::error_code::internal_error; +use crate::error_code::invalid_request; use base64::Engine; use base64::engine::general_purpose::STANDARD; use codex_app_server_protocol::FsCopyParams; @@ -158,22 +158,10 @@ impl FsApi { } } -pub(crate) fn invalid_request(message: impl Into) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: message.into(), - data: None, - } -} - pub(crate) fn map_fs_error(err: io::Error) -> JSONRPCErrorError { if err.kind() == io::ErrorKind::InvalidInput { invalid_request(err.to_string()) } else { - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: err.to_string(), - data: None, - } + internal_error(err.to_string()) } } diff --git a/codex-rs/app-server/src/fs_watch.rs b/codex-rs/app-server/src/fs_watch.rs index ff0005147..4ae1ca149 100644 --- a/codex-rs/app-server/src/fs_watch.rs +++ b/codex-rs/app-server/src/fs_watch.rs @@ -1,4 +1,4 @@ -use crate::fs_api::invalid_request; +use crate::error_code::invalid_request; use crate::outgoing_message::ConnectionId; use crate::outgoing_message::OutgoingMessageSender; use codex_app_server_protocol::FsChangedNotification; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 2def169c4..6cdb93936 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -10,7 +10,7 @@ use crate::codex_message_processor::CodexMessageProcessorArgs; use crate::config_api::ConfigApi; use crate::config_manager::ConfigManager; use crate::device_key_api::DeviceKeyApi; -use crate::error_code::INVALID_REQUEST_ERROR_CODE; +use crate::error_code::invalid_request; use crate::external_agent_config_api::ExternalAgentConfigApi; use crate::fs_api::FsApi; use crate::fs_watch::FsWatchManager; @@ -34,7 +34,6 @@ use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientNotification; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ConfigBatchWriteParams; -use codex_app_server_protocol::ConfigReadParams; use codex_app_server_protocol::ConfigValueWriteParams; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::DeviceKeyCreateParams; @@ -42,20 +41,10 @@ use codex_app_server_protocol::DeviceKeyPublicParams; use codex_app_server_protocol::DeviceKeySignParams; use codex_app_server_protocol::ExperimentalApi; use codex_app_server_protocol::ExperimentalFeatureEnablementSetParams; -use codex_app_server_protocol::ExternalAgentConfigDetectParams; use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification; use codex_app_server_protocol::ExternalAgentConfigImportParams; use codex_app_server_protocol::ExternalAgentConfigImportResponse; use codex_app_server_protocol::ExternalAgentConfigMigrationItemType; -use codex_app_server_protocol::FsCopyParams; -use codex_app_server_protocol::FsCreateDirectoryParams; -use codex_app_server_protocol::FsGetMetadataParams; -use codex_app_server_protocol::FsReadDirectoryParams; -use codex_app_server_protocol::FsReadFileParams; -use codex_app_server_protocol::FsRemoveParams; -use codex_app_server_protocol::FsUnwatchParams; -use codex_app_server_protocol::FsWatchParams; -use codex_app_server_protocol::FsWriteFileParams; use codex_app_server_protocol::InitializeResponse; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCErrorError; @@ -390,43 +379,28 @@ impl MessageProcessor { Arc::clone(&self.outgoing), request_context.clone(), async { - let request_json = match serde_json::to_value(&request) { - Ok(request_json) => request_json, - Err(err) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("Invalid request: {err}"), - data: None, - }; - self.outgoing.send_error(request_id.clone(), error).await; - return; - } - }; - - let codex_request = match serde_json::from_value::(request_json) { - Ok(codex_request) => codex_request, - Err(err) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("Invalid request: {err}"), - data: None, - }; - self.outgoing.send_error(request_id.clone(), error).await; - return; - } - }; - // 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(), - ) + let result = async { + let request_json = serde_json::to_value(&request) + .map_err(|err| invalid_request(format!("Invalid request: {err}")))?; + let codex_request = serde_json::from_value::(request_json) + .map_err(|err| invalid_request(format!("Invalid request: {err}")))?; + // 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 + } .await; + if let Err(error) = result { + self.outgoing.send_error(request_id.clone(), error).await; + } }, ) .await; @@ -463,14 +437,18 @@ impl MessageProcessor { // 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. - self.handle_client_request( - request_id.clone(), - request, - Arc::clone(&session), - Some(outbound_initialized), - request_context.clone(), - ) - .await; + 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; @@ -598,7 +576,7 @@ impl MessageProcessor { // 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 { // Handle Initialize internally so CodexMessageProcessor does not have to concern @@ -608,13 +586,7 @@ impl MessageProcessor { request_id, }; if session.initialized() { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "Already initialized".to_string(), - data: None, - }; - self.outgoing.send_error(connection_request_id, error).await; - return; + return Err(invalid_request("Already initialized")); } // TODO(maxj): Revisit capability scoping for `experimental_api_enabled`. @@ -642,17 +614,9 @@ impl MessageProcessor { // Validate before committing; set_default_originator validates while // mutating process-global metadata. if HeaderValue::from_str(&name).is_err() { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "Invalid clientInfo.name: '{name}'. Must be a valid HTTP header value." - ), - data: None, - }; - self.outgoing - .send_error(connection_request_id.clone(), error) - .await; - return; + return Err(invalid_request(format!( + "Invalid clientInfo.name: '{name}'. Must be a valid HTTP header value." + ))); } let originator = name.clone(); let user_agent_suffix = format!("{name}; {version}"); @@ -668,13 +632,7 @@ impl MessageProcessor { }) .is_err() { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "Already initialized".to_string(), - data: None, - }; - self.outgoing.send_error(connection_request_id, error).await; - return; + return Err(invalid_request("Already initialized")); } // Only the request that wins session initialization may mutate @@ -729,7 +687,7 @@ impl MessageProcessor { .connection_initialized(connection_id) .await; } - return; + return Ok(()); } self.dispatch_initialized_client_request( @@ -738,7 +696,7 @@ impl MessageProcessor { session, request_context, ) - .await; + .await } async fn dispatch_initialized_client_request( @@ -747,27 +705,15 @@ impl MessageProcessor { codex_request: ClientRequest, session: Arc, request_context: RequestContext, - ) { + ) -> Result<(), JSONRPCErrorError> { if !session.initialized() { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "Not initialized".to_string(), - data: None, - }; - self.outgoing.send_error(connection_request_id, error).await; - return; + return Err(invalid_request("Not initialized")); } if let Some(reason) = codex_request.experimental_reason() && !session.experimental_api_enabled() { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: experimental_required_message(reason), - data: None, - }; - self.outgoing.send_error(connection_request_id, error).await; - return; + return Err(invalid_request(experimental_required_message(reason))); } let connection_id = connection_request_id.connection_id; if self.config.features.enabled(Feature::GeneralAnalytics) @@ -793,7 +739,7 @@ impl MessageProcessor { client_version, device_key_requests_allowed, ) - .await; + .await } async fn handle_initialized_client_request( @@ -804,66 +750,48 @@ impl MessageProcessor { app_server_client_name: Option, client_version: Option, device_key_requests_allowed: bool, - ) { + ) -> Result<(), JSONRPCErrorError> { let connection_id = connection_request_id.connection_id; + let request_id_for_connection = |request_id| ConnectionRequestId { + connection_id, + request_id, + }; match codex_request { ClientRequest::ConfigRead { request_id, params } => { - self.handle_config_read( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.config_api.read(params).await, + ) + .await; } ClientRequest::ExternalAgentConfigDetect { request_id, params } => { - self.handle_external_agent_config_detect( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.external_agent_config_api.detect(params).await, + ) + .await; } ClientRequest::ExternalAgentConfigImport { request_id, params } => { self.handle_external_agent_config_import( - ConnectionRequestId { - connection_id, - request_id, - }, + request_id_for_connection(request_id), params, ) - .await; + .await?; } ClientRequest::ConfigValueWrite { request_id, params } => { - self.handle_config_value_write( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.handle_config_value_write(request_id_for_connection(request_id), params) + .await; } ClientRequest::ConfigBatchWrite { request_id, params } => { - self.handle_config_batch_write( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.handle_config_batch_write(request_id_for_connection(request_id), params) + .await; } ClientRequest::ExperimentalFeatureEnablementSet { request_id, params } => { self.handle_experimental_feature_enablement_set( - ConnectionRequestId { - connection_id, - request_id, - }, + request_id_for_connection(request_id), params, ) .await; @@ -872,133 +800,105 @@ impl MessageProcessor { request_id, params: _, } => { - self.handle_config_requirements_read(ConnectionRequestId { - connection_id, - request_id, - }) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.config_api.config_requirements_read().await, + ) + .await; } ClientRequest::DeviceKeyCreate { request_id, params } => { self.handle_device_key_create( - ConnectionRequestId { - connection_id, - request_id, - }, + request_id_for_connection(request_id), params, device_key_requests_allowed, ); } ClientRequest::DeviceKeyPublic { request_id, params } => { self.handle_device_key_public( - ConnectionRequestId { - connection_id, - request_id, - }, + request_id_for_connection(request_id), params, device_key_requests_allowed, ); } ClientRequest::DeviceKeySign { request_id, params } => { self.handle_device_key_sign( - ConnectionRequestId { - connection_id, - request_id, - }, + request_id_for_connection(request_id), params, device_key_requests_allowed, ); } ClientRequest::FsReadFile { request_id, params } => { - self.handle_fs_read_file( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.read_file(params).await, + ) + .await; } ClientRequest::FsWriteFile { request_id, params } => { - self.handle_fs_write_file( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.write_file(params).await, + ) + .await; } ClientRequest::FsCreateDirectory { request_id, params } => { - self.handle_fs_create_directory( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.create_directory(params).await, + ) + .await; } ClientRequest::FsGetMetadata { request_id, params } => { - self.handle_fs_get_metadata( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.get_metadata(params).await, + ) + .await; } ClientRequest::FsReadDirectory { request_id, params } => { - self.handle_fs_read_directory( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.read_directory(params).await, + ) + .await; } ClientRequest::FsRemove { request_id, params } => { - self.handle_fs_remove( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.remove(params).await, + ) + .await; } ClientRequest::FsCopy { request_id, params } => { - self.handle_fs_copy( - ConnectionRequestId { - connection_id, - request_id, - }, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_api.copy(params).await, + ) + .await; } ClientRequest::FsWatch { request_id, params } => { - self.handle_fs_watch( - ConnectionRequestId { - connection_id, - request_id, - }, - connection_id, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_watch_manager.watch(connection_id, params).await, + ) + .await; } ClientRequest::FsUnwatch { request_id, params } => { - self.handle_fs_unwatch( - ConnectionRequestId { - connection_id, - request_id, - }, - connection_id, - params, - ) - .await; + self.outgoing + .send_result( + request_id_for_connection(request_id), + self.fs_watch_manager.unwatch(connection_id, params).await, + ) + .await; } other => { // Box the delegated future so this wrapper's async state machine does not @@ -1016,13 +916,7 @@ impl MessageProcessor { .await; } } - } - - async fn handle_config_read(&self, request_id: ConnectionRequestId, params: ConfigReadParams) { - match self.config_api.read(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } + Ok(()) } async fn handle_config_value_write( @@ -1167,13 +1061,6 @@ impl MessageProcessor { } } - async fn handle_config_requirements_read(&self, request_id: ConnectionRequestId) { - match self.config_api.config_requirements_read().await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - fn handle_device_key_create( &self, request_id: ConnectionRequestId, @@ -1230,193 +1117,80 @@ impl MessageProcessor { let device_key_api = self.device_key_api.clone(); let outgoing = Arc::clone(&self.outgoing); tokio::spawn(async move { - if !device_key_requests_allowed { - outgoing - .send_error( - request_id, - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("{method} is not available over remote transports"), - data: None, - }, - ) - .await; - return; - } - - match run_request(device_key_api).await { - Ok(response) => outgoing.send_response(request_id, response).await, - Err(error) => outgoing.send_error(request_id, error).await, + let result = async { + if !device_key_requests_allowed { + return Err(invalid_request(format!( + "{method} is not available over remote transports" + ))); + } + run_request(device_key_api).await } + .await; + outgoing.send_result(request_id, result).await; }); } - async fn handle_external_agent_config_detect( - &self, - request_id: ConnectionRequestId, - params: ExternalAgentConfigDetectParams, - ) { - match self.external_agent_config_api.detect(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - async fn handle_external_agent_config_import( &self, request_id: ConnectionRequestId, params: ExternalAgentConfigImportParams, - ) { + ) -> Result<(), JSONRPCErrorError> { let has_plugin_imports = params.migration_items.iter().any(|item| { matches!( item.item_type, ExternalAgentConfigMigrationItemType::Plugins ) }); - match self.external_agent_config_api.import(params).await { - Ok(pending_plugin_imports) => { - if has_plugin_imports { - self.handle_config_mutation().await; - } - self.outgoing - .send_response(request_id, ExternalAgentConfigImportResponse {}) - .await; - if !has_plugin_imports { - return; - } + let pending_plugin_imports = self.external_agent_config_api.import(params).await?; + if has_plugin_imports { + self.handle_config_mutation().await; + } + self.outgoing + .send_response(request_id, ExternalAgentConfigImportResponse {}) + .await; - if pending_plugin_imports.is_empty() { - self.outgoing - .send_server_notification( - ServerNotification::ExternalAgentConfigImportCompleted( - ExternalAgentConfigImportCompletedNotification {}, - ), - ) - .await; - return; - } + if !has_plugin_imports { + return Ok(()); + } - let external_agent_config_api = self.external_agent_config_api.clone(); - let outgoing = Arc::clone(&self.outgoing); - let thread_manager = Arc::clone(&self.thread_manager); - tokio::spawn(async move { - for pending_plugin_import in pending_plugin_imports { - match external_agent_config_api - .complete_pending_plugin_import(pending_plugin_import) - .await - { - Ok(()) => {} - Err(error) => { - tracing::warn!( - error = %error.message, - "external agent config plugin import failed" - ); - } - } + if pending_plugin_imports.is_empty() { + self.outgoing + .send_server_notification(ServerNotification::ExternalAgentConfigImportCompleted( + ExternalAgentConfigImportCompletedNotification {}, + )) + .await; + return Ok(()); + } + + let external_agent_config_api = self.external_agent_config_api.clone(); + let outgoing = Arc::clone(&self.outgoing); + let thread_manager = Arc::clone(&self.thread_manager); + tokio::spawn(async move { + for pending_plugin_import in pending_plugin_imports { + match external_agent_config_api + .complete_pending_plugin_import(pending_plugin_import) + .await + { + Ok(()) => {} + Err(error) => { + tracing::warn!( + error = %error.message, + "external agent config plugin import failed" + ); } - thread_manager.plugins_manager().clear_cache(); - thread_manager.skills_manager().clear_cache(); - outgoing - .send_server_notification( - ServerNotification::ExternalAgentConfigImportCompleted( - ExternalAgentConfigImportCompletedNotification {}, - ), - ) - .await; - }); + } } - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } + thread_manager.plugins_manager().clear_cache(); + thread_manager.skills_manager().clear_cache(); + outgoing + .send_server_notification(ServerNotification::ExternalAgentConfigImportCompleted( + ExternalAgentConfigImportCompletedNotification {}, + )) + .await; + }); - async fn handle_fs_read_file(&self, request_id: ConnectionRequestId, params: FsReadFileParams) { - match self.fs_api.read_file(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_write_file( - &self, - request_id: ConnectionRequestId, - params: FsWriteFileParams, - ) { - match self.fs_api.write_file(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_create_directory( - &self, - request_id: ConnectionRequestId, - params: FsCreateDirectoryParams, - ) { - match self.fs_api.create_directory(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_get_metadata( - &self, - request_id: ConnectionRequestId, - params: FsGetMetadataParams, - ) { - match self.fs_api.get_metadata(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_read_directory( - &self, - request_id: ConnectionRequestId, - params: FsReadDirectoryParams, - ) { - match self.fs_api.read_directory(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_remove(&self, request_id: ConnectionRequestId, params: FsRemoveParams) { - match self.fs_api.remove(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_copy(&self, request_id: ConnectionRequestId, params: FsCopyParams) { - match self.fs_api.copy(params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_watch( - &self, - request_id: ConnectionRequestId, - connection_id: ConnectionId, - params: FsWatchParams, - ) { - match self.fs_watch_manager.watch(connection_id, params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } - } - - async fn handle_fs_unwatch( - &self, - request_id: ConnectionRequestId, - connection_id: ConnectionId, - params: FsUnwatchParams, - ) { - match self.fs_watch_manager.unwatch(connection_id, params).await { - Ok(response) => self.outgoing.send_response(request_id, response).await, - Err(error) => self.outgoing.send_error(request_id, error).await, - } + Ok(()) } } diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 4d073fc5a..6ca1fcaba 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -22,6 +22,7 @@ use tracing::Span; use tracing::warn; use crate::error_code::INTERNAL_ERROR_CODE; +use crate::error_code::internal_error; use crate::server_request_error::TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON; #[cfg(test)] @@ -196,7 +197,7 @@ impl ThreadScopedOutgoingMessageSender { pub(crate) async fn send_error( &self, request_id: ConnectionRequestId, - error: JSONRPCErrorError, + error: impl Into, ) { self.outgoing.send_error(request_id, error).await; } @@ -493,11 +494,7 @@ impl OutgoingMessageSender { self.send_error_inner( request_context, request_id, - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to serialize response: {err}"), - data: None, - }, + internal_error(format!("failed to serialize response: {err}")), ) .await; } @@ -571,13 +568,27 @@ impl OutgoingMessageSender { pub(crate) async fn send_error( &self, request_id: ConnectionRequestId, - error: JSONRPCErrorError, + error: impl Into, ) { let request_context = self.take_request_context(&request_id).await; - self.send_error_inner(request_context, request_id, error) + self.send_error_inner(request_context, request_id, error.into()) .await; } + pub(crate) async fn send_result( + &self, + request_id: ConnectionRequestId, + result: std::result::Result, + ) where + T: Serialize, + E: Into, + { + match result { + Ok(response) => self.send_response(request_id, response).await, + Err(error) => self.send_error(request_id, error).await, + } + } + async fn send_error_inner( &self, request_context: Option,