Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 //! Shared in-process app-server client facade for CLI surfaces. 2 //! 3 //! This crate wraps [`codex_app_server::in_process`] behind a single async API 4 //! used by surfaces like TUI and exec. It centralizes: 5 //! 6 //! - Runtime startup and initialize-capabilities handshake. 7 //! - Typed caller-provided startup identity (`SessionSource` + client name). 8 //! - Typed and raw request/notification dispatch. 9 //! - Server request resolution and rejection. 10 //! - Ordered, lossless event consumption that cannot block request processing. 11 //! - Bounded graceful shutdown with abort fallback. 12 //! 13 //! The facade interposes a worker task between the caller and the underlying 14 //! [`InProcessClientHandle`](codex_app_server::in_process::InProcessClientHandle), 15 //! bridging async `mpsc` channels on both sides. Commands and the underlying 16 //! runtime remain bounded; the local consumer event queue is unbounded so 17 //! unread notifications cannot prevent request responses from being delivered. 18 19 mod path; 20 mod remote; 21 22 use std::error::Error; 23 use std::fmt; 24 use std::io::Error as IoError; 25 use std::io::ErrorKind; 26 use std::io::Result as IoResult; 27 use std::sync::Arc; 28 use std::time::Duration; 29 30 pub use codex_app_server::app_server_control_socket_path; 31 pub use codex_app_server::in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY; 32 pub use codex_app_server::in_process::InProcessServerEvent; 33 use codex_app_server::in_process::InProcessStartArgs; 34 use codex_app_server::in_process::LogDbLayer; 35 pub use codex_app_server::in_process::StateDbHandle; 36 use codex_app_server_protocol::ClientInfo; 37 use codex_app_server_protocol::ClientNotification; 38 use codex_app_server_protocol::ClientRequest; 39 use codex_app_server_protocol::ConfigWarningNotification; 40 use codex_app_server_protocol::InitializeCapabilities; 41 use codex_app_server_protocol::InitializeParams; 42 use codex_app_server_protocol::JSONRPCErrorError; 43 use codex_app_server_protocol::RequestId; 44 use codex_app_server_protocol::Result as JsonRpcResult; 45 use codex_app_server_protocol::ServerNotification; 46 use codex_app_server_protocol::ServerRequest; 47 use codex_arg0::Arg0DispatchPaths; 48 use codex_config::CloudConfigBundleLoader; 49 use codex_config::LoaderOverrides; 50 use codex_config::NoopThreadConfigLoader; 51 use codex_core::config::Config; 52 pub use codex_core::otel_init::build_provider as build_otel_provider; 53 pub use codex_exec_server::EnvironmentManager; 54 pub use codex_exec_server::ExecServerRuntimePaths; 55 use codex_feedback::CodexFeedback; 56 use codex_protocol::protocol::SessionSource; 57 use codex_utils_absolute_path::AbsolutePathBuf; 58 use serde::de::DeserializeOwned; 59 use tokio::sync::mpsc; 60 use tokio::sync::oneshot; 61 use tokio::time::timeout; 62 use toml::Value as TomlValue; 63 use tracing::warn; 64 65 pub use crate::path::AppServerPath; 66 pub use crate::remote::RemoteAppServerClient; 67 pub use crate::remote::RemoteAppServerConnectArgs; 68 pub use crate::remote::RemoteAppServerEndpoint; 69 70 /// Transitional access to core-only embedded app-server types. 71 /// 72 /// New TUI behavior should prefer the app-server protocol methods. This 73 /// module exists so clients can remove a direct `codex-core` dependency 74 /// while legacy startup/config paths are migrated to RPCs. 75 pub mod legacy_core { 76 pub mod config { 77 pub use codex_core::config::*; 78 79 pub mod edit { 80 pub use codex_core::config::edit::*; 81 } 82 } 83 } 84 85 const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); 86 // Covers the embedded drain, its analytics flush, and final task join. 87 const IN_PROCESS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(45); 88 89 /// Raw app-server request result for typed in-process requests. 90 /// 91 /// Even on the in-process path, successful responses still travel back through 92 /// the same JSON-RPC result envelope used by socket/stdio transports because 93 /// `MessageProcessor` continues to produce that shape internally. 94 pub type RequestResult = std::result::Result<JsonRpcResult, JSONRPCErrorError>; 95 96 #[derive(Debug, Clone)] 97 pub enum AppServerEvent { 98 Lagged { skipped: usize }, 99 ServerNotification(Box<ServerNotification>), 100 ServerRequest(Box<ServerRequest>), 101 Disconnected { message: String }, 102 } 103 104 impl From<InProcessServerEvent> for AppServerEvent { 105 fn from(value: InProcessServerEvent) -> Self { 106 match value { 107 InProcessServerEvent::Lagged { skipped } => Self::Lagged { skipped }, 108 InProcessServerEvent::ServerNotification(notification) => { 109 Self::ServerNotification(notification) 110 } 111 InProcessServerEvent::ServerRequest(request) => Self::ServerRequest(request), 112 } 113 } 114 } 115 116 /// Layered error for [`InProcessAppServerClient::request_typed`]. 117 /// 118 /// This keeps transport failures, server-side JSON-RPC failures, and response 119 /// decode failures distinct so callers can decide whether to retry, surface a 120 /// server error, or treat the response as an internal request/response mismatch. 121 #[derive(Debug)] 122 pub enum TypedRequestError { 123 Transport { 124 method: String, 125 source: IoError, 126 }, 127 Server { 128 method: String, 129 source: JSONRPCErrorError, 130 }, 131 Deserialize { 132 method: String, 133 source: serde_json::Error, 134 }, 135 } 136 137 impl fmt::Display for TypedRequestError { 138 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 139 match self { 140 Self::Transport { method, source } => { 141 write!(f, "{method} transport error: {source}") 142 } 143 Self::Server { method, source } => { 144 write!( 145 f, 146 "{method} failed: {} (code {})", 147 source.message, source.code 148 )?; 149 if let Some(data) = source.data.as_ref() { 150 write!(f, ", data: {data}")?; 151 } 152 Ok(()) 153 } 154 Self::Deserialize { method, source } => { 155 write!(f, "{method} response decode error: {source}") 156 } 157 } 158 } 159 } 160 161 impl Error for TypedRequestError { 162 fn source(&self) -> Option<&(dyn Error + 'static)> { 163 match self { 164 Self::Transport { source, .. } => Some(source), 165 Self::Server { .. } => None, 166 Self::Deserialize { source, .. } => Some(source), 167 } 168 } 169 } 170 171 #[derive(Clone)] 172 pub struct InProcessClientStartArgs { 173 /// Resolved argv0 dispatch paths used by command execution internals. 174 pub arg0_paths: Arg0DispatchPaths, 175 /// Shared config used to initialize app-server runtime. 176 pub config: Arc<Config>, 177 /// CLI config overrides that are already parsed into TOML values. 178 pub cli_overrides: Vec<(String, TomlValue)>, 179 /// Loader override knobs used by config API paths. 180 pub loader_overrides: LoaderOverrides, 181 /// Whether config API paths should reject unknown config fields. 182 pub strict_config: bool, 183 /// Preloaded cloud config bundle provider. 184 pub cloud_config_bundle: CloudConfigBundleLoader, 185 /// Feedback sink used by app-server/core telemetry and logs. 186 pub feedback: CodexFeedback, 187 /// SQLite tracing layer used to flush recently emitted logs before feedback upload. 188 pub log_db: Option<LogDbLayer>, 189 /// Process-wide SQLite state handle shared with the embedded app-server. 190 pub state_db: Option<StateDbHandle>, 191 /// Environment manager used by core execution and filesystem operations. 192 pub environment_manager: Arc<EnvironmentManager>, 193 /// Startup warnings emitted after initialize succeeds. 194 pub config_warnings: Vec<ConfigWarningNotification>, 195 /// Session source recorded in app-server thread metadata. 196 pub session_source: SessionSource, 197 /// Whether auth loading should honor the `CODEX_API_KEY` environment variable. 198 pub enable_codex_api_key_env: bool, 199 /// Client name reported during initialize. 200 pub client_name: String, 201 /// Client version reported during initialize. 202 pub client_version: String, 203 /// Whether experimental APIs are requested at initialize time. 204 pub experimental_api: bool, 205 /// Whether MCP servers may send `openai/form` elicitation requests. 206 pub mcp_server_openai_form_elicitation: bool, 207 /// Notification methods this client opts out of receiving. 208 pub opt_out_notification_methods: Vec<String>, 209 /// Queue capacity for command and embedded-runtime channels (clamped to at least 1). 210 pub channel_capacity: usize, 211 } 212 213 impl InProcessClientStartArgs { 214 /// Builds initialize params from caller-provided metadata. 215 pub fn initialize_params(&self) -> InitializeParams { 216 let capabilities = InitializeCapabilities { 217 experimental_api: self.experimental_api, 218 request_attestation: false, 219 extensions: None, 220 opt_out_notification_methods: if self.opt_out_notification_methods.is_empty() { 221 None 222 } else { 223 Some(self.opt_out_notification_methods.clone()) 224 }, 225 mcp_server_openai_form_elicitation: self.mcp_server_openai_form_elicitation, 226 }; 227 228 InitializeParams { 229 client_info: ClientInfo { 230 name: self.client_name.clone(), 231 title: None, 232 version: self.client_version.clone(), 233 }, 234 capabilities: Some(capabilities), 235 } 236 } 237 238 fn into_runtime_start_args(self) -> InProcessStartArgs { 239 let initialize = self.initialize_params(); 240 InProcessStartArgs { 241 arg0_paths: self.arg0_paths, 242 config: self.config, 243 cli_overrides: self.cli_overrides, 244 loader_overrides: self.loader_overrides, 245 strict_config: self.strict_config, 246 cloud_config_bundle: self.cloud_config_bundle, 247 thread_config_loader: Arc::new(NoopThreadConfigLoader), 248 feedback: self.feedback, 249 log_db: self.log_db, 250 state_db: self.state_db, 251 environment_manager: self.environment_manager, 252 config_warnings: self.config_warnings, 253 session_source: self.session_source, 254 enable_codex_api_key_env: self.enable_codex_api_key_env, 255 initialize, 256 channel_capacity: self.channel_capacity, 257 } 258 } 259 } 260 261 /// Internal command sent from public facade methods to the worker task. 262 /// 263 /// Each variant carries a oneshot sender so the caller can `await` the 264 /// result without holding a mutable reference to the client. 265 enum ClientCommand { 266 Request { 267 request: Box<ClientRequest>, 268 response_tx: oneshot::Sender<IoResult<RequestResult>>, 269 }, 270 Notify { 271 notification: ClientNotification, 272 response_tx: oneshot::Sender<IoResult<()>>, 273 }, 274 ResolveServerRequest { 275 request_id: RequestId, 276 result: JsonRpcResult, 277 response_tx: oneshot::Sender<IoResult<()>>, 278 }, 279 RejectServerRequest { 280 request_id: RequestId, 281 error: JSONRPCErrorError, 282 response_tx: oneshot::Sender<IoResult<()>>, 283 }, 284 Shutdown { 285 response_tx: oneshot::Sender<IoResult<()>>, 286 }, 287 } 288 289 /// Async facade over the in-process app-server runtime. 290 /// 291 /// This type owns a worker task that bridges between: 292 /// - caller-facing async `mpsc` channels used by TUI/exec 293 /// - [`codex_app_server::in_process::InProcessClientHandle`], which speaks to 294 /// the embedded `MessageProcessor` 295 /// 296 /// The facade intentionally preserves the server's request/notification/event 297 /// model instead of exposing direct core runtime handles. That keeps in-process 298 /// callers aligned with app-server behavior while still avoiding a process 299 /// boundary. 300 pub struct InProcessAppServerClient { 301 command_tx: mpsc::Sender<ClientCommand>, 302 event_rx: mpsc::UnboundedReceiver<InProcessServerEvent>, 303 worker_handle: tokio::task::JoinHandle<()>, 304 } 305 306 #[derive(Clone)] 307 pub struct InProcessAppServerRequestHandle { 308 command_tx: mpsc::Sender<ClientCommand>, 309 } 310 311 #[derive(Clone)] 312 pub enum AppServerRequestHandle { 313 InProcess(InProcessAppServerRequestHandle), 314 Remote(crate::remote::RemoteAppServerRequestHandle), 315 } 316 317 pub enum AppServerClient { 318 InProcess(InProcessAppServerClient), 319 Remote(RemoteAppServerClient), 320 } 321 322 impl InProcessAppServerClient { 323 /// Starts the in-process runtime and facade worker task. 324 /// 325 /// The returned client is ready for requests and ordered event consumption. 326 /// Request queues remain bounded without blocking on unread notifications. 327 pub async fn start(args: InProcessClientStartArgs) -> IoResult<Self> { 328 let channel_capacity = args.channel_capacity.max(1); 329 let mut handle = 330 codex_app_server::in_process::start(args.into_runtime_start_args()).await?; 331 let request_sender = handle.sender(); 332 let (command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(channel_capacity); 333 // e9996ec62a preserved transcript events by awaiting a bounded queue, but that can 334 // deadlock a foreground request whose response is behind unread notifications. 335 // Match the remote-client fix in 79ea57715636: only this local consumer queue is 336 // unbounded; commands and the embedded runtime stay bounded and events remain ordered. 337 let (event_tx, event_rx) = mpsc::unbounded_channel::<InProcessServerEvent>(); 338 339 let worker_handle = tokio::spawn(async move { 340 let mut event_stream_enabled = true; 341 loop { 342 tokio::select! { 343 command = command_rx.recv() => { 344 match command { 345 Some(ClientCommand::Request { request, response_tx }) => { 346 let request_sender = request_sender.clone(); 347 // Request waits happen on a detached task so 348 // this loop can keep draining runtime events 349 // while the request is blocked on client input. 350 tokio::spawn(async move { 351 // Device ceremonies belong to the waiting UI. Preserve 352 // its cancellation through this buffering task. 353 let cancellable = matches!(*request, 354 ClientRequest::UserVerificationStatus { .. } 355 | ClientRequest::UserVerificationEnroll { .. } 356 | ClientRequest::UserVerificationDelete { .. } 357 | ClientRequest::UserVerificationVerify { .. }); 358 let mut response_tx = response_tx; 359 tokio::select! { 360 _ = response_tx.closed(), if cancellable => {} 361 result = request_sender.request(*request) => { 362 let _ = response_tx.send(result); 363 } 364 } 365 }); 366 } 367 Some(ClientCommand::Notify { 368 notification, 369 response_tx, 370 }) => { 371 let result = request_sender.notify(notification); 372 let _ = response_tx.send(result); 373 } 374 Some(ClientCommand::ResolveServerRequest { 375 request_id, 376 result, 377 response_tx, 378 }) => { 379 let send_result = 380 request_sender.respond_to_server_request(request_id, result); 381 let _ = response_tx.send(send_result); 382 } 383 Some(ClientCommand::RejectServerRequest { 384 request_id, 385 error, 386 response_tx, 387 }) => { 388 let send_result = request_sender.fail_server_request(request_id, error); 389 let _ = response_tx.send(send_result); 390 } 391 Some(ClientCommand::Shutdown { response_tx }) => { 392 let shutdown_result = handle.shutdown().await; 393 let _ = response_tx.send(shutdown_result); 394 break; 395 } 396 None => { 397 let _ = handle.shutdown().await; 398 break; 399 } 400 } 401 } 402 event = handle.next_event(), if event_stream_enabled => { 403 let Some(event) = event else { 404 break; 405 }; 406 if let InProcessServerEvent::ServerRequest(request) = &event 407 && let ServerRequest::ChatgptAuthTokensRefresh { request_id, .. } = 408 request.as_ref() 409 { 410 let send_result = request_sender.fail_server_request( 411 request_id.clone(), 412 JSONRPCErrorError { 413 code: -32000, 414 message: "chatgpt auth token refresh is not supported for in-process app-server clients".to_string(), 415 data: None, 416 }, 417 ); 418 if let Err(err) = send_result { 419 warn!( 420 "failed to reject unsupported chatgpt auth token refresh request: {err}" 421 ); 422 } 423 continue; 424 } 425 426 if event_tx.send(event).is_err() { 427 event_stream_enabled = false; 428 } 429 } 430 } 431 } 432 }); 433 434 Ok(Self { 435 command_tx, 436 event_rx, 437 worker_handle, 438 }) 439 } 440 441 pub fn request_handle(&self) -> InProcessAppServerRequestHandle { 442 InProcessAppServerRequestHandle { 443 command_tx: self.command_tx.clone(), 444 } 445 } 446 447 /// Sends a typed client request and returns raw JSON-RPC result. 448 /// 449 /// Callers that expect a concrete response type should usually prefer 450 /// [`request_typed`](Self::request_typed). 451 pub async fn request(&self, request: ClientRequest) -> IoResult<RequestResult> { 452 let (response_tx, response_rx) = oneshot::channel(); 453 self.command_tx 454 .send(ClientCommand::Request { 455 request: Box::new(request), 456 response_tx, 457 }) 458 .await 459 .map_err(|_| { 460 IoError::new( 461 ErrorKind::BrokenPipe, 462 "in-process app-server worker channel is closed", 463 ) 464 })?; 465 response_rx.await.map_err(|_| { 466 IoError::new( 467 ErrorKind::BrokenPipe, 468 "in-process app-server request channel is closed", 469 ) 470 })? 471 } 472 473 /// Sends a typed client request and decodes the successful response body. 474 /// 475 /// This still deserializes from a JSON value produced by app-server's 476 /// JSON-RPC result envelope. Because the caller chooses `T`, `Deserialize` 477 /// failures indicate an internal request/response mismatch at the call site 478 /// (or an in-process bug), not transport skew from an external client. 479 pub async fn request_typed<T>(&self, request: ClientRequest) -> Result<T, TypedRequestError> 480 where 481 T: DeserializeOwned, 482 { 483 let method = request.method_name(); 484 let response = 485 self.request(request) 486 .await 487 .map_err(|source| TypedRequestError::Transport { 488 method: method.to_string(), 489 source, 490 })?; 491 let result = response.map_err(|source| TypedRequestError::Server { 492 method: method.to_string(), 493 source, 494 })?; 495 serde_json::from_value(result).map_err(|source| TypedRequestError::Deserialize { 496 method: method.to_string(), 497 source, 498 }) 499 } 500 501 /// Sends a typed client notification. 502 pub async fn notify(&self, notification: ClientNotification) -> IoResult<()> { 503 let (response_tx, response_rx) = oneshot::channel(); 504 self.command_tx 505 .send(ClientCommand::Notify { 506 notification, 507 response_tx, 508 }) 509 .await 510 .map_err(|_| { 511 IoError::new( 512 ErrorKind::BrokenPipe, 513 "in-process app-server worker channel is closed", 514 ) 515 })?; 516 response_rx.await.map_err(|_| { 517 IoError::new( 518 ErrorKind::BrokenPipe, 519 "in-process app-server notify channel is closed", 520 ) 521 })? 522 } 523 524 /// Resolves a pending server request. 525 /// 526 /// This should only be called with request IDs obtained from the current 527 /// client's event stream. 528 pub async fn resolve_server_request( 529 &self, 530 request_id: RequestId, 531 result: JsonRpcResult, 532 ) -> IoResult<()> { 533 let (response_tx, response_rx) = oneshot::channel(); 534 self.command_tx 535 .send(ClientCommand::ResolveServerRequest { 536 request_id, 537 result, 538 response_tx, 539 }) 540 .await 541 .map_err(|_| { 542 IoError::new( 543 ErrorKind::BrokenPipe, 544 "in-process app-server worker channel is closed", 545 ) 546 })?; 547 response_rx.await.map_err(|_| { 548 IoError::new( 549 ErrorKind::BrokenPipe, 550 "in-process app-server resolve channel is closed", 551 ) 552 })? 553 } 554 555 /// Rejects a pending server request with JSON-RPC error payload. 556 pub async fn reject_server_request( 557 &self, 558 request_id: RequestId, 559 error: JSONRPCErrorError, 560 ) -> IoResult<()> { 561 let (response_tx, response_rx) = oneshot::channel(); 562 self.command_tx 563 .send(ClientCommand::RejectServerRequest { 564 request_id, 565 error, 566 response_tx, 567 }) 568 .await 569 .map_err(|_| { 570 IoError::new( 571 ErrorKind::BrokenPipe, 572 "in-process app-server worker channel is closed", 573 ) 574 })?; 575 response_rx.await.map_err(|_| { 576 IoError::new( 577 ErrorKind::BrokenPipe, 578 "in-process app-server reject channel is closed", 579 ) 580 })? 581 } 582 583 /// Returns the next in-process event, or `None` when worker exits. 584 /// 585 /// Events remain ordered and are retained while callers await requests. 586 pub async fn next_event(&mut self) -> Option<InProcessServerEvent> { 587 self.event_rx.recv().await 588 } 589 590 /// Shuts down worker and in-process runtime with bounded wait. 591 /// 592 /// If graceful shutdown exceeds timeout, the worker task is aborted to 593 /// avoid leaking background tasks in embedding callers. 594 pub async fn shutdown(self) -> IoResult<()> { 595 let Self { 596 command_tx, 597 event_rx, 598 worker_handle, 599 } = self; 600 let mut worker_handle = worker_handle; 601 // Stop forwarding caller-facing events before asking the worker to shut down. 602 drop(event_rx); 603 let (response_tx, response_rx) = oneshot::channel(); 604 if command_tx 605 .send(ClientCommand::Shutdown { response_tx }) 606 .await 607 .is_ok() 608 && let Ok(command_result) = timeout(IN_PROCESS_SHUTDOWN_TIMEOUT, response_rx).await 609 { 610 command_result.map_err(|_| { 611 IoError::new( 612 ErrorKind::BrokenPipe, 613 "in-process app-server shutdown channel is closed", 614 ) 615 })??; 616 } 617 618 if let Err(_elapsed) = timeout(IN_PROCESS_SHUTDOWN_TIMEOUT, &mut worker_handle).await { 619 worker_handle.abort(); 620 let _ = worker_handle.await; 621 } 622 Ok(()) 623 } 624 } 625 626 impl InProcessAppServerRequestHandle { 627 pub async fn request(&self, request: ClientRequest) -> IoResult<RequestResult> { 628 let (response_tx, response_rx) = oneshot::channel(); 629 self.command_tx 630 .send(ClientCommand::Request { 631 request: Box::new(request), 632 response_tx, 633 }) 634 .await 635 .map_err(|_| { 636 IoError::new( 637 ErrorKind::BrokenPipe, 638 "in-process app-server worker channel is closed", 639 ) 640 })?; 641 response_rx.await.map_err(|_| { 642 IoError::new( 643 ErrorKind::BrokenPipe, 644 "in-process app-server request channel is closed", 645 ) 646 })? 647 } 648 649 pub async fn request_typed<T>(&self, request: ClientRequest) -> Result<T, TypedRequestError> 650 where 651 T: DeserializeOwned, 652 { 653 let method = request.method_name(); 654 let response = 655 self.request(request) 656 .await 657 .map_err(|source| TypedRequestError::Transport { 658 method: method.to_string(), 659 source, 660 })?; 661 let result = response.map_err(|source| TypedRequestError::Server { 662 method: method.to_string(), 663 source, 664 })?; 665 serde_json::from_value(result).map_err(|source| TypedRequestError::Deserialize { 666 method: method.to_string(), 667 source, 668 }) 669 } 670 } 671 672 impl AppServerRequestHandle { 673 pub async fn request(&self, request: ClientRequest) -> IoResult<RequestResult> { 674 match self { 675 Self::InProcess(handle) => handle.request(request).await, 676 Self::Remote(handle) => handle.request(request).await, 677 } 678 } 679 680 pub async fn request_typed<T>(&self, request: ClientRequest) -> Result<T, TypedRequestError> 681 where 682 T: DeserializeOwned, 683 { 684 match self { 685 Self::InProcess(handle) => handle.request_typed(request).await, 686 Self::Remote(handle) => handle.request_typed(request).await, 687 } 688 } 689 } 690 691 impl AppServerClient { 692 /// App-server platform family, which can differ from the executor's platform. 693 /// Older remote servers may omit this metadata. 694 pub fn platform_family(&self) -> Option<&str> { 695 match self { 696 Self::InProcess(_) => Some(std::env::consts::FAMILY), 697 Self::Remote(client) => client.platform_family(), 698 } 699 } 700 701 /// App-server operating system as reported at initialization, including unknown values. 702 pub fn platform_os(&self) -> Option<&str> { 703 match self { 704 Self::InProcess(_) => Some(std::env::consts::OS), 705 Self::Remote(client) => client.platform_os(), 706 } 707 } 708 709 pub fn codex_home(&self, local_codex_home: &AbsolutePathBuf) -> Option<AppServerPath> { 710 match self { 711 Self::InProcess(_) => Some(AppServerPath::from_app_server( 712 local_codex_home.display().to_string(), 713 )), 714 Self::Remote(client) => client.codex_home().map(AppServerPath::from_app_server), 715 } 716 } 717 718 pub async fn request(&self, request: ClientRequest) -> IoResult<RequestResult> { 719 match self { 720 Self::InProcess(client) => client.request(request).await, 721 Self::Remote(client) => client.request(request).await, 722 } 723 } 724 725 pub async fn request_typed<T>(&self, request: ClientRequest) -> Result<T, TypedRequestError> 726 where 727 T: DeserializeOwned, 728 { 729 match self { 730 Self::InProcess(client) => client.request_typed(request).await, 731 Self::Remote(client) => client.request_typed(request).await, 732 } 733 } 734 735 pub async fn notify(&self, notification: ClientNotification) -> IoResult<()> { 736 match self { 737 Self::InProcess(client) => client.notify(notification).await, 738 Self::Remote(client) => client.notify(notification).await, 739 } 740 } 741 742 pub async fn resolve_server_request( 743 &self, 744 request_id: RequestId, 745 result: JsonRpcResult, 746 ) -> IoResult<()> { 747 match self { 748 Self::InProcess(client) => client.resolve_server_request(request_id, result).await, 749 Self::Remote(client) => client.resolve_server_request(request_id, result).await, 750 } 751 } 752 753 pub async fn reject_server_request( 754 &self, 755 request_id: RequestId, 756 error: JSONRPCErrorError, 757 ) -> IoResult<()> { 758 match self { 759 Self::InProcess(client) => client.reject_server_request(request_id, error).await, 760 Self::Remote(client) => client.reject_server_request(request_id, error).await, 761 } 762 } 763 764 pub async fn next_event(&mut self) -> Option<AppServerEvent> { 765 match self { 766 Self::InProcess(client) => client.next_event().await.map(Into::into), 767 Self::Remote(client) => client.next_event().await, 768 } 769 } 770 771 pub async fn shutdown(self) -> IoResult<()> { 772 match self { 773 Self::InProcess(client) => client.shutdown().await, 774 Self::Remote(client) => client.shutdown().await, 775 } 776 } 777 778 pub fn request_handle(&self) -> AppServerRequestHandle { 779 match self { 780 Self::InProcess(client) => AppServerRequestHandle::InProcess(client.request_handle()), 781 Self::Remote(client) => AppServerRequestHandle::Remote(client.request_handle()), 782 } 783 } 784 } 785 786 #[cfg(test)] 787 mod tests { 788 use super::*; 789 use codex_app_server_protocol::AccountUpdatedNotification; 790 use codex_app_server_protocol::ConfigRequirementsReadResponse; 791 use codex_app_server_protocol::GetAccountResponse; 792 use codex_app_server_protocol::JSONRPCMessage; 793 use codex_app_server_protocol::JSONRPCRequest; 794 use codex_app_server_protocol::JSONRPCResponse; 795 use codex_app_server_protocol::ServerNotification; 796 use codex_app_server_protocol::SessionSource as ApiSessionSource; 797 use codex_app_server_protocol::ThreadSettingsUpdateParams; 798 use codex_app_server_protocol::ThreadSettingsUpdateResponse; 799 use codex_app_server_protocol::ThreadStartParams; 800 use codex_app_server_protocol::ThreadStartResponse; 801 use codex_app_server_protocol::ToolRequestUserInputParams; 802 use codex_app_server_protocol::ToolRequestUserInputQuestion; 803 use codex_core::config::ConfigBuilder; 804 use codex_core::init_state_db; 805 use codex_protocol::config_types::Personality; 806 use codex_uds::UnixListener; 807 use codex_utils_absolute_path::AbsolutePathBuf; 808 use futures::SinkExt; 809 use futures::StreamExt; 810 use pretty_assertions::assert_eq; 811 use std::ops::Deref; 812 use std::path::Path; 813 use tempfile::TempDir; 814 use tokio::net::TcpListener; 815 use tokio::time::Duration; 816 use tokio::time::timeout; 817 use tokio_tungstenite::accept_async; 818 use tokio_tungstenite::accept_hdr_async; 819 use tokio_tungstenite::tungstenite::Message; 820 use tokio_tungstenite::tungstenite::handshake::server::Request as WebSocketRequest; 821 use tokio_tungstenite::tungstenite::handshake::server::Response as WebSocketResponse; 822 use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION; 823 824 async fn build_test_config() -> Config { 825 match ConfigBuilder::default().build().await { 826 Ok(config) => config, 827 Err(_) => Config::load_default_with_cli_overrides(Vec::new()) 828 .await 829 .expect("default config should load"), 830 } 831 } 832 833 async fn build_test_config_for_codex_home(codex_home: &Path) -> Config { 834 match ConfigBuilder::default() 835 .codex_home(codex_home.to_path_buf()) 836 .build() 837 .await 838 { 839 Ok(config) => config, 840 Err(_) => Config::load_default_with_cli_overrides_for_codex_home( 841 codex_home.to_path_buf(), 842 Vec::new(), 843 ) 844 .await 845 .expect("default config should load"), 846 } 847 } 848 849 struct TestClient { 850 _codex_home: TempDir, 851 client: InProcessAppServerClient, 852 } 853 854 impl Deref for TestClient { 855 type Target = InProcessAppServerClient; 856 857 fn deref(&self) -> &Self::Target { 858 &self.client 859 } 860 } 861 862 impl TestClient { 863 async fn shutdown(self) -> IoResult<()> { 864 self.client.shutdown().await 865 } 866 } 867 868 async fn start_test_client_with_capacity( 869 session_source: SessionSource, 870 channel_capacity: usize, 871 ) -> TestClient { 872 let codex_home = TempDir::new().expect("temp dir"); 873 let config = Arc::new(build_test_config_for_codex_home(codex_home.path()).await); 874 let state_db = init_state_db(config.as_ref()) 875 .await 876 .expect("state db should initialize for in-process test"); 877 let client = InProcessAppServerClient::start(InProcessClientStartArgs { 878 arg0_paths: Arg0DispatchPaths::default(), 879 config, 880 cli_overrides: Vec::new(), 881 loader_overrides: LoaderOverrides::default(), 882 strict_config: false, 883 cloud_config_bundle: CloudConfigBundleLoader::default(), 884 feedback: CodexFeedback::new(), 885 log_db: None, 886 state_db: Some(state_db), 887 environment_manager: Arc::new(EnvironmentManager::default_for_tests()), 888 config_warnings: Vec::new(), 889 session_source, 890 enable_codex_api_key_env: false, 891 client_name: "codex-app-server-client-test".to_string(), 892 client_version: "0.0.0-test".to_string(), 893 experimental_api: true, 894 mcp_server_openai_form_elicitation: false, 895 opt_out_notification_methods: Vec::new(), 896 channel_capacity, 897 }) 898 .await 899 .expect("in-process app-server client should start"); 900 901 TestClient { 902 _codex_home: codex_home, 903 client, 904 } 905 } 906 907 async fn start_test_client(session_source: SessionSource) -> TestClient { 908 start_test_client_with_capacity(session_source, DEFAULT_IN_PROCESS_CHANNEL_CAPACITY).await 909 } 910 911 async fn start_test_remote_server<F, Fut>(handler: F) -> String 912 where 913 F: FnOnce(tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) -> Fut 914 + Send 915 + 'static, 916 Fut: std::future::Future<Output = ()> + Send + 'static, 917 { 918 start_test_remote_server_with_auth(/*expected_auth_token*/ None, handler).await 919 } 920 921 async fn start_test_remote_server_with_auth<F, Fut>( 922 expected_auth_token: Option<String>, 923 handler: F, 924 ) -> String 925 where 926 F: FnOnce(tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) -> Fut 927 + Send 928 + 'static, 929 Fut: std::future::Future<Output = ()> + Send + 'static, 930 { 931 let listener = TcpListener::bind("127.0.0.1:0") 932 .await 933 .expect("listener should bind"); 934 let addr = listener.local_addr().expect("listener address"); 935 tokio::spawn(async move { 936 let (stream, _) = listener.accept().await.expect("accept should succeed"); 937 let websocket = accept_hdr_async( 938 stream, 939 move |request: &WebSocketRequest, response: WebSocketResponse| { 940 let provided_auth_token = request 941 .headers() 942 .get(AUTHORIZATION) 943 .and_then(|value| value.to_str().ok()) 944 .map(str::to_owned); 945 let expected_auth_token = expected_auth_token 946 .as_ref() 947 .map(|token| format!("Bearer {token}")); 948 assert_eq!(provided_auth_token, expected_auth_token); 949 Ok(response) 950 }, 951 ) 952 .await 953 .expect("websocket upgrade should succeed"); 954 handler(websocket).await; 955 }); 956 format!("ws://{addr}") 957 } 958 959 async fn expect_remote_initialize<S>(websocket: &mut tokio_tungstenite::WebSocketStream<S>) 960 where 961 S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, 962 { 963 expect_remote_initialize_with_metadata( 964 websocket, 965 serde_json::json!({ 966 "userAgent": "codex_cli_rs/9.8.7-test (Test OS; x86_64) rust", 967 "codexHome": "/server/.codex", 968 }), 969 ) 970 .await; 971 } 972 973 async fn expect_remote_initialize_with_metadata<S>( 974 websocket: &mut tokio_tungstenite::WebSocketStream<S>, 975 metadata: serde_json::Value, 976 ) where 977 S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, 978 { 979 let JSONRPCMessage::Request(request) = read_websocket_message(websocket).await else { 980 panic!("expected initialize request"); 981 }; 982 assert_eq!(request.method, "initialize"); 983 write_websocket_message( 984 websocket, 985 JSONRPCMessage::Response(JSONRPCResponse { 986 id: request.id, 987 result: metadata, 988 }), 989 ) 990 .await; 991 992 let JSONRPCMessage::Notification(notification) = read_websocket_message(websocket).await 993 else { 994 panic!("expected initialized notification"); 995 }; 996 assert_eq!(notification.method, "initialized"); 997 } 998 999 async fn read_websocket_message<S>( 1000 websocket: &mut tokio_tungstenite::WebSocketStream<S>, 1001 ) -> JSONRPCMessage 1002 where 1003 S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, 1004 { 1005 loop { 1006 let frame = websocket 1007 .next() 1008 .await 1009 .expect("frame should be available") 1010 .expect("frame should decode"); 1011 match frame { 1012 Message::Text(text) => { 1013 return serde_json::from_str::<JSONRPCMessage>(&text) 1014 .expect("text frame should be valid JSON-RPC"); 1015 } 1016 Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => { 1017 continue; 1018 } 1019 Message::Close(_) => panic!("unexpected close frame"), 1020 } 1021 } 1022 } 1023 1024 async fn write_websocket_message<S>( 1025 websocket: &mut tokio_tungstenite::WebSocketStream<S>, 1026 message: JSONRPCMessage, 1027 ) where 1028 S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, 1029 { 1030 websocket 1031 .send(Message::Text( 1032 serde_json::to_string(&message) 1033 .expect("message should serialize") 1034 .into(), 1035 )) 1036 .await 1037 .expect("message should send"); 1038 } 1039 1040 fn command_execution_output_delta_notification(delta: &str) -> ServerNotification { 1041 ServerNotification::CommandExecutionOutputDelta( 1042 codex_app_server_protocol::CommandExecutionOutputDeltaNotification { 1043 thread_id: "thread".to_string(), 1044 turn_id: "turn".to_string(), 1045 item_id: "item".to_string(), 1046 delta: delta.to_string(), 1047 }, 1048 ) 1049 } 1050 1051 fn agent_message_delta_notification(delta: &str) -> ServerNotification { 1052 ServerNotification::AgentMessageDelta( 1053 codex_app_server_protocol::AgentMessageDeltaNotification { 1054 thread_id: "thread".to_string(), 1055 turn_id: "turn".to_string(), 1056 item_id: "item".to_string(), 1057 delta: delta.to_string(), 1058 }, 1059 ) 1060 } 1061 1062 fn item_completed_notification(text: &str) -> ServerNotification { 1063 ServerNotification::ItemCompleted(codex_app_server_protocol::ItemCompletedNotification { 1064 thread_id: "thread".to_string(), 1065 turn_id: "turn".to_string(), 1066 completed_at_ms: 0, 1067 item: codex_app_server_protocol::ThreadItem::AgentMessage { 1068 id: "item".to_string(), 1069 text: text.to_string(), 1070 phase: None, 1071 memory_citation: None, 1072 delivery: None, 1073 questions: None, 1074 }, 1075 }) 1076 } 1077 1078 fn turn_completed_notification() -> ServerNotification { 1079 ServerNotification::TurnCompleted(codex_app_server_protocol::TurnCompletedNotification { 1080 thread_id: "thread".to_string(), 1081 turn: codex_app_server_protocol::Turn { 1082 id: "turn".to_string(), 1083 items_view: codex_app_server_protocol::TurnItemsView::Full, 1084 items: Vec::new(), 1085 status: codex_app_server_protocol::TurnStatus::Completed, 1086 error: None, 1087 started_at: None, 1088 completed_at: Some(0), 1089 duration_ms: Some(1), 1090 }, 1091 }) 1092 } 1093 1094 fn test_remote_connect_args(websocket_url: String) -> RemoteAppServerConnectArgs { 1095 RemoteAppServerConnectArgs { 1096 endpoint: RemoteAppServerEndpoint::WebSocket { 1097 websocket_url, 1098 auth_token: None, 1099 }, 1100 client_name: "codex-app-server-client-test".to_string(), 1101 client_version: "0.0.0-test".to_string(), 1102 experimental_api: true, 1103 mcp_server_openai_form_elicitation: false, 1104 opt_out_notification_methods: Vec::new(), 1105 channel_capacity: 8, 1106 } 1107 } 1108 1109 #[test] 1110 fn remote_initialize_params_forward_openai_form_capability() { 1111 let mut args = test_remote_connect_args("ws://localhost/rpc".to_string()); 1112 args.mcp_server_openai_form_elicitation = true; 1113 1114 assert!( 1115 args.initialize_params() 1116 .capabilities 1117 .expect("initialize capabilities") 1118 .mcp_server_openai_form_elicitation 1119 ); 1120 } 1121 1122 #[tokio::test] 1123 async fn typed_request_roundtrip_works() { 1124 let TestClient { 1125 _codex_home, 1126 client, 1127 } = start_test_client(SessionSource::Exec).await; 1128 let client = AppServerClient::InProcess(client); 1129 assert_eq!( 1130 (client.platform_family(), client.platform_os()), 1131 (Some(std::env::consts::FAMILY), Some(std::env::consts::OS)) 1132 ); 1133 let _response: ConfigRequirementsReadResponse = client 1134 .request_typed(ClientRequest::ConfigRequirementsRead { 1135 request_id: RequestId::Integer(1), 1136 params: None, 1137 }) 1138 .await 1139 .expect("typed request should succeed"); 1140 client.shutdown().await.expect("shutdown should complete"); 1141 } 1142 1143 #[tokio::test] 1144 async fn typed_request_reports_json_rpc_errors() { 1145 let client = start_test_client(SessionSource::Exec).await; 1146 let err = client 1147 .request_typed::<ConfigRequirementsReadResponse>(ClientRequest::ThreadRead { 1148 request_id: RequestId::Integer(99), 1149 params: codex_app_server_protocol::ThreadReadParams { 1150 thread_id: "missing-thread".to_string(), 1151 include_turns: false, 1152 }, 1153 }) 1154 .await 1155 .expect_err("missing thread should return a JSON-RPC error"); 1156 assert!( 1157 err.to_string().starts_with("thread/read failed:"), 1158 "expected method-qualified JSON-RPC failure message" 1159 ); 1160 client.shutdown().await.expect("shutdown should complete"); 1161 } 1162 1163 #[tokio::test] 1164 async fn caller_provided_session_source_is_applied() { 1165 for (session_source, expected_source) in [ 1166 (SessionSource::Exec, ApiSessionSource::Exec), 1167 (SessionSource::Cli, ApiSessionSource::Cli), 1168 ] { 1169 let client = start_test_client(session_source).await; 1170 let parsed: ThreadStartResponse = client 1171 .request_typed(ClientRequest::ThreadStart { 1172 request_id: RequestId::Integer(2), 1173 params: ThreadStartParams { 1174 ephemeral: Some(true), 1175 ..ThreadStartParams::default() 1176 }, 1177 }) 1178 .await 1179 .expect("thread/start should succeed"); 1180 assert_eq!(parsed.thread.source, expected_source); 1181 client.shutdown().await.expect("shutdown should complete"); 1182 } 1183 } 1184 1185 #[tokio::test] 1186 async fn threads_started_via_app_server_are_visible_through_typed_requests() { 1187 let client = start_test_client(SessionSource::Cli).await; 1188 1189 let response: ThreadStartResponse = client 1190 .request_typed(ClientRequest::ThreadStart { 1191 request_id: RequestId::Integer(3), 1192 params: ThreadStartParams { 1193 ephemeral: Some(true), 1194 ..ThreadStartParams::default() 1195 }, 1196 }) 1197 .await 1198 .expect("thread/start should succeed"); 1199 let read = client 1200 .request_typed::<codex_app_server_protocol::ThreadReadResponse>( 1201 ClientRequest::ThreadRead { 1202 request_id: RequestId::Integer(4), 1203 params: codex_app_server_protocol::ThreadReadParams { 1204 thread_id: response.thread.id.clone(), 1205 include_turns: false, 1206 }, 1207 }, 1208 ) 1209 .await 1210 .expect("thread/read should return the newly started thread"); 1211 assert_eq!(read.thread.id, response.thread.id); 1212 1213 client.shutdown().await.expect("shutdown should complete"); 1214 } 1215 1216 #[tokio::test] 1217 async fn tiny_channel_capacity_still_supports_request_roundtrip() { 1218 let client = 1219 start_test_client_with_capacity(SessionSource::Exec, /*channel_capacity*/ 1).await; 1220 let _response: ConfigRequirementsReadResponse = client 1221 .request_typed(ClientRequest::ConfigRequirementsRead { 1222 request_id: RequestId::Integer(1), 1223 params: None, 1224 }) 1225 .await 1226 .expect("typed request should succeed"); 1227 client.shutdown().await.expect("shutdown should complete"); 1228 } 1229 1230 #[tokio::test] 1231 async fn unread_lossless_notifications_do_not_block_in_process_requests() { 1232 let mut client = 1233 start_test_client_with_capacity(SessionSource::Cli, /*channel_capacity*/ 1).await; 1234 let thread: ThreadStartResponse = client 1235 .request_typed(ClientRequest::ThreadStart { 1236 request_id: RequestId::Integer(1), 1237 params: ThreadStartParams { 1238 ephemeral: Some(true), 1239 personality: Some(Personality::None), 1240 ..ThreadStartParams::default() 1241 }, 1242 }) 1243 .await 1244 .expect("thread/start should succeed"); 1245 let request_handle = client.request_handle(); 1246 1247 timeout(Duration::from_secs(2), async { 1248 for (index, personality) in [ 1249 Personality::Friendly, 1250 Personality::Pragmatic, 1251 Personality::Friendly, 1252 Personality::Pragmatic, 1253 ] 1254 .into_iter() 1255 .enumerate() 1256 { 1257 let _: ThreadSettingsUpdateResponse = request_handle 1258 .request_typed(ClientRequest::ThreadSettingsUpdate { 1259 request_id: RequestId::Integer((index + 2) as i64), 1260 params: ThreadSettingsUpdateParams { 1261 thread_id: thread.thread.id.clone(), 1262 personality: Some(personality), 1263 ..ThreadSettingsUpdateParams::default() 1264 }, 1265 }) 1266 .await 1267 .expect("thread/settings/update should succeed"); 1268 } 1269 1270 let _: ConfigRequirementsReadResponse = request_handle 1271 .request_typed(ClientRequest::ConfigRequirementsRead { 1272 request_id: RequestId::Integer(10), 1273 params: None, 1274 }) 1275 .await 1276 .expect("configuration request should succeed"); 1277 }) 1278 .await 1279 .expect("unread lossless notifications must not block app-server requests"); 1280 1281 let mut personalities = Vec::new(); 1282 timeout(Duration::from_secs(2), async { 1283 while personalities.len() < 4 { 1284 if let Some(InProcessServerEvent::ServerNotification(notification)) = 1285 client.client.next_event().await 1286 && let ServerNotification::ThreadSettingsUpdated(notification) = 1287 notification.as_ref() 1288 { 1289 personalities.push(notification.thread_settings.personality); 1290 } 1291 } 1292 }) 1293 .await 1294 .expect("queued settings notifications should remain readable"); 1295 assert_eq!( 1296 personalities, 1297 vec![ 1298 Some(Personality::Friendly), 1299 Some(Personality::Pragmatic), 1300 Some(Personality::Friendly), 1301 Some(Personality::Pragmatic), 1302 ] 1303 ); 1304 1305 client.shutdown().await.expect("shutdown should complete"); 1306 } 1307 1308 #[tokio::test] 1309 async fn remote_platform_metadata_preserves_reported_and_missing_values() { 1310 for (family, os) in [ 1311 (Some("windows"), Some("windows")), 1312 (Some("unix"), Some("linux")), 1313 (Some("future-family"), Some("future-os")), 1314 (Some("unix"), None), 1315 (None, Some("linux")), 1316 (None, None), 1317 ] { 1318 let websocket_url = start_test_remote_server(move |mut websocket| async move { 1319 let mut metadata = serde_json::json!({}); 1320 if let Some(family) = family { 1321 metadata["platformFamily"] = family.into(); 1322 } 1323 if let Some(os) = os { 1324 metadata["platformOs"] = os.into(); 1325 } 1326 expect_remote_initialize_with_metadata(&mut websocket, metadata).await; 1327 websocket.close(None).await.expect("close should succeed"); 1328 }) 1329 .await; 1330 let client = AppServerClient::Remote( 1331 RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1332 .await 1333 .expect("remote client should connect"), 1334 ); 1335 assert_eq!( 1336 (client.platform_family(), client.platform_os()), 1337 (family, os) 1338 ); 1339 client.shutdown().await.expect("shutdown should complete"); 1340 } 1341 } 1342 1343 #[tokio::test] 1344 async fn remote_typed_request_roundtrip_works() { 1345 let websocket_url = start_test_remote_server(|mut websocket| async move { 1346 expect_remote_initialize(&mut websocket).await; 1347 let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await 1348 else { 1349 panic!("expected account/read request"); 1350 }; 1351 assert_eq!(request.method, "account/read"); 1352 write_websocket_message( 1353 &mut websocket, 1354 JSONRPCMessage::Response(JSONRPCResponse { 1355 id: request.id, 1356 result: serde_json::to_value(GetAccountResponse { 1357 workspace_routing: None, 1358 account: None, 1359 requires_openai_auth: false, 1360 }) 1361 .expect("response should serialize"), 1362 }), 1363 ) 1364 .await; 1365 websocket.close(None).await.expect("close should succeed"); 1366 }) 1367 .await; 1368 let client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1369 .await 1370 .expect("remote client should connect"); 1371 1372 assert_eq!(client.server_version(), Some("9.8.7-test")); 1373 assert_eq!(client.codex_home(), Some("/server/.codex")); 1374 let response: GetAccountResponse = client 1375 .request_typed(ClientRequest::GetAccount { 1376 request_id: RequestId::Integer(1), 1377 params: codex_app_server_protocol::GetAccountParams { 1378 refresh_token: false, 1379 }, 1380 }) 1381 .await 1382 .expect("typed request should succeed"); 1383 assert_eq!(response.account, None); 1384 1385 client.shutdown().await.expect("shutdown should complete"); 1386 } 1387 1388 #[tokio::test] 1389 async fn remote_unix_socket_typed_request_roundtrip_works() { 1390 let socket_dir = TempDir::new().expect("socket dir"); 1391 let socket_path = AbsolutePathBuf::from_absolute_path(socket_dir.path().join("codex.sock")) 1392 .expect("socket path should resolve"); 1393 let mut listener = UnixListener::bind(socket_path.as_path()) 1394 .await 1395 .expect("listener should bind"); 1396 tokio::spawn(async move { 1397 let stream = listener.accept().await.expect("accept should succeed"); 1398 let mut websocket = accept_async(stream) 1399 .await 1400 .expect("websocket upgrade should succeed"); 1401 expect_remote_initialize(&mut websocket).await; 1402 let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await 1403 else { 1404 panic!("expected account/read request"); 1405 }; 1406 assert_eq!(request.method, "account/read"); 1407 write_websocket_message( 1408 &mut websocket, 1409 JSONRPCMessage::Response(JSONRPCResponse { 1410 id: request.id, 1411 result: serde_json::to_value(GetAccountResponse { 1412 workspace_routing: None, 1413 account: None, 1414 requires_openai_auth: false, 1415 }) 1416 .expect("response should serialize"), 1417 }), 1418 ) 1419 .await; 1420 websocket.close(None).await.expect("close should succeed"); 1421 }); 1422 let client = RemoteAppServerClient::connect(RemoteAppServerConnectArgs { 1423 endpoint: RemoteAppServerEndpoint::UnixSocket { socket_path }, 1424 client_name: "codex-app-server-client-test".to_string(), 1425 client_version: "0.0.0-test".to_string(), 1426 experimental_api: true, 1427 mcp_server_openai_form_elicitation: false, 1428 opt_out_notification_methods: Vec::new(), 1429 channel_capacity: 8, 1430 }) 1431 .await 1432 .expect("remote client should connect"); 1433 1434 let response: GetAccountResponse = client 1435 .request_typed(ClientRequest::GetAccount { 1436 request_id: RequestId::Integer(1), 1437 params: codex_app_server_protocol::GetAccountParams { 1438 refresh_token: false, 1439 }, 1440 }) 1441 .await 1442 .expect("typed request should succeed"); 1443 assert_eq!(response.account, None); 1444 1445 client.shutdown().await.expect("shutdown should complete"); 1446 } 1447 1448 #[tokio::test] 1449 async fn remote_typed_request_accepts_large_single_frame_response() { 1450 let padding = "x".repeat((17 << 20) + 1024); 1451 let websocket_url = start_test_remote_server(move |mut websocket| async move { 1452 expect_remote_initialize(&mut websocket).await; 1453 let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await 1454 else { 1455 panic!("expected account/read request"); 1456 }; 1457 assert_eq!(request.method, "account/read"); 1458 write_websocket_message( 1459 &mut websocket, 1460 JSONRPCMessage::Response(JSONRPCResponse { 1461 id: request.id, 1462 result: serde_json::json!({ 1463 "account": null, 1464 "requiresOpenaiAuth": false, 1465 "padding": padding, 1466 }), 1467 }), 1468 ) 1469 .await; 1470 websocket.close(None).await.expect("close should succeed"); 1471 }) 1472 .await; 1473 let client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1474 .await 1475 .expect("remote client should connect"); 1476 1477 let response: GetAccountResponse = client 1478 .request_typed(ClientRequest::GetAccount { 1479 request_id: RequestId::Integer(1), 1480 params: codex_app_server_protocol::GetAccountParams { 1481 refresh_token: false, 1482 }, 1483 }) 1484 .await 1485 .expect("large typed request should succeed"); 1486 assert_eq!( 1487 response, 1488 GetAccountResponse { 1489 workspace_routing: None, 1490 account: None, 1491 requires_openai_auth: false, 1492 } 1493 ); 1494 1495 client.shutdown().await.expect("shutdown should complete"); 1496 } 1497 1498 #[tokio::test] 1499 async fn remote_connect_includes_auth_header_when_configured() { 1500 let auth_token = "remote-bearer-token".to_string(); 1501 let websocket_url = start_test_remote_server_with_auth( 1502 Some(auth_token.clone()), 1503 |mut websocket| async move { 1504 expect_remote_initialize(&mut websocket).await; 1505 websocket.close(None).await.expect("close should succeed"); 1506 }, 1507 ) 1508 .await; 1509 let client = RemoteAppServerClient::connect(RemoteAppServerConnectArgs { 1510 endpoint: RemoteAppServerEndpoint::WebSocket { 1511 websocket_url, 1512 auth_token: Some(auth_token), 1513 }, 1514 client_name: "codex-app-server-client-test".to_string(), 1515 client_version: "0.0.0-test".to_string(), 1516 experimental_api: true, 1517 mcp_server_openai_form_elicitation: false, 1518 opt_out_notification_methods: Vec::new(), 1519 channel_capacity: 8, 1520 }) 1521 .await 1522 .expect("remote client should connect"); 1523 1524 client.shutdown().await.expect("shutdown should complete"); 1525 } 1526 1527 #[tokio::test] 1528 async fn remote_connect_rejects_non_loopback_ws_when_auth_configured() { 1529 let result = RemoteAppServerClient::connect(RemoteAppServerConnectArgs { 1530 endpoint: RemoteAppServerEndpoint::WebSocket { 1531 websocket_url: "ws://example.com:4500".to_string(), 1532 auth_token: Some("remote-bearer-token".to_string()), 1533 }, 1534 client_name: "codex-app-server-client-test".to_string(), 1535 client_version: "0.0.0-test".to_string(), 1536 experimental_api: true, 1537 mcp_server_openai_form_elicitation: false, 1538 opt_out_notification_methods: Vec::new(), 1539 channel_capacity: 8, 1540 }) 1541 .await; 1542 let err = match result { 1543 Ok(_) => panic!("non-loopback ws should be rejected before connect"), 1544 Err(err) => err, 1545 }; 1546 assert_eq!(err.kind(), ErrorKind::InvalidInput); 1547 assert!( 1548 err.to_string() 1549 .contains("remote auth tokens require `wss://` or loopback `ws://` URLs") 1550 ); 1551 } 1552 1553 #[test] 1554 fn remote_auth_token_transport_policy_allows_wss_and_loopback_ws() { 1555 assert!(crate::remote::websocket_url_supports_auth_token( 1556 &url::Url::parse("wss://example.com:443").expect("wss URL should parse") 1557 )); 1558 assert!(crate::remote::websocket_url_supports_auth_token( 1559 &url::Url::parse("ws://127.0.0.1:4500").expect("loopback ws URL should parse") 1560 )); 1561 assert!(!crate::remote::websocket_url_supports_auth_token( 1562 &url::Url::parse("ws://example.com:4500").expect("non-loopback ws URL should parse") 1563 )); 1564 } 1565 1566 #[tokio::test] 1567 async fn remote_duplicate_request_id_keeps_original_waiter() { 1568 let (first_request_seen_tx, first_request_seen_rx) = tokio::sync::oneshot::channel(); 1569 let websocket_url = start_test_remote_server(|mut websocket| async move { 1570 expect_remote_initialize(&mut websocket).await; 1571 let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await 1572 else { 1573 panic!("expected account/read request"); 1574 }; 1575 assert_eq!(request.method, "account/read"); 1576 first_request_seen_tx 1577 .send(request.id.clone()) 1578 .expect("request id should send"); 1579 assert!( 1580 timeout( 1581 Duration::from_millis(100), 1582 read_websocket_message(&mut websocket) 1583 ) 1584 .await 1585 .is_err(), 1586 "duplicate request should not be forwarded to the server" 1587 ); 1588 write_websocket_message( 1589 &mut websocket, 1590 JSONRPCMessage::Response(JSONRPCResponse { 1591 id: request.id, 1592 result: serde_json::to_value(GetAccountResponse { 1593 workspace_routing: None, 1594 account: None, 1595 requires_openai_auth: false, 1596 }) 1597 .expect("response should serialize"), 1598 }), 1599 ) 1600 .await; 1601 let _ = websocket.next().await; 1602 }) 1603 .await; 1604 let client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1605 .await 1606 .expect("remote client should connect"); 1607 let first_request_handle = client.request_handle(); 1608 let second_request_handle = first_request_handle.clone(); 1609 1610 let first_request = tokio::spawn(async move { 1611 first_request_handle 1612 .request_typed::<GetAccountResponse>(ClientRequest::GetAccount { 1613 request_id: RequestId::Integer(1), 1614 params: codex_app_server_protocol::GetAccountParams { 1615 refresh_token: false, 1616 }, 1617 }) 1618 .await 1619 }); 1620 1621 let first_request_id = first_request_seen_rx 1622 .await 1623 .expect("server should observe the first request"); 1624 assert_eq!(first_request_id, RequestId::Integer(1)); 1625 1626 let second_err = second_request_handle 1627 .request_typed::<GetAccountResponse>(ClientRequest::GetAccount { 1628 request_id: RequestId::Integer(1), 1629 params: codex_app_server_protocol::GetAccountParams { 1630 refresh_token: false, 1631 }, 1632 }) 1633 .await 1634 .expect_err("duplicate request id should be rejected"); 1635 assert_eq!( 1636 second_err.to_string(), 1637 "account/read transport error: duplicate remote app-server request id `1`" 1638 ); 1639 1640 let first_response = first_request 1641 .await 1642 .expect("first request task should join") 1643 .expect("first request should succeed"); 1644 assert_eq!( 1645 first_response, 1646 GetAccountResponse { 1647 workspace_routing: None, 1648 account: None, 1649 requires_openai_auth: false, 1650 } 1651 ); 1652 1653 client.shutdown().await.expect("shutdown should complete"); 1654 } 1655 1656 #[tokio::test] 1657 async fn remote_notifications_arrive_over_websocket() { 1658 let websocket_url = start_test_remote_server(|mut websocket| async move { 1659 expect_remote_initialize(&mut websocket).await; 1660 write_websocket_message( 1661 &mut websocket, 1662 JSONRPCMessage::Notification( 1663 serde_json::from_value( 1664 serde_json::to_value(ServerNotification::AccountUpdated( 1665 AccountUpdatedNotification { 1666 auth_mode: None, 1667 plan_type: None, 1668 }, 1669 )) 1670 .expect("notification should serialize"), 1671 ) 1672 .expect("notification should convert to JSON-RPC"), 1673 ), 1674 ) 1675 .await; 1676 }) 1677 .await; 1678 let mut client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1679 .await 1680 .expect("remote client should connect"); 1681 1682 let event = client.next_event().await.expect("event should arrive"); 1683 assert!(matches!( 1684 event, 1685 AppServerEvent::ServerNotification(notification) 1686 if matches!(notification.as_ref(), ServerNotification::AccountUpdated(_)) 1687 )); 1688 1689 client.shutdown().await.expect("shutdown should complete"); 1690 } 1691 1692 #[tokio::test] 1693 async fn remote_backpressure_preserves_transcript_notifications() { 1694 let (done_tx, done_rx) = tokio::sync::oneshot::channel(); 1695 let websocket_url = start_test_remote_server(|mut websocket| async move { 1696 expect_remote_initialize(&mut websocket).await; 1697 for notification in [ 1698 command_execution_output_delta_notification("stdout-1"), 1699 command_execution_output_delta_notification("stdout-2"), 1700 agent_message_delta_notification("hello"), 1701 item_completed_notification("hello"), 1702 turn_completed_notification(), 1703 ] { 1704 write_websocket_message( 1705 &mut websocket, 1706 JSONRPCMessage::Notification( 1707 serde_json::from_value( 1708 serde_json::to_value(notification) 1709 .expect("notification should serialize"), 1710 ) 1711 .expect("notification should convert to JSON-RPC"), 1712 ), 1713 ) 1714 .await; 1715 } 1716 let _ = done_rx.await; 1717 }) 1718 .await; 1719 let mut client = RemoteAppServerClient::connect(RemoteAppServerConnectArgs { 1720 channel_capacity: 1, 1721 ..test_remote_connect_args(websocket_url) 1722 }) 1723 .await 1724 .expect("remote client should connect"); 1725 1726 let first_event = timeout(Duration::from_secs(2), client.next_event()) 1727 .await 1728 .expect("first event should arrive before timeout") 1729 .expect("event stream should stay open"); 1730 assert!(matches!( 1731 first_event, 1732 AppServerEvent::ServerNotification(notification) 1733 if matches!( 1734 notification.as_ref(), 1735 ServerNotification::CommandExecutionOutputDelta(notification) 1736 if notification.delta == "stdout-1" 1737 ) 1738 )); 1739 1740 let mut remaining_events = Vec::new(); 1741 for _ in 0..4 { 1742 remaining_events.push( 1743 timeout(Duration::from_secs(2), client.next_event()) 1744 .await 1745 .expect("event should arrive before timeout") 1746 .expect("event stream should stay open"), 1747 ); 1748 } 1749 1750 let mut transcript_event_names = Vec::new(); 1751 for event in &remaining_events { 1752 match event { 1753 AppServerEvent::Lagged { skipped: 1 } => {} 1754 AppServerEvent::ServerNotification(notification) => match notification.as_ref() { 1755 ServerNotification::CommandExecutionOutputDelta(notification) 1756 if notification.delta == "stdout-2" => {} 1757 ServerNotification::AgentMessageDelta(notification) 1758 if notification.delta == "hello" => 1759 { 1760 transcript_event_names.push("agent_message_delta"); 1761 } 1762 ServerNotification::ItemCompleted(notification) 1763 if matches!( 1764 ¬ification.item, 1765 codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } 1766 if text == "hello" 1767 ) => 1768 { 1769 transcript_event_names.push("item_completed"); 1770 } 1771 ServerNotification::TurnCompleted(notification) 1772 if notification.turn.status 1773 == codex_app_server_protocol::TurnStatus::Completed => 1774 { 1775 transcript_event_names.push("turn_completed"); 1776 } 1777 _ => panic!("unexpected remaining event: {event:?}"), 1778 }, 1779 _ => panic!("unexpected remaining event: {event:?}"), 1780 } 1781 } 1782 assert_eq!( 1783 transcript_event_names, 1784 vec!["agent_message_delta", "item_completed", "turn_completed"] 1785 ); 1786 1787 done_tx 1788 .send(()) 1789 .expect("server completion signal should send"); 1790 client.shutdown().await.expect("shutdown should complete"); 1791 } 1792 1793 #[tokio::test] 1794 async fn remote_server_request_resolution_roundtrip_works() { 1795 let websocket_url = start_test_remote_server(|mut websocket| async move { 1796 expect_remote_initialize(&mut websocket).await; 1797 let request_id = RequestId::String("srv-1".to_string()); 1798 let server_request = JSONRPCRequest { 1799 id: request_id.clone(), 1800 method: "item/tool/requestUserInput".to_string(), 1801 params: Some( 1802 serde_json::to_value(ToolRequestUserInputParams { 1803 thread_id: "thread-1".to_string(), 1804 turn_id: "turn-1".to_string(), 1805 item_id: "call-1".to_string(), 1806 questions: vec![ToolRequestUserInputQuestion { 1807 id: "question-1".to_string(), 1808 header: "Mode".to_string(), 1809 question: "Pick one".to_string(), 1810 is_other: false, 1811 is_secret: false, 1812 options: Some(vec![]), 1813 }], 1814 is_blocking: true, 1815 auto_resolution_ms: None, 1816 }) 1817 .expect("params should serialize"), 1818 ), 1819 trace: None, 1820 }; 1821 write_websocket_message(&mut websocket, JSONRPCMessage::Request(server_request)).await; 1822 1823 let JSONRPCMessage::Response(response) = read_websocket_message(&mut websocket).await 1824 else { 1825 panic!("expected server request response"); 1826 }; 1827 assert_eq!(response.id, request_id); 1828 }) 1829 .await; 1830 let mut client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1831 .await 1832 .expect("remote client should connect"); 1833 1834 let AppServerEvent::ServerRequest(request) = client 1835 .next_event() 1836 .await 1837 .expect("request event should arrive") 1838 else { 1839 panic!("expected server request event"); 1840 }; 1841 client 1842 .resolve_server_request(request.id().clone(), serde_json::json!({})) 1843 .await 1844 .expect("server request should resolve"); 1845 1846 client.shutdown().await.expect("shutdown should complete"); 1847 } 1848 1849 #[tokio::test] 1850 async fn remote_server_request_received_during_initialize_is_delivered() { 1851 let websocket_url = start_test_remote_server(|mut websocket| async move { 1852 let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await 1853 else { 1854 panic!("expected initialize request"); 1855 }; 1856 assert_eq!(request.method, "initialize"); 1857 1858 let request_id = RequestId::String("srv-init".to_string()); 1859 write_websocket_message( 1860 &mut websocket, 1861 JSONRPCMessage::Request(JSONRPCRequest { 1862 id: request_id.clone(), 1863 method: "item/tool/requestUserInput".to_string(), 1864 params: Some( 1865 serde_json::to_value(ToolRequestUserInputParams { 1866 thread_id: "thread-1".to_string(), 1867 turn_id: "turn-1".to_string(), 1868 item_id: "call-1".to_string(), 1869 questions: vec![ToolRequestUserInputQuestion { 1870 id: "question-1".to_string(), 1871 header: "Mode".to_string(), 1872 question: "Pick one".to_string(), 1873 is_other: false, 1874 is_secret: false, 1875 options: Some(vec![]), 1876 }], 1877 is_blocking: true, 1878 auto_resolution_ms: None, 1879 }) 1880 .expect("params should serialize"), 1881 ), 1882 trace: None, 1883 }), 1884 ) 1885 .await; 1886 write_websocket_message( 1887 &mut websocket, 1888 JSONRPCMessage::Response(JSONRPCResponse { 1889 id: request.id, 1890 result: serde_json::json!({}), 1891 }), 1892 ) 1893 .await; 1894 1895 let JSONRPCMessage::Notification(notification) = 1896 read_websocket_message(&mut websocket).await 1897 else { 1898 panic!("expected initialized notification"); 1899 }; 1900 assert_eq!(notification.method, "initialized"); 1901 1902 let JSONRPCMessage::Response(response) = read_websocket_message(&mut websocket).await 1903 else { 1904 panic!("expected server request response"); 1905 }; 1906 assert_eq!(response.id, request_id); 1907 }) 1908 .await; 1909 let mut client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1910 .await 1911 .expect("remote client should connect"); 1912 1913 let AppServerEvent::ServerRequest(request) = client 1914 .next_event() 1915 .await 1916 .expect("request event should arrive") 1917 else { 1918 panic!("expected server request event"); 1919 }; 1920 client 1921 .resolve_server_request(request.id().clone(), serde_json::json!({})) 1922 .await 1923 .expect("server request should resolve"); 1924 1925 client.shutdown().await.expect("shutdown should complete"); 1926 } 1927 1928 #[tokio::test] 1929 async fn remote_unknown_server_request_is_rejected() { 1930 let websocket_url = start_test_remote_server(|mut websocket| async move { 1931 expect_remote_initialize(&mut websocket).await; 1932 let request_id = RequestId::String("srv-unknown".to_string()); 1933 write_websocket_message( 1934 &mut websocket, 1935 JSONRPCMessage::Request(JSONRPCRequest { 1936 id: request_id.clone(), 1937 method: "thread/unknown".to_string(), 1938 params: None, 1939 trace: None, 1940 }), 1941 ) 1942 .await; 1943 1944 let JSONRPCMessage::Error(response) = read_websocket_message(&mut websocket).await 1945 else { 1946 panic!("expected JSON-RPC error response"); 1947 }; 1948 assert_eq!(response.id, request_id); 1949 assert_eq!(response.error.code, -32601); 1950 assert_eq!( 1951 response.error.message, 1952 "unsupported remote app-server request `thread/unknown`" 1953 ); 1954 }) 1955 .await; 1956 let client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1957 .await 1958 .expect("remote client should connect"); 1959 1960 client.shutdown().await.expect("shutdown should complete"); 1961 } 1962 1963 #[tokio::test] 1964 async fn remote_disconnect_surfaces_as_event() { 1965 let websocket_url = start_test_remote_server(|mut websocket| async move { 1966 expect_remote_initialize(&mut websocket).await; 1967 websocket.close(None).await.expect("close should succeed"); 1968 }) 1969 .await; 1970 let mut client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url)) 1971 .await 1972 .expect("remote client should connect"); 1973 1974 let event = client 1975 .next_event() 1976 .await 1977 .expect("disconnect event should arrive"); 1978 assert!(matches!(event, AppServerEvent::Disconnected { .. })); 1979 } 1980 1981 #[test] 1982 fn typed_request_error_exposes_sources() { 1983 let transport = TypedRequestError::Transport { 1984 method: "config/read".to_string(), 1985 source: IoError::new(ErrorKind::BrokenPipe, "closed"), 1986 }; 1987 assert_eq!(std::error::Error::source(&transport).is_some(), true); 1988 1989 let server = TypedRequestError::Server { 1990 method: "thread/read".to_string(), 1991 source: JSONRPCErrorError { 1992 code: -32603, 1993 data: Some(serde_json::json!({"detail": "config lock mismatch"})), 1994 message: "internal".to_string(), 1995 }, 1996 }; 1997 assert_eq!(std::error::Error::source(&server).is_some(), false); 1998 assert_eq!( 1999 server.to_string(), 2000 "thread/read failed: internal (code -32603), data: {\"detail\":\"config lock mismatch\"}" 2001 ); 2002 2003 let deserialize = TypedRequestError::Deserialize { 2004 method: "thread/start".to_string(), 2005 source: serde_json::from_str::<u32>("\"nope\"") 2006 .expect_err("invalid integer should return deserialize error"), 2007 }; 2008 assert_eq!(std::error::Error::source(&deserialize).is_some(), true); 2009 } 2010 2011 #[tokio::test] 2012 async fn next_event_surfaces_lagged_markers() { 2013 let (command_tx, _command_rx) = mpsc::channel(1); 2014 let (event_tx, event_rx) = mpsc::unbounded_channel(); 2015 let worker_handle = tokio::spawn(async {}); 2016 event_tx 2017 .send(InProcessServerEvent::Lagged { skipped: 3 }) 2018 .expect("lagged marker should enqueue"); 2019 drop(event_tx); 2020 2021 let mut client = InProcessAppServerClient { 2022 command_tx, 2023 event_rx, 2024 worker_handle, 2025 }; 2026 2027 let event = timeout(Duration::from_secs(2), client.next_event()) 2028 .await 2029 .expect("lagged marker should arrive before timeout"); 2030 assert!(matches!( 2031 event, 2032 Some(InProcessServerEvent::Lagged { skipped: 3 }) 2033 )); 2034 2035 client.shutdown().await.expect("shutdown should complete"); 2036 } 2037 2038 #[tokio::test] 2039 async fn runtime_start_args_forward_environment_manager_and_openai_form_capability() { 2040 let config = Arc::new(build_test_config().await); 2041 let environment_manager = Arc::new( 2042 EnvironmentManager::create_for_tests( 2043 Some("ws://127.0.0.1:8765".to_string()), 2044 Some( 2045 ExecServerRuntimePaths::new( 2046 std::env::current_exe().expect("current exe"), 2047 /*codex_linux_sandbox_exe*/ None, 2048 ) 2049 .expect("runtime paths"), 2050 ), 2051 ) 2052 .await, 2053 ); 2054 2055 let runtime_args = InProcessClientStartArgs { 2056 arg0_paths: Arg0DispatchPaths::default(), 2057 config: config.clone(), 2058 cli_overrides: Vec::new(), 2059 loader_overrides: LoaderOverrides::default(), 2060 strict_config: false, 2061 cloud_config_bundle: CloudConfigBundleLoader::default(), 2062 feedback: CodexFeedback::new(), 2063 log_db: None, 2064 state_db: None, 2065 environment_manager: environment_manager.clone(), 2066 config_warnings: Vec::new(), 2067 session_source: SessionSource::Exec, 2068 enable_codex_api_key_env: false, 2069 client_name: "codex-app-server-client-test".to_string(), 2070 client_version: "0.0.0-test".to_string(), 2071 experimental_api: true, 2072 mcp_server_openai_form_elicitation: true, 2073 opt_out_notification_methods: Vec::new(), 2074 channel_capacity: DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, 2075 } 2076 .into_runtime_start_args(); 2077 2078 assert_eq!(runtime_args.config, config); 2079 assert!( 2080 runtime_args 2081 .initialize 2082 .capabilities 2083 .expect("initialize capabilities") 2084 .mcp_server_openai_form_elicitation 2085 ); 2086 assert!(Arc::ptr_eq( 2087 &runtime_args.environment_manager, 2088 &environment_manager 2089 )); 2090 assert!( 2091 runtime_args 2092 .environment_manager 2093 .default_environment() 2094 .expect("default environment") 2095 .is_remote() 2096 ); 2097 } 2098 2099 #[tokio::test] 2100 async fn shutdown_completes_promptly_without_retained_managers() { 2101 let client = start_test_client(SessionSource::Cli).await; 2102 2103 timeout(Duration::from_secs(1), client.shutdown()) 2104 .await 2105 .expect("shutdown should not wait for the 5s fallback timeout") 2106 .expect("shutdown should complete"); 2107 } 2108 2109 #[tokio::test(start_paused = true)] 2110 async fn shutdown_waits_for_in_process_drain() { 2111 use std::sync::atomic::AtomicBool; 2112 use std::sync::atomic::Ordering; 2113 2114 let (command_tx, mut command_rx) = mpsc::channel(1); 2115 let (_event_tx, event_rx) = mpsc::unbounded_channel(); 2116 let completed = Arc::new(AtomicBool::new(false)); 2117 let worker_completed = Arc::clone(&completed); 2118 let worker_handle = tokio::spawn(async move { 2119 let response_tx = match command_rx.recv().await { 2120 Some(ClientCommand::Shutdown { response_tx }) => response_tx, 2121 _ => panic!("expected shutdown command"), 2122 }; 2123 tokio::time::sleep(Duration::from_secs(30)).await; 2124 worker_completed.store(true, Ordering::Release); 2125 let _ = response_tx.send(Ok(())); 2126 }); 2127 let client = InProcessAppServerClient { 2128 command_tx, 2129 event_rx, 2130 worker_handle, 2131 }; 2132 2133 client.shutdown().await.expect("shutdown should complete"); 2134 assert!(completed.load(Ordering::Acquire)); 2135 } 2136 }