From baaea5229eb7b4b7f845d328b80c0db394aee982 Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Tue, 5 May 2026 10:39:35 -0700 Subject: [PATCH] Use shared app-server JSON-RPC error helpers --- codex-rs/app-server/src/in_process.rs | 63 +++---- codex-rs/app-server/src/outgoing_message.rs | 32 ++-- codex-rs/app-server/src/request_processors.rs | 3 - .../request_processors/account_processor.rs | 120 ++++--------- .../src/request_processors/apps_processor.rs | 27 +-- .../request_processors/catalog_processor.rs | 6 +- .../src/request_processors/config_errors.rs | 8 +- .../request_processors/config_processor.rs | 13 +- .../marketplace_processor.rs | 6 +- .../src/request_processors/mcp_processor.rs | 37 ++--- .../src/request_processors/plugins.rs | 22 +-- .../src/request_processors/search.rs | 20 +-- .../request_processors/thread_lifecycle.rs | 28 +--- .../request_processors/thread_processor.rs | 157 +++++------------- .../src/request_processors/turn_processor.rs | 60 ++----- 15 files changed, 174 insertions(+), 428 deletions(-) diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 09d57a60a338..e3eeb90349e0 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -52,9 +52,9 @@ use std::time::Duration; use crate::analytics_utils::analytics_events_client_from_config; use crate::config_manager::ConfigManager; -use crate::error_code::INTERNAL_ERROR_CODE; -use crate::error_code::INVALID_REQUEST_ERROR_CODE; use crate::error_code::OVERLOADED_ERROR_CODE; +use crate::error_code::internal_error; +use crate::error_code::invalid_request; use crate::message_processor::ConnectionSessionState; use crate::message_processor::MessageProcessor; use crate::message_processor::MessageProcessorArgs; @@ -526,11 +526,9 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { entry.insert(response_tx); } Entry::Occupied(_) => { - let _ = response_tx.send(Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("duplicate request id: {request_id:?}"), - data: None, - })); + let _ = response_tx.send(Err(invalid_request(format!( + "duplicate request id: {request_id:?}" + )))); continue; } } @@ -553,13 +551,9 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { if let Some(response_tx) = pending_request_responses.remove(&request_id) { - let _ = response_tx.send(Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: - "in-process app-server request processor is closed" - .to_string(), - data: None, - })); + let _ = response_tx.send(Err(internal_error( + "in-process app-server request processor is closed", + ))); } break; } @@ -627,15 +621,20 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { if let Err(send_error) = event_tx .try_send(InProcessServerEvent::ServerRequest(request)) { - let (code, message, inner) = match send_error { + let (error, inner) = match send_error { mpsc::error::TrySendError::Full(inner) => ( - OVERLOADED_ERROR_CODE, - "in-process server request queue is full", + JSONRPCErrorError { + code: OVERLOADED_ERROR_CODE, + message: + "in-process server request queue is full".to_string(), + data: None, + }, inner, ), mpsc::error::TrySendError::Closed(inner) => ( - INTERNAL_ERROR_CODE, - "in-process server request consumer is closed", + internal_error( + "in-process server request consumer is closed", + ), inner, ), }; @@ -644,14 +643,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { _ => unreachable!("we just sent a ServerRequest variant"), }; outgoing_message_sender - .notify_client_error( - request_id, - JSONRPCErrorError { - code, - message: message.to_string(), - data: None, - }, - ) + .notify_client_error(request_id, error) .await; } } @@ -688,21 +680,17 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { drop(writer_rx); drop(processor_tx); outgoing_message_sender - .cancel_all_requests(Some(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "in-process app-server runtime is shutting down".to_string(), - data: None, - })) + .cancel_all_requests(Some(internal_error( + "in-process app-server runtime is shutting down", + ))) .await; // Drop the runtime's last sender before awaiting the router task so // `outgoing_rx.recv()` can observe channel closure and exit cleanly. drop(outgoing_message_sender); for (_, response_tx) in pending_request_responses { - let _ = response_tx.send(Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "in-process app-server runtime is shutting down".to_string(), - data: None, - })); + let _ = response_tx.send(Err(internal_error( + "in-process app-server runtime is shutting down", + ))); } if let Err(_elapsed) = timeout(SHUTDOWN_TIMEOUT, &mut processor_handle).await { @@ -731,6 +719,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { #[cfg(test)] mod tests { use super::*; + use crate::error_code::INVALID_REQUEST_ERROR_CODE; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ConfigRequirementsReadResponse; use codex_app_server_protocol::DeviceKeyPublicParams; diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index f7a90538c2e8..2807d2532e3a 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -21,7 +21,6 @@ use tracing::Instrument; 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; pub(crate) use codex_app_server_transport::ConnectionId; @@ -157,11 +156,14 @@ impl ThreadScopedOutgoingMessageSender { self.outgoing .cancel_requests_for_thread( self.thread_id, - Some(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "client request resolved because the turn state was changed" - .to_string(), - data: Some(serde_json::json!({ "reason": TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON })), + Some({ + let mut error = internal_error( + "client request resolved because the turn state was changed", + ); + error.data = Some(serde_json::json!({ + "reason": TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON, + })); + error }), ) .await @@ -1011,11 +1013,7 @@ mod tests { connection_id: ConnectionId(9), request_id: RequestId::Integer(3), }; - let error = JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "boom".to_string(), - data: None, - }; + let error = internal_error("boom"); outgoing.send_error(request_id.clone(), error.clone()).await; @@ -1139,11 +1137,7 @@ mod tests { )) .await; - let error = JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "refresh failed".to_string(), - data: None, - }; + let error = internal_error("refresh failed"); outgoing .notify_client_error(request_id, error.clone()) @@ -1253,11 +1247,7 @@ mod tests { }, )) .await; - let error = JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "tracked request cancelled".to_string(), - data: None, - }; + let error = internal_error("tracked request cancelled"); outgoing .cancel_requests_for_thread(thread_id, Some(error.clone())) diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 0154b61734cc..faf9fc3f68f3 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -4,9 +4,6 @@ use crate::command_exec::CommandExecManager; use crate::command_exec::StartCommandExecParams; use crate::config_manager::ConfigManager; use crate::error_code::INPUT_TOO_LARGE_ERROR_CODE; -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::invalid_params; use crate::models::supported_models; use crate::outgoing_message::ConnectionId; diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index b26d1e3e1f66..2609e25ac4de 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -243,12 +243,9 @@ impl AccountRequestProcessor { } fn external_auth_active_error(&self) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "External auth is active. Use account/login/start (chatgptAuthTokens) to update it or account/logout to clear it." - .to_string(), - data: None, - } + invalid_request( + "External auth is active. Use account/login/start (chatgptAuthTokens) to update it or account/logout to clear it.", + ) } async fn login_api_key_common( @@ -263,11 +260,9 @@ impl AccountRequestProcessor { self.config.forced_login_method, Some(ForcedLoginMethod::Chatgpt) ) { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "API key login is disabled. Use ChatGPT login instead.".to_string(), - data: None, - }); + return Err(invalid_request( + "API key login is disabled. Use ChatGPT login instead.", + )); } // Cancel any active login attempt. @@ -287,11 +282,7 @@ impl AccountRequestProcessor { self.auth_manager.reload().await; Ok(()) } - Err(err) => Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to save api key: {err}"), - data: None, - }), + Err(err) => Err(internal_error(format!("failed to save api key: {err}"))), } } @@ -321,11 +312,9 @@ impl AccountRequestProcessor { } if matches!(config.forced_login_method, Some(ForcedLoginMethod::Api)) { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "ChatGPT login is disabled. Use API key login instead.".to_string(), - data: None, - }); + return Err(invalid_request( + "ChatGPT login is disabled. Use API key login instead.", + )); } let opts = LoginServerOptions { @@ -354,18 +343,10 @@ impl AccountRequestProcessor { fn login_chatgpt_device_code_start_error(err: IoError) -> JSONRPCErrorError { let is_not_found = err.kind() == std::io::ErrorKind::NotFound; - JSONRPCErrorError { - code: if is_not_found { - INVALID_REQUEST_ERROR_CODE - } else { - INTERNAL_ERROR_CODE - }, - message: if is_not_found { - err.to_string() - } else { - format!("failed to request device code: {err}") - }, - data: None, + if is_not_found { + invalid_request(err.to_string()) + } else { + internal_error(format!("failed to request device code: {err}")) } } @@ -698,11 +679,7 @@ impl AccountRequestProcessor { match self.auth_manager.logout_with_revoke().await { Ok(_) => {} Err(err) => { - return Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("logout failed: {err}"), - data: None, - }); + return Err(internal_error(format!("logout failed: {err}"))); } } @@ -885,28 +862,19 @@ impl AccountRequestProcessor { params: SendAddCreditsNudgeEmailParams, ) -> Result { let Some(auth) = self.auth_manager.auth().await else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "codex account authentication required to notify workspace owner" - .to_string(), - data: None, - }); + return Err(invalid_request( + "codex account authentication required to notify workspace owner", + )); }; if !auth.uses_codex_backend() { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "chatgpt authentication required to notify workspace owner".to_string(), - data: None, - }); + return Err(invalid_request( + "chatgpt authentication required to notify workspace owner", + )); } let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to construct backend client: {err}"), - data: None, - })?; + .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; match client .send_add_credits_nudge_email(Self::backend_credit_type(params.credit_type)) @@ -916,11 +884,9 @@ impl AccountRequestProcessor { Err(err) if err.status().is_some_and(|status| status.as_u16() == 429) => { Ok(AddCreditsNudgeEmailStatus::CooldownActive) } - Err(err) => Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to notify workspace owner: {err}"), - data: None, - }), + Err(err) => Err(internal_error(format!( + "failed to notify workspace owner: {err}" + ))), } } @@ -941,42 +907,28 @@ impl AccountRequestProcessor { JSONRPCErrorError, > { let Some(auth) = self.auth_manager.auth().await else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "codex account authentication required to read rate limits".to_string(), - data: None, - }); + return Err(invalid_request( + "codex account authentication required to read rate limits", + )); }; if !auth.uses_codex_backend() { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "chatgpt authentication required to read rate limits".to_string(), - data: None, - }); + return Err(invalid_request( + "chatgpt authentication required to read rate limits", + )); } let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to construct backend client: {err}"), - data: None, - })?; + .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; let snapshots = client .get_rate_limits_many() .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to fetch codex rate limits: {err}"), - data: None, - })?; + .map_err(|err| internal_error(format!("failed to fetch codex rate limits: {err}")))?; if snapshots.is_empty() { - return Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: "failed to fetch codex rate limits: no snapshots returned".to_string(), - data: None, - }); + return Err(internal_error( + "failed to fetch codex rate limits: no snapshots returned", + )); } let rate_limits_by_limit_id: HashMap = snapshots diff --git a/codex-rs/app-server/src/request_processors/apps_processor.rs b/codex-rs/app-server/src/request_processors/apps_processor.rs index 917641dd3ede..da2956dbab69 100644 --- a/codex-rs/app-server/src/request_processors/apps_processor.rs +++ b/codex-rs/app-server/src/request_processors/apps_processor.rs @@ -231,21 +231,14 @@ impl AppsRequestProcessor { &self, thread_id: &str, ) -> Result<(ThreadId, Arc), JSONRPCErrorError> { - let thread_id = ThreadId::from_string(thread_id).map_err(|err| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid thread id: {err}"), - data: None, - })?; + let thread_id = ThreadId::from_string(thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; let thread = self .thread_manager .get_thread(thread_id) .await - .map_err(|_| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("thread not found: {thread_id}"), - data: None, - })?; + .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?; Ok((thread_id, thread)) } @@ -257,11 +250,7 @@ impl AppsRequestProcessor { self.config_manager .load_latest_config(fallback_cwd) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to reload config: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to reload config: {err}"))) } async fn workspace_codex_plugins_enabled( @@ -319,11 +308,9 @@ fn paginate_apps( ) -> Result { let total = connectors.len(); if start > total { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("cursor {start} exceeds total apps {total}"), - data: None, - }); + return Err(invalid_request(format!( + "cursor {start} exceeds total apps {total}" + ))); } let effective_limit = limit.unwrap_or(total as u32).max(1) as usize; diff --git a/codex-rs/app-server/src/request_processors/catalog_processor.rs b/codex-rs/app-server/src/request_processors/catalog_processor.rs index c2876514a55e..fac819c4c307 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -190,11 +190,7 @@ impl CatalogRequestProcessor { self.config_manager .load_latest_config(fallback_cwd) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to reload config: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to reload config: {err}"))) } async fn workspace_codex_plugins_enabled( diff --git a/codex-rs/app-server/src/request_processors/config_errors.rs b/codex-rs/app-server/src/request_processors/config_errors.rs index ea1664031b39..63fe2b3d2cfc 100644 --- a/codex-rs/app-server/src/request_processors/config_errors.rs +++ b/codex-rs/app-server/src/request_processors/config_errors.rs @@ -29,9 +29,7 @@ pub(super) fn config_load_error(err: &std::io::Error) -> JSONRPCErrorError { data }); - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("failed to load configuration: {err}"), - data, - } + let mut error = invalid_request(format!("failed to load configuration: {err}")); + error.data = data; + error } diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index 8043675e0d2d..f08486771938 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use crate::config_manager::ConfigManager; use crate::config_manager_service::ConfigManagerError; -use crate::error_code::INVALID_REQUEST_ERROR_CODE; use crate::error_code::internal_error; use crate::error_code::invalid_request; use crate::outgoing_message::ConnectionRequestId; @@ -612,11 +611,9 @@ fn map_error(err: ConfigManagerError) -> JSONRPCErrorError { } fn config_write_error(code: ConfigWriteErrorCode, message: impl Into) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: message.into(), - data: Some(json!({ - "config_write_error_code": code, - })), - } + let mut error = invalid_request(message); + error.data = Some(json!({ + "config_write_error_code": code, + })); + error } diff --git a/codex-rs/app-server/src/request_processors/marketplace_processor.rs b/codex-rs/app-server/src/request_processors/marketplace_processor.rs index fa5bc3d82753..1a095074180b 100644 --- a/codex-rs/app-server/src/request_processors/marketplace_processor.rs +++ b/codex-rs/app-server/src/request_processors/marketplace_processor.rs @@ -132,10 +132,6 @@ impl MarketplaceRequestProcessor { self.config_manager .load_latest_config(fallback_cwd) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to reload config: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to reload config: {err}"))) } } diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index e161fc4fa4dc..22d43d87a2a9 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -89,32 +89,21 @@ impl McpRequestProcessor { self.config_manager .load_latest_config(fallback_cwd) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to reload config: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to reload config: {err}"))) } async fn load_thread( &self, thread_id: &str, ) -> Result<(ThreadId, Arc), JSONRPCErrorError> { - let thread_id = ThreadId::from_string(thread_id).map_err(|err| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid thread id: {err}"), - data: None, - })?; + let thread_id = ThreadId::from_string(thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; let thread = self .thread_manager .get_thread(thread_id) .await - .map_err(|_| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("thread not found: {thread_id}"), - data: None, - })?; + .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?; Ok((thread_id, thread)) } @@ -130,11 +119,9 @@ impl McpRequestProcessor { let mcp_servers = match serde_json::to_value(configured_servers) { Ok(value) => value, Err(err) => { - return Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to serialize MCP servers: {err}"), - data: None, - }); + return Err(internal_error(format!( + "failed to serialize MCP servers: {err}" + ))); } }; @@ -142,13 +129,9 @@ impl McpRequestProcessor { match serde_json::to_value(config.mcp_oauth_credentials_store_mode) { Ok(value) => value, Err(err) => { - return Err(JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!( - "failed to serialize MCP OAuth credentials store mode: {err}" - ), - data: None, - }); + return Err(internal_error(format!( + "failed to serialize MCP OAuth credentials store mode: {err}" + ))); } }; diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 068dc04ca995..40f2a412ace4 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -219,11 +219,7 @@ impl PluginRequestProcessor { self.config_manager .load_latest_config(fallback_cwd) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to reload config: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to reload config: {err}"))) } async fn workspace_codex_plugins_enabled( @@ -1307,16 +1303,17 @@ fn remote_plugin_catalog_error_to_jsonrpc( err: RemotePluginCatalogError, context: &str, ) -> JSONRPCErrorError { - let code = match &err { + let message = format!("{context}: {err}"); + match &err { RemotePluginCatalogError::AuthRequired | RemotePluginCatalogError::UnsupportedAuthMode => { - INVALID_REQUEST_ERROR_CODE + invalid_request(message) } RemotePluginCatalogError::UnexpectedStatus { status, .. } if status.as_u16() == 404 => { - INVALID_REQUEST_ERROR_CODE + invalid_request(message) } RemotePluginCatalogError::InvalidPluginPath { .. } | RemotePluginCatalogError::ArchiveTooLarge { .. } - | RemotePluginCatalogError::UnknownMarketplace { .. } => INVALID_REQUEST_ERROR_CODE, + | RemotePluginCatalogError::UnknownMarketplace { .. } => invalid_request(message), RemotePluginCatalogError::AuthToken(_) | RemotePluginCatalogError::Request { .. } | RemotePluginCatalogError::UnexpectedStatus { .. } @@ -1330,12 +1327,7 @@ fn remote_plugin_catalog_error_to_jsonrpc( | RemotePluginCatalogError::ArchiveJoin(_) | RemotePluginCatalogError::MissingUploadEtag | RemotePluginCatalogError::UnexpectedResponse(_) - | RemotePluginCatalogError::CacheRemove(_) => INTERNAL_ERROR_CODE, - }; - JSONRPCErrorError { - code, - message: format!("{context}: {err}"), - data: None, + | RemotePluginCatalogError::CacheRemove(_) => internal_error(message), } } diff --git a/codex-rs/app-server/src/request_processors/search.rs b/codex-rs/app-server/src/request_processors/search.rs index 85f1efe68d6a..d683c6f10a87 100644 --- a/codex-rs/app-server/src/request_processors/search.rs +++ b/codex-rs/app-server/src/request_processors/search.rs @@ -3,8 +3,8 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; -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 crate::fuzzy_file_search::FuzzyFileSearchSession; use crate::fuzzy_file_search::run_fuzzy_file_search; use crate::fuzzy_file_search::start_fuzzy_file_search_session; @@ -132,19 +132,3 @@ impl SearchRequestProcessor { Ok(FuzzyFileSearchSessionStopResponse {}) } } - -fn invalid_request(message: impl Into) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: message.into(), - data: None, - } -} - -fn internal_error(message: impl Into) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: message.into(), - data: None, - } -} diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index fae6fa8555d8..4a677d91ab4f 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -147,23 +147,17 @@ pub(super) async fn ensure_conversation_listener( { Ok(conv) => conv, Err(_) => { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("thread not found: {conversation_id}"), - data: None, - }); + return Err(invalid_request(format!( + "thread not found: {conversation_id}" + ))); } }; let thread_state = { let pending_thread_unloads = listener_task_context.pending_thread_unloads.lock().await; if pending_thread_unloads.contains(&conversation_id) { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "thread {conversation_id} is closing; retry after the thread is closed" - ), - data: None, - }); + return Err(invalid_request(format!( + "thread {conversation_id} is closing; retry after the thread is closed" + ))); } let Some(thread_state) = listener_task_context .thread_state_manager @@ -229,13 +223,9 @@ pub(super) async fn ensure_listener_task_running( ) .await else { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "thread {conversation_id} is closing; retry after the thread is closed" - ), - data: None, - }); + return Err(invalid_request(format!( + "thread {conversation_id} is closing; retry after the thread is closed" + ))); }; let (mut listener_command_rx, listener_generation) = { let mut thread_state = thread_state.lock().await; diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 3d8746092dcb..178d13ce3a74 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -165,10 +165,8 @@ fn normalize_thread_list_cwd_filters( for cwd in cwds { let cwd = AbsolutePathBuf::relative_to_current_dir(cwd.as_str()) .map(AbsolutePathBuf::into_path_buf) - .map_err(|err| JSONRPCErrorError { - code: INVALID_PARAMS_ERROR_CODE, - message: format!("invalid thread/list cwd filter `{cwd}`: {err}"), - data: None, + .map_err(|err| { + invalid_params(format!("invalid thread/list cwd filter `{cwd}`: {err}")) })?; normalized_cwds.push(cwd); } @@ -567,21 +565,14 @@ impl ThreadRequestProcessor { thread_id: &str, ) -> Result<(ThreadId, Arc), JSONRPCErrorError> { // Resolve the core conversation handle from a v2 thread id string. - let thread_id = ThreadId::from_string(thread_id).map_err(|err| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid thread id: {err}"), - data: None, - })?; + let thread_id = ThreadId::from_string(thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; let thread = self .thread_manager .get_thread(thread_id) .await - .map_err(|_| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("thread not found: {thread_id}"), - data: None, - })?; + .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?; Ok((thread_id, thread)) } @@ -604,11 +595,7 @@ impl ThreadRequestProcessor { thread .set_app_server_client_info(app_server_client_name, app_server_client_version) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to set app server client info: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to set app server client info: {err}"))) } async fn finalize_thread_teardown(&self, thread_id: ThreadId) { @@ -1536,10 +1523,8 @@ impl ThreadRequestProcessor { .and_then(|stored_thread| { summary_from_stored_thread(stored_thread, fallback_provider.as_str()) .map(|summary| summary_to_thread(summary, &self.config.cwd)) - .ok_or_else(|| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to read unarchived thread {thread_id}"), - data: None, + .ok_or_else(|| { + internal_error(format!("failed to read unarchived thread {thread_id}")) }) })?; @@ -3239,11 +3224,9 @@ fn paginate_thread_turns( .as_ref() .and_then(|anchor| turns.iter().position(|turn| turn.id == anchor.turn_id)); if anchor.is_some() && anchor_index.is_none() { - return Err(JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: "invalid cursor: anchor turn is no longer present".to_string(), - data: None, - }); + return Err(invalid_request( + "invalid cursor: anchor turn is no longer present", + )); } let mut keyed_turns: Vec<_> = turns.into_iter().enumerate().collect(); @@ -3304,19 +3287,11 @@ fn serialize_thread_turns_cursor( turn_id: turn_id.to_string(), include_anchor, }) - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to serialize cursor: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to serialize cursor: {err}"))) } fn parse_thread_turns_cursor(cursor: &str) -> Result { - serde_json::from_str(cursor).map_err(|_| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid cursor: {cursor}"), - data: None, - }) + serde_json::from_str(cursor).map_err(|_| invalid_request(format!("invalid cursor: {cursor}"))) } fn reconstruct_thread_turns_for_turns_list( @@ -3367,36 +3342,18 @@ fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError { match err { - ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }, - err => JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to list threads: {err}"), - data: None, - }, + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + err => internal_error(format!("failed to list threads: {err}")), } } fn thread_store_resume_read_error(err: ThreadStoreError) -> JSONRPCErrorError { match err { - ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }, - ThreadStoreError::ThreadNotFound { thread_id } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("no rollout found for thread id {thread_id}"), - data: None, - }, - err => JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to read thread: {err}"), - data: None, - }, + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("no rollout found for thread id {thread_id}")) + } + err => internal_error(format!("failed to read thread: {err}")), } } @@ -3459,25 +3416,17 @@ fn conversation_summary_thread_id_read_error( ThreadStoreError::ThreadNotFound { thread_id } if thread_id == conversation_id => { conversation_summary_not_found_error(conversation_id) } - ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }, - err => JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to load conversation summary for {conversation_id}: {err}"), - data: None, - }, + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + err => internal_error(format!( + "failed to load conversation summary for {conversation_id}: {err}" + )), } } fn conversation_summary_not_found_error(conversation_id: ThreadId) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("no rollout found for conversation id {conversation_id}"), - data: None, - } + invalid_request(format!( + "no rollout found for conversation id {conversation_id}" + )) } fn conversation_summary_rollout_path_read_error( @@ -3485,55 +3434,29 @@ fn conversation_summary_rollout_path_read_error( err: ThreadStoreError, ) -> JSONRPCErrorError { match err { - ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }, - err => JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!( - "failed to load conversation summary from {}: {}", - path.display(), - err - ), - data: None, - }, + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + err => internal_error(format!( + "failed to load conversation summary from {}: {}", + path.display(), + err + )), } } fn thread_store_write_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { match err { - ThreadStoreError::ThreadNotFound { thread_id } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("thread not found: {thread_id}"), - data: None, - }, - ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }, - err => JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to {operation}: {err}"), - data: None, - }, + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("thread not found: {thread_id}")) + } + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + err => internal_error(format!("failed to {operation}: {err}")), } } fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { match err { - ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - }, - err => JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to {operation} thread: {err}"), - data: None, - }, + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + err => internal_error(format!("failed to {operation} thread: {err}")), } } diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 1629f25c23fa..033ecf600ddf 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -171,21 +171,14 @@ impl TurnRequestProcessor { thread_id: &str, ) -> Result<(ThreadId, Arc), JSONRPCErrorError> { // Resolve the core conversation handle from a v2 thread id string. - let thread_id = ThreadId::from_string(thread_id).map_err(|err| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid thread id: {err}"), - data: None, - })?; + let thread_id = ThreadId::from_string(thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; let thread = self .thread_manager .get_thread(thread_id) .await - .map_err(|_| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("thread not found: {thread_id}"), - data: None, - })?; + .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?; Ok((thread_id, thread)) } @@ -209,14 +202,6 @@ impl TurnRequestProcessor { fn review_request_from_target( target: ApiReviewTarget, ) -> Result<(ReviewRequest, String), JSONRPCErrorError> { - fn invalid_request(message: String) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message, - data: None, - } - } - let cleaned_target = match target { ApiReviewTarget::UncommittedChanges => ApiReviewTarget::UncommittedChanges, ApiReviewTarget::BaseBranch { branch } => { @@ -305,17 +290,15 @@ impl TurnRequestProcessor { } fn input_too_large_error(actual_chars: usize) -> JSONRPCErrorError { - JSONRPCErrorError { - code: INVALID_PARAMS_ERROR_CODE, - message: format!( - "Input exceeds the maximum length of {MAX_USER_INPUT_TEXT_CHARS} characters." - ), - data: Some(serde_json::json!({ - "input_error_code": INPUT_TOO_LARGE_ERROR_CODE, - "max_chars": MAX_USER_INPUT_TEXT_CHARS, - "actual_chars": actual_chars, - })), - } + let mut error = invalid_params(format!( + "Input exceeds the maximum length of {MAX_USER_INPUT_TEXT_CHARS} characters." + )); + error.data = Some(serde_json::json!({ + "input_error_code": INPUT_TOO_LARGE_ERROR_CODE, + "max_chars": MAX_USER_INPUT_TEXT_CHARS, + "actual_chars": actual_chars, + })); + error } fn validate_v2_input_limit(items: &[V2UserInput]) -> Result<(), JSONRPCErrorError> { @@ -564,11 +547,7 @@ impl TurnRequestProcessor { thread .set_app_server_client_info(app_server_client_name, app_server_client_version) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to set app server client info: {err}"), - data: None, - }) + .map_err(|err| internal_error(format!("failed to set app server client info: {err}"))) } async fn turn_steer_inner( @@ -612,9 +591,8 @@ impl TurnRequestProcessor { ) .await .map_err(|err| { - let (code, message, data, error_type) = match err { + let (message, data, error_type) = match err { SteerInputError::NoActiveTurn(_) => ( - INVALID_REQUEST_ERROR_CODE, "no active turn to steer".to_string(), None, Some(AnalyticsJsonRpcError::TurnSteer( @@ -622,7 +600,6 @@ impl TurnRequestProcessor { )), ), SteerInputError::ExpectedTurnMismatch { expected, actual } => ( - INVALID_REQUEST_ERROR_CODE, format!("expected active turn id `{expected}` but found `{actual}`"), None, Some(AnalyticsJsonRpcError::TurnSteer( @@ -658,24 +635,19 @@ impl TurnRequestProcessor { } }; ( - INVALID_REQUEST_ERROR_CODE, message, data, Some(AnalyticsJsonRpcError::TurnSteer(turn_steer_error)), ) } SteerInputError::EmptyInput => ( - INVALID_REQUEST_ERROR_CODE, "input must not be empty".to_string(), None, Some(AnalyticsJsonRpcError::Input(InputError::Empty)), ), }; - let error = JSONRPCErrorError { - code, - message, - data, - }; + let mut error = invalid_request(message); + error.data = data; self.track_error_response(request_id, &error, error_type); error })?;