From 9fc715c0861c956c894a91890b78dc05b304ba29 Mon Sep 17 00:00:00 2001 From: jif Date: Wed, 22 Jul 2026 09:21:41 +0000 Subject: [PATCH] Order unified exec lifecycle events reliably (#34713) ## What changed - Treat output-task closure as the signal that trailing output is complete, drain any remaining chunks before publishing command completion, and retain the grace period as a fallback. - Wait for deferred network-denial classification before emitting the final command result. - Serialize `write_stdin` interaction and completion events so an interaction that exits a process is published first, and avoid pruning processes while their terminal events are being finalized. ## Testing - Add coverage for late output, missing output-close signals, late network denials, interaction/completion ordering, cross-platform aggregated output, and pruning during finalization. GitOrigin-RevId: 46a118552bb1a4658b78aa5a5ca76eb8c5fba571 --- .../handlers/unified_exec/write_stdin.rs | 23 +- .../core/src/unified_exec/async_watcher.rs | 63 +++++- .../src/unified_exec/async_watcher_tests.rs | 196 ++++++++++++++++++ codex-rs/core/src/unified_exec/mod.rs | 12 ++ codex-rs/core/src/unified_exec/mod_tests.rs | 1 + codex-rs/core/src/unified_exec/process.rs | 22 +- .../core/src/unified_exec/process_manager.rs | 56 ++++- .../src/unified_exec/process_manager_tests.rs | 63 ++++++ .../core/src/unified_exec/process_tests.rs | 2 +- codex-rs/core/tests/suite/unified_exec.rs | 42 +++- 10 files changed, 445 insertions(+), 35 deletions(-) diff --git a/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs b/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs index 7c949481885c..7f1709b0450f 100644 --- a/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs +++ b/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs @@ -7,9 +7,8 @@ use crate::tools::registry::CoreToolRuntime; use crate::tools::registry::PostToolUsePayload; use crate::tools::registry::PreToolUsePayload; use crate::tools::registry::ToolExecutor; +use crate::unified_exec::WriteStdinInteractionEvent; use crate::unified_exec::WriteStdinRequest; -use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::TerminalInteractionEvent; use codex_tools::ToolName; use codex_tools::ToolSpec; use serde::Deserialize; @@ -80,28 +79,16 @@ impl WriteStdinHandler { yield_time_ms: args.yield_time_ms, max_output_tokens: args.max_output_tokens, truncation_policy: turn.model_info.truncation_policy.into(), + interaction_event: Some(WriteStdinInteractionEvent { + session: &session, + turn: &turn, + }), }) .await .map_err(|err| { FunctionCallError::RespondToModel(format!("write_stdin failed: {err}")) })?; - // Empty stdin is a background poll, so emit it only while there is - // still a live process for the UI to wait on. Non-empty stdin is a real - // terminal interaction and should remain visible even if it completes - // the process before the response returns. - if !args.chars.is_empty() || response.process_id.is_some() { - let process_id = response.process_id.unwrap_or(args.session_id); - let interaction = TerminalInteractionEvent { - call_id: response.event_call_id.clone(), - process_id: process_id.to_string(), - stdin: args.chars.clone(), - }; - session - .send_event(turn.as_ref(), EventMsg::TerminalInteraction(interaction)) - .await; - } - Ok(boxed_tool_output(response)) } } diff --git a/codex-rs/core/src/unified_exec/async_watcher.rs b/codex-rs/core/src/unified_exec/async_watcher.rs index c1d61ebdcee6..9a582a6fae66 100644 --- a/codex-rs/core/src/unified_exec/async_watcher.rs +++ b/codex-rs/core/src/unified_exec/async_watcher.rs @@ -1,5 +1,6 @@ use std::pin::Pin; use std::sync::Arc; +use std::sync::atomic::Ordering; use tokio::sync::Mutex; use tokio::time::Duration; @@ -7,6 +8,7 @@ use tokio::time::Instant; use tokio::time::Sleep; use super::UnifiedExecContext; +use super::process::OutputHandles; use super::process::UnifiedExecProcess; use crate::exec::MAX_EXEC_OUTPUT_DELTAS_PER_CALL; use crate::session::session::Session; @@ -45,6 +47,11 @@ pub(crate) fn start_streaming_output( let mut receiver = process.output_receiver(); let output_drained = process.output_drained_notify(); let exit_token = process.cancellation_token(); + let OutputHandles { + output_closed, + output_closed_notify, + .. + } = process.output_handles(); let session_ref = Arc::clone(&context.session); let turn_ref = Arc::clone(&context.turn); @@ -57,8 +64,19 @@ pub(crate) fn start_streaming_output( let mut emitted_deltas: usize = 0; let mut grace_sleep: Option>> = None; + let output_closed_notified = output_closed_notify.notified(); + tokio::pin!(output_closed_notified); + let mut output_complete = false; loop { + // Register before checking the atomic so a close between the check + // and the select cannot miss the notification. + output_closed_notified.as_mut().enable(); + if grace_sleep.is_some() && output_closed.load(Ordering::Acquire) { + output_complete = true; + break; + } + tokio::select! { _ = exit_token.cancelled(), if grace_sleep.is_none() => { let deadline = Instant::now() + TRAILING_OUTPUT_GRACE; @@ -70,10 +88,13 @@ pub(crate) fn start_streaming_output( sleep.as_mut().await; } }, if grace_sleep.is_some() => { - output_drained.notify_one(); break; } + _ = &mut output_closed_notified, if grace_sleep.is_some() => { + output_closed_notified.set(output_closed_notify.notified()); + } + received = receiver.recv() => { let chunk = match received { Ok(chunk) => chunk, @@ -81,7 +102,7 @@ pub(crate) fn start_streaming_output( continue; }, Err(RecvError::Closed) => { - output_drained.notify_one(); + output_complete = true; break; } }; @@ -98,6 +119,35 @@ pub(crate) fn start_streaming_output( } } } + + output_complete |= output_closed.load(Ordering::Acquire); + if output_complete { + // Output producers publish all chunks before setting output_closed + // with Release ordering, so the Acquire above makes this a final + // safe drain. + loop { + let chunk = match receiver.try_recv() { + Ok(chunk) => chunk, + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => continue, + Err( + tokio::sync::broadcast::error::TryRecvError::Empty + | tokio::sync::broadcast::error::TryRecvError::Closed, + ) => break, + }; + + process_chunk( + &mut pending, + &transcript, + &call_id, + &session_ref, + &turn_ref, + &mut emitted_deltas, + chunk, + ) + .await; + } + } + output_drained.notify_one(); }); } @@ -114,13 +164,22 @@ pub(crate) fn spawn_exit_watcher( process_id: i32, transcript: Arc>, started_at: Instant, + network_denial_monitor: Option>, ) { let exit_token = process.cancellation_token(); let output_drained = process.output_drained_notify(); + let interaction_lock = process.interaction_lock(); tokio::spawn(async move { exit_token.cancelled().await; output_drained.notified().await; + // Deferred network denial deliberately remains observable for a short + // window after process exit. Do not classify the terminal event until + // that monitor has settled, even when output closes immediately. + if let Some(network_denial_monitor) = network_denial_monitor { + let _ = network_denial_monitor.await; + } + let _interaction_guard = interaction_lock.lock_owned().await; let duration = Instant::now().saturating_duration_since(started_at); if let Some(message) = process.failure_message() { diff --git a/codex-rs/core/src/unified_exec/async_watcher_tests.rs b/codex-rs/core/src/unified_exec/async_watcher_tests.rs index 07d1bec60f26..01be4b7802d2 100644 --- a/codex-rs/core/src/unified_exec/async_watcher_tests.rs +++ b/codex-rs/core/src/unified_exec/async_watcher_tests.rs @@ -1,6 +1,202 @@ +use std::sync::Arc; + +use super::TRAILING_OUTPUT_GRACE; +use super::spawn_exit_watcher; use super::split_valid_utf8_prefix_with_max; +use super::start_streaming_output; +use crate::session::tests::make_session_and_context_with_rx; +use crate::unified_exec::UnifiedExecContext; +use crate::unified_exec::head_tail_buffer::HeadTailBuffer; +use crate::unified_exec::process::NoopSpawnLifecycle; +use crate::unified_exec::process::UnifiedExecProcess; +use codex_protocol::items::CommandExecutionStatus; +use codex_protocol::items::TurnItem; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; +use codex_sandboxing::SandboxType; use pretty_assertions::assert_eq; +use tokio::time::Duration; +use tokio::time::Instant; + +struct StreamingOutputHarness { + process: Arc, + stdout_tx: tokio::sync::broadcast::Sender>, + exit_tx: tokio::sync::oneshot::Sender, + transcript: Arc>, + context: UnifiedExecContext, + rx_event: async_channel::Receiver, +} + +async fn streaming_output_harness() -> anyhow::Result { + let (writer_tx, _writer_rx) = tokio::sync::mpsc::channel::>(1); + let (stdout_tx, stdout_rx) = tokio::sync::broadcast::channel::>(8); + let (exit_tx, exit_rx) = tokio::sync::oneshot::channel::(); + let spawned = codex_utils_pty::spawn_from_driver(codex_utils_pty::ProcessDriver { + writer_tx, + stdout_rx, + stderr_rx: None, + exit_rx, + terminator: None, + writer_handle: None, + resizer: None, + }); + let process = Arc::new( + UnifiedExecProcess::from_spawned(spawned, SandboxType::None, Box::new(NoopSpawnLifecycle)) + .await?, + ); + let (session, turn, rx_event) = make_session_and_context_with_rx().await; + let context = UnifiedExecContext::new(session, turn, "streaming-output-test".to_string()); + let transcript = Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default())); + start_streaming_output(&process, &context, Arc::clone(&transcript)); + + Ok(StreamingOutputHarness { + process, + stdout_tx, + exit_tx, + transcript, + context, + rx_event, + }) +} + +#[tokio::test] +async fn streaming_output_finishes_on_close_without_waiting_for_grace() -> anyhow::Result<()> { + let StreamingOutputHarness { + process, + stdout_tx, + exit_tx, + transcript, + .. + } = streaming_output_harness().await?; + let output_drained = process.output_drained_notify(); + let drained = output_drained.notified(); + tokio::pin!(drained); + + tokio::time::pause(); + let exited_at = Instant::now(); + exit_tx.send(0).expect("send exit"); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(50)).await; + stdout_tx + .send(b"LATE-OUTPUT-MARKER".to_vec()) + .expect("send late output"); + }); + + (&mut drained).await; + let elapsed = Instant::now().saturating_duration_since(exited_at); + tokio::time::resume(); + + assert!( + elapsed >= Duration::from_millis(50) && elapsed < TRAILING_OUTPUT_GRACE, + "output close should finish before the grace fallback: {elapsed:?}" + ); + assert_eq!( + transcript.lock().await.to_bytes_with_omission_marker(), + b"LATE-OUTPUT-MARKER" + ); + + Ok(()) +} + +#[tokio::test] +async fn streaming_output_keeps_grace_as_fallback_without_close() -> anyhow::Result<()> { + let StreamingOutputHarness { + process, + stdout_tx: _stdout_tx, + exit_tx, + .. + } = streaming_output_harness().await?; + let output_drained = process.output_drained_notify(); + let drained = output_drained.notified(); + tokio::pin!(drained); + + tokio::time::pause(); + let exited_at = Instant::now(); + exit_tx.send(0).expect("send exit"); + (&mut drained).await; + let elapsed = Instant::now().saturating_duration_since(exited_at); + tokio::time::resume(); + + assert!( + elapsed >= TRAILING_OUTPUT_GRACE + && elapsed <= TRAILING_OUTPUT_GRACE + Duration::from_millis(10), + "missing output close should use the grace fallback: {elapsed:?}" + ); + + Ok(()) +} + +#[tokio::test] +async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() -> anyhow::Result<()> { + let StreamingOutputHarness { + process, + stdout_tx, + exit_tx, + transcript, + context, + rx_event, + } = streaming_output_harness().await?; + + tokio::time::pause(); + let process_for_late_denial = Arc::clone(&process); + let (late_denial_armed_tx, late_denial_armed_rx) = tokio::sync::oneshot::channel(); + let network_denial_monitor = tokio::spawn(async move { + let sleep = tokio::time::sleep(Duration::from_millis(10)); + tokio::pin!(sleep); + late_denial_armed_tx.send(()).expect("arm late denial"); + sleep.await; + process_for_late_denial.fail_and_terminate("LATE_DENIAL".to_string()); + }); + late_denial_armed_rx.await.expect("late denial armed"); + + #[allow(deprecated)] + let cwd = context.turn.cwd.clone().into(); + spawn_exit_watcher( + Arc::clone(&process), + Arc::clone(&context.session), + Arc::clone(&context.turn), + context.call_id, + vec!["proof".to_string()], + cwd, + /*process_id*/ 123, + transcript, + Instant::now(), + Some(network_denial_monitor), + ); + + let exited_at = Instant::now(); + exit_tx.send(0).expect("send exit"); + drop(stdout_tx); + + let event = rx_event.recv().await.expect("command end event"); + let elapsed = Instant::now().saturating_duration_since(exited_at); + tokio::time::resume(); + let EventMsg::ItemCompleted(completed) = event.msg else { + panic!("expected ItemCompleted"); + }; + let TurnItem::CommandExecution(item) = completed.item else { + panic!("expected CommandExecution"); + }; + assert_eq!( + ( + item.status, + item.exit_code, + item.aggregated_output.as_deref() + ), + ( + CommandExecutionStatus::Failed, + Some(-1), + Some("LATE_DENIAL") + ) + ); + assert!( + elapsed >= Duration::from_millis(10) && elapsed < TRAILING_OUTPUT_GRACE, + "completion should wait for denial without falling back to the output grace: {elapsed:?}" + ); + + Ok(()) +} #[test] fn split_valid_utf8_prefix_respects_max_bytes_for_ascii() { diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index 21c1a17f0965..e8dd31c2ce8f 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/codex-rs/core/src/unified_exec/mod.rs @@ -116,6 +116,18 @@ pub(crate) struct WriteStdinRequest<'a> { pub yield_time_ms: u64, pub max_output_tokens: Option, pub truncation_policy: TruncationPolicy, + pub interaction_event: Option>, +} + +pub(crate) struct WriteStdinInteractionEvent<'a> { + pub session: &'a Arc, + pub turn: &'a Arc, +} + +impl std::fmt::Debug for WriteStdinInteractionEvent<'_> { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("WriteStdinInteractionEvent") + } } #[derive(Default)] diff --git a/codex-rs/core/src/unified_exec/mod_tests.rs b/codex-rs/core/src/unified_exec/mod_tests.rs index 22c08f4aeb8d..04cfa747cabc 100644 --- a/codex-rs/core/src/unified_exec/mod_tests.rs +++ b/codex-rs/core/src/unified_exec/mod_tests.rs @@ -322,6 +322,7 @@ async fn write_stdin( yield_time_ms, max_output_tokens: None, truncation_policy: TruncationPolicy::Tokens(10_000), + interaction_event: None, }) .await } diff --git a/codex-rs/core/src/unified_exec/process.rs b/codex-rs/core/src/unified_exec/process.rs index 670e94e9b6ff..0b6937f4a73e 100644 --- a/codex-rs/core/src/unified_exec/process.rs +++ b/codex-rs/core/src/unified_exec/process.rs @@ -65,6 +65,18 @@ pub(crate) struct OutputHandles { pub(crate) cancellation_token: CancellationToken, } +struct OutputTaskGuard { + output_closed: Arc, + output_closed_notify: Arc, +} + +impl Drop for OutputTaskGuard { + fn drop(&mut self) { + self.output_closed.store(true, Ordering::Release); + self.output_closed_notify.notify_waiters(); + } +} + /// Transport-specific process handle used by unified exec. enum ProcessHandle { Local(Box), @@ -203,8 +215,6 @@ impl UnifiedExecProcess { } fn finish_termination(&self) { - self.output_closed.store(true, Ordering::Release); - self.output_closed_notify.notify_waiters(); self.cancellation_token.cancel(); if let Some(output_task) = &self.output_task { output_task.abort(); @@ -434,6 +444,10 @@ impl UnifiedExecProcess { let process = started.process; let mut events = process.subscribe_events(); tokio::spawn(async move { + let _output_task_guard = OutputTaskGuard { + output_closed: Arc::clone(&output_closed), + output_closed_notify: Arc::clone(&output_closed_notify), + }; let mut last_seq: u64 = 0; loop { let event = match events.recv().await { @@ -590,6 +604,10 @@ impl UnifiedExecProcess { output_tx: broadcast::Sender>, ) -> JoinHandle<()> { tokio::spawn(async move { + let _output_task_guard = OutputTaskGuard { + output_closed: Arc::clone(&output_closed), + output_closed_notify: Arc::clone(&output_closed_notify), + }; loop { match receiver.recv().await { Ok(chunk) => { diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 27f889da7acc..96c324adc011 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -45,6 +45,7 @@ use crate::unified_exec::ProcessStore; use crate::unified_exec::UnifiedExecContext; use crate::unified_exec::UnifiedExecError; use crate::unified_exec::UnifiedExecProcessManager; +use crate::unified_exec::WriteStdinInteractionEvent; use crate::unified_exec::WriteStdinRequest; use crate::unified_exec::async_watcher::emit_exec_end_for_unified_exec; use crate::unified_exec::async_watcher::emit_failed_exec_end_for_unified_exec; @@ -61,7 +62,9 @@ use codex_network_proxy::NetworkProxy; use codex_protocol::config_types::ShellEnvironmentPolicy; use codex_protocol::error::CodexErr; use codex_protocol::error::SandboxErr; +use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecCommandSource; +use codex_protocol::protocol::TerminalInteractionEvent; use codex_sandboxing::SandboxCommand; use codex_tools::ToolName; use codex_utils_output_truncation::approx_tokens_from_byte_count; @@ -350,7 +353,7 @@ fn terminate_process_on_network_denial( process: Arc, session: std::sync::Weak, deferred: DeferredNetworkApproval, -) { +) -> tokio::task::JoinHandle<()> { let network_cancelled = deferred.cancellation_token(); let process_exited = process.cancellation_token(); tokio::spawn(async move { @@ -366,7 +369,7 @@ fn terminate_process_on_network_denial( let session = session.upgrade(); let message = network_denial_message_for_session(session.as_ref(), Some(deferred)).await; process.fail_and_terminate(message); - }); + }) } impl UnifiedExecProcessManager { @@ -426,13 +429,13 @@ impl UnifiedExecProcessManager { return Err(err); } }; - if let Some(deferred) = deferred_network_approval.as_ref() { + let network_denial_monitor = deferred_network_approval.as_ref().map(|deferred| { terminate_process_on_network_denial( Arc::clone(&process), Arc::downgrade(&context.session), deferred.clone(), - ); - } + ) + }); let transcript = Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default())); let event_ctx = ToolEventCtx::new( @@ -466,6 +469,7 @@ impl UnifiedExecProcessManager { request.process_id, request.tty, deferred_network_approval.clone(), + network_denial_monitor, Arc::clone(&transcript), Arc::clone(&initial_exec_command_active), ) @@ -826,6 +830,23 @@ impl UnifiedExecProcessManager { hook_command: Some(hook_command), }; + let should_emit_interaction = !request.input.is_empty() || response.process_id.is_some(); + if should_emit_interaction + && let Some(WriteStdinInteractionEvent { session, turn }) = request.interaction_event + { + let interaction = TerminalInteractionEvent { + call_id: response.event_call_id.clone(), + process_id: response + .process_id + .unwrap_or(request.process_id) + .to_string(), + stdin: request.input.to_string(), + }; + session + .send_event(turn.as_ref(), EventMsg::TerminalInteraction(interaction)) + .await; + } + Ok(response) } @@ -911,6 +932,7 @@ impl UnifiedExecProcessManager { process_id: i32, tty: bool, network_approval: Option, + network_denial_monitor: Option>, transcript: Arc>, initial_exec_command_active: Arc, ) { @@ -949,6 +971,7 @@ impl UnifiedExecProcessManager { process_id, transcript, started_at, + network_denial_monitor, ); } @@ -1330,17 +1353,34 @@ impl UnifiedExecProcessManager { .iter() .map(|(id, entry)| (*id, entry.last_used, entry.process.has_exited())) .collect(); + let mut found_locked_exited_process = false; while let Some(process_id) = Self::process_id_to_prune_from_meta(&meta) { - // Do not prune processes being held by write_stdin. - if let Some(interaction_lock) = store + let candidate_process = store .processes .get(&process_id) - .map(|entry| entry.process.interaction_lock()) + .map(|entry| Arc::clone(&entry.process)); + let candidate_has_exited = candidate_process + .as_ref() + .is_some_and(|process| process.has_exited()); + if found_locked_exited_process && !candidate_has_exited { + // The store may temporarily exceed its soft cap while an exited + // process is publishing its terminal event. Do not evict a live + // process just because that exited process is briefly locked. + return None; + } + + // Do not prune processes while write_stdin or terminal event + // publication holds their interaction lock. + if let Some(interaction_lock) = candidate_process + .as_ref() + .map(|process| process.interaction_lock()) && let Ok(_interaction_guard) = interaction_lock.try_lock_owned() { return store.remove(process_id); } + found_locked_exited_process |= candidate_has_exited + || candidate_process.is_some_and(|process| process.has_exited()); meta.retain(|(id, _, _)| *id != process_id); } diff --git a/codex-rs/core/src/unified_exec/process_manager_tests.rs b/codex-rs/core/src/unified_exec/process_manager_tests.rs index 5c117d4dd611..ebd1e10a5518 100644 --- a/codex-rs/core/src/unified_exec/process_manager_tests.rs +++ b/codex-rs/core/src/unified_exec/process_manager_tests.rs @@ -487,3 +487,66 @@ fn pruning_protects_recent_processes_even_if_exited() { // (10) is exited but among the last 8; we should drop the LRU outside that set. assert_eq!(candidate, Some(1)); } + +#[cfg(unix)] +#[tokio::test] +async fn pruning_does_not_evict_live_process_while_exited_process_is_finalizing() { + let exited_process = Arc::new( + crate::unified_exec::process_tests::remote_process( + codex_exec_server::WriteStatus::Accepted, + /*terminate_error*/ None, + ) + .await, + ); + exited_process + .terminate_confirmed() + .await + .expect("exited process should terminate"); + let live_process = Arc::new( + crate::unified_exec::process_tests::remote_process( + codex_exec_server::WriteStatus::Accepted, + /*terminate_error*/ None, + ) + .await, + ); + let _interaction_guard = exited_process.interaction_lock().lock_owned().await; + let now = Instant::now(); + let cwd = PathUri::parse("file:///tmp").expect("test cwd should be valid"); + let mut store = ProcessStore::default(); + let max_process_id = + i32::try_from(MAX_UNIFIED_EXEC_PROCESSES).expect("process cap should fit in i32"); + + for process_id in 1..=max_process_id { + let is_exited = process_id == 1; + store.processes.insert( + process_id, + ProcessEntry { + process: if is_exited { + Arc::clone(&exited_process) + } else { + Arc::clone(&live_process) + }, + call_id: format!("call-{process_id}"), + process_id, + cwd: cwd.clone(), + initial_exec_command_active: Arc::new(AtomicBool::new(false)), + hook_command: format!("command-{process_id}"), + tty: false, + network_approval: None, + session: std::sync::Weak::new(), + last_used: if is_exited { + now - Duration::from_secs(1) + } else { + now + }, + }, + ); + } + + let pruned = UnifiedExecProcessManager::prune_processes_if_needed(&mut store); + + assert_eq!( + (pruned.map(|entry| entry.process_id), store.processes.len()), + (None, MAX_UNIFIED_EXEC_PROCESSES) + ); +} diff --git a/codex-rs/core/src/unified_exec/process_tests.rs b/codex-rs/core/src/unified_exec/process_tests.rs index af0cb627413c..494f29bde0f1 100644 --- a/codex-rs/core/src/unified_exec/process_tests.rs +++ b/codex-rs/core/src/unified_exec/process_tests.rs @@ -85,7 +85,7 @@ impl ExecProcess for MockExecProcess { } } -async fn remote_process( +pub(super) async fn remote_process( write_status: WriteStatus, terminate_error: Option, ) -> UnifiedExecProcess { diff --git a/codex-rs/core/tests/suite/unified_exec.rs b/codex-rs/core/tests/suite/unified_exec.rs index 5fa4f265d273..06523c546e58 100644 --- a/codex-rs/core/tests/suite/unified_exec.rs +++ b/codex-rs/core/tests/suite/unified_exec.rs @@ -1950,10 +1950,36 @@ async fn write_stdin_returns_exit_metadata_and_clears_session() -> Result<()> { ) .await?; - wait_for_event(&test.codex, |event| { - matches!(event, EventMsg::TurnComplete(_)) - }) - .await; + let mut exit_lifecycle_order = Vec::new(); + let mut turn_completed = false; + loop { + let event = wait_for_event(&test.codex, |_| true).await; + match event { + EventMsg::TerminalInteraction(event) + if event.call_id == start_call_id + && event.stdin + == exit_args + .get("chars") + .and_then(Value::as_str) + .expect("exit chars") => + { + exit_lifecycle_order.push("terminal_interaction"); + } + EventMsg::ExecCommandEnd(event) if event.call_id == start_call_id => { + exit_lifecycle_order.push("exec_command_end"); + } + EventMsg::TurnComplete(_) => turn_completed = true, + _ => {} + } + if turn_completed && exit_lifecycle_order.len() == 2 { + break; + } + } + assert_eq!( + exit_lifecycle_order, + ["terminal_interaction", "exec_command_end"], + "the interaction that exits a process must precede command completion" + ); let requests = request_log.requests(); assert!(!requests.is_empty(), "expected at least one POST request"); @@ -3392,6 +3418,14 @@ async fn unified_exec_runs_on_all_platforms() -> Result<()> { submit_unified_exec_turn(&test, "summarize large output", PermissionProfile::Disabled).await?; + let end_event = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::ExecCommandEnd(event) if event.call_id == call_id => Some(event.clone()), + _ => None, + }) + .await; + assert_eq!(end_event.exit_code, 0); + assert_regex_match(".*hello crossplat.*", &end_event.aggregated_output); + wait_for_event(&test.codex, |event| { matches!(event, EventMsg::TurnComplete(_)) })