merge

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

@@ -0,0 +1,70 @@
+use crate::error_code::INTERNAL_ERROR_CODE;
+use crate::error_code::INVALID_REQUEST_ERROR_CODE;
+use codex_app_server_protocol::ConfigBatchWriteParams;
+use codex_app_server_protocol::ConfigReadParams;
+use codex_app_server_protocol::ConfigReadResponse;
+use codex_app_server_protocol::ConfigValueWriteParams;
+use codex_app_server_protocol::ConfigWriteErrorCode;
+use codex_app_server_protocol::ConfigWriteResponse;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_core::config::ConfigService;
+use codex_core::config::ConfigServiceError;
+use serde_json::json;
+use std::path::PathBuf;
+use toml::Value as TomlValue;
+
+#[derive(Clone)]
+pub(crate) struct ConfigApi {
+    service: ConfigService,
+}
+
+impl ConfigApi {
+    pub(crate) fn new(codex_home: PathBuf, cli_overrides: Vec<(String, TomlValue)>) -> Self {
+        Self {
+            service: ConfigService::new(codex_home, cli_overrides),
+        }
+    }
+
+    pub(crate) async fn read(
+        &self,
+        params: ConfigReadParams,
+    ) -> Result<ConfigReadResponse, JSONRPCErrorError> {
+        self.service.read(params).await.map_err(map_error)
+    }
+
+    pub(crate) async fn write_value(
+        &self,
+        params: ConfigValueWriteParams,
+    ) -> Result<ConfigWriteResponse, JSONRPCErrorError> {
+        self.service.write_value(params).await.map_err(map_error)
+    }
+
+    pub(crate) async fn batch_write(
+        &self,
+        params: ConfigBatchWriteParams,
+    ) -> Result<ConfigWriteResponse, JSONRPCErrorError> {
+        self.service.batch_write(params).await.map_err(map_error)
+    }
+}
+
+fn map_error(err: ConfigServiceError) -> JSONRPCErrorError {
+    if let Some(code) = err.write_error_code() {
+        return config_write_error(code, err.to_string());
+    }
+
+    JSONRPCErrorError {
+        code: INTERNAL_ERROR_CODE,
+        message: err.to_string(),
+        data: None,
+    }
+}
+
+fn config_write_error(code: ConfigWriteErrorCode, message: impl Into<String>) -> JSONRPCErrorError {
+    JSONRPCErrorError {
+        code: INVALID_REQUEST_ERROR_CODE,
+        message: message.into(),
+        data: Some(json!({
+            "config_write_error_code": code,
+        })),
+    }
+}

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

