github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 04:45:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve conversation history when compaction fails (#53932)

Browse Source 
## Why

Compaction could record partial output before the response completed, leaving failed summaries in conversation history and persisted rollouts.

## What changed

Buffer compaction output until `response.completed` so failed or incomplete attempts leave no partial output in history. Record successful output before the response completion notification, preserving its compaction metadata.

## Testing

Add regression coverage for preserving messages, tool results, and token usage after a disconnected compaction stream, including after restart. Verify incomplete compaction output is excluded from subsequent requests and successful output remains observable and persisted.

GitOrigin-RevId: 8651200b2aae616000c55c1ed63283f3e2c1b8d5
This commit is contained in:
jif
2026-10-14 09:44:24 +00:00
committed by copyberry
parent 95273b84fb
commit 400f2356d6
2 changed files with 239 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/compact.rs
+18 -20
View File
@@ -858,26 +858,8 @@ async fn drain_to_completed(
));
};
match event {
Ok(ResponseEvent::OutputItemDone(item)) => {
if matches!(phase, CompactionPhase::PostTurn) {
// Commit post-turn summaries only after success; failures must leave both
// the live history and persisted rollout intact.
output.push(item);
} else {
sess.record_annotated_conversation_items(
turn_context,
turn_context.model_info(),
vec![ResponseItemEnvelope {
item,
metadata: Some(CodexHarnessMetadata {
compaction_output: true,
..Default::default()
}),
}],
)
.await;
}
}
// Keep each attempt isolated until the response completes successfully.
Ok(ResponseEvent::OutputItemDone(item)) => output.push(item),
Ok(ResponseEvent::ServerReasoningIncluded(included)) => {
sess.set_server_reasoning_included(included).await;
}
@@ -890,6 +872,22 @@ async fn drain_to_completed(
usage_metadata,
..
}) => {
if !matches!(phase, CompactionPhase::PostTurn) {
for item in std::mem::take(&mut output) {
sess.record_annotated_conversation_items(
turn_context,
turn_context.model_info(),
vec![ResponseItemEnvelope {
item,
metadata: Some(CodexHarnessMetadata {
compaction_output: true,
..Default::default()
}),
}],
)
.await;
}
}
sess.record_observed_response_completed(
turn_context,
&response_id,
codex-rs/core/tests/suite/compact.rs
+221 -22
View File
@@ -19,6 +19,7 @@ use codex_protocol::config_types::Settings;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ImageReference;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelServiceTier;
use codex_protocol::openai_models::ModelsResponse;
@@ -1174,6 +1175,17 @@ async fn incomplete_response_during_compaction_is_terminal_and_counted(
let server = start_mock_server().await;
let previous_model = "gpt-5.4";
let selected_model = "gpt-5.2";
let incomplete = sse(vec![
ev_assistant_message("discarded", "DISCARDED_INCOMPLETE_SUMMARY"),
json!({
"type": "response.incomplete",
"response": {
"id": "incomplete-compaction",
"incomplete_details": {"reason": "max_output_tokens"},
"usage": reported_usage
}
}),
]);
let response_mock = mount_sse_sequence(
&server,
vec![
@@ -1181,14 +1193,8 @@ async fn incomplete_response_during_compaction_is_terminal_and_counted(
ev_assistant_message("initial", "Initial reply."),
ev_completed_with_tokens("initial", /*total_tokens*/ 120_000),
]),
sse(vec![json!({
"type": "response.incomplete",
"response": {
"id": "incomplete-compaction",
"incomplete_details": {"reason": "max_output_tokens"},
"usage": reported_usage
}
})]),
incomplete.clone(),
incomplete,
],
)
.await;
@@ -1215,7 +1221,9 @@ async fn incomplete_response_during_compaction_is_terminal_and_counted(
})
.build_with_auto_env(&server)
.await?;
test.submit_turn("hello").await?;
// A real large history remains installed after failure. Server usage also
// includes request overhead that the local history estimate does not count.
test.submit_turn(&"x".repeat(440_000)).await?;
// A smaller selected model forces compaction with the previous model.
// An incomplete result must stop both retries and fallback to the selected model.
test.codex
@@ -1272,6 +1280,138 @@ async fn incomplete_response_during_compaction_is_terminal_and_counted(
.collect::<Vec<_>>(),
vec![json!(previous_model), json!(previous_model)],
);
// Failure has not made the original oversized history fit the smaller model.
// The next turn must attempt compaction again without retaining partial output.
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "continue after the failed compaction".into(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = response_mock.requests();
assert_eq!(requests.len(), 3);
assert_eq!(requests[2].body_json()["model"], json!(previous_model));
assert!(
body_contains_text(&requests[2].body_json().to_string(), SUMMARIZATION_PROMPT)
|| !requests[2].inputs_of_type("compaction_trigger").is_empty()
);
assert!(!body_contains_text(
&requests[2].body_json().to_string(),
"DISCARDED_INCOMPLETE_SUMMARY",
));
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_compaction_preserves_history_after_restart() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let initial = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_assistant_message("before-tool", "Checking the original state."),
ev_exec_command_call("original-tool", "echo ORIGINAL_TOOL_RESULT"),
ev_completed_with_tokens("tool-response", /*total_tokens*/ 1_000),
]),
sse(vec![
ev_assistant_message("after-tool", FIRST_REPLY),
ev_completed_with_tokens("initial-response", /*total_tokens*/ 2_000),
]),
],
)
.await;
let model_provider = non_openai_model_provider(&server);
let mut builder = test_codex()
.with_pre_build_hook(allow_echo_commands)
.with_config(move |config| {
config.model_provider = model_provider;
config.model_provider.stream_max_retries = Some(0);
set_test_compact_prompt(config);
});
let test = builder.build_with_auto_env(&server).await?;
test.submit_turn("inspect the original state").await?;
assert_eq!(initial.requests().len(), 2);
let original_usage = test.codex.token_usage_info().await.expect("original usage");
let failed_compaction = mount_sse_once(
&server,
sse(vec![
ev_reasoning_item("discarded-reasoning", &["Summarizing"], &[]),
ev_assistant_message("discarded-summary", "DISCARDED_PARTIAL_SUMMARY"),
]),
)
.await;
test.codex.submit(Op::Compact).await?;
let mut errors = Vec::new();
wait_for_event(&test.codex, |event| {
if let EventMsg::Error(error) = event {
errors.push(error.message.clone());
}
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(
errors,
vec!["stream disconnected before completion: stream closed before response.completed"]
);
assert_eq!(test.codex.token_usage_info().await, Some(original_usage));
let original_request = failed_compaction.single_request();
let original_items = original_request
.input()
.into_iter()
.filter(|item| {
item["role"] == "assistant"
|| item["type"] == "reasoning"
|| item["call_id"] == "original-tool"
})
.collect::<Vec<_>>();
assert_eq!(original_items.len(), 4);
assert!(
original_request
.function_call_output("original-tool")
.to_string()
.contains("ORIGINAL_TOOL_RESULT")
);
// A live follow-up and a compaction retry after restart must both see the original evidence.
let follow_up = mount_sse_once(&server, sse(vec![ev_completed("follow-up")])).await;
test.submit_text_turn("continue after the failed compaction")
.await?;
let model_provider = test.config.model_provider.clone();
builder = builder.with_config(move |config| {
config.model_provider = model_provider;
set_test_compact_prompt(config);
});
let resumed = builder.restart_with_auto_env(&server, &test).await?;
let retry = mount_sse_once(
&server,
sse(vec![
ev_assistant_message("valid-summary", SUMMARY_TEXT),
ev_completed("successful-compaction"),
]),
)
.await;
resumed.codex.submit(Op::Compact).await?;
assert_compaction_uses_turn_lifecycle_id(&resumed.codex).await;
for request in [follow_up.single_request(), retry.single_request()] {
let retained_items = request
.input()
.into_iter()
.filter(|item| {
item["role"] == "assistant"
|| item["type"] == "reasoning"
|| item["call_id"] == "original-tool"
})
.collect::<Vec<_>>();
assert_eq!(retained_items, original_items);
}
Ok(())
}
@@ -1289,19 +1429,34 @@ async fn manual_compact_records_durable_and_local_token_usage() {
ev_assistant_message("m1", SUMMARY_TEXT),
ev_completed_with_tokens("r1", /*total_tokens*/ 0),
]);
mount_sse_once(&server, sse_compact).await;
mount_sse_sequence(
&server,
vec![
sse(vec![
ev_assistant_message("initial", FIRST_REPLY),
ev_completed_with_tokens("initial", /*total_tokens*/ 1_000),
]),
sse_compact,
],
)
.await;
let model_provider = non_openai_model_provider(&server);
let mut builder = test_codex().with_config(move |config| {
config.model_provider = model_provider;
set_test_compact_prompt(config);
});
let test = builder.build(&server).await.unwrap();
let test = builder.build_with_auto_env(&server).await.unwrap();
test.submit_turn("hello").await.unwrap();
let rollout_path = test.codex.rollout_path().expect("rollout path");
let codex = test.codex;
// Trigger manual compact and collect TokenCount events for the compact turn.
codex.submit(Op::Compact).await.unwrap();
wait_for_event(&codex, |event| {
matches!(event, EventMsg::RawResponseCompleted(response) if response.response_id == "r1")
})
.await;
// First TokenCount: from the compact API call (usage.total_tokens = 0).
let first = wait_for_event_match(&codex, |ev| match ev {
@@ -1347,9 +1502,9 @@ async fn manual_compact_records_durable_and_local_token_usage() {
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(records.len(), 1);
assert_eq!(records[0].response_id, "r1");
assert_eq!(records[0].usage.total_tokens, 0);
assert_eq!(records.len(), 2);
assert_eq!(records[1].response_id, "r1");
assert_eq!(records[1].usage.total_tokens, 0);
let compacted = rollout_items
.iter()
.find_map(|item| match item {
@@ -1362,7 +1517,7 @@ async fn manual_compact_records_durable_and_local_token_usage() {
compacted.compaction_response_id.as_deref(),
compacted.latest_token_usage_record.as_ref(),
),
(Some("r1"), records.first().copied())
(Some("r1"), records.last().copied())
);
}
@@ -1376,10 +1531,15 @@ async fn manual_compact_emits_context_compaction_items() {
ev_assistant_message("m1", FIRST_REPLY),
ev_completed("r1"),
]);
let sse2 = sse(vec![
ev_assistant_message("m2", SUMMARY_TEXT),
ev_completed("r2"),
]);
let reasoning = ev_reasoning_item("compact-reasoning", &["Summarizing the conversation"], &[]);
let summary = ev_assistant_message("m2", SUMMARY_TEXT);
let mut expected_summary = summary["item"].clone();
expected_summary["internal_chat_message_metadata_passthrough"] =
json!({ "content_item_kinds": ["unknown"] });
let mut expected_raw_items = [reasoning["item"].clone(), expected_summary]
.map(|item| serde_json::from_value::<ResponseItem>(item).unwrap())
.to_vec();
let sse2 = sse(vec![reasoning, summary, ev_completed("r2")]);
mount_sse_sequence(&server, vec![sse1, sse2]).await;
let model_provider = non_openai_model_provider(&server);
@@ -1387,7 +1547,8 @@ async fn manual_compact_emits_context_compaction_items() {
config.model_provider = model_provider;
set_test_compact_prompt(config);
});
let codex = builder.build(&server).await.unwrap().codex;
let test = builder.build_with_auto_env(&server).await.unwrap();
let codex = &test.codex;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
@@ -1396,14 +1557,19 @@ async fn manual_compact_emits_context_compaction_items() {
}]))
.await
.unwrap();
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
wait_for_event(codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
codex.submit(Op::Compact).await.unwrap();
let compact_turn_id = codex.submit(Op::Compact).await.unwrap();
for item in &mut expected_raw_items {
item.set_turn_id_if_missing(&compact_turn_id);
}
let mut started_item = None;
let mut completed_item = None;
let mut legacy_event = false;
let mut saw_turn_complete = false;
let mut raw_items = Vec::new();
let mut raw_items_at_response_completion = None;
while !saw_turn_complete || started_item.is_none() || completed_item.is_none() || !legacy_event
{
@@ -1427,6 +1593,10 @@ async fn manual_compact_emits_context_compaction_items() {
EventMsg::TurnComplete(_) => {
saw_turn_complete = true;
}
EventMsg::RawResponseItem(event) => raw_items.push(event.item),
EventMsg::RawResponseCompleted(_) => {
raw_items_at_response_completion = Some(raw_items.clone());
}
_ => {}
}
}
@@ -1435,6 +1605,35 @@ async fn manual_compact_emits_context_compaction_items() {
let completed_item = completed_item.expect("context compaction item completed");
assert_eq!(started_item.id, completed_item.id);
assert!(legacy_event);
// Raw output remains observable before the terminal response notification,
// independently of when the summary is installed into conversation history.
assert_eq!(
raw_items_at_response_completion,
Some(expected_raw_items.clone())
);
let rollout =
fs::read_to_string(test.codex.rollout_path().expect("rollout path")).expect("read rollout");
let persisted_output = rollout
.lines()
.map(|line| {
codex_rollout::parse_rollout_line(line)
.expect("rollout item")
.item
})
.filter_map(|item| match item {
RolloutItem::ResponseItem(envelope)
if envelope
.metadata
.as_ref()
.is_some_and(|metadata| metadata.compaction_output) =>
{
Some(envelope.item)
}
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(persisted_output, expected_raw_items);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]