Preserve incoming prompts when pre-turn compaction fails (#44487)

## Why

Pre-turn compaction runs before incoming input is recorded, so failures could leave an accepted prompt out of conversation history. Reporting the error before prompt hooks finish also lets clients steer follow-up input into a turn still preserving its prompt.

## What changed

Record input and run prompt hooks on every pre-turn compaction failure. Defer local and remote compaction error events to `run_turn`, after prompt preservation completes, while retaining the remote error prefix.

## Testing

Extend the remote compaction output-validation test to trigger automatic compaction and assert that the prompt is emitted before the error, saved exactly once, and followed by no additional model request. Retain the assertion that usage is recorded for invalid compaction output.

GitOrigin-RevId: ab7ccb259836c084173e0e81dbbd4c610e95a6c8

codex-rs/core/src/compact.rs

@@ -306,8 +306,11 @@ async fn run_compact_task_inner_impl(
             }
             Err(e) if matches!(e.details(), CodexErrorDetails::SessionBudgetExceeded) => {
                 sess.track_turn_codex_error(turn_context.as_ref(), &e);
-                let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
-                sess.send_event(&turn_context, event).await;
+                // Pre-turn failures are reported after preserving the incoming prompt.
+                if !matches!(compaction_metadata.phase(), CompactionPhase::PreTurn) {
+                    let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
+                    sess.send_event(&turn_context, event).await;
+                }
                 return Err(e);
             }
             Err(e) if matches!(e.details(), CodexErrorDetails::ContextWindowExceeded) => {
@@ -322,8 +325,10 @@ async fn run_compact_task_inner_impl(
                 }
                 sess.set_total_tokens_full(turn_context.as_ref()).await;
                 sess.track_turn_codex_error(turn_context.as_ref(), &e);
-                let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
-                sess.send_event(&turn_context, event).await;
+                if !matches!(compaction_metadata.phase(), CompactionPhase::PreTurn) {
+                    let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
+                    sess.send_event(&turn_context, event).await;
+                }
                 return Err(e);
             }
             Err(e) => {
@@ -340,8 +345,10 @@ async fn run_compact_task_inner_impl(
                     continue;
                 } else {
                     sess.track_turn_codex_error(turn_context.as_ref(), &e);
-                    let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
-                    sess.send_event(&turn_context, event).await;
+                    if !matches!(compaction_metadata.phase(), CompactionPhase::PreTurn) {
+                        let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
+                        sess.send_event(&turn_context, event).await;
+                    }
                     return Err(e);
                 }
             }

codex-rs/core/src/compact_remote_v2.rs

@@ -208,10 +208,13 @@ async fn run_remote_compact_task_inner(
         Err(err) if matches!(err.details(), CodexErrorDetails::TurnAborted) => Err(err),
         Err(err) => {
             sess.track_turn_codex_error(turn_context, &err);
-            let event = EventMsg::Error(
-                err.to_error_event(Some("Error running remote compact task".to_string())),
-            );
-            sess.send_event(turn_context, event).await;
+            // Pre-turn failures are reported by run_turn after preserving the incoming prompt.
+            if !matches!(phase, CompactionPhase::PreTurn) {
+                let event = EventMsg::Error(
+                    err.to_error_event(Some("Error running remote compact task".to_string())),
+                );
+                sess.send_event(turn_context, event).await;
+            }
             Err(err)
         }
     }

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

@@ -188,15 +188,16 @@ pub(crate) async fn run_turn(
     )
     .await
     {
+        // Compaction runs before the new input is recorded, so preserve it on every failure.
+        run_hooks_and_record_inputs(
+            &sess,
+            &turn_context,
+            &turn_context.capture_current_model_info(),
+            &input,
+            PersistContext::Standard,
+        )
+        .await;
         if matches!(err.details(), CodexErrorDetails::TurnAborted) {
-            run_hooks_and_record_inputs(
-                &sess,
-                &turn_context,
-                &turn_context.capture_current_model_info(),
-                &input,
-                PersistContext::Standard,
-            )
-            .await;
             return Err(err);
         }
         if matches!(err.details(), CodexErrorDetails::ToolCollision(_)) {
@@ -205,6 +206,17 @@ pub(crate) async fn run_turn(
         let error = err.to_codex_protocol_error();
         sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone())
             .await;
+        // Publish the failure only after prompt hooks finish, so clients cannot react to
+        // an error by steering follow-up input into a turn still preserving its prompt.
+        let message_prefix = match turn_context.provider.capabilities().remote_compaction {
+            RemoteCompactionSupport::V2 => Some("Error running remote compact task".to_string()),
+            RemoteCompactionSupport::Unsupported => None,
+        };
+        sess.send_event(
+            turn_context.as_ref(),
+            EventMsg::Error(err.to_error_event(message_prefix)),
+        )
+        .await;
         error!("Failed to run pre-sampling compact");
         return Ok(None);
     }

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

@@ -436,15 +436,22 @@ async fn remote_compact_v2_records_usage_before_output_validation() -> Result<()
     skip_if_no_network!(Ok(()));
 
     let harness = TestCodexHarness::with_auto_env_builder(
-        test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()),
+        test_codex()
+            .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
+            .with_config(|config| {
+                config.model_auto_compact_token_limit = Some(200);
+            }),
     )
     .await?;
     let codex = &harness.test().codex;
     let rollout_path = codex.rollout_path().context("rollout path")?;
-    responses::mount_sse_sequence(
+    let responses_mock = responses::mount_sse_sequence(
         harness.server(),
         vec![
-            sse(vec![responses::ev_completed("before-compact")]),
+            sse(vec![responses::ev_completed_with_tokens(
+                "before-compact",
+                /*total_tokens*/ 500,
+            )]),
             sse(vec![
                 json!({
                     "type": "response.output_item.done",
@@ -467,9 +474,42 @@ async fn remote_compact_v2_records_usage_before_output_validation() -> Result<()
     .await;
 
     harness.test().submit_turn("before compact").await?;
-    codex.submit(Op::Compact).await?;
-    wait_for_event(codex, |event| matches!(event, EventMsg::Error(_))).await;
+    codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "turn that triggers auto compact".into(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+    let mut preserved_prompts = Vec::new();
+    wait_for_event(codex, |event| {
+        if let EventMsg::UserMessage(message) = event {
+            preserved_prompts.push(message.message.clone());
+        }
+        matches!(event, EventMsg::Error(_))
+    })
+    .await;
+    assert_eq!(preserved_prompts, vec!["turn that triggers auto compact"]);
     wait_for_event(codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
+    assert_eq!(responses_mock.requests().len(), 2);
+
+    codex.flush_rollout().await?;
+    let history = codex.load_history(/*include_archived*/ false).await?;
+    assert_eq!(
+        history
+            .items
+            .iter()
+            .filter(|item| matches!(
+                item,
+                RolloutItem::ResponseItem(envelope)
+                    if is_retained_user_message(
+                        &envelope.item,
+                        "turn that triggers auto compact",
+                    )
+            ))
+            .count(),
+        1,
+        "the accepted prompt should be saved exactly once after compaction fails"
+    );
     codex.shutdown_and_wait().await?;
 
     let record = fs::read_to_string(&rollout_path)?