diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 0b1859f14a..a0d03a9de1 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -192,9 +192,9 @@ pub enum InProcessServerEvent { /// Internal message sent from [`InProcessClientHandle`] methods to the runtime task. /// -/// Requests carry a oneshot sender for the response; notifications and server-request -/// replies are fire-and-forget from the caller's perspective (transport errors are -/// caught by `try_send` on the outer channel). +/// Requests carry a oneshot sender for the response. Startup notifications can +/// acknowledge processing; ordinary notifications and server-request replies are +/// fire-and-forget (transport errors are caught by `try_send` on the outer channel). enum InProcessClientMessage { Request { request: Box, @@ -203,6 +203,7 @@ enum InProcessClientMessage { }, Notification { notification: ClientNotification, + processed_tx: Option>, }, ServerRequestResponse { request_id: RequestId, @@ -219,7 +220,10 @@ enum InProcessClientMessage { enum ProcessorCommand { Request(Box, tokio_util::sync::CancellationToken), - Notification(ClientNotification), + Notification { + notification: ClientNotification, + processed_tx: Option>, + }, } #[derive(Clone)] @@ -246,7 +250,10 @@ impl InProcessClientSender { } pub fn notify(&self, notification: ClientNotification) -> IoResult<()> { - self.try_send_client_message(InProcessClientMessage::Notification { notification }) + self.try_send_client_message(InProcessClientMessage::Notification { + notification, + processed_tx: None, + }) } pub fn respond_to_server_request(&self, request_id: RequestId, result: Result) -> IoResult<()> { @@ -379,9 +386,10 @@ impl InProcessClientHandle { /// Starts an in-process app-server runtime and performs initialize handshake. /// -/// This function sends `initialize` followed by `initialized` before returning -/// the handle, so callers receive a ready-to-use runtime. If initialize fails, -/// the runtime is shut down and an `InvalidData` error is returned. +/// This function waits for `initialize` and `initialized` to be processed before +/// returning the handle, so callers receive a ready-to-use runtime. Rejected +/// initialization shuts down the runtime and returns an `InvalidData` error. +/// Transport failures while processing `initialized` also shut down the runtime. pub async fn start(mut args: InProcessStartArgs) -> IoResult { if let Ok(Some(err)) = check_execpolicy_for_warnings(&args.config.config_layer_stack).await { let (path, range) = crate::exec_policy_warning_location(&err); @@ -409,7 +417,26 @@ pub async fn start(mut args: InProcessStartArgs) -> IoResult IoResult { + Some(ProcessorCommand::Notification { notification, processed_tx }) => { processor.process_client_notification(notification).await; + if let Some(processed_tx) = processed_tx { + let _ = processed_tx.send(()); + } } None => { break; @@ -678,8 +708,8 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult { - match processor_tx.try_send(ProcessorCommand::Notification(notification)) { + Some(InProcessClientMessage::Notification { notification, processed_tx }) => { + match processor_tx.try_send(ProcessorCommand::Notification { notification, processed_tx }) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(_)) => { warn!("dropping in-process client notification (queue full)"); @@ -1047,29 +1077,24 @@ mod tests { } } + #[test_case::test_case(0; "zero_capacity_is_clamped")] + #[test_case::test_case(1; "one_slot_capacity")] #[tokio::test] - async fn in_process_start_clamps_zero_channel_capacity() { + async fn in_process_start_accepts_immediate_request(channel_capacity: usize) { let client = start_test_client_with_options( SessionSource::Cli, - /*channel_capacity*/ 0, + channel_capacity, /*capabilities*/ None, ) .await; - let response = loop { - match client - .request(ClientRequest::ConfigRequirementsRead { - request_id: RequestId::Integer(4), - params: None, - }) - .await - { - Ok(response) => break response.expect("request should succeed"), - Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { - tokio::task::yield_now().await; - } - Err(err) => panic!("request transport should work: {err}"), - } - }; + let response = client + .request(ClientRequest::ConfigRequirementsRead { + request_id: RequestId::Integer(4), + params: None, + }) + .await + .expect("the first request should not compete with the initialize handshake") + .expect("request should succeed"); let _parsed: ConfigRequirementsReadResponse = serde_json::from_value(response).expect("response should match v2 schema"); client