Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
richardopenai merged 4 commits into from
Jun 27, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
005450e
unified-exec: consume pushed process events
richardopenai Jun 26, 2026
9fa6c12
unified-exec: add pushed event integration coverage
richardopenai Jun 26, 2026
5f1454e
Merge remote-tracking branch 'origin/main' into codex/unified-exec-pr…
richardopenai Jun 26, 2026
680f1cb
Fix thread history test initializer
richardopenai Jun 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions codex-rs/core/src/session/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7016,6 +7016,7 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() {
dynamic_tools: Vec::new(),
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
Expand Down
182 changes: 125 additions & 57 deletions codex-rs/core/src/unified_exec/process.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Add integration coverage for pushed exec events

This switches unified exec from wake/read polling to consuming pushed process events, including lag and legacy fallback behavior, but the diff only adds mock-based unit coverage in process_tests.rs; no core/suite TestCodex integration test exercises the real agent tool path. Please add integration coverage for the new completion path so the actual session wiring and user-facing exec behavior are protected.

AGENTS.md reference: AGENTS.md:L114-L116

Useful? React with 👍 / 👎.

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))

@richardopenai richardopenai Jun 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After we get a wakeup, we call process.read which introduces an extra round trip

.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;
}
}
})
}
Expand Down
37 changes: 0 additions & 37 deletions codex-rs/core/src/unified_exec/process_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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));
}
1 change: 1 addition & 0 deletions codex-rs/core/tests/suite/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading