diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 3054610330..d97a0ac6f4 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -3402,6 +3402,13 @@ "format": "int32", "type": "integer" }, + "outputDrainTimedOut": { + "description": "Whether output collection hit its post-exit deadline before seeing EOF or reaching the requested byte cap on either stream. Older servers omit this field.", + "type": [ + "boolean", + "null" + ] + }, "processHandle": { "description": "Client-supplied, connection-scoped `processHandle` from `process/spawn`.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index a0c7a30a13..f028a393cc 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -18117,6 +18117,13 @@ "format": "int32", "type": "integer" }, + "outputDrainTimedOut": { + "description": "Whether output collection hit its post-exit deadline before seeing EOF or reaching the requested byte cap on either stream. Older servers omit this field.", + "type": [ + "boolean", + "null" + ] + }, "processHandle": { "description": "Client-supplied, connection-scoped `processHandle` from `process/spawn`.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index c35df0e580..f39124d246 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -14109,6 +14109,13 @@ "format": "int32", "type": "integer" }, + "outputDrainTimedOut": { + "description": "Whether output collection hit its post-exit deadline before seeing EOF or reaching the requested byte cap on either stream. Older servers omit this field.", + "type": [ + "boolean", + "null" + ] + }, "processHandle": { "description": "Client-supplied, connection-scoped `processHandle` from `process/spawn`.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/json/v2/ProcessExitedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ProcessExitedNotification.json index 3a0a81d316..ba6696617b 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ProcessExitedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ProcessExitedNotification.json @@ -7,6 +7,13 @@ "format": "int32", "type": "integer" }, + "outputDrainTimedOut": { + "description": "Whether output collection hit its post-exit deadline before seeing EOF or reaching the requested byte cap on either stream. Older servers omit this field.", + "type": [ + "boolean", + "null" + ] + }, "processHandle": { "description": "Client-supplied, connection-scoped `processHandle` from `process/spawn`.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index a5d97cda12..22316631ce 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 065fefedd0..73e21e8ce4 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ProcessExitedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ProcessExitedNotification.ts index 0d82633421..40194f55bf 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ProcessExitedNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ProcessExitedNotification.ts @@ -14,6 +14,11 @@ processHandle: string, * Process exit code. */ exitCode: number, +/** + * Whether output collection hit its post-exit deadline before seeing EOF or reaching the + * requested byte cap on either stream. Older servers omit this field. + */ +outputDrainTimedOut?: boolean | null, /** * Buffered stdout capture. * diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 0e0ebe9e5d..07447b4190 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -2415,6 +2415,8 @@ mod tests { && matches!(field_prefix.trim(), "startedAtMs?" | "completedAtMs?")) || (path == Path::new("v2/McpServerOauthLoginCompletedNotification.ts") && field_prefix.trim() == "loginId?") + || (path == Path::new("v2/ProcessExitedNotification.ts") + && field_prefix.trim() == "outputDrainTimedOut?") || (path == Path::new("v2/MisalignmentErrorDetails.ts") && field_prefix.trim() == "reviewTarget?"); if field_prefix.chars().rev().find(|c| !c.is_whitespace()) == Some('?') diff --git a/codex-rs/app-server-protocol/src/protocol/v2/process.rs b/codex-rs/app-server-protocol/src/protocol/v2/process.rs index 11a8d687d4..600e5cc22a 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/process.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/process.rs @@ -183,6 +183,10 @@ pub struct ProcessExitedNotification { pub process_handle: String, /// Process exit code. pub exit_code: i32, + /// Whether output collection hit its post-exit deadline before seeing EOF or reaching the + /// requested byte cap on either stream. Older servers omit this field. + #[ts(optional, as = "Option>")] + pub output_drain_timed_out: Option, /// Buffered stdout capture. /// /// Empty when stdout was streamed via `process/outputDelta`. diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 27b9236f7f..c33bb6a634 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -1945,6 +1945,7 @@ fn process_notifications_round_trip() { let exited = ProcessExitedNotification { process_handle: "proc-1".to_string(), exit_code: 0, + output_drain_timed_out: Some(true), stdout: "out".to_string(), stdout_cap_reached: false, stderr: "err".to_string(), @@ -1956,15 +1957,25 @@ fn process_notifications_round_trip() { json!({ "processHandle": "proc-1", "exitCode": 0, + "outputDrainTimedOut": true, "stdout": "out", "stdoutCapReached": false, "stderr": "err", "stderrCapReached": true, }) ); - let decoded = serde_json::from_value::(value) + let decoded = serde_json::from_value::(value.clone()) .expect("deserialize process/exited"); assert_eq!(decoded, exited); + + let mut legacy = value; + legacy + .as_object_mut() + .unwrap() + .remove("outputDrainTimedOut"); + let decoded = serde_json::from_value::(legacy) + .expect("deserialize process/exited from an older server"); + assert_eq!(decoded.output_drain_timed_out, None); } #[test] diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index ffdc5bc29e..8073046e59 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -114,7 +114,9 @@ type PendingClientRequestResponse = std::result::Result bool { matches!( notification, - ServerNotification::TurnCompleted(_) + ServerNotification::ProcessOutputDelta(_) + | ServerNotification::ProcessExited(_) + | ServerNotification::TurnCompleted(_) | ServerNotification::ThreadQueueChanged(_) | ServerNotification::ThreadSettingsUpdated(_) | ServerNotification::ThreadAttachmentUpdated(_) @@ -460,10 +462,12 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult IoResult>::new(); let mut shutdown_ack = None; + let mut pending_event: Option<(InProcessServerEvent, Option>)> = None; loop { tokio::select! { @@ -693,7 +698,27 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult { + permit = event_tx.reserve(), if pending_event.is_some() => { + let Ok(permit) = permit else { break }; + let Some((event, written)) = pending_event.take() else { continue }; + permit.send(event); + if let Some(written) = written { + let _ = written.send(()); + } + } + queued_message = async { + tokio::select! { + reply = process_reply_rx.recv() => match reply { + Some(OutgoingEnvelope::ToConnection { message, write_complete_tx, .. }) => { + Some(QueuedOutgoingMessage { message, write_complete_tx }) + } + Some(OutgoingEnvelope::Broadcast { .. }) => unreachable!("replies are connection scoped"), + None => None, + }, + // Leave ordinary replies and events queued until the pending event fits. + event = writer_rx.recv(), if pending_event.is_none() => event, + } + } => { let Some(queued_message) = queued_message else { break; }; @@ -757,16 +782,13 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult { let notification = envelope.notification; if server_notification_requires_delivery(¬ification) { - if event_tx - .send(InProcessServerEvent::ServerNotification(Box::new( - notification, - ))) - .await - .is_err() - { - break; - } - } else if let Err(send_error) = + pending_event = Some(( + InProcessServerEvent::ServerNotification(Box::new(notification)), + queued_message.write_complete_tx, + )); + continue; + } + if let Err(send_error) = event_tx.try_send(InProcessServerEvent::ServerNotification( Box::new(notification), )) @@ -790,7 +812,9 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult, ) -> InProcessClientHandle { let codex_home = TempDir::new().expect("temp dir"); let config = Arc::new(build_test_config(codex_home.path()).await); @@ -904,7 +929,7 @@ mod tests { title: None, version: "0.0.0".to_string(), }, - capabilities: None, + capabilities, }, channel_capacity, }; @@ -914,7 +939,12 @@ mod tests { } async fn start_test_client(session_source: SessionSource) -> InProcessClientHandle { - start_test_client_with_capacity(session_source, DEFAULT_IN_PROCESS_CHANNEL_CAPACITY).await + start_test_client_with_options( + session_source, + DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + /*capabilities*/ None, + ) + .await } #[tokio::test] @@ -938,6 +968,47 @@ mod tests { .expect("in-process runtime should shutdown cleanly"); } + /// Importing a project makes its change visible to an embedder before the import reply. + #[tokio::test] + async fn project_import_delivers_change_before_reply() { + let mut client = start_test_client_with_options( + SessionSource::Cli, + DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + Some(codex_app_server_protocol::InitializeCapabilities { + experimental_api: true, + ..Default::default() + }), + ) + .await; + let params = codex_app_server_protocol::ProjectImportParams { + name: "Imported".into(), + roots: Vec::new(), + metadata: None, + threads: None, + idempotency_key: "test-import".into(), + }; + client + .request(ClientRequest::ProjectImport { + request_id: RequestId::Integer(1), + params, + }) + .await + .unwrap() + .unwrap(); + loop { + let event = client + .event_rx + .try_recv() + .expect("project/changed must already be queued"); + if let InProcessServerEvent::ServerNotification(notification) = event + && matches!(*notification, ServerNotification::ProjectChanged(_)) + { + break; + } + } + client.shutdown().await.unwrap(); + } + #[tokio::test] async fn in_process_start_uses_requested_session_source_for_thread_start() { for (requested_source, expected_source) in [ @@ -968,8 +1039,12 @@ mod tests { #[tokio::test] async fn in_process_start_clamps_zero_channel_capacity() { - let client = - start_test_client_with_capacity(SessionSource::Cli, /*channel_capacity*/ 0).await; + let client = start_test_client_with_options( + SessionSource::Cli, + /*channel_capacity*/ 0, + /*capabilities*/ None, + ) + .await; let response = loop { match client .request(ClientRequest::ConfigRequirementsRead { @@ -1104,6 +1179,156 @@ mod tests { )); } + #[cfg(unix)] + async fn spawn_streaming_test_process( + cwd: &Path, + handle: &str, + script: &str, + ) -> InProcessClientHandle { + use codex_app_server_protocol::InitializeCapabilities; + use codex_app_server_protocol::ProcessSpawnParams; + + let client = start_test_client_with_options( + SessionSource::Cli, + /*channel_capacity*/ 2, + Some(InitializeCapabilities { + experimental_api: true, + ..Default::default() + }), + ) + .await; + client + .request(ClientRequest::ProcessSpawn { + request_id: RequestId::Integer(1), + params: ProcessSpawnParams { + process_handle: handle.into(), + command: vec!["/bin/sh".into(), "-c".into(), script.into()], + cwd: cwd.try_into().unwrap(), + tty: false, + stream_stdin: false, + stream_stdout_stderr: true, + output_bytes_cap: Some(None), + timeout_ms: None, + env: None, + size: None, + }, + }) + .await + .unwrap() + .unwrap(); + client + } + + /// A kill after child exit must promptly report it has exited while the client is still + /// waiting to consume the complete download and its successful exit notification. + #[cfg(unix)] + #[tokio::test] + async fn process_output_and_exit_survive_a_full_in_process_event_queue() { + use base64::Engine; + use base64::engine::general_purpose::STANDARD; + use codex_app_server_protocol::ProcessKillParams; + + let dir = TempDir::new().unwrap(); + let contents: Vec = (0..128 * 1024u32).flat_map(u32::to_le_bytes).collect(); + std::fs::write(dir.path().join("input"), &contents).unwrap(); + let mut client = spawn_streaming_test_process( + dir.path(), + "download", + concat!( + "p=$$; (while kill -0 $p 2>/dev/null; do sleep 0.05; done; touch exited) ", + "/dev/null 2>&1 & exec cat input", + ), + ) + .await; + + timeout(Duration::from_secs(10), async { + // The pipe readers can hold this payload while delivery is stalled. + while client.event_rx.capacity() != 0 || !dir.path().join("exited").exists() { + tokio::task::yield_now().await; + } + let error = client + .request(ClientRequest::ProcessKill { + request_id: RequestId::Integer(2), + params: ProcessKillParams { + process_handle: "download".into(), + }, + }) + .await + .unwrap() + .unwrap_err(); + assert_eq!(error.message, "process \"download\" is no longer running"); + let mut received = Vec::new(); + loop { + if let InProcessServerEvent::ServerNotification(notification) = + client.next_event().await.unwrap() + { + match *notification { + ServerNotification::ProcessOutputDelta(delta) => { + received.extend(STANDARD.decode(delta.delta_base64).unwrap()); + } + ServerNotification::ProcessExited(exit) => { + assert_eq!(exit.exit_code, 0); + assert_eq!(received.len(), contents.len()); + assert_eq!(received, contents); + break; + } + _ => {} + } + } + } + }) + .await + .expect("process output and exit must both reach the consumer"); + client.shutdown().await.unwrap(); + } + + /// Unread output must not block process control replies or a subsequent kill on the same handle. + #[cfg(unix)] + #[tokio::test] + async fn in_process_kill_completes_while_process_output_is_stalled() { + use codex_app_server_protocol::ProcessKillParams; + use codex_app_server_protocol::ProcessResizePtyParams; + use codex_app_server_protocol::ProcessTerminalSize; + + let dir = TempDir::new().unwrap(); + let client = spawn_streaming_test_process( + dir.path(), + "writer", + "head -c 524288 /dev/zero; head -c 524288 /dev/zero >&2; touch written; exec sleep 30", + ) + .await; + timeout(Duration::from_secs(10), async { + while client.event_rx.capacity() != 0 || !dir.path().join("written").exists() { + tokio::task::yield_now().await; + } + // A resize on a pipe fails, but its error still has to release the handle for kill. + client + .request(ClientRequest::ProcessResizePty { + request_id: RequestId::Integer(2), + params: ProcessResizePtyParams { + process_handle: "writer".into(), + size: ProcessTerminalSize { rows: 24, cols: 80 }, + }, + }) + .await + .unwrap() + .unwrap_err(); + client + .request(ClientRequest::ProcessKill { + request_id: RequestId::Integer(3), + params: ProcessKillParams { + process_handle: "writer".into(), + }, + }) + .await + .unwrap() + .unwrap(); + }) + .await + .expect("kill must finish before the client resumes reading output"); + client.shutdown().await.unwrap(); + } + #[tokio::test] async fn in_process_shutdown_completes_with_full_request_and_event_queues() { let capacity = DEFAULT_IN_PROCESS_CHANNEL_CAPACITY; @@ -1132,10 +1357,10 @@ mod tests { .expect("thread/name/set transport should work") .expect("thread/name/set should succeed"); - // Drain through the first attachment notification to establish a clean boundary, - // then leave capacity + 1 notifications unread. Reuse one identity to stay below - // the attachment limit. Each mutation responds before sending its notification. - for index in 0..(capacity + 2) { + // Observe both mutation replies at the point their updates become visible, then + // leave capacity + 1 updates unread to exercise shutdown with full queues. + let sender = client.sender(); + for index in 0..(capacity + 3) { let request_id = RequestId::Integer((index + 3) as i64); let request = if index % 2 == 0 { ClientRequest::ThreadAttachmentAdd { @@ -1157,12 +1382,9 @@ mod tests { }, } }; - client - .request(request) - .await - .expect("attachment request transport should work") - .expect("attachment mutation should succeed"); - if index == 0 { + let mut reply = std::pin::pin!(sender.request(request)); + if index < 2 { + assert!(reply.as_mut().now_or_never().is_none()); loop { let event = client .next_event() @@ -1174,26 +1396,33 @@ mod tests { ServerNotification::ThreadAttachmentUpdated(_) ) { + reply + .as_mut() + .now_or_never() + .expect("the attachment reply must be ready before its update") + .expect("attachment request transport should work") + .expect("attachment mutation should succeed"); break; } } + } else { + reply + .await + .expect("attachment request transport should work") + .expect("attachment mutation should succeed"); } } - let sender = client.sender(); + assert_eq!(client.event_rx.capacity(), 0); let mut queued_requests = Vec::with_capacity(capacity); for index in 0..capacity { let mut request = Box::pin(sender.request(ClientRequest::ConfigRequirementsRead { request_id: RequestId::Integer((1000 + index) as i64), params: None, })); - if index == 0 { - // Verify a normal read stalls behind the unread notification before - // filling the remaining request slots. - assert!(timeout(Duration::from_secs(1), &mut request).await.is_err()); - } else { - assert!(request.as_mut().now_or_never().is_none()); - } + // Fill the command queue on this single-thread runtime without giving the + // dispatcher a turn between submissions. + assert!(request.as_mut().now_or_never().is_none()); queued_requests.push(request); } assert_eq!(sender.client_tx.capacity(), 0); diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index f893d3d20e..35f5027f7c 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -66,6 +66,7 @@ pub(crate) struct RequestContext { request_id: ConnectionRequestId, pub(crate) cancellation: tokio_util::sync::CancellationToken, cancellation_scope: RequestCancellationScope, + is_process_request: bool, span: Span, parent_trace: Option, _diagnostics_guard: Arc, @@ -88,6 +89,10 @@ impl RequestContext { Self { request_id, cancellation: tokio_util::sync::CancellationToken::new(), + is_process_request: matches!( + method, + "process/spawn" | "process/writeStdin" | "process/resizePty" | "process/kill" + ), cancellation_scope: match method { "userVerification/status" | "userVerification/enroll" @@ -133,6 +138,7 @@ pub(crate) struct OutgoingMessageSender { verification_connections: Mutex>, next_server_request_id: AtomicI64, sender: mpsc::Sender, + process_reply_sender: mpsc::Sender, request_id_to_callback: Mutex>, /// Incoming requests that are still waiting on a final response or error. /// We keep them here because this is where responses, errors, and @@ -254,6 +260,7 @@ impl OutgoingMessageSender { verification_auth: OnceLock::new(), verification_connections: Mutex::new(HashSet::new()), next_server_request_id: AtomicI64::new(0), + process_reply_sender: sender.clone(), sender, request_id_to_callback: Mutex::new(HashMap::new()), request_contexts: Mutex::new(HashMap::new()), @@ -261,6 +268,14 @@ impl OutgoingMessageSender { } } + pub(crate) fn with_process_reply_sender( + mut self, + sender: mpsc::Sender, + ) -> Self { + self.process_reply_sender = sender; + self + } + pub(crate) async fn register_request_context(&self, request_context: RequestContext) { let mut request_contexts = self.request_contexts.lock().await; if request_contexts @@ -889,10 +904,23 @@ impl OutgoingMessageSender { message: OutgoingMessage, message_kind: &'static str, ) { - let send_fut = self.sender.send(OutgoingEnvelope::ToConnection { + // Process requests serialize by handle. A blocked resize reply would also block the next kill. + let sender = if request_context + .as_ref() + .is_some_and(|ctx| ctx.is_process_request) + { + &self.process_reply_sender + } else { + &self.sender + }; + // The spawn reply must reach the embedder before the collectors can publish output. + let (written_tx, written_rx) = (!sender.same_channel(&self.sender)) + .then(oneshot::channel) + .unzip(); + let send_fut = sender.send(OutgoingEnvelope::ToConnection { connection_id, message, - write_complete_tx: None, + write_complete_tx: written_tx, }); let send_result = if let Some(request_context) = request_context { send_fut.instrument(request_context.span()).await @@ -902,6 +930,8 @@ impl OutgoingMessageSender { if let Err(err) = send_result { warn!("failed to send {message_kind} to client: {err:?}"); + } else if let Some(written_rx) = written_rx { + let _ = written_rx.await; } } } diff --git a/codex-rs/app-server/src/request_processors/process_exec_processor.rs b/codex-rs/app-server/src/request_processors/process_exec_processor.rs index 91e6f94eef..e1f6b9d7ad 100644 --- a/codex-rs/app-server/src/request_processors/process_exec_processor.rs +++ b/codex-rs/app-server/src/request_processors/process_exec_processor.rs @@ -34,7 +34,7 @@ use codex_utils_pty::TerminalSize; use tokio::sync::Mutex; use tokio::sync::mpsc; use tokio::sync::oneshot; -use tokio::sync::watch; +use tokio::time::Instant; use tokio_util::sync::CancellationToken; use crate::error_code::internal_error; @@ -250,7 +250,8 @@ struct SpawnProcessOutputParams { connection_id: ConnectionId, process_handle: String, output_rx: mpsc::Receiver>, - stdio_timeout_rx: watch::Receiver, + reader: Option, + process_exited: CancellationToken, outgoing: Arc, stream: ProcessOutputStream, stream_output: bool, @@ -261,6 +262,7 @@ struct SpawnProcessOutputParams { struct ProcessOutputCapture { text: String, cap_reached: bool, + output_incomplete: bool, } impl ProcessExecManager { @@ -508,13 +510,14 @@ async fn run_process(params: RunProcessParams) { } = spawned; tokio::pin!(exit_rx); let mut expiration_outcome = None; - let (stdio_timeout_tx, stdio_timeout_rx) = watch::channel(false); + let process_exited = CancellationToken::new(); let stdout_handle = collect_spawn_process_output(SpawnProcessOutputParams { connection_id: request_id.connection_id, process_handle: process_handle.clone(), output_rx: stdout_rx, - stdio_timeout_rx: stdio_timeout_rx.clone(), + reader: session.stdout_reader(), + process_exited: process_exited.clone(), outgoing: Arc::clone(&outgoing), stream: ProcessOutputStream::Stdout, stream_output: stream_stdout_stderr, @@ -524,7 +527,8 @@ async fn run_process(params: RunProcessParams) { connection_id: request_id.connection_id, process_handle: process_handle.clone(), output_rx: stderr_rx, - stdio_timeout_rx, + reader: session.stderr_reader(), + process_exited: process_exited.clone(), outgoing: Arc::clone(&outgoing), stream: ProcessOutputStream::Stderr, stream_output: stream_stdout_stderr, @@ -582,22 +586,17 @@ async fn run_process(params: RunProcessParams) { } }; - // Give stdout/stderr readers a bounded grace period to drain after process exit. - let timeout_handle = tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(IO_DRAIN_TIMEOUT_MS)).await; - let _ = stdio_timeout_tx.send(true); - }); - + drop(control_rx); + process_exited.cancel(); let stdout = stdout_handle.await.unwrap_or_default(); let stderr = stderr_handle.await.unwrap_or_default(); - timeout_handle.abort(); - outgoing .send_server_notification_to_connection_and_wait( request_id.connection_id, ServerNotification::ProcessExited(ProcessExitedNotification { process_handle, exit_code, + output_drain_timed_out: Some(stdout.output_incomplete || stderr.output_incomplete), stdout: stdout.text, stdout_cap_reached: stdout.cap_reached, stderr: stderr.text, @@ -605,6 +604,10 @@ async fn run_process(params: RunProcessParams) { }), ) .await; + + // On older Windows, ClosePseudoConsole can block until surviving clients disconnect. + #[cfg(windows)] + tokio::task::spawn_blocking(move || drop(session)); } fn collect_spawn_process_output( @@ -614,7 +617,8 @@ fn collect_spawn_process_output( connection_id, process_handle, mut output_rx, - mut stdio_timeout_rx, + reader, + process_exited, outgoing, stream, stream_output, @@ -624,14 +628,33 @@ fn collect_spawn_process_output( let mut buffer: Vec = Vec::new(); let mut observed_num_bytes = 0usize; let mut cap_reached = false; + let mut drain_timed_out = false; + let mut draining = false; + let drain_timer = tokio::time::sleep(Duration::ZERO); + tokio::pin!(drain_timer); loop { - let mut chunk = tokio::select! { - chunk = output_rx.recv() => match chunk { - Some(chunk) => chunk, - None => break, - }, - _ = stdio_timeout_rx.wait_for(|&v| v) => break, + let chunk = tokio::select! { + biased; + _ = process_exited.cancelled(), if !draining => { + draining = true; + drain_timer.as_mut().reset(Instant::now() + Duration::from_millis(IO_DRAIN_TIMEOUT_MS)); + continue; + } + _ = &mut drain_timer, if draining && !output_rx.is_closed() => { + drain_timed_out = !output_rx.is_closed(); + output_rx.close(); + if let Some(reader) = &reader { + reader.abort(); + } + continue; + } + chunk = output_rx.recv() => chunk, }; + let Some(mut chunk) = chunk else { break }; + // The byte cap limits delivery. Keep draining until EOF or the post-exit deadline. + if cap_reached { + continue; + } while chunk.len() < OUTPUT_CHUNK_SIZE_HINT && let Ok(next_chunk) = output_rx.try_recv() { @@ -649,6 +672,7 @@ fn collect_spawn_process_output( }; cap_reached = Some(observed_num_bytes) == output_bytes_cap; if stream_output { + let write_started = Instant::now(); outgoing .send_server_notification_to_connection_and_wait( connection_id, @@ -660,16 +684,19 @@ fn collect_spawn_process_output( }), ) .await; + // A slow client can take longer than the pipe drain budget to accept one chunk. + if draining { + let deadline = drain_timer.deadline() + write_started.elapsed(); + drain_timer.as_mut().reset(deadline); + } } else { buffer.extend_from_slice(capped_chunk); } - if cap_reached { - break; - } } ProcessOutputCapture { text: bytes_to_string_smart(&buffer), cap_reached, + output_incomplete: drain_timed_out && output_bytes_cap != Some(observed_num_bytes), } }) } @@ -732,3 +759,7 @@ fn no_active_process_error(process_handle: &str) -> JSONRPCErrorError { fn process_no_longer_running_error(process_handle: &str) -> JSONRPCErrorError { invalid_request(format!("process {process_handle:?} is no longer running")) } + +#[cfg(test)] +#[path = "process_exec_processor_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/request_processors/process_exec_processor_tests.rs b/codex-rs/app-server/src/request_processors/process_exec_processor_tests.rs new file mode 100644 index 0000000000..0b694b1dd5 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/process_exec_processor_tests.rs @@ -0,0 +1,234 @@ +//! Coverage for post-exit output delivery over a backpressured connection. +#![cfg(test)] + +use super::*; +use crate::outgoing_message::OutgoingEnvelope; +use crate::outgoing_message::OutgoingMessage; +use codex_analytics::AnalyticsEventsClient; +use pretty_assertions::assert_eq; + +fn start_collector( + process_exited: CancellationToken, +) -> ( + mpsc::Sender>, + mpsc::Receiver, + tokio::task::JoinHandle, +) { + let (output_tx, output_rx) = mpsc::channel(2); + let (tx, notifications) = mpsc::channel(1); + let collector = collect_spawn_process_output(SpawnProcessOutputParams { + connection_id: ConnectionId(1), + process_handle: "test".to_string(), + output_rx, + reader: None, + process_exited, + outgoing: Arc::new(OutgoingMessageSender::new( + tx, + AnalyticsEventsClient::disabled(), + )), + stream: ProcessOutputStream::Stdout, + stream_output: true, + output_bytes_cap: None, + }); + (output_tx, notifications, collector) +} + +async fn read_delta( + notifications: &mut mpsc::Receiver, +) -> (Vec, oneshot::Sender<()>) { + let Some(OutgoingEnvelope::ToConnection { + message: OutgoingMessage::AppServerNotification(envelope), + write_complete_tx: Some(written), + .. + }) = notifications.recv().await + else { + panic!("expected an output notification"); + }; + let ServerNotification::ProcessOutputDelta(delta) = envelope.notification else { + panic!("expected an output delta"); + }; + (STANDARD.decode(delta.delta_base64).unwrap(), written) +} + +/// A client can take longer than the drain grace to accept each of several final chunks. +#[tokio::test(start_paused = true)] +async fn slow_writes_deliver_the_entire_tail_after_exit() { + let exited = CancellationToken::new(); + let (output, mut notifications, collector) = start_collector(exited.clone()); + output.send(vec![1]).await.unwrap(); + let (first, written) = read_delta(&mut notifications).await; + exited.cancel(); + output.send(vec![2]).await.unwrap(); + let slow_write = Duration::from_millis(IO_DRAIN_TIMEOUT_MS * 2); + tokio::time::advance(slow_write).await; + written.send(()).unwrap(); + + let (second, written) = read_delta(&mut notifications).await; + tokio::time::advance(slow_write).await; + // A descendant can still produce bytes while the previous write is blocked. + output.send(vec![3]).await.unwrap(); + written.send(()).unwrap(); + + let (third, written) = read_delta(&mut notifications).await; + drop(output); + written.send(()).unwrap(); + assert_eq!([first, second, third], [vec![1], vec![2], vec![3]]); + let capture = collector.await.unwrap(); + assert!(!capture.output_incomplete); +} + +/// A descendant that holds a pipe open gets one shared grace period across reads. +#[tokio::test(start_paused = true)] +async fn inherited_pipe_does_not_restart_the_grace_period_for_each_chunk() { + let exited = CancellationToken::new(); + exited.cancel(); + let (output, mut notifications, collector) = start_collector(exited); + tokio::task::yield_now().await; + let grace = Duration::from_millis(IO_DRAIN_TIMEOUT_MS); + tokio::time::advance(grace / 2).await; + output.send(vec![1]).await.unwrap(); + let (_, written) = read_delta(&mut notifications).await; + written.send(()).unwrap(); + + let capture = tokio::time::timeout(grace / 2 + Duration::from_millis(1), collector) + .await + .expect("the collector must finish within the original drain budget") + .unwrap(); + assert!(capture.output_incomplete); +} + +/// Even a capped stream closes its OS reader on a genuine timeout while its sibling stays open. +#[cfg(unix)] +#[tokio::test] +async fn timed_out_stderr_closes_while_stdout_is_still_open() { + let dir = tempfile::tempdir().unwrap(); + let mut child = codex_utils_pty::spawn_pipe_process_no_stdin( + "/bin/sh", + &[ + "-c".into(), + concat!( + "printf cap >&2; (while [ ! -f release ]; do sleep 0.05; done; ", + "if dd if=/dev/zero bs=65536 count=32 >&2 2>/dev/null; ", + "then printf open; else printf closed; fi) &", + ) + .into(), + ], + dir.path(), + &HashMap::new(), + &None, + &[], + ) + .await + .unwrap(); + let (tx, _notifications) = mpsc::channel(1); + let process_exited = CancellationToken::new(); + let collector = collect_spawn_process_output(SpawnProcessOutputParams { + connection_id: ConnectionId(1), + process_handle: "probe".into(), + output_rx: child.stderr_rx, + reader: child.session.stderr_reader(), + process_exited: process_exited.clone(), + outgoing: Arc::new(OutgoingMessageSender::new( + tx, + AnalyticsEventsClient::disabled(), + )), + stream: ProcessOutputStream::Stderr, + stream_output: false, + output_bytes_cap: Some(3), + }); + assert_eq!(child.exit_rx.await.unwrap(), 0); + process_exited.cancel(); + let capture = collector.await.unwrap(); + assert_eq!(capture.text, "cap"); + assert!(capture.cap_reached); + assert!(!capture.output_incomplete); + std::fs::write(dir.path().join("release"), b"").unwrap(); + let mut stdout = Vec::new(); + tokio::time::timeout(Duration::from_secs(10), async { + while let Some(bytes) = child.stdout_rx.recv().await { + stdout.extend(bytes); + } + }) + .await + .expect("the descendant must observe the closed stderr pipe"); + assert_eq!(stdout, b"closed"); +} + +/// The client must receive the root's exit while a surviving Windows console child can still run. +#[cfg(windows)] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn windows_terminal_reports_exit_before_closing_the_console() { + let dir = tempfile::tempdir().unwrap(); + // Bound the lifetime of the descendant even if the server fails to deliver the exit. + std::fs::write( + dir.path().join("child.cmd"), + concat!( + "@echo off\r\n", + "for /l %%i in (1,1,30) do (\r\n", + " if exist received goto ack\r\n", + " ping.exe -n 2 127.0.0.1 >nul\r\n", + ")\r\nexit /b 1\r\n:ack\r\ntype nul >acknowledged\r\n", + ), + ) + .unwrap(); + let spawned = codex_utils_pty::spawn_pty_process( + "cmd.exe", + &[ + "/d".into(), + "/c".into(), + "echo complete & start /b cmd.exe /d /c child.cmd".into(), + ], + dir.path(), + &std::env::vars().collect(), + /*arg0*/ &None, + TerminalSize::default(), + codex_utils_pty::ChildFds::Inherited(&[]), + ) + .await + .unwrap(); + let (_control_tx, control_rx) = mpsc::channel(1); + let (tx, mut notifications) = mpsc::channel(1); + let process = tokio::spawn(run_process(RunProcessParams { + outgoing: Arc::new(OutgoingMessageSender::new( + tx, + AnalyticsEventsClient::disabled(), + )), + request_id: ConnectionRequestId { + connection_id: ConnectionId(1), + request_id: codex_app_server_protocol::RequestId::Integer(1), + }, + process_handle: "console".into(), + spawned, + control_rx, + stream_stdin: true, + stream_stdout_stderr: false, + expiration: ExecExpiration::DefaultTimeout, + output_bytes_cap: None, + })); + let Some(OutgoingEnvelope::ToConnection { + message: OutgoingMessage::AppServerNotification(envelope), + write_complete_tx: Some(written), + .. + }) = tokio::time::timeout(Duration::from_secs(15), notifications.recv()) + .await + .expect("exit should arrive while the descendant is still attached") + else { + panic!("expected an exit notification") + }; + let ServerNotification::ProcessExited(exit) = envelope.notification else { + panic!("expected process/exited"); + }; + assert_eq!(exit.exit_code, 0); + assert!(exit.stdout.contains("complete")); + assert_eq!(exit.output_drain_timed_out, Some(true)); + std::fs::write(dir.path().join("received"), b"").unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + while !dir.path().join("acknowledged").exists() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("the attached child should still be alive when the client receives process/exited"); + written.send(()).unwrap(); + process.await.unwrap(); +} diff --git a/codex-rs/app-server/tests/suite/v2/process_exec.rs b/codex-rs/app-server/tests/suite/v2/process_exec.rs index 9ac5bc28de..6deeb8a422 100644 --- a/codex-rs/app-server/tests/suite/v2/process_exec.rs +++ b/codex-rs/app-server/tests/suite/v2/process_exec.rs @@ -102,6 +102,7 @@ async fn process_spawn_returns_before_exit_and_emits_exit_notification() -> Resu ProcessExitedNotification { process_handle, exit_code: 0, + output_drain_timed_out: Some(false), stdout: "process-out||".to_string(), stdout_cap_reached: false, stderr: "process-err".to_string(), @@ -111,6 +112,90 @@ async fn process_spawn_returns_before_exit_and_emits_exit_notification() -> Resu Ok(()) } +/// Held-open streams report the drain deadline separately from the child status, unless the requested prefix is complete. +#[cfg(unix)] +#[tokio::test] +async fn process_spawn_preserves_exit_code_when_output_drain_times_out() -> Result<()> { + let codex_home = TempDir::new()?; + let (_server, mut mcp) = initialized_mcp(codex_home.path()).await?; + for (handle, cap, script, exit_code, timed_out, stdout, stderr, stdout_cap_reached) in [ + ( + "unlimited", + None, + "printf prefix; sleep 30 2>/dev/null &", + 0, + true, + "prefix", + "", + false, + ), + ( + "full", + Some(3), + "printf abcdef; sleep 30 2>/dev/null &", + 0, + false, + "abc", + "", + true, + ), + ( + "partial-stderr", + Some(3), + "printf abcdef; printf x >&2; sleep 30 >/dev/null &", + 0, + true, + "abc", + "x", + true, + ), + ( + "zero", + Some(0), + "sleep 30 2>/dev/null &", + 0, + false, + "", + "", + false, + ), + ( + "child-error", + None, + "printf prefix; sleep 30 2>/dev/null & exit 7", + 7, + true, + "prefix", + "", + false, + ), + ] { + let command = vec!["sh".into(), "-c".into(), script.into()]; + let request_id = mcp + .send_process_spawn_request(ProcessSpawnParams { + output_bytes_cap: Some(cap), + ..process_spawn_params(handle.into(), codex_home.path(), command)? + }) + .await?; + mcp.read_stream_until_response_message(RequestId::Integer(request_id)) + .await?; + let exited = timeout(DEFAULT_READ_TIMEOUT, read_process_exited(&mut mcp)).await??; + assert_eq!( + exited, + ProcessExitedNotification { + process_handle: handle.into(), + exit_code, + output_drain_timed_out: Some(timed_out), + stdout: stdout.into(), + stdout_cap_reached, + stderr: stderr.into(), + stderr_cap_reached: false, + } + ); + } + Ok(()) +} + #[tokio::test] async fn process_spawn_returns_error_when_local_environment_is_disabled() -> Result<()> { let codex_home = TempDir::new()?; @@ -139,6 +224,7 @@ async fn process_spawn_returns_error_when_local_environment_is_disabled() -> Res Ok(()) } +/// Filling a capture cap must leave the child free to write the remainder and exit successfully. #[tokio::test] async fn process_spawn_reports_buffered_output_cap_reached() -> Result<()> { let codex_home = TempDir::new()?; @@ -157,7 +243,8 @@ async fn process_spawn_reports_buffered_output_cap_reached() -> Result<()> { vec![ "sh".to_string(), "-lc".to_string(), - "printf abcde; printf 12345 >&2".to_string(), + "printf abcde; dd if=/dev/zero bs=65536 count=32 2>/dev/null && printf 12345 >&2" + .to_string(), ] }; let spawn_request_id = mcp @@ -178,6 +265,7 @@ async fn process_spawn_reports_buffered_output_cap_reached() -> Result<()> { ProcessExitedNotification { process_handle, exit_code: 0, + output_drain_timed_out: Some(false), stdout: "abc".to_string(), stdout_cap_reached: true, stderr: "123".to_string(), @@ -226,6 +314,7 @@ async fn process_spawn_exec_failure_releases_handle_for_retry() -> Result<()> { ProcessExitedNotification { process_handle, exit_code: 19, + output_drain_timed_out: Some(false), stdout: "retried".to_string(), stdout_cap_reached: false, stderr: String::new(), diff --git a/codex-rs/utils/pty/src/pipe.rs b/codex-rs/utils/pty/src/pipe.rs index 802dfa5280..1088c01fd1 100644 --- a/codex-rs/utils/pty/src/pipe.rs +++ b/codex-rs/utils/pty/src/pipe.rs @@ -242,21 +242,6 @@ async fn spawn_process_with_stdin_mode( read_output_stream(BufReader::new(stderr), stderr_tx).await; }) }); - let mut reader_abort_handles = Vec::new(); - if let Some(handle) = stdout_handle.as_ref() { - reader_abort_handles.push(handle.abort_handle()); - } - if let Some(handle) = stderr_handle.as_ref() { - reader_abort_handles.push(handle.abort_handle()); - } - let reader_handle = tokio::spawn(async move { - if let Some(handle) = stdout_handle { - let _ = handle.await; - } - if let Some(handle) = stderr_handle { - let _ = handle.await; - } - }); let (exit_tx, exit_rx) = oneshot::channel::(); let exit_status = Arc::new(AtomicBool::new(false)); @@ -298,8 +283,8 @@ async fn spawn_process_with_stdin_mode( #[cfg(unix)] process_group_id, }), - reader_handle, - reader_abort_handles, + stdout_handle, + stderr_handle, writer_handle, wait_handle, exit_status, diff --git a/codex-rs/utils/pty/src/process.rs b/codex-rs/utils/pty/src/process.rs index 5593450b89..4ad710f63f 100644 --- a/codex-rs/utils/pty/src/process.rs +++ b/codex-rs/utils/pty/src/process.rs @@ -111,8 +111,8 @@ type ResizeFn = Box anyhow::Result<()> + Send>; pub struct ProcessHandle { writer_tx: StdMutex>>>, killer: StdMutex>>, - reader_handle: StdMutex>>, - reader_abort_handles: StdMutex>, + stdout_reader: Option>, + stderr_reader: Option>, writer_handle: StdMutex>>, wait_handle: StdMutex>>, exit_status: Arc, @@ -136,8 +136,8 @@ impl ProcessHandle { pub(crate) fn new( writer_tx: mpsc::Sender>, killer: Box, - reader_handle: JoinHandle<()>, - reader_abort_handles: Vec, + stdout_reader: Option>, + stderr_reader: Option>, writer_handle: JoinHandle<()>, wait_handle: JoinHandle<()>, exit_status: Arc, @@ -148,8 +148,8 @@ impl ProcessHandle { Self { writer_tx: StdMutex::new(Some(writer_tx)), killer: StdMutex::new(Some(killer)), - reader_handle: StdMutex::new(Some(reader_handle)), - reader_abort_handles: StdMutex::new(reader_abort_handles), + stdout_reader, + stderr_reader, writer_handle: StdMutex::new(Some(writer_handle)), wait_handle: StdMutex::new(Some(wait_handle)), exit_status, @@ -159,6 +159,14 @@ impl ProcessHandle { } } + pub fn stdout_reader(&self) -> Option { + self.stdout_reader.as_ref().map(JoinHandle::abort_handle) + } + + pub fn stderr_reader(&self) -> Option { + self.stderr_reader.as_ref().map(JoinHandle::abort_handle) + } + /// Returns a channel sender for writing raw bytes to the child stdin. pub fn writer_sender(&self) -> mpsc::Sender> { if let Ok(writer_tx) = self.writer_tx.lock() @@ -246,15 +254,11 @@ impl ProcessHandle { pub fn terminate(&self) { self.request_terminate(); - if let Ok(mut h) = self.reader_handle.lock() - && let Some(handle) = h.take() + for reader in [&self.stdout_reader, &self.stderr_reader] + .into_iter() + .flatten() { - handle.abort(); - } - if let Ok(mut handles) = self.reader_abort_handles.lock() { - for handle in handles.drain(..) { - handle.abort(); - } + reader.abort(); } if let Ok(mut h) = self.writer_handle.lock() && let Some(handle) = h.take() @@ -459,11 +463,8 @@ pub fn spawn_from_driver(driver: ProcessDriver) -> SpawnedProcess { #[cfg(windows)] interrupt_terminates, }), - reader_handle, - stderr_reader_handle - .map(|handle| handle.abort_handle()) - .into_iter() - .collect(), + Some(reader_handle), + stderr_reader_handle, writer_handle, wait_handle, exit_status, diff --git a/codex-rs/utils/pty/src/pty.rs b/codex-rs/utils/pty/src/pty.rs index 035fad8841..7bee13adee 100644 --- a/codex-rs/utils/pty/src/pty.rs +++ b/codex-rs/utils/pty/src/pty.rs @@ -300,8 +300,8 @@ async fn spawn_process_portable( #[cfg(unix)] process_group_id, }), - reader_handle, - Vec::new(), + Some(reader_handle), + /*stderr_reader*/ None, writer_handle, wait_handle, exit_status, @@ -459,8 +459,8 @@ async fn spawn_process_preserving_fds( let handle = ProcessHandle::new( writer_tx, Box::new(RawPidTerminator { process_group_id }), - reader_handle, - Vec::new(), + Some(reader_handle), + /*stderr_reader*/ None, writer_handle, wait_handle, exit_status, diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 0176b9f852..8c265a7d2b 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -3511,6 +3511,13 @@ class ProcessExitedNotification(BaseModel): populate_by_name=True, ) exit_code: Annotated[int, Field(alias="exitCode", description="Process exit code.")] + output_drain_timed_out: Annotated[ + bool | None, + Field( + alias="outputDrainTimedOut", + description="Whether output collection hit its post-exit deadline before seeing EOF or reaching the requested byte cap on either stream. Older servers omit this field.", + ), + ] = None process_handle: Annotated[ str, Field(