agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Lift app-server JSON-RPC error handling to request boundary (#19484)

Browse Source 
## 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.
This commit is contained in:
pakrym-oai
2026-04-26 15:10:35 -07:00
committed by GitHub
Unverified
parent deaa307fb2
commit 2a020f1a0a
9 changed files with 270 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/command_exec.rs
+13 -36
View File
@@ -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<CommandExecWriteResponse, JSONRPCErrorError> {
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<TerminalSize, JSONRPCErrorError> {
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;
codex-rs/app-server/src/config_api.rs
+13 -28
View File
@@ -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<String>) -> JSONRPCErrorError {
codex-rs/app-server/src/device_key_api.rs
+7 -10
View File
@@ -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())
}
}
}
codex-rs/app-server/src/error_code.rs
+22
View File
@@ -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<String>) -> JSONRPCErrorError {
error(INVALID_REQUEST_ERROR_CODE, message)
}
pub(crate) fn invalid_params(message: impl Into<String>) -> JSONRPCErrorError {
error(INVALID_PARAMS_ERROR_CODE, message)
}
pub(crate) fn internal_error(message: impl Into<String>) -> JSONRPCErrorError {
error(INTERNAL_ERROR_CODE, message)
}
fn error(code: i64, message: impl Into<String>) -> JSONRPCErrorError {
JSONRPCErrorError {
code,
message: message.into(),
data: None,
}
}
codex-rs/app-server/src/external_agent_config_api.rs
+4 -13
View File
@@ -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()))
}
}
codex-rs/app-server/src/fs_api.rs
+3 -15
View File
@@ -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<String>) -> 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())
}
}
codex-rs/app-server/src/fs_watch.rs
+1 -1
View File
@@ -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;
codex-rs/app-server/src/message_processor.rs
+188 -414
View File
@@ -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::<ClientRequest>(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::<ClientRequest>(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<ConnectionSessionState>,
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<String>,
client_version: Option<String>,
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(())
}
}
codex-rs/app-server/src/outgoing_message.rs
+19 -8
View File
@@ -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<JSONRPCErrorError>,
) {
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<JSONRPCErrorError>,
) {
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<T, E>(
&self,
request_id: ConnectionRequestId,
result: std::result::Result<T, E>,
) where
T: Serialize,
E: Into<JSONRPCErrorError>,
{
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<RequestContext>,