From d013576f8bd8f03144881a3756332e3e1079283c Mon Sep 17 00:00:00 2001 From: Rasmus Rygaard Date: Tue, 14 Apr 2026 09:53:17 -0700 Subject: [PATCH 1/3] Redirect debug client output to a file (#17234) In the app-server debug client, allow redirecting output to a file in addition to just stdout. Shell redirecting works OK but is a bit weird with the interactive mode of the debug client since a bunch of newlines get dumped into the shell. With async messages from MCPs starting it's also tricky to actually type in a prompt. --- codex-rs/debug-client/README.md | 7 ++-- codex-rs/debug-client/src/client.rs | 4 +-- codex-rs/debug-client/src/main.rs | 19 +++++++++- codex-rs/debug-client/src/output.rs | 55 ++++++++++++++++++++++++++++- codex-rs/debug-client/src/reader.rs | 4 +-- 5 files changed, 81 insertions(+), 8 deletions(-) diff --git a/codex-rs/debug-client/README.md b/codex-rs/debug-client/README.md index bd310d9056..447ed12360 100644 --- a/codex-rs/debug-client/README.md +++ b/codex-rs/debug-client/README.md @@ -12,7 +12,8 @@ Start the app-server client (it will spawn `codex app-server` itself): ``` cargo run -p codex-debug-client -- \ --codex-bin codex \ - --approval-policy on-request + --approval-policy on-request \ + --output-file /tmp/app-server-server-json.jsonl ``` You can resume a specific thread: @@ -29,6 +30,7 @@ cargo run -p codex-debug-client -- --thread-id thr_123 - `--approval-policy `: `untrusted`, `on-failure` (deprecated), `on-request`, `never`. - `--auto-approve`: auto-approve command/file-change approvals (default: decline). - `--final-only`: only show completed assistant messages and tool items. +- `--output-file `: write raw server JSONL to this file instead of stdout. - `--model `: optional model override for thread start/resume. - `--model-provider `: optional provider override. - `--cwd `: optional working directory override. @@ -46,7 +48,8 @@ Type a line to send it as a new turn. Commands are prefixed with `:`: The prompt shows the active thread id. Client messages (help, errors, approvals) print to stderr; raw server JSON prints to stdout so you can pipe/record it -unless `--final-only` is set. +unless `--final-only` is set. Pass `--output-file ` to record raw server +JSONL to a file instead of stdout. ## Notes diff --git a/codex-rs/debug-client/src/client.rs b/codex-rs/debug-client/src/client.rs index 2ada10e377..8962a1f02f 100644 --- a/codex-rs/debug-client/src/client.rs +++ b/codex-rs/debug-client/src/client.rs @@ -299,8 +299,8 @@ impl AppServerClient { } let line = buffer.trim_end_matches(['\n', '\r']); - if !line.is_empty() && !self.filtered_output { - let _ = output.server_line(line); + if !line.is_empty() { + let _ = output.server_json_line(line, self.filtered_output); } let message = match serde_json::from_str::(line) { diff --git a/codex-rs/debug-client/src/main.rs b/codex-rs/debug-client/src/main.rs index 6350331d6d..a09d9d04f5 100644 --- a/codex-rs/debug-client/src/main.rs +++ b/codex-rs/debug-client/src/main.rs @@ -4,8 +4,10 @@ mod output; mod reader; mod state; +use std::fs::File; use std::io; use std::io::BufRead; +use std::path::PathBuf; use std::sync::mpsc; use anyhow::Context; @@ -50,6 +52,10 @@ struct Cli { #[arg(long, default_value_t = false)] final_only: bool, + /// Write raw server JSONL to this file instead of stdout. + #[arg(long, value_name = "PATH")] + output_file: Option, + /// Optional model override when starting/resuming a thread. #[arg(long)] model: Option, @@ -65,7 +71,18 @@ struct Cli { fn main() -> Result<()> { let cli = Cli::parse(); - let output = Output::new(); + let jsonl_file = cli + .output_file + .as_ref() + .map(File::create) + .transpose() + .with_context(|| { + let Some(path) = cli.output_file.as_ref() else { + return "open output file".to_string(); + }; + format!("open output file {}", path.display()) + })?; + let output = Output::new(jsonl_file); let approval_policy = parse_approval_policy(&cli.approval_policy)?; let mut client = AppServerClient::spawn( diff --git a/codex-rs/debug-client/src/output.rs b/codex-rs/debug-client/src/output.rs index ca3ac9d9cb..f164a3ade8 100644 --- a/codex-rs/debug-client/src/output.rs +++ b/codex-rs/debug-client/src/output.rs @@ -1,4 +1,5 @@ #![allow(clippy::expect_used)] +use std::fs::File; use std::io; use std::io::IsTerminal; use std::io::Write; @@ -24,19 +25,41 @@ pub struct Output { lock: Arc>, prompt: Arc>, color: bool, + jsonl_file: Option>>, } impl Output { - pub fn new() -> Self { + pub fn new(jsonl_file: Option) -> Self { let no_color = std::env::var_os("NO_COLOR").is_some(); let color = !no_color && io::stdout().is_terminal() && io::stderr().is_terminal(); Self { lock: Arc::new(Mutex::new(())), prompt: Arc::new(Mutex::new(PromptState::default())), color, + jsonl_file: jsonl_file.map(|file| Arc::new(Mutex::new(file))), } } + pub fn server_json_line(&self, line: &str, filtered_output: bool) -> io::Result<()> { + let _guard = self.lock.lock().expect("output lock poisoned"); + + if let Some(file) = self.jsonl_file.as_ref() { + let mut file = file.lock().expect("jsonl file lock poisoned"); + writeln!(file, "{line}")?; + file.flush()?; + } + + if self.jsonl_file.is_none() && !filtered_output { + self.clear_prompt_line_locked()?; + let mut stdout = io::stdout(); + writeln!(stdout, "{line}")?; + stdout.flush()?; + self.redraw_prompt_locked()?; + } + + Ok(()) + } + pub fn server_line(&self, line: &str) -> io::Result<()> { let _guard = self.lock.lock().expect("output lock poisoned"); self.clear_prompt_line_locked()?; @@ -120,3 +143,33 @@ impl Output { Ok(()) } } + +#[cfg(test)] +mod tests { + use std::fs; + + use super::*; + + #[test] + fn server_json_line_writes_to_configured_file() { + let path = std::env::temp_dir().join(format!( + "codex-debug-client-output-{}.jsonl", + std::process::id() + )); + let file = File::create(&path).expect("create output file"); + let output = Output::new(Some(file)); + + output + .server_json_line(r#"{"id":1}"#, false) + .expect("write unfiltered line"); + output + .server_json_line(r#"{"id":2}"#, true) + .expect("write filtered line"); + + assert_eq!( + fs::read_to_string(&path).expect("read output file"), + "{\"id\":1}\n{\"id\":2}\n" + ); + let _ = fs::remove_file(path); + } +} diff --git a/codex-rs/debug-client/src/reader.rs b/codex-rs/debug-client/src/reader.rs index ed401bb100..cb51641633 100644 --- a/codex-rs/debug-client/src/reader.rs +++ b/codex-rs/debug-client/src/reader.rs @@ -67,8 +67,8 @@ pub fn start_reader( } let line = buffer.trim_end_matches(['\n', '\r']); - if !line.is_empty() && !filtered_output { - let _ = output.server_line(line); + if !line.is_empty() { + let _ = output.server_json_line(line, filtered_output); } let Ok(message) = serde_json::from_str::(line) else { From 769b1c3d7e40195f5a4048f82245e3fc19441c4d Mon Sep 17 00:00:00 2001 From: Curtis 'Fjord' Hawthorne Date: Tue, 14 Apr 2026 11:06:50 -0700 Subject: [PATCH 2/3] Keep image_detail_original as a removed feature flag (#17803) --- codex-rs/cli/src/main.rs | 13 +++++++++++++ codex-rs/core/config.schema.json | 6 ++++++ codex-rs/features/src/lib.rs | 9 +++++++++ codex-rs/features/src/tests.rs | 14 ++++++++++++++ 4 files changed, 42 insertions(+) diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 563790c827..5ad86a80fd 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -2205,6 +2205,19 @@ mod tests { ); } + #[test] + fn feature_toggles_accept_removed_image_detail_original_flag() { + let toggles = FeatureToggles { + enable: vec!["image_detail_original".to_string()], + disable: Vec::new(), + }; + let overrides = toggles.to_overrides().expect("valid features"); + assert_eq!( + overrides, + vec!["features.image_detail_original=true".to_string(),] + ); + } + #[test] fn feature_toggles_unknown_feature_errors() { let toggles = FeatureToggles { diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 58c1eb4d12..7c3f6dd99a 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -398,6 +398,9 @@ "guardian_approval": { "type": "boolean" }, + "image_detail_original": { + "type": "boolean" + }, "image_generation": { "type": "boolean" }, @@ -2251,6 +2254,9 @@ "guardian_approval": { "type": "boolean" }, + "image_detail_original": { + "type": "boolean" + }, "image_generation": { "type": "boolean" }, diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index aa32a45718..727c32ccd4 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -178,6 +178,9 @@ pub enum Feature { RealtimeConversation, /// Connect app-server to the ChatGPT remote control service. RemoteControl, + /// Removed compatibility flag retained as a no-op so old wrappers can + /// still pass `--enable image_detail_original`. + ImageDetailOriginal, /// Removed compatibility flag. The TUI now always uses the app-server implementation. TuiAppServer, /// Prevent idle system sleep while a turn is actively running. @@ -874,6 +877,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::ImageDetailOriginal, + key: "image_detail_original", + stage: Stage::Removed, + default_enabled: false, + }, FeatureSpec { id: Feature::TuiAppServer, key: "tui_app_server", diff --git a/codex-rs/features/src/tests.rs b/codex-rs/features/src/tests.rs index 3ecab0997c..818ffc8bd7 100644 --- a/codex-rs/features/src/tests.rs +++ b/codex-rs/features/src/tests.rs @@ -53,6 +53,12 @@ fn use_linux_sandbox_bwrap_is_removed_and_disabled_by_default() { assert_eq!(Feature::UseLinuxSandboxBwrap.default_enabled(), false); } +#[test] +fn image_detail_original_is_removed_and_disabled_by_default() { + assert_eq!(Feature::ImageDetailOriginal.stage(), Stage::Removed); + assert_eq!(Feature::ImageDetailOriginal.default_enabled(), false); +} + #[test] fn js_repl_is_experimental_and_user_toggleable() { let spec = Feature::JsRepl.info(); @@ -145,6 +151,14 @@ fn use_linux_sandbox_bwrap_is_a_removed_feature_key() { ); } +#[test] +fn image_detail_original_is_a_removed_feature_key() { + assert_eq!( + feature_for_key("image_detail_original"), + Some(Feature::ImageDetailOriginal) + ); +} + #[test] fn image_generation_is_under_development() { assert_eq!(Feature::ImageGeneration.stage(), Stage::UnderDevelopment); From 23d4098c0f5ef26deca8aaff4851055aadd82e37 Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Tue, 14 Apr 2026 11:24:34 -0700 Subject: [PATCH 3/3] app-server: prepare to run initialized rpcs concurrently (#17372) ## Summary - Refactors `MessageProcessor` and per-connection session state so initialized service RPC handling can be moved into spawned tasks in a follow-up PR. - Shares the processor and initialized session data with `Arc`/`OnceLock` instead of mutable borrowed connection state. - Keeps initialized request handling synchronous in this PR; it does **not** call `tokio::spawn` for service RPCs yet. ## Testing - `just fmt` - `cargo test -p codex-app-server` *(fails on existing hardening gaps covered by #17375, #17376, and #17377; the pipelined config regression passed before the unrelated failures)* - `just fix -p codex-app-server` --- codex-rs/app-server/src/app_server_tracing.rs | 8 +- codex-rs/app-server/src/in_process.rs | 23 +- codex-rs/app-server/src/lib.rs | 37 +- codex-rs/app-server/src/message_processor.rs | 362 +++++++++++------- .../src/message_processor/tracing_tests.rs | 16 +- codex-rs/app-server/src/transport/mod.rs | 4 +- 6 files changed, 282 insertions(+), 168 deletions(-) diff --git a/codex-rs/app-server/src/app_server_tracing.rs b/codex-rs/app-server/src/app_server_tracing.rs index b06a8e52c4..2118e77300 100644 --- a/codex-rs/app-server/src/app_server_tracing.rs +++ b/codex-rs/app-server/src/app_server_tracing.rs @@ -72,10 +72,10 @@ pub(crate) fn typed_request_span( &span, client_info .map(|(client_name, _)| client_name) - .or(session.app_server_client_name.as_deref()), + .or(session.app_server_client_name()), client_info .map(|(_, client_version)| client_version) - .or(session.client_version.as_deref()), + .or(session.client_version()), ); attach_parent_context(&span, &method, request.id(), /*parent_trace*/ None); @@ -147,7 +147,7 @@ fn client_name<'a>( if let Some(params) = initialize_client_info { return Some(params.client_info.name.as_str()); } - session.app_server_client_name.as_deref() + session.app_server_client_name() } fn client_version<'a>( @@ -157,7 +157,7 @@ fn client_version<'a>( if let Some(params) = initialize_client_info { return Some(params.client_info.version.as_str()); } - session.client_version.as_deref() + session.client_version() } fn initialize_client_info(request: &JSONRPCRequest) -> Option { diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index eb76848c57..9ac0e6cf42 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -386,7 +386,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { AuthManager::shared_from_config(args.config.as_ref(), args.enable_codex_api_key_env); let (processor_tx, mut processor_rx) = mpsc::channel::(channel_capacity); let mut processor_handle = tokio::spawn(async move { - let processor = MessageProcessor::new(MessageProcessorArgs { + let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs { outgoing: Arc::clone(&processor_outgoing), arg0_paths: args.arg0_paths, config: args.config, @@ -401,9 +401,9 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { auth_manager, rpc_transport: AppServerRpcTransport::InProcess, remote_control_handle: None, - }); + })); let mut thread_created_rx = processor.thread_created_receiver(); - let mut session = ConnectionSessionState::default(); + let session = Arc::new(ConnectionSessionState::default()); let mut listen_for_threads = true; loop { @@ -411,28 +411,33 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { command = processor_rx.recv() => { match command { Some(ProcessorCommand::Request(request)) => { - let was_initialized = session.initialized; + let was_initialized = session.initialized(); processor .process_client_request( IN_PROCESS_CONNECTION_ID, *request, - &mut session, + Arc::clone(&session), &outbound_initialized, ) .await; + let opted_out_notification_methods_snapshot = + session.opted_out_notification_methods(); + let experimental_api_enabled = + session.experimental_api_enabled(); + let is_initialized = session.initialized(); if let Ok(mut opted_out_notification_methods) = outbound_opted_out_notification_methods.write() { *opted_out_notification_methods = - session.opted_out_notification_methods.clone(); + opted_out_notification_methods_snapshot; } else { warn!("failed to update outbound opted-out notifications"); } outbound_experimental_api_enabled.store( - session.experimental_api_enabled, + experimental_api_enabled, Ordering::Release, ); - if !was_initialized && session.initialized { + if !was_initialized && is_initialized { processor.send_initialize_notifications().await; } } @@ -447,7 +452,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { created = thread_created_rx.recv(), if listen_for_threads => { match created { Ok(thread_id) => { - let connection_ids = if session.initialized { + let connection_ids = if session.initialized() { vec![IN_PROCESS_CONNECTION_ID] } else { Vec::::new() diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 918108f16b..ca48ad0f41 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -652,7 +652,7 @@ pub async fn run_main_with_transport( AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false); let cli_overrides: Vec<(String, TomlValue)> = cli_kv_overrides.clone(); let loader_overrides = loader_overrides_for_config_api; - let processor = MessageProcessor::new(MessageProcessorArgs { + let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs { outgoing: outgoing_message_sender, arg0_paths, config: Arc::new(config), @@ -667,7 +667,7 @@ pub async fn run_main_with_transport( auth_manager, rpc_transport: analytics_rpc_transport(transport), remote_control_handle: Some(remote_control_handle), - }); + })); let mut thread_created_rx = processor.thread_created_receiver(); let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count(); let mut connections = HashMap::::new(); @@ -769,23 +769,28 @@ pub async fn run_main_with_transport( warn!("dropping request from unknown connection: {connection_id:?}"); continue; }; - let was_initialized = connection_state.session.initialized; + let was_initialized = + connection_state.session.initialized(); processor .process_request( connection_id, request, transport, - &mut connection_state.session, + Arc::clone(&connection_state.session), ) .await; + let opted_out_notification_methods_snapshot = connection_state + .session + .opted_out_notification_methods(); + let experimental_api_enabled = + connection_state.session.experimental_api_enabled(); + let is_initialized = connection_state.session.initialized(); if let Ok(mut opted_out_notification_methods) = connection_state .outbound_opted_out_notification_methods .write() { - *opted_out_notification_methods = connection_state - .session - .opted_out_notification_methods - .clone(); + *opted_out_notification_methods = + opted_out_notification_methods_snapshot; } else { warn!( "failed to update outbound opted-out notifications" @@ -794,10 +799,10 @@ pub async fn run_main_with_transport( connection_state .outbound_experimental_api_enabled .store( - connection_state.session.experimental_api_enabled, + experimental_api_enabled, std::sync::atomic::Ordering::Release, ); - if !was_initialized && connection_state.session.initialized { + if !was_initialized && is_initialized { processor .send_initialize_notifications_to_connection( connection_id, @@ -837,12 +842,12 @@ pub async fn run_main_with_transport( created = thread_created_rx.recv(), if listen_for_threads => { match created { Ok(thread_id) => { - let initialized_connection_ids: Vec = connections - .iter() - .filter_map(|(connection_id, connection_state)| { - connection_state.session.initialized.then_some(*connection_id) - }) - .collect(); + let mut initialized_connection_ids = Vec::new(); + for (connection_id, connection_state) in &connections { + if connection_state.session.initialized() { + initialized_connection_ids.push(*connection_id); + } + } processor .try_attach_thread_listener( thread_id, diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 7bfc01b2a1..48072cce6d 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -2,6 +2,7 @@ use std::collections::BTreeMap; use std::collections::HashSet; use std::future::Future; use std::sync::Arc; +use std::sync::OnceLock; use std::sync::RwLock; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -20,6 +21,7 @@ use crate::outgoing_message::RequestContext; use crate::transport::AppServerTransport; use crate::transport::RemoteControlHandle; use async_trait::async_trait; +use axum::http::HeaderValue; use codex_analytics::AnalyticsEventsClient; use codex_analytics::AppServerRpcTransport; use codex_app_server_protocol::AppListUpdatedNotification; @@ -173,13 +175,52 @@ pub(crate) struct MessageProcessor { remote_control_handle: Option, } -#[derive(Clone, Debug, Default)] +#[derive(Debug, Default)] pub(crate) struct ConnectionSessionState { - pub(crate) initialized: bool, - pub(crate) experimental_api_enabled: bool, - pub(crate) opted_out_notification_methods: HashSet, - pub(crate) app_server_client_name: Option, - pub(crate) client_version: Option, + initialized: OnceLock, +} + +#[derive(Debug)] +struct InitializedConnectionSessionState { + experimental_api_enabled: bool, + opted_out_notification_methods: HashSet, + app_server_client_name: String, + client_version: String, +} + +impl ConnectionSessionState { + pub(crate) fn initialized(&self) -> bool { + self.initialized.get().is_some() + } + + pub(crate) fn experimental_api_enabled(&self) -> bool { + self.initialized + .get() + .is_some_and(|session| session.experimental_api_enabled) + } + + pub(crate) fn opted_out_notification_methods(&self) -> HashSet { + self.initialized + .get() + .map(|session| session.opted_out_notification_methods.clone()) + .unwrap_or_default() + } + + pub(crate) fn app_server_client_name(&self) -> Option<&str> { + self.initialized + .get() + .map(|session| session.app_server_client_name.as_str()) + } + + pub(crate) fn client_version(&self) -> Option<&str> { + self.initialized + .get() + .map(|session| session.client_version.as_str()) + } + + fn initialize(&self, session: InitializedConnectionSessionState) -> Result<(), ()> { + self.initialized.set(session).map_err(|_| ()) + } } pub(crate) struct MessageProcessorArgs { @@ -299,11 +340,11 @@ impl MessageProcessor { } pub(crate) async fn process_request( - &self, + self: &Arc, connection_id: ConnectionId, request: JSONRPCRequest, transport: AppServerTransport, - session: &mut ConnectionSessionState, + session: Arc, ) { let request_method = request.method.as_str(); tracing::trace!( @@ -316,7 +357,7 @@ impl MessageProcessor { request_id: request.id.clone(), }; let request_span = - crate::app_server_tracing::request_span(&request, transport, connection_id, session); + crate::app_server_tracing::request_span(&request, transport, connection_id, &session); let request_trace = request.trace.as_ref().map(|trace| W3cTraceContext { traceparent: trace.traceparent.clone(), tracestate: trace.tracestate.clone(), @@ -358,7 +399,7 @@ impl MessageProcessor { self.handle_client_request( request_id.clone(), codex_request, - session, + Arc::clone(&session), /*outbound_initialized*/ None, request_context.clone(), ) @@ -373,10 +414,10 @@ impl MessageProcessor { /// This bypasses JSON request deserialization but keeps identical request /// semantics by delegating to `handle_client_request`. pub(crate) async fn process_client_request( - &self, + self: &Arc, connection_id: ConnectionId, request: ClientRequest, - session: &mut ConnectionSessionState, + session: Arc, outbound_initialized: &AtomicBool, ) { let request_id = ConnectionRequestId { @@ -384,7 +425,7 @@ impl MessageProcessor { request_id: request.id().clone(), }; let request_span = - crate::app_server_tracing::typed_request_span(&request, connection_id, session); + crate::app_server_tracing::typed_request_span(&request, connection_id, &session); let request_context = RequestContext::new(request_id.clone(), request_span, /*parent_trace*/ None); tracing::trace!( @@ -402,7 +443,7 @@ impl MessageProcessor { self.handle_client_request( request_id.clone(), request, - session, + Arc::clone(&session), Some(outbound_initialized), request_context.clone(), ) @@ -525,10 +566,10 @@ impl MessageProcessor { } async fn handle_client_request( - &self, + self: &Arc, connection_request_id: ConnectionRequestId, codex_request: ClientRequest, - session: &mut ConnectionSessionState, + session: Arc, // `Some(...)` means the caller wants initialize to immediately mark the // connection outbound-ready. Websocket JSON-RPC calls pass `None` so // lib.rs can deliver connection-scoped initialize notifications first. @@ -536,126 +577,166 @@ impl MessageProcessor { request_context: RequestContext, ) { let connection_id = connection_request_id.connection_id; - match codex_request { + if let ClientRequest::Initialize { request_id, params } = codex_request { // Handle Initialize internally so CodexMessageProcessor does not have to concern // itself with the `initialized` bool. - ClientRequest::Initialize { request_id, params } => { - let connection_request_id = ConnectionRequestId { - connection_id, - request_id, + let connection_request_id = ConnectionRequestId { + connection_id, + request_id, + }; + if session.initialized() { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: "Already initialized".to_string(), + data: None, }; - 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; - } - - // TODO(maxj): Revisit capability scoping for `experimental_api_enabled`. - // Current behavior is per-connection. Reviewer feedback notes this can - // create odd cross-client behavior (for example dynamic tool calls on a - // shared thread when another connected client did not opt into - // experimental API). Proposed direction is instance-global first-write-wins - // with initialize-time mismatch rejection. - let analytics_initialize_params = params.clone(); - let (experimental_api_enabled, opt_out_notification_methods) = - match params.capabilities { - Some(capabilities) => ( - capabilities.experimental_api, - capabilities - .opt_out_notification_methods - .unwrap_or_default(), - ), - None => (false, Vec::new()), - }; - session.experimental_api_enabled = experimental_api_enabled; - session.opted_out_notification_methods = - opt_out_notification_methods.into_iter().collect(); - let ClientInfo { - name, - title: _title, - version, - } = params.client_info; - session.app_server_client_name = Some(name.clone()); - session.client_version = Some(version.clone()); - let originator = name.clone(); - if let Err(error) = set_default_originator(originator.clone()) { - match error { - SetOriginatorError::InvalidHeaderValue => { - 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; - } - SetOriginatorError::AlreadyInitialized => { - // No-op. This is expected to happen if the originator is already set via env var. - // TODO(owen): Once we remove support for CODEX_INTERNAL_ORIGINATOR_OVERRIDE, - // this will be an unexpected state and we can return a JSON-RPC error indicating - // internal server error. - } - } - } - if self.config.features.enabled(Feature::GeneralAnalytics) { - self.analytics_events_client.track_initialize( - connection_id.0, - analytics_initialize_params, - originator, - self.rpc_transport, - ); - } - set_default_client_residency_requirement(self.config.enforce_residency.value()); - let user_agent_suffix = format!("{name}; {version}"); - if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() { - *suffix = Some(user_agent_suffix); - } - - let user_agent = get_codex_user_agent(); - let response = InitializeResponse { - user_agent, - codex_home: self.config.codex_home.clone(), - platform_family: std::env::consts::FAMILY.to_string(), - platform_os: std::env::consts::OS.to_string(), - }; - self.outgoing - .send_response(connection_request_id, response) - .await; - - session.initialized = true; - if let Some(outbound_initialized) = outbound_initialized { - // In-process clients can complete readiness immediately here. The - // websocket path defers this until lib.rs finishes transport-layer - // initialize handling for the specific connection. - outbound_initialized.store(true, Ordering::Release); - self.codex_message_processor - .connection_initialized(connection_id) - .await; - } + self.outgoing.send_error(connection_request_id, error).await; return; } - _ => { - 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; + + // TODO(maxj): Revisit capability scoping for `experimental_api_enabled`. + // Current behavior is per-connection. Reviewer feedback notes this can + // create odd cross-client behavior (for example dynamic tool calls on a + // shared thread when another connected client did not opt into + // experimental API). Proposed direction is instance-global first-write-wins + // with initialize-time mismatch rejection. + let analytics_initialize_params = params.clone(); + let (experimental_api_enabled, opt_out_notification_methods) = match params.capabilities + { + Some(capabilities) => ( + capabilities.experimental_api, + capabilities + .opt_out_notification_methods + .unwrap_or_default(), + ), + None => (false, Vec::new()), + }; + let ClientInfo { + name, + title: _title, + version, + } = params.client_info; + // 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; + } + let originator = name.clone(); + let user_agent_suffix = format!("{name}; {version}"); + let codex_home = self.config.codex_home.clone(); + if session + .initialize(InitializedConnectionSessionState { + experimental_api_enabled, + opted_out_notification_methods: opt_out_notification_methods + .into_iter() + .collect(), + app_server_client_name: name.clone(), + client_version: version, + }) + .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; + } + + // Only the request that wins session initialization may mutate + // process-global client metadata. + if let Err(error) = set_default_originator(originator.clone()) { + match error { + SetOriginatorError::InvalidHeaderValue => { + tracing::warn!( + client_info_name = %name, + "validated clientInfo.name was rejected while setting originator" + ); + } + SetOriginatorError::AlreadyInitialized => { + // No-op. This is expected to happen if the originator is already set via env var. + // TODO(owen): Once we remove support for CODEX_INTERNAL_ORIGINATOR_OVERRIDE, + // this will be an unexpected state and we can return a JSON-RPC error indicating + // internal server error. + } } } + if self.config.features.enabled(Feature::GeneralAnalytics) { + self.analytics_events_client.track_initialize( + connection_id.0, + analytics_initialize_params, + originator, + self.rpc_transport, + ); + } + set_default_client_residency_requirement(self.config.enforce_residency.value()); + if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() { + *suffix = Some(user_agent_suffix); + } + + let user_agent = get_codex_user_agent(); + let response = InitializeResponse { + user_agent, + codex_home, + platform_family: std::env::consts::FAMILY.to_string(), + platform_os: std::env::consts::OS.to_string(), + }; + + self.outgoing + .send_response(connection_request_id, response) + .await; + + if let Some(outbound_initialized) = outbound_initialized { + // In-process clients can complete readiness immediately here. The + // websocket path defers this until lib.rs finishes transport-layer + // initialize handling for the specific connection. + outbound_initialized.store(true, Ordering::Release); + self.codex_message_processor + .connection_initialized(connection_id) + .await; + } + return; } + + self.dispatch_initialized_client_request( + connection_request_id, + codex_request, + session, + request_context, + ) + .await; + } + + async fn dispatch_initialized_client_request( + self: &Arc, + connection_request_id: ConnectionRequestId, + codex_request: ClientRequest, + session: Arc, + request_context: RequestContext, + ) { + 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; + } + if let Some(reason) = codex_request.experimental_reason() - && !session.experimental_api_enabled + && !session.experimental_api_enabled() { let error = JSONRPCErrorError { code: INVALID_REQUEST_ERROR_CODE, @@ -676,6 +757,29 @@ impl MessageProcessor { ); } + let app_server_client_name = session.app_server_client_name().map(str::to_string); + let client_version = session.client_version().map(str::to_string); + Arc::clone(self) + .handle_initialized_client_request( + connection_request_id, + codex_request, + request_context, + app_server_client_name, + client_version, + ) + .await; + } + + async fn handle_initialized_client_request( + self: Arc, + connection_request_id: ConnectionRequestId, + codex_request: ClientRequest, + request_context: RequestContext, + app_server_client_name: Option, + client_version: Option, + ) { + let connection_id = connection_request_id.connection_id; + match codex_request { ClientRequest::ConfigRead { request_id, params } => { self.handle_config_read( @@ -847,8 +951,8 @@ impl MessageProcessor { .process_request( connection_id, other, - session.app_server_client_name.clone(), - session.client_version.clone(), + app_server_client_name, + client_version, request_context, ) .boxed() diff --git a/codex-rs/app-server/src/message_processor/tracing_tests.rs b/codex-rs/app-server/src/message_processor/tracing_tests.rs index d0fe22e9b0..c1bb995bab 100644 --- a/codex-rs/app-server/src/message_processor/tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor/tracing_tests.rs @@ -109,9 +109,9 @@ fn tracing_test_guard() -> &'static tokio::sync::Mutex<()> { struct TracingHarness { _server: MockServer, _codex_home: TempDir, - processor: MessageProcessor, + processor: Arc, outgoing_rx: mpsc::Receiver, - session: ConnectionSessionState, + session: Arc, tracing: &'static TestTracing, } @@ -129,7 +129,7 @@ impl TracingHarness { _codex_home: codex_home, processor, outgoing_rx, - session: ConnectionSessionState::default(), + session: Arc::new(ConnectionSessionState::default()), tracing, }; @@ -152,7 +152,7 @@ impl TracingHarness { /*trace*/ None, ) .await; - assert!(harness.session.initialized); + assert!(harness.session.initialized()); Ok(harness) } @@ -182,7 +182,7 @@ impl TracingHarness { TEST_CONNECTION_ID, request, AppServerTransport::Stdio, - &mut self.session, + Arc::clone(&self.session), ) .await; read_response(&mut self.outgoing_rx, request_id).await @@ -230,14 +230,14 @@ async fn build_test_config(codex_home: &Path, server_uri: &str) -> Result, ) -> ( - MessageProcessor, + Arc, mpsc::Receiver, ) { let (outgoing_tx, outgoing_rx) = mpsc::channel(16); let outgoing = Arc::new(OutgoingMessageSender::new(outgoing_tx)); let auth_manager = AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false); - let processor = MessageProcessor::new(MessageProcessorArgs { + let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs { outgoing, arg0_paths: Arg0DispatchPaths::default(), config, @@ -252,7 +252,7 @@ fn build_test_processor( auth_manager, rpc_transport: AppServerRpcTransport::Stdio, remote_control_handle: None, - }); + })); (processor, outgoing_rx) } diff --git a/codex-rs/app-server/src/transport/mod.rs b/codex-rs/app-server/src/transport/mod.rs index 92383cb78f..44e2cb0f92 100644 --- a/codex-rs/app-server/src/transport/mod.rs +++ b/codex-rs/app-server/src/transport/mod.rs @@ -121,7 +121,7 @@ pub(crate) struct ConnectionState { pub(crate) outbound_initialized: Arc, pub(crate) outbound_experimental_api_enabled: Arc, pub(crate) outbound_opted_out_notification_methods: Arc>>, - pub(crate) session: ConnectionSessionState, + pub(crate) session: Arc, } impl ConnectionState { @@ -134,7 +134,7 @@ impl ConnectionState { outbound_initialized, outbound_experimental_api_enabled, outbound_opted_out_notification_methods, - session: ConnectionSessionState::default(), + session: Arc::new(ConnectionSessionState::default()), } } }