openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-23 18:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Block Guardian reviews on eager compaction

Browse Source
This commit is contained in:
Ray Morgan
2026-06-17 10:35:58 -07:00
parent 8a7111f98d
commit 412c4c6eac
2 changed files with 28 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/guardian/review_session.rs
+16 -37
View File
@@ -62,7 +62,6 @@ mod eager_compaction;
use eager_compaction::GuardianEagerCompaction;
use eager_compaction::GuardianEagerCompactionOutcome;
const GUARDIAN_EAGER_COMPACTION_WAIT_TIMEOUT: Duration = Duration::from_secs(5);
const GUARDIAN_INTERRUPT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug)]
pub(crate) enum GuardianReviewSessionOutcome {
@@ -95,9 +94,9 @@ pub(crate) struct GuardianReviewSessionParams {
pub(crate) deadline: tokio::time::Instant,
}
#[derive(Default)]
pub(crate) struct GuardianReviewSessionManager {
state: Arc<Mutex<GuardianReviewSessionState>>,
eager_compaction_wait_timeout: Duration,
}
#[derive(Default)]
@@ -297,20 +296,7 @@ impl Drop for EphemeralReviewCleanup {
}
}
impl Default for GuardianReviewSessionManager {
fn default() -> Self {
Self::new(GUARDIAN_EAGER_COMPACTION_WAIT_TIMEOUT)
}
}
impl GuardianReviewSessionManager {
pub(crate) fn new(eager_compaction_wait_timeout: Duration) -> Self {
Self {
state: Arc::new(Mutex::new(GuardianReviewSessionState::default())),
eager_compaction_wait_timeout,
}
}
pub(crate) async fn trunk_rollout_path(&self) -> Option<PathBuf> {
let trunk = self.state.lock().await.trunk.clone()?;
trunk.codex.session.ensure_rollout_materialized().await;
@@ -433,36 +419,29 @@ impl GuardianReviewSessionManager {
let eager_compaction_guard = match run_before_review_deadline(
deadline,
params.external_cancel.as_ref(),
tokio::time::timeout(
self.eager_compaction_wait_timeout,
trunk.wait_for_eager_compaction(),
),
trunk.wait_for_eager_compaction(),
)
.await
{
Ok(Ok(guard)) => Some(guard),
Ok(Err(_)) => None,
Ok(guard) => guard,
Err(outcome) => {
return (outcome, GuardianReviewAnalyticsResult::without_session());
}
};
let eager_compaction_guard = match eager_compaction_guard {
Some(guard) if *guard == GuardianEagerCompactionOutcome::Reusable => guard,
guard => {
let review_session = self.remove_trunk_if_current(&trunk).await;
drop(guard);
if let Some(review_session) = review_session {
review_session.shutdown_in_background();
}
return Box::pin(self.run_ephemeral_review(
params,
next_reuse_key,
deadline,
/*fork_snapshot*/ None,
))
.await;
if *eager_compaction_guard == GuardianEagerCompactionOutcome::DiscardSession {
let review_session = self.remove_trunk_if_current(&trunk).await;
drop(eager_compaction_guard);
if let Some(review_session) = review_session {
review_session.shutdown_in_background();
}
};
return Box::pin(self.run_ephemeral_review(
params,
next_reuse_key,
deadline,
/*fork_snapshot*/ None,
))
.await;
}
// Hold the maintenance latch until review ownership is decided.
let state = self.state.lock().await;
let trunk_pointer_is_current = state
codex-rs/core/src/guardian/tests.rs
+12 -49
View File
@@ -2236,17 +2236,8 @@ async fn run_eager_compaction_review(
}
}
#[derive(Clone, Copy)]
enum EagerCompactionRecoveryCase {
Failure,
Timeout,
}
async fn run_eager_compaction_recovery_case(
recovery_case: EagerCompactionRecoveryCase,
) -> anyhow::Result<()> {
const FAILURE_WAIT_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 3_600);
const TIMEOUT_WAIT_TIMEOUT: Duration = Duration::from_millis(/*millis*/ 50);
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_recovers_from_eager_compaction_failure() -> anyhow::Result<()> {
const FIRST_REVIEW_TOTAL_TOKENS: i64 = 500_000;
let (compaction_tx, compaction_rx) = tokio::sync::oneshot::channel();
let failed_compaction_sse = |id: &str| {
@@ -2255,38 +2246,23 @@ async fn run_eager_compaction_recovery_case(
ev_assistant_message(&format!("msg-eager-compact-{id}"), "partial summary"),
])
};
let compaction_body = match recovery_case {
EagerCompactionRecoveryCase::Failure => failed_compaction_sse("failed"),
EagerCompactionRecoveryCase::Timeout => {
sse(vec![ev_response_created("resp-eager-compact-timeout")])
}
};
let mut responses = vec![
let responses = vec![
guardian_review_sse_with_body_tokens(
"1",
FIRST_REVIEW_TOTAL_TOKENS - EAGER_COMPACTION_GUARDIAN_PREFILL_TOKENS,
),
vec![StreamingSseChunk {
gate: Some(compaction_rx),
body: compaction_body,
body: failed_compaction_sse("failed"),
}],
];
if matches!(recovery_case, EagerCompactionRecoveryCase::Failure) {
responses.push(vec![StreamingSseChunk {
vec![StreamingSseChunk {
gate: None,
body: failed_compaction_sse("retry-failed"),
}]);
}
responses.push(guardian_review_sse("2", ev_completed));
}],
guardian_review_sse("2", ev_completed),
];
let (server, _) = start_streaming_sse_server(responses).await;
let (mut session, mut turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
let eager_compaction_wait_timeout = match recovery_case {
EagerCompactionRecoveryCase::Failure => FAILURE_WAIT_TIMEOUT,
EagerCompactionRecoveryCase::Timeout => TIMEOUT_WAIT_TIMEOUT,
};
Arc::get_mut(&mut session)
.expect("guardian parent session should be uniquely owned")
.guardian_review_session = GuardianReviewSessionManager::new(eager_compaction_wait_timeout);
let (session, mut turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
configure_guardian_eager_compaction_test(&mut turn);
seed_guardian_parent_history(&session, &turn).await;
@@ -2296,22 +2272,15 @@ async fn run_eager_compaction_recovery_case(
server.wait_for_request_count(/*count*/ 2),
)
.await?;
let mut compaction_tx = Some(compaction_tx);
if matches!(recovery_case, EagerCompactionRecoveryCase::Failure) {
let compaction_tx = compaction_tx.take().expect("compaction response gate");
compaction_tx
.send(())
.expect("failed compaction response gate should still be open");
}
compaction_tx
.send(())
.expect("failed compaction response gate should still be open");
let metadata = run_eager_compaction_review(
&session,
&turn,
Some("Retry after eager compaction maintenance."),
)
.await?;
if let Some(compaction_tx) = compaction_tx {
let _ = compaction_tx.send(());
}
assert!(matches!(
metadata.guardian_session_kind,
Some(GuardianReviewSessionKind::EphemeralForked)
@@ -2319,12 +2288,6 @@ async fn run_eager_compaction_recovery_case(
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_recovers_from_eager_compaction_failure_and_timeout() -> anyhow::Result<()> {
run_eager_compaction_recovery_case(EagerCompactionRecoveryCase::Failure).await?;
run_eager_compaction_recovery_case(EagerCompactionRecoveryCase::Timeout).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_reused_trunk_ignores_stale_prior_turn_completion() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));