From ee93abb690291b19fcabb14255437f5cc5493bdc Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 10 Sep 2026 09:58:58 +0000 Subject: [PATCH] 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 | 19 +++++--- codex-rs/core/src/compact_remote_v2.rs | 11 +++-- codex-rs/core/src/session/turn.rs | 28 ++++++++---- codex-rs/core/tests/suite/compact_remote.rs | 50 ++++++++++++++++++--- 4 files changed, 85 insertions(+), 23 deletions(-) diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index f39a57169f..55cc1718ba 100644 --- a/codex-rs/core/src/compact.rs +++ b/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); } } diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 153fc0a899..376ad06c99 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/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) } } diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 3950d34d13..8940adb91f 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/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); } diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index ac674f6b9f..53dee6ea97 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/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)?