Avoid panics after background compaction fails before a checkpoint (#55581)

## Why

Polling a failed background compaction consumes the closed oneshot receiver.
Keeping that receiver in session state can cause a later await to panic instead
of falling back to foreground compaction.

## What changed

Remove the receiver from session state before polling it, restoring it only when
its result is still pending. Discard closed receivers so compaction can fall back
normally.

## Testing

Add a regression scenario where background compaction fails before the next
checkpoint and automatic compaction falls back successfully.

GitOrigin-RevId: 1461bb403f4b0b8b7e171d1b3cba41411f2014b4

codex-rs/core/src/session/async_compaction.rs

@@ -149,8 +149,16 @@ pub(super) async fn take_ready(
         return None;
     }
     let mut state = sess.state.lock().await;
-    let candidate = state.async_compaction.result.as_mut()?.try_recv().ok()?;
-    state.async_compaction.result = None;
+    let mut result = state.async_compaction.result.take()?;
+    let candidate = match result.try_recv() {
+        Ok(candidate) => candidate,
+        Err(oneshot::error::TryRecvError::Empty) => {
+            state.async_compaction.result = Some(result);
+            return None;
+        }
+        // try_recv consumes a closed receiver; awaiting it later would panic.
+        Err(oneshot::error::TryRecvError::Closed) => return None,
+    };
     (candidate.prefix.generation == state.history.history_version()).then_some(candidate)
 }
 

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

@@ -3010,11 +3010,13 @@ enum PendingCompactionOutcome {
     SuccessOverBudget,
     StaleSuccess,
     Failure,
+    ReadyFailure,
     Interrupted,
 }
 
 #[test_case::test_case(PendingCompactionOutcome::Success, CompactionTrigger::Auto; "auto uses pending result")]
 #[test_case::test_case(PendingCompactionOutcome::Failure, CompactionTrigger::Auto; "auto falls back after failure")]
+#[test_case::test_case(PendingCompactionOutcome::ReadyFailure, CompactionTrigger::Auto; "auto falls back after failure before checkpoint")]
 #[test_case::test_case(PendingCompactionOutcome::StaleSuccess, CompactionTrigger::Auto; "auto falls back after account changes")]
 #[test_case::test_case(PendingCompactionOutcome::Interrupted, CompactionTrigger::Auto; "auto can be interrupted")]
 #[test_case::test_case(PendingCompactionOutcome::Absent, CompactionTrigger::Manual; "manual without background")]
@@ -3042,12 +3044,16 @@ async fn async_compaction_reuses_pending_work(
     } else {
         10_000
     };
-    let valid_output = outcome != PendingCompactionOutcome::Failure;
+    let valid_output = !matches!(
+        outcome,
+        PendingCompactionOutcome::Failure | PendingCompactionOutcome::ReadyFailure
+    );
     let succeeds = !matches!(
         outcome,
         PendingCompactionOutcome::Absent
             | PendingCompactionOutcome::SuccessOverBudget
             | PendingCompactionOutcome::Failure
+            | PendingCompactionOutcome::ReadyFailure
             | PendingCompactionOutcome::StaleSuccess
     );
     let response = |id, event, tokens| vec![event, ev_completed_with_tokens(id, tokens)];
@@ -3113,6 +3119,20 @@ async fn async_compaction_reuses_pending_work(
     if has_background {
         tokio::time::timeout(timeout, streaming.wait_for_request_count(/*count*/ 2)).await?;
     }
+    let release_background = if outcome == PendingCompactionOutcome::ReadyFailure {
+        release_background
+            .send(())
+            .expect("fail background compaction before the next checkpoint");
+        // The client must observe the failure before take_ready polls the closed receiver.
+        tokio::time::timeout(
+            timeout,
+            test.codex.wait_for_background_compaction_for_test(),
+        )
+        .await?;
+        None
+    } else {
+        Some(release_background)
+    };
     if manual {
         test.codex.submit(Op::Compact).await?;
     } else if over_budget {
@@ -3132,7 +3152,7 @@ async fn async_compaction_reuses_pending_work(
             if matches!(event.item, TurnItem::ContextCompaction(_)))
     })
     .await;
-    if has_background {
+    if has_background && outcome != PendingCompactionOutcome::ReadyFailure {
         assert_eq!(streaming.requests().await.len(), if manual { 2 } else { 3 });
     }
     if outcome == PendingCompactionOutcome::Interrupted {
@@ -3141,7 +3161,10 @@ async fn async_compaction_reuses_pending_work(
             matches!(event, EventMsg::TurnAborted(_))
         })
         .await;
-        release_background.send(()).expect("release unused result");
+        release_background
+            .expect("pending background compaction")
+            .send(())
+            .expect("release unused result");
         completions.remove(1).await?;
         test.codex.shutdown_and_wait().await?;
         assert_eq!(streaming.requests().await.len(), if manual { 2 } else { 3 });
@@ -3161,7 +3184,7 @@ async fn async_compaction_reuses_pending_work(
         )?;
         auth.reload().await;
     }
-    if has_background {
+    if has_background && let Some(release_background) = release_background {
         release_background.send(()).expect("release pending result");
     }
     let EventMsg::ItemStarted(started) = started else {