github/codex
4
0 Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-09 00:39:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move synchronous Guardian orchestration into the reviewer extension (#44252)

Browse Source 
## What changed

Move the synchronous review loop, outcome mapping, deadline helpers, and session pool into `codex-guardian-reviewer`. Core supplies adapters for evidence capture, authorization checks, session creation, and event publication through the new host interfaces.

Have `guardian-v2` initialize the reviewer pool and host separately, with the pool managing prewarming, session reuse, concurrent forks, invalidation, and shutdown.

## Testing

Extend the app-server reviewer lifecycle test to cover interrupted concurrent reviews as well as completed reviews, including resuming a reviewer after parent shutdown. Adapt core reviewer tests to use the pool and factory interfaces.

GitOrigin-RevId: 2d69bd20c169b20534764b98cbdb63f38564a530
This commit is contained in:
jif authored and copyberry committed 2026-09-09 17:59:10 +00:00
1 parent eb7bd64ef9
commit 2617ed2e1c
27 files changed
+1570 -1417
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

codex-rs/core/src/session/tests.rs
-139
View File
@@ -8609,145 +8609,6 @@ async fn shutdown_and_wait_waits_when_shutdown_is_already_in_progress() {
.expect("shutdown waiter");
}
#[tokio::test]
async fn shutdown_and_wait_shuts_down_cached_guardian_subagent() {
let (parent_session, parent_turn_context) = make_session_and_context().await;
let parent_session = Arc::new(parent_session);
let parent_config = Arc::clone(&parent_turn_context.config);
let (parent_tx_sub, parent_rx_sub) = async_channel::bounded(4);
let (_parent_tx_event, parent_rx_event) = async_channel::unbounded();
let parent_session_for_loop = Arc::clone(&parent_session);
let parent_session_loop_handle = tokio::spawn(async move {
submission_loop(parent_session_for_loop, parent_config, parent_rx_sub).await;
});
let parent_io = SessionIo {
tx_sub: parent_tx_sub,
rx_event: parent_rx_event,
agent_status: watch::channel(AgentStatus::PendingInit).1,
session_loop_termination: session_loop_termination_from_handle(parent_session_loop_handle),
};
let (child_session, _child_turn_context) = make_session_and_context().await;
let (child_tx_sub, child_rx_sub) = async_channel::bounded::<Submission>(4);
let (_child_tx_event, child_rx_event) = async_channel::unbounded();
let (child_shutdown_tx, child_shutdown_rx) = tokio::sync::oneshot::channel();
let child_session_loop_handle = tokio::spawn(async move {
let shutdown = child_rx_sub
.recv()
.await
.expect("child shutdown submission");
assert!(matches!(shutdown.op, Op::Shutdown));
child_shutdown_tx
.send(())
.expect("child shutdown signal should be delivered");
});
let child_session = Arc::new(child_session);
let child_io = SessionIo {
tx_sub: child_tx_sub,
rx_event: child_rx_event,
agent_status: watch::channel(AgentStatus::PendingInit).1,
session_loop_termination: session_loop_termination_from_handle(child_session_loop_handle),
};
parent_session
.guardian_review_session()
.cache_for_test(child_session, child_io)
.await;
parent_io
.shutdown_and_wait()
.await
.expect("parent shutdown should succeed");
child_shutdown_rx
.await
.expect("guardian subagent should receive a shutdown op");
}
#[tokio::test]
async fn cached_guardian_subagent_exposes_its_rollout_path() {
let (parent_session, _parent_turn_context) = make_session_and_context().await;
let parent_session = Arc::new(parent_session);
let (mut child_session, _child_turn_context) = make_session_and_context().await;
let child_rollout_path = attach_thread_persistence(&mut child_session).await;
let (child_tx_sub, _child_rx_sub) = async_channel::bounded(4);
let (_child_tx_event, child_rx_event) = async_channel::unbounded();
let child_session_loop_handle = tokio::spawn(async {});
let child_session = Arc::new(child_session);
let child_io = SessionIo {
tx_sub: child_tx_sub,
rx_event: child_rx_event,
agent_status: watch::channel(AgentStatus::PendingInit).1,
session_loop_termination: session_loop_termination_from_handle(child_session_loop_handle),
};
parent_session
.guardian_review_session()
.cache_for_test(child_session, child_io)
.await;
assert_eq!(
parent_session
.guardian_review_session()
.trunk_rollout_path()
.await,
Some(child_rollout_path)
);
}
#[tokio::test]
async fn shutdown_and_wait_shuts_down_tracked_ephemeral_guardian_review() {
let (parent_session, parent_turn_context) = make_session_and_context().await;
let parent_session = Arc::new(parent_session);
let parent_config = Arc::clone(&parent_turn_context.config);
let (parent_tx_sub, parent_rx_sub) = async_channel::bounded(4);
let (_parent_tx_event, parent_rx_event) = async_channel::unbounded();
let parent_session_for_loop = Arc::clone(&parent_session);
let parent_session_loop_handle = tokio::spawn(async move {
submission_loop(parent_session_for_loop, parent_config, parent_rx_sub).await;
});
let parent_io = SessionIo {
tx_sub: parent_tx_sub,
rx_event: parent_rx_event,
agent_status: watch::channel(AgentStatus::PendingInit).1,
session_loop_termination: session_loop_termination_from_handle(parent_session_loop_handle),
};
let (child_session, _child_turn_context) = make_session_and_context().await;
let (child_tx_sub, child_rx_sub) = async_channel::bounded::<Submission>(4);
let (_child_tx_event, child_rx_event) = async_channel::unbounded();
let (child_shutdown_tx, child_shutdown_rx) = tokio::sync::oneshot::channel();
let child_session_loop_handle = tokio::spawn(async move {
let shutdown = child_rx_sub
.recv()
.await
.expect("child shutdown submission");
assert!(matches!(shutdown.op, Op::Shutdown));
child_shutdown_tx
.send(())
.expect("child shutdown signal should be delivered");
});
let child_session = Arc::new(child_session);
let child_io = SessionIo {
tx_sub: child_tx_sub,
rx_event: child_rx_event,
agent_status: watch::channel(AgentStatus::PendingInit).1,
session_loop_termination: session_loop_termination_from_handle(child_session_loop_handle),
};
parent_session
.guardian_review_session()
.register_ephemeral_for_test(child_session, child_io)
.await;
parent_io
.shutdown_and_wait()
.await
.expect("parent shutdown should succeed");
child_shutdown_rx
.await
.expect("ephemeral guardian review should receive a shutdown op");
}
pub(crate) async fn make_session_and_context_with_auth_and_config_and_rx<F>(
auth: CodexAuth,
dynamic_tools: Vec<DynamicToolSpec>,