Simplify Guardian eager compaction

This commit is contained in:
Ray Morgan
2026-06-17 10:35:57 -07:00
parent 7afecd191a
commit 5acc3a40e5
3 changed files with 8 additions and 83 deletions
+1 -1
View File
@@ -529,7 +529,7 @@ impl GuardianReviewSessionManager {
&& matches!(outcome, GuardianReviewSessionOutcome::Completed(_))
&& trunk.refresh_last_committed_fork_snapshot().await.is_some()
{
trunk.schedule_eager_compaction(params).await;
trunk.schedule_eager_compaction().await;
}
if keep_review_session {
@@ -1,6 +1,5 @@
use std::sync::Arc;
use codex_protocol::protocol::CompactedItem;
use tokio::sync::Mutex;
use tokio::sync::OwnedMutexGuard;
use tracing::warn;
@@ -9,10 +8,8 @@ use crate::guardian::GUARDIAN_REVIEW_TIMEOUT;
use crate::session::turn;
use super::GuardianReviewSession;
use super::GuardianReviewSessionParams;
use super::InitialHistory;
use super::RolloutItem;
use super::spawn_guardian_review_session;
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) enum GuardianEagerCompactionOutcome {
@@ -25,10 +22,7 @@ pub(super) enum GuardianEagerCompactionOutcome {
pub(super) type GuardianEagerCompaction = Arc<Mutex<GuardianEagerCompactionOutcome>>;
impl GuardianReviewSession {
pub(super) async fn schedule_eager_compaction(
self: &Arc<Self>,
params: GuardianReviewSessionParams,
) {
pub(super) async fn schedule_eager_compaction(self: &Arc<Self>) {
let turn_context = self.codex.session.new_default_turn().await;
if !turn::auto_compact_needed(self.codex.session.as_ref(), turn_context.as_ref()).await {
return;
@@ -38,10 +32,7 @@ impl GuardianReviewSession {
let review_session = Arc::clone(self);
drop(tokio::spawn(async move {
if review_session
.run_eager_compaction(turn_context, params)
.await
{
if review_session.run_eager_compaction(turn_context).await {
*outcome_guard = GuardianEagerCompactionOutcome::Reusable;
}
}));
@@ -56,7 +47,6 @@ impl GuardianReviewSession {
async fn run_eager_compaction(
self: &Arc<Self>,
turn_context: Arc<crate::session::turn_context::TurnContext>,
params: GuardianReviewSessionParams,
) -> bool {
let Ok(_review_guard) = self.review_lock.acquire().await else {
return false;
@@ -68,42 +58,14 @@ impl GuardianReviewSession {
return false;
};
let rollout_item_count = items.len();
let shadow_cancel = self.cancel_token.child_token();
let mut shadow_config = params.spawn_config.clone();
shadow_config.ephemeral = true;
let shadow = match spawn_guardian_review_session(
&params,
shadow_config,
self.reuse_key.clone(),
shadow_cancel.clone(),
Some(fork_snapshot),
)
.await
{
Ok(shadow) => Arc::new(shadow),
Err(err) => {
warn!(
guardian_thread_id = %self.codex.session.thread_id,
"failed to spawn eager guardian compaction session: {err}"
);
return false;
}
};
let shadow_for_events = Arc::clone(&shadow);
let event_drain =
tokio::spawn(
async move { while shadow_for_events.codex.next_event().await.is_ok() {} },
);
let shadow_turn = shadow.codex.session.new_default_turn().await;
let mut client_session = shadow.codex.session.services.model_client.new_session();
let mut client_session = self.codex.session.services.model_client.new_session();
let compact_result = tokio::select! {
_ = self.cancel_token.cancelled() => None,
result = tokio::time::timeout(
GUARDIAN_REVIEW_TIMEOUT,
turn::run_pre_turn_auto_compact(
&shadow.codex.session,
&shadow_turn,
&self.codex.session,
&turn_context,
&mut client_session,
),
) => match result {
@@ -117,31 +79,8 @@ impl GuardianReviewSession {
}
},
};
let reusable = match compact_result {
match compact_result {
Some(Ok(())) => {
let history = shadow
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
let compacted_item = CompactedItem {
message: String::new(),
replacement_history: Some(history.clone()),
};
self.codex
.session
.replace_compacted_history(
history,
shadow.codex.session.reference_context_item().await,
compacted_item,
)
.await;
self.codex
.session
.recompute_token_usage(&turn_context)
.await;
let durable = self
.refresh_last_committed_fork_snapshot()
.await
@@ -175,10 +114,6 @@ impl GuardianReviewSession {
false
}
None => false,
};
shadow_cancel.cancel();
shadow.shutdown().await;
let _ = event_drain.await;
reusable
}
}
}
-10
View File
@@ -2279,12 +2279,6 @@ async fn run_eager_compaction_recovery_case(
seed_guardian_parent_history(&session, &turn).await;
run_eager_compaction_review(&session, &turn, /*retry_reason*/ None).await?;
let discarded_rollout_path = session
.guardian_review_session
.trunk_rollout_path()
.await
.expect("discarded guardian rollout path");
let committed_rollout = tokio::fs::read(&discarded_rollout_path).await?;
tokio::time::timeout(
EAGER_COMPACTION_TEST_TIMEOUT,
server.wait_for_request_count(/*count*/ 2),
@@ -2306,10 +2300,6 @@ async fn run_eager_compaction_recovery_case(
if let Some(compaction_tx) = compaction_tx {
let _ = compaction_tx.send(());
}
assert_eq!(
tokio::fs::read(&discarded_rollout_path).await?,
committed_rollout
);
assert!(matches!(
metadata.guardian_session_kind,
Some(GuardianReviewSessionKind::EphemeralForked)