[codex] consume pushed exec-server process events (#30273)

## Summary

- complete unified-exec processes from the ordered event stream instead
of issuing a final zero-wait `process/read`
- add optional executor sandbox-denial state to `process/exited`
- retain `process/read` as a retained-output and compatibility fallback
for receiver lag, sequence gaps, and legacy servers
- recover sandbox-denial state across transport reconnection
- cover the real `TestCodex` remote-exec path without adding a public
test-only event constructor

## Why

A successful one-shot tool call currently receives its output and
terminal notifications, then pays another wide-area `process/read` round
trip before returning. Staging traces showed that remote response wait
accounted for more than 99.8% of RPC time; local serialization,
queueing, and deserialization were below 0.6 ms.

## Measured impact

A direct staging A/B used the same build and route and changed only
completion mode. Each arm ran three times with 30 one-shot
`/usr/bin/true` calls per run. The table reports the median of the three
per-run percentiles.

| Metric | Final `process/read` | Pushed events | Change |
| --- | ---: | ---: | ---: |
| End-to-end completion p50 | 159.5 ms | 118.7 ms | -40.8 ms (-25.6%) |
| End-to-end completion p95 | 182.4 ms | 131.7 ms | -50.6 ms (-27.8%) |
| Completion-wait p50 | 80.1 ms | 41.5 ms | -38.5 ms (-48.1%) |
| Final `process/read` RPC p50 | 79.9 ms | eliminated | -79.9 ms |

TCP_NODELAY was enabled in both A/B arms, so its effect cancels out. The
successful, complete, in-order event path issued zero final
`process/read` calls.

## Compatibility and recovery

- new servers send `sandboxDenied` on `process/exited`
- legacy servers omit it, which triggers one compatibility
`process/read`
- broadcast lag or a sequence gap triggers a retained-output read
- recovery remains bounded by the server's existing 1 MiB
retained-output window
- complete, in-order event streams issue no completion read
- sandbox denial is attached to the exit event before consumers can
observe process completion
- server-first and client-first rollouts remain wire-compatible;
server-first realizes the latency win immediately

## Integration coverage

The `TestCodex` suite exercises four distinct remote-exec contracts:

- complete pushed output/exit/close with zero reads
- direct pushed sandbox denial with zero reads
- legacy missing denial metadata with exactly one compatibility read
- count-bounded replay eviction recovered from retained output without
duplication

## Validation

- `just test -p codex-core
exec_command_consumes_pushed_remote_process_events`: 4 passed
- `just test -p codex-core unified_exec::process_tests::`: 4 passed
- `just test -p codex-exec-server`: 294 passed, 2 skipped
- `just test -p codex-exec-server-protocol`: 5 passed
- `just test -p codex-rmcp-client`: 89 passed, 2 skipped
- focused Bazel `//codex-rs/core:core-all-test`: passed across 16 shards
- scoped `just fix` passed for core and exec-server
- `just fmt` passed

The complete workspace suite was not rerun; focused Cargo and Bazel
coverage passed for the changed behavior.

Author: richardopenai

codex-rs/core/src/unified_exec/process.rs

@@ -14,6 +14,7 @@ use tokio_util::sync::CancellationToken;
 
 use crate::exec::is_likely_sandbox_denied;
 use codex_exec_server::ExecProcess;
+use codex_exec_server::ExecProcessEvent;
 use codex_exec_server::ProcessSignal as ExecServerProcessSignal;
 use codex_exec_server::ReadResponse as ExecReadResponse;
 use codex_exec_server::StartedExecProcess;
@@ -425,84 +426,151 @@ impl UnifiedExecProcess {
             cancellation_token,
         } = output_handles;
         let process = started.process;
-        let mut wake_rx = process.subscribe_wake();
+        let mut events = process.subscribe_events();
         tokio::spawn(async move {
-            let mut after_seq = None;
+            let mut last_seq: u64 = 0;
             loop {
-                match process
-                    .read(after_seq, /*max_bytes*/ None, /*wait_ms*/ Some(0))
-                    .await
+                let event = match events.recv().await {
+                    Ok(event) => Some(event),
+                    Err(broadcast::error::RecvError::Lagged(_)) => None,
+                    Err(broadcast::error::RecvError::Closed) => {
+                        let state = state_tx.borrow().clone();
+                        let _ = state_tx.send_replace(
+                            state.failed("exec-server process event stream closed".to_string()),
+                        );
+                        output_closed.store(true, Ordering::Release);
+                        output_closed_notify.notify_waiters();
+                        cancellation_token.cancel();
+                        break;
+                    }
+                };
+                let event_seq = event.as_ref().and_then(|event| match event {
+                    ExecProcessEvent::Output(chunk) => Some(chunk.seq),
+                    ExecProcessEvent::Exited { seq, .. } | ExecProcessEvent::Closed { seq } => {
+                        Some(*seq)
+                    }
+                    ExecProcessEvent::Failed(_) => None,
+                });
+                let missing_sandbox_denial = matches!(
+                    event.as_ref(),
+                    Some(ExecProcessEvent::Exited {
+                        sandbox_denied: None,
+                        ..
+                    })
+                );
+                if event.is_none()
+                    || event_seq.is_some_and(|seq| seq > last_seq.saturating_add(1))
+                    || missing_sandbox_denial
                 {
-                    Ok(response) => {
-                        let ExecReadResponse {
-                            chunks,
-                            next_seq,
-                            exited,
-                            exit_code,
-                            closed,
-                            failure,
-                            sandbox_denied,
-                        } = response;
-
-                        for chunk in chunks {
-                            let bytes = chunk.chunk.into_inner();
-                            let mut guard = output_buffer.lock().await;
-                            guard.push_chunk(bytes.clone());
-                            drop(guard);
-                            let _ = output_tx.send(bytes);
-                            output_notify.notify_waiters();
-                        }
-
-                        if let Some(message) = failure {
+                    let response = match process
+                        .read(
+                            Some(last_seq),
+                            /*max_bytes*/ None,
+                            /*wait_ms*/ Some(0),
+                        )
+                        .await
+                    {
+                        Ok(response) => response,
+                        Err(err) => {
                             let state = state_tx.borrow().clone();
-                            let _ = state_tx.send_replace(state.failed(message));
+                            let _ = state_tx.send_replace(state.failed(err.to_string()));
                             output_closed.store(true, Ordering::Release);
                             output_closed_notify.notify_waiters();
                             cancellation_token.cancel();
                             break;
                         }
+                    };
+                    let ExecReadResponse {
+                        chunks,
+                        next_seq,
+                        exited,
+                        exit_code,
+                        closed,
+                        failure,
+                        sandbox_denied,
+                    } = response;
+                    for chunk in chunks.into_iter().filter(|chunk| chunk.seq > last_seq) {
+                        let bytes = chunk.chunk.into_inner();
+                        let mut guard = output_buffer.lock().await;
+                        guard.push_chunk(bytes.clone());
+                        drop(guard);
+                        let _ = output_tx.send(bytes);
+                        output_notify.notify_waiters();
+                    }
+                    last_seq = last_seq.max(next_seq.saturating_sub(1));
+                    if let Some(message) = failure {
+                        let state = state_tx.borrow().clone();
+                        let _ = state_tx.send_replace(state.failed(message));
+                        output_closed.store(true, Ordering::Release);
+                        output_closed_notify.notify_waiters();
+                        cancellation_token.cancel();
+                        break;
+                    }
+                    if sandbox_denied || exited {
+                        let mut state = state_tx.borrow().clone();
+                        state.sandbox_denied |= sandbox_denied;
+                        let _ = state_tx.send_replace(if exited {
+                            state.exited(exit_code)
+                        } else {
+                            state
+                        });
+                    }
+                    if closed {
+                        output_closed.store(true, Ordering::Release);
+                        output_closed_notify.notify_waiters();
+                        cancellation_token.cancel();
+                        break;
+                    }
+                    continue;
+                }
 
-                        if sandbox_denied {
-                            let mut state = state_tx.borrow().clone();
-                            state.sandbox_denied = true;
-                            let _ = state_tx.send_replace(state);
-                        }
-
-                        if exited {
-                            let state = state_tx.borrow().clone();
-                            let _ = state_tx.send_replace(state.exited(exit_code));
+                let Some(event) = event else {
+                    continue;
+                };
+                match event {
+                    ExecProcessEvent::Output(chunk) => {
+                        if chunk.seq <= last_seq {
+                            continue;
                         }
-
-                        if closed {
-                            output_closed.store(true, Ordering::Release);
-                            output_closed_notify.notify_waiters();
-                            cancellation_token.cancel();
+                        last_seq = chunk.seq;
+                        let bytes = chunk.chunk.into_inner();
+                        let mut guard = output_buffer.lock().await;
+                        guard.push_chunk(bytes.clone());
+                        drop(guard);
+                        let _ = output_tx.send(bytes);
+                        output_notify.notify_waiters();
+                    }
+                    ExecProcessEvent::Exited {
+                        seq,
+                        exit_code,
+                        sandbox_denied,
+                    } => {
+                        if seq <= last_seq {
+                            continue;
                         }
-
-                        after_seq = next_seq.checked_sub(1);
-                        if output_closed.load(Ordering::Acquire) {
-                            break;
+                        last_seq = seq;
+                        let mut state = state_tx.borrow().clone();
+                        state.sandbox_denied |= sandbox_denied.unwrap_or(false);
+                        let _ = state_tx.send_replace(state.exited(Some(exit_code)));
+                    }
+                    ExecProcessEvent::Closed { seq } => {
+                        if seq <= last_seq {
+                            continue;
                         }
+                        output_closed.store(true, Ordering::Release);
+                        output_closed_notify.notify_waiters();
+                        cancellation_token.cancel();
+                        break;
                     }
-                    Err(err) => {
+                    ExecProcessEvent::Failed(message) => {
                         let state = state_tx.borrow().clone();
-                        let _ = state_tx.send_replace(state.failed(err.to_string()));
+                        let _ = state_tx.send_replace(state.failed(message));
                         output_closed.store(true, Ordering::Release);
                         output_closed_notify.notify_waiters();
                         cancellation_token.cancel();
                         break;
                     }
                 }
-
-                if wake_rx.changed().await.is_err() {
-                    let state = state_tx.borrow().clone();
-                    let _ = state_tx
-                        .send_replace(state.failed("exec-server wake channel closed".to_string()));
-                    output_closed.store(true, Ordering::Release);
-                    output_closed_notify.notify_waiters();
-                    cancellation_token.cancel();
-                    break;
-                }
             }
         })
     }

codex-rs/core/src/unified_exec/process_tests.rs

@@ -15,7 +15,6 @@ use std::collections::VecDeque;
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use tokio::sync::watch;
-use tokio::time::Duration;
 
 struct MockExecProcess {
     process_id: ProcessId,
@@ -173,39 +172,3 @@ async fn remote_terminate_confirmed_updates_state_on_success_only() {
 
     assert!(process.has_exited());
 }
-
-#[tokio::test]
-async fn remote_process_waits_for_early_exit_event() {
-    let (wake_tx, _wake_rx) = watch::channel(0);
-    let started = StartedExecProcess {
-        process: Arc::new(MockExecProcess {
-            process_id: "test-process".to_string().into(),
-            write_response: WriteResponse {
-                status: WriteStatus::Accepted,
-            },
-            read_responses: Mutex::new(VecDeque::from([ReadResponse {
-                chunks: Vec::new(),
-                next_seq: 2,
-                exited: true,
-                exit_code: Some(17),
-                closed: true,
-                failure: None,
-                sandbox_denied: false,
-            }])),
-            terminate_error: None,
-            wake_tx: wake_tx.clone(),
-        }),
-    };
-
-    tokio::spawn(async move {
-        tokio::time::sleep(Duration::from_millis(10)).await;
-        let _ = wake_tx.send(1);
-    });
-
-    let process = UnifiedExecProcess::from_exec_server_started(started)
-        .await
-        .expect("remote process should observe early exit");
-
-    assert!(process.has_exited());
-    assert_eq!(process.exit_code(), Some(17));
-}

codex-rs/core/tests/suite/mod.rs

@@ -126,6 +126,7 @@ mod tools;
 mod truncation;
 mod turn_state;
 mod unified_exec;
+mod unified_exec_process_events;
 #[cfg(unix)]
 mod unified_exec_zsh_fork_approvals;
 mod unstable_features_warning;