@@ -19,6 +19,10 @@ pub(crate) async fn run_fuzzy_file_search(
     roots: Vec<String>,
     cancellation_flag: Arc<AtomicBool>,
 ) -> Vec<FuzzyFileSearchResult> {
+    if roots.is_empty() {
+        return Vec::new();
+    }
+
     #[expect(clippy::expect_used)]
     let limit_per_root =
         NonZero::new(LIMIT_PER_ROOT).expect("LIMIT_PER_ROOT should be a valid non-zero usize");

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

@@ -3,7 +3,6 @@
 use codex_common::CliConfigOverrides;
 use codex_core::config::Config;
 use codex_core::config::ConfigOverrides;
-use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use std::io::ErrorKind;
 use std::io::Result as IoResult;
 use std::path::PathBuf;
@@ -18,6 +17,7 @@ use tokio::io::AsyncWriteExt;
 use tokio::io::BufReader;
 use tokio::io::{self};
 use tokio::sync::mpsc;
+use toml::Value as TomlValue;
 use tracing::Level;
 use tracing::debug;
 use tracing::error;
@@ -28,7 +28,9 @@ use tracing_subscriber::filter::Targets;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
 
+mod bespoke_event_handling;
 mod codex_message_processor;
+mod config_api;
 mod error_code;
 mod fuzzy_file_search;
 mod message_processor;
@@ -46,7 +48,7 @@ pub async fn run_main(
 ) -> IoResult<()> {
     // Set up channels.
     let (incoming_tx, mut incoming_rx) = mpsc::channel::<JSONRPCMessage>(CHANNEL_CAPACITY);
-    let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
+    let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingMessage>(CHANNEL_CAPACITY);
 
     // Task: read from stdin, push to `incoming_tx`.
     let stdin_reader_handle = tokio::spawn({
@@ -79,11 +81,12 @@ pub async fn run_main(
             format!("error parsing -c overrides: {e}"),
         )
     })?;
-    let config = Config::load_with_cli_overrides(cli_kv_overrides, ConfigOverrides::default())
-        .await
-        .map_err(|e| {
-            std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
-        })?;
+    let config =
+        Config::load_with_cli_overrides(cli_kv_overrides.clone(), ConfigOverrides::default())
+            .await
+            .map_err(|e| {
+                std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
+            })?;
 
     let feedback = CodexFeedback::new();
 
@@ -99,6 +102,7 @@ pub async fn run_main(
     // control the log level with `RUST_LOG`.
     let stderr_fmt = tracing_subscriber::fmt::layer()
         .with_writer(std::io::stderr)
+        .with_span_events(tracing_subscriber::fmt::format::FmtSpan::FULL)
         .with_filter(EnvFilter::from_default_env());
 
     let feedback_layer = tracing_subscriber::fmt::layer()
@@ -107,23 +111,26 @@ pub async fn run_main(
         .with_target(false)
         .with_filter(Targets::new().with_default(Level::TRACE));
 
+    let otel_logger_layer = otel.as_ref().and_then(|o| o.logger_layer());
+
+    let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer());
+
     let _ = tracing_subscriber::registry()
         .with(stderr_fmt)
         .with(feedback_layer)
-        .with(otel.as_ref().map(|provider| {
-            OpenTelemetryTracingBridge::new(&provider.logger).with_filter(
-                tracing_subscriber::filter::filter_fn(codex_core::otel_init::codex_export_filter),
-            )
-        }))
+        .with(otel_logger_layer)
+        .with(otel_tracing_layer)
         .try_init();
 
     // Task: process incoming messages.
     let processor_handle = tokio::spawn({
         let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
+        let cli_overrides: Vec<(String, TomlValue)> = cli_kv_overrides.clone();
         let mut processor = MessageProcessor::new(
             outgoing_message_sender,
             codex_linux_sandbox_exe,
             std::sync::Arc::new(config),
+            cli_overrides,
             feedback.clone(),
         );
         async move {

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

@@ -1,17 +1,22 @@
 use std::path::PathBuf;
+use std::sync::Arc;
 
 use crate::codex_message_processor::CodexMessageProcessor;
+use crate::config_api::ConfigApi;
 use crate::error_code::INVALID_REQUEST_ERROR_CODE;
 use crate::outgoing_message::OutgoingMessageSender;
 use codex_app_server_protocol::ClientInfo;
 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::InitializeResponse;
-
 use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCErrorError;
 use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCRequest;
 use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
 use codex_core::AuthManager;
 use codex_core::ConversationManager;
 use codex_core::config::Config;
@@ -19,11 +24,12 @@ use codex_core::default_client::USER_AGENT_SUFFIX;
 use codex_core::default_client::get_codex_user_agent;
 use codex_feedback::CodexFeedback;
 use codex_protocol::protocol::SessionSource;
-use std::sync::Arc;
+use toml::Value as TomlValue;
 
 pub(crate) struct MessageProcessor {
     outgoing: Arc<OutgoingMessageSender>,
     codex_message_processor: CodexMessageProcessor,
+    config_api: ConfigApi,
     initialized: bool,
 }
 
@@ -34,6 +40,7 @@ impl MessageProcessor {
         outgoing: OutgoingMessageSender,
         codex_linux_sandbox_exe: Option<PathBuf>,
         config: Arc<Config>,
+        cli_overrides: Vec<(String, TomlValue)>,
         feedback: CodexFeedback,
     ) -> Self {
         let outgoing = Arc::new(outgoing);
@@ -51,13 +58,16 @@ impl MessageProcessor {
             conversation_manager,
             outgoing.clone(),
             codex_linux_sandbox_exe,
-            config,
+            Arc::clone(&config),
+            cli_overrides.clone(),
             feedback,
         );
+        let config_api = ConfigApi::new(config.codex_home.clone(), cli_overrides);
 
         Self {
             outgoing,
             codex_message_processor,
+            config_api,
             initialized: false,
         }
     }
@@ -118,6 +128,7 @@ impl MessageProcessor {
                     self.outgoing.send_response(request_id, response).await;
 
                     self.initialized = true;
+
                     return;
                 }
             }
@@ -134,9 +145,20 @@ impl MessageProcessor {
             }
         }
 
-        self.codex_message_processor
-            .process_request(codex_request)
-            .await;
+        match codex_request {
+            ClientRequest::ConfigRead { request_id, params } => {
+                self.handle_config_read(request_id, params).await;
+            }
+            ClientRequest::ConfigValueWrite { request_id, params } => {
+                self.handle_config_value_write(request_id, params).await;
+            }
+            ClientRequest::ConfigBatchWrite { request_id, params } => {
+                self.handle_config_batch_write(request_id, params).await;
+            }
+            other => {
+                self.codex_message_processor.process_request(other).await;
+            }
+        }
     }
 
     pub(crate) async fn process_notification(&self, notification: JSONRPCNotification) {
@@ -156,4 +178,33 @@ impl MessageProcessor {
     pub(crate) fn process_error(&mut self, err: JSONRPCError) {
         tracing::error!("<- error: {:?}", err);
     }
+
+    async fn handle_config_read(&self, request_id: RequestId, 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,
+        }
+    }
+
+    async fn handle_config_value_write(
+        &self,
+        request_id: RequestId,
+        params: ConfigValueWriteParams,
+    ) {
+        match self.config_api.write_value(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_config_batch_write(
+        &self,
+        request_id: RequestId,
+        params: ConfigBatchWriteParams,
+    ) {
+        match self.config_api.batch_write(params).await {
+            Ok(response) => self.outgoing.send_response(request_id, response).await,
+            Err(error) => self.outgoing.send_error(request_id, error).await,
+        }
+    }
 }

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

@@ -1,12 +1,19 @@
-use codex_app_server_protocol::AuthMode;
+use std::sync::Arc;
+
 use codex_app_server_protocol::Model;
 use codex_app_server_protocol::ReasoningEffortOption;
-use codex_common::model_presets::ModelPreset;
-use codex_common::model_presets::ReasoningEffortPreset;
-use codex_common::model_presets::builtin_model_presets;
+use codex_core::ConversationManager;
+use codex_core::config::Config;
+use codex_protocol::openai_models::ModelPreset;
+use codex_protocol::openai_models::ReasoningEffortPreset;
 
-pub fn supported_models(auth_mode: Option<AuthMode>) -> Vec<Model> {
-    builtin_model_presets(auth_mode)
+pub async fn supported_models(
+    conversation_manager: Arc<ConversationManager>,
+    config: &Config,
+) -> Vec<Model> {
+    conversation_manager
+        .list_models(config)
+        .await
         .into_iter()
         .map(model_from_preset)
         .collect()
@@ -27,7 +34,7 @@ fn model_from_preset(preset: ModelPreset) -> Model {
 }
 
 fn reasoning_efforts_from_preset(
-    efforts: &'static [ReasoningEffortPreset],
+    efforts: Vec<ReasoningEffortPreset>,
 ) -> Vec<ReasoningEffortOption> {
     efforts
         .iter()

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

@@ -16,15 +16,18 @@ use tracing::warn;
 
 use crate::error_code::INTERNAL_ERROR_CODE;
 
+#[cfg(test)]
+use codex_protocol::account::PlanType;
+
 /// Sends messages to the client and manages request callbacks.
 pub(crate) struct OutgoingMessageSender {
     next_request_id: AtomicI64,
-    sender: mpsc::UnboundedSender<OutgoingMessage>,
+    sender: mpsc::Sender<OutgoingMessage>,
     request_id_to_callback: Mutex<HashMap<RequestId, oneshot::Sender<Result>>>,
 }
 
 impl OutgoingMessageSender {
-    pub(crate) fn new(sender: mpsc::UnboundedSender<OutgoingMessage>) -> Self {
+    pub(crate) fn new(sender: mpsc::Sender<OutgoingMessage>) -> Self {
         Self {
             next_request_id: AtomicI64::new(0),
             sender,
@@ -45,8 +48,12 @@ impl OutgoingMessageSender {
         }
 
         let outgoing_message =
-            OutgoingMessage::Request(request.request_with_id(outgoing_message_id));
-        let _ = self.sender.send(outgoing_message);
+            OutgoingMessage::Request(request.request_with_id(outgoing_message_id.clone()));
+        if let Err(err) = self.sender.send(outgoing_message).await {
+            warn!("failed to send request {outgoing_message_id:?} to client: {err:?}");
+            let mut request_id_to_callback = self.request_id_to_callback.lock().await;
+            request_id_to_callback.remove(&outgoing_message_id);
+        }
         rx_approve
     }
 
@@ -72,7 +79,9 @@ impl OutgoingMessageSender {
         match serde_json::to_value(response) {
             Ok(result) => {
                 let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result });
-                let _ = self.sender.send(outgoing_message);
+                if let Err(err) = self.sender.send(outgoing_message).await {
+                    warn!("failed to send response to client: {err:?}");
+                }
             }
             Err(err) => {
                 self.send_error(
@@ -89,21 +98,29 @@ impl OutgoingMessageSender {
     }
 
     pub(crate) async fn send_server_notification(&self, notification: ServerNotification) {
-        let _ = self
+        if let Err(err) = self
             .sender
-            .send(OutgoingMessage::AppServerNotification(notification));
+            .send(OutgoingMessage::AppServerNotification(notification))
+            .await
+        {
+            warn!("failed to send server notification to client: {err:?}");
+        }
     }
 
     /// All notifications should be migrated to [`ServerNotification`] and
     /// [`OutgoingMessage::Notification`] should be removed.
     pub(crate) async fn send_notification(&self, notification: OutgoingNotification) {
         let outgoing_message = OutgoingMessage::Notification(notification);
-        let _ = self.sender.send(outgoing_message);
+        if let Err(err) = self.sender.send(outgoing_message).await {
+            warn!("failed to send notification to client: {err:?}");
+        }
     }
 
     pub(crate) async fn send_error(&self, id: RequestId, error: JSONRPCErrorError) {
         let outgoing_message = OutgoingMessage::Error(OutgoingError { id, error });
-        let _ = self.sender.send(outgoing_message);
+        if let Err(err) = self.sender.send(outgoing_message).await {
+            warn!("failed to send error to client: {err:?}");
+        }
     }
 }
 
@@ -215,6 +232,8 @@ mod tests {
                         resets_at: Some(123),
                     }),
                     secondary: None,
+                    credits: None,
+                    plan_type: Some(PlanType::Plus),
                 },
             });
 
@@ -229,7 +248,9 @@ mod tests {
                             "windowDurationMins": 15,
                             "resetsAt": 123
                         },
-                        "secondary": null
+                        "secondary": null,
+                        "credits": null,
+                        "planType": "plus"
                     }
                 },
             }),