Budget retained images during remote compaction (#40280)

Remote compaction's retained-message budget counted text but not images, so
image-heavy history could retain more context than the budget represented.

- Add the opt-in `compaction_image_budget` feature to charge retained images
  using the existing image size estimate.
- Keep images and their adjacent labels atomic when truncating a boundary
  message, while preserving existing text, audio, metadata, and
  client-authored developer-message behavior.
- Stop backfilling older messages when an image at the truncation boundary
  does not fit.

- Add unit coverage for image boundaries, annotations, audio, text-only
  messages, and client-authored developer messages.
- Add an integration test covering repeated remote compaction with the feature
  enabled, disabled, and left at its default.

GitOrigin-RevId: 88868bbd09d2858939dbae3f8402e1f497b4e3c6
This commit is contained in:
ashwinnathan-openai
2026-08-23 14:42:10 -07:00
committed by Owen Lin
parent 052728fc22
commit afe97cba99
9 changed files with 503 additions and 23 deletions
+6
View File
@@ -529,6 +529,9 @@
"collaboration_modes": {
"type": "boolean"
},
"compaction_image_budget": {
"type": "boolean"
},
"computer_use": {
"type": "boolean"
},
@@ -5426,6 +5429,9 @@
"collaboration_modes": {
"type": "boolean"
},
"compaction_image_budget": {
"type": "boolean"
},
"computer_use": {
"type": "boolean"
},
+68 -11
View File
@@ -60,6 +60,15 @@ mod attempt;
use attempt::RemoteCompactV2Attempt;
use attempt::run_remote_compact_v2_attempt;
#[path = "compact_remote_v2_images.rs"]
mod images;
#[derive(Clone, Copy, PartialEq, Eq)]
enum RetainedImageBudget {
Disabled,
Enabled,
}
// Mirror the current /responses/compact retained-message default while the
// server-side path remains the reference implementation.
pub(crate) const RETAINED_MESSAGE_TOKEN_BUDGET: usize = 64_000;
@@ -299,6 +308,11 @@ async fn run_remote_compact_task_inner_impl(
prompt_input_metadata,
compaction_output,
sess.enabled(Feature::RetainClientDeveloperMessages),
if sess.enabled(Feature::CompactionImageBudget) {
RetainedImageBudget::Enabled
} else {
RetainedImageBudget::Disabled
},
);
analytics_details.retained_image_count = Some(retained_images);
let (new_window_number, new_window_ids) = sess.advance_auto_compact_window().await;
@@ -461,6 +475,7 @@ fn build_v2_compacted_history(
prompt_input_metadata: Vec<Option<CodexHarnessMetadata>>,
compaction_output: ResponseItem,
retain_client_developer_messages: bool,
image_budget: RetainedImageBudget,
) -> (Vec<ResponseItemEnvelope>, usize) {
debug_assert_eq!(prompt_input.len(), prompt_input_metadata.len());
let prompt_input = prompt_input
@@ -478,7 +493,7 @@ fn build_v2_compacted_history(
.flat_map(HistoryItemGroup::into_items)
.collect::<Vec<_>>();
let mut retained =
truncate_retained_messages_for_remote_compaction(retained, RETAINED_MESSAGE_TOKEN_BUDGET);
truncate_retained_messages(retained, RETAINED_MESSAGE_TOKEN_BUDGET, image_budget);
let retained_image_count = retained
.iter()
.map(|envelope| retained_input_image_count(&envelope.item))
@@ -556,6 +571,14 @@ fn retained_input_image_count(item: &ResponseItem) -> usize {
pub(crate) fn truncate_retained_messages_for_remote_compaction(
items: Vec<ResponseItemEnvelope>,
max_tokens: usize,
) -> Vec<ResponseItemEnvelope> {
truncate_retained_messages(items, max_tokens, RetainedImageBudget::Disabled)
}
fn truncate_retained_messages(
items: Vec<ResponseItemEnvelope>,
max_tokens: usize,
image_budget: RetainedImageBudget,
) -> Vec<ResponseItemEnvelope> {
let mut remaining = max_tokens;
let mut truncated_reversed = Vec::with_capacity(items.len());
@@ -569,14 +592,22 @@ pub(crate) fn truncate_retained_messages_for_remote_compaction(
}
let client_developer = is_client_authored_developer_message(&group.source);
let charge_images = image_budget == RetainedImageBudget::Enabled && !client_developer;
let notice_tokens = group
.attached_notice
.as_ref()
.map_or(0, |notice| message_text_token_count(&notice.item).max(1));
// Client-authored developer messages already charge non-text content via
// the serialized estimate. Preserve their text-only boundary correction.
let content_tokens = if charge_images {
message_content_token_count(&group.source.item)
} else {
message_text_token_count(&group.source.item)
};
let source_tokens = if client_developer {
usize::try_from(estimate_item_token_count(&group.source.item)).unwrap_or(usize::MAX)
} else {
message_text_token_count(&group.source.item).max(1)
content_tokens.max(1)
};
let token_count = source_tokens.saturating_add(notice_tokens);
if token_count <= remaining {
@@ -587,23 +618,30 @@ pub(crate) fn truncate_retained_messages_for_remote_compaction(
remaining = remaining.saturating_sub(token_count);
} else if remaining > notice_tokens {
let available_tokens = remaining - notice_tokens;
let text_budget = if client_developer {
available_tokens.saturating_sub(
source_tokens.saturating_sub(message_text_token_count(&group.source.item)),
)
let content_budget = if client_developer {
available_tokens.saturating_sub(source_tokens.saturating_sub(content_tokens))
} else {
available_tokens
};
let Some(mut truncated_item) =
truncate_message_text_to_token_budget(group.source, text_budget)
else {
let image_count = retained_input_image_count(&group.source.item);
if charge_images && image_count > 0 {
// An oversized image can leave no boundary content. Do not backfill
// the remaining budget with older messages in that case.
remaining = 0;
}
let truncated_item = if charge_images && image_count > 0 {
images::truncate_message_to_token_budget(group.source, content_budget)
} else {
truncate_message_text_to_token_budget(group.source, content_budget)
};
let Some(mut truncated_item) = truncated_item else {
continue;
};
if client_developer {
let item_tokens = usize::try_from(estimate_item_token_count(&truncated_item.item))
.unwrap_or(usize::MAX);
if item_tokens > available_tokens {
let adjusted_budget = text_budget
let adjusted_budget = content_budget
.saturating_sub(item_tokens - available_tokens)
.saturating_sub(1);
let Some(adjusted) =
@@ -625,12 +663,22 @@ pub(crate) fn truncate_retained_messages_for_remote_compaction(
}
truncated_reversed.push(truncated_item);
remaining = 0;
} else if charge_images && retained_input_image_count(&group.source.item) > 0 {
remaining = 0;
}
}
truncated_reversed.reverse();
truncated_reversed
}
fn message_content_token_count(item: &ResponseItem) -> usize {
let ResponseItem::Message { content, .. } = item else {
return usize::try_from(estimate_item_token_count(item)).unwrap_or(usize::MAX);
};
content.iter().map(images::content_item_token_count).sum()
}
fn message_text_token_count(item: &ResponseItem) -> usize {
let ResponseItem::Message { content, .. } = item else {
return usize::try_from(estimate_item_token_count(item)).unwrap_or(usize::MAX);
@@ -735,7 +783,11 @@ mod tests {
) -> (Vec<ResponseItemEnvelope>, usize) {
let metadata = vec![None; input.len()];
build_v2_compacted_history(
input, metadata, output, /*retain_client_developer_messages*/ false,
input,
metadata,
output,
/*retain_client_developer_messages*/ false,
RetainedImageBudget::Disabled,
)
}
@@ -846,6 +898,7 @@ mod tests {
],
output.clone(),
enabled,
RetainedImageBudget::Disabled,
);
let mut expected = vec![
ResponseItemEnvelope {
@@ -1102,3 +1155,7 @@ mod tests {
);
}
}
#[cfg(test)]
#[path = "compact_remote_v2_image_budget_tests.rs"]
mod image_budget_tests;
@@ -0,0 +1,162 @@
use super::*;
use codex_protocol::models::ContentItemKind;
use codex_protocol::models::InternalChatMessageMetadataPassthrough;
use codex_protocol::models::image_close_tag_text;
use codex_protocol::models::local_image_open_tag_text_with_path;
use pretty_assertions::assert_eq;
fn message(content: Vec<ContentItem>) -> ResponseItem {
ResponseItem::Message {
id: None,
role: "user".to_string(),
content,
phase: None,
internal_chat_message_metadata_passthrough: None,
}
}
fn image() -> ContentItem {
ContentItem::InputImage {
image_url: "data:image/png;base64,abc".to_string(),
detail: None,
}
}
fn text(value: &str) -> ContentItem {
ContentItem::InputText {
text: value.to_string(),
}
}
fn trim(items: Vec<ResponseItem>, max_tokens: usize) -> Vec<ResponseItem> {
truncate_retained_messages(
items.into_iter().map(ResponseItemEnvelope::new).collect(),
max_tokens,
RetainedImageBudget::Enabled,
)
.into_iter()
.map(ResponseItemEnvelope::into_item)
.collect()
}
#[test]
fn image_only_boundary_is_atomic_and_does_not_backfill_older_messages() {
let newest = message(vec![text("new")]);
let items = vec![
message(vec![text("old")]),
message(vec![image()]),
newest.clone(),
];
let image_tokens = images::content_item_token_count(&image());
for (max_tokens, expected) in [
(
image_tokens + 1,
vec![message(vec![image()]), newest.clone()],
),
(image_tokens, vec![newest.clone()]),
(1, vec![newest]),
] {
assert_eq!(trim(items.clone(), max_tokens), expected);
}
}
#[test]
fn later_image_parts_preserve_labels_audio_and_annotations() {
let parts = vec![
text("earlier text"),
text(&local_image_open_tag_text_with_path(
/*label_number*/ 7,
std::path::Path::new("image.png"),
)),
image(),
text(&image_close_tag_text()),
text("keep"),
ContentItem::InputAudio {
audio_url: "data:audio/wav;base64,abc".to_string(),
},
];
let kinds = (0..parts.len())
.map(|i| ContentItemKind(format!("part.{i}")))
.collect::<Vec<_>>();
let mut source = message(parts.clone());
let ResponseItem::Message {
internal_chat_message_metadata_passthrough,
..
} = &mut source
else {
unreachable!()
};
*internal_chat_message_metadata_passthrough = Some(InternalChatMessageMetadataPassthrough {
turn_id: Some("turn-1".to_string()),
content_item_kinds: Some(kinds.clone()),
..Default::default()
});
let image_tokens = parts[1..4]
.iter()
.map(images::content_item_token_count)
.sum::<usize>();
for (max_tokens, start) in [(image_tokens, 4), (image_tokens + 1, 1)] {
let mut expected = source.clone();
let ResponseItem::Message {
content,
internal_chat_message_metadata_passthrough,
..
} = &mut expected
else {
unreachable!()
};
*content = parts[start..].to_vec();
internal_chat_message_metadata_passthrough
.as_mut()
.unwrap()
.content_item_kinds = Some(kinds[start..].to_vec());
assert_eq!(trim(vec![source.clone()], max_tokens), vec![expected]);
}
}
#[test]
fn image_treatment_preserves_text_only_boundary_behavior() {
let source = ResponseItemEnvelope::new(message(vec![
text("earlier text"),
text(&"middle".repeat(20)),
text("later text"),
ContentItem::InputAudio {
audio_url: "data:audio/wav;base64,abc".to_string(),
},
]));
for max_tokens in [0, 1, 4, 16, 100] {
assert_eq!(
truncate_retained_messages(
vec![source.clone()],
max_tokens,
RetainedImageBudget::Enabled,
),
truncate_retained_messages_for_remote_compaction(vec![source.clone()], max_tokens),
);
}
}
#[test]
fn image_treatment_preserves_client_developer_boundary_behavior() {
let mut item = message(vec![image(), text(&"a".repeat(1000))]);
let ResponseItem::Message { role, .. } = &mut item else {
unreachable!()
};
*role = "developer".to_string();
let source = ResponseItemEnvelope {
item,
metadata: Some(CodexHarnessMetadata {
client_authored: true,
}),
};
for max_tokens in [1871, 1879, 1880, 1950, 2200] {
assert_eq!(
truncate_retained_messages(
vec![source.clone()],
max_tokens,
RetainedImageBudget::Enabled
),
truncate_retained_messages_for_remote_compaction(vec![source.clone()], max_tokens),
);
}
}
@@ -0,0 +1,100 @@
use crate::context_manager::estimate_image_bytes;
use codex_context_fragments::AnnotatedContent;
use codex_context_fragments::set_annotated_content;
use codex_context_fragments::to_annotated_content;
use codex_history::ResponseItemEnvelope;
use codex_protocol::models::ContentItem;
use codex_protocol::models::is_image_close_tag_text;
use codex_protocol::models::is_image_open_tag_text;
use codex_protocol::models::is_local_image_open_tag_text;
use codex_protocol::protocol::TruncationPolicy;
use codex_utils_output_truncation::approx_token_count;
use codex_utils_output_truncation::approx_tokens_from_byte_count_i64;
use codex_utils_output_truncation::truncate_text;
pub(super) fn content_item_token_count(item: &ContentItem) -> usize {
match item {
ContentItem::InputText { text } | ContentItem::OutputText { text } => {
approx_token_count(text)
}
ContentItem::InputImage { image_url, detail } => usize::try_from(
approx_tokens_from_byte_count_i64(estimate_image_bytes(image_url, *detail)),
)
.unwrap_or(usize::MAX),
ContentItem::InputAudio { .. } => 0,
}
}
/// Retain later parts of an image-containing boundary message, keeping each
/// image and its adjacent harness labels atomic. Text keeps its existing middle
/// truncation policy; audio remains uncharged and is preserved as before.
pub(super) fn truncate_message_to_token_budget(
mut envelope: ResponseItemEnvelope,
max_tokens: usize,
) -> Option<ResponseItemEnvelope> {
let mut content = to_annotated_content(&mut envelope.item)?;
let mut remaining = max_tokens;
let mut retained = Vec::with_capacity(content.len());
while !content.is_empty() {
let last = content.len() - 1;
let image_index = match content[last].content() {
ContentItem::InputImage { .. } => Some(last),
ContentItem::InputText { text }
if is_image_close_tag_text(text)
&& last > 0
&& matches!(content[last - 1].content(), ContentItem::InputImage { .. }) =>
{
Some(last - 1)
}
_ => None,
};
if let Some(image_index) = image_index {
let has_open_tag = image_index > 0
&& matches!(
content[image_index - 1].content(),
ContentItem::InputText { text }
if is_local_image_open_tag_text(text) || is_image_open_tag_text(text)
);
let start = image_index - usize::from(has_open_tag);
let token_count = content[start..]
.iter()
.map(AnnotatedContent::content)
.map(content_item_token_count)
.sum::<usize>();
let fits = token_count <= remaining;
remaining = if fits { remaining - token_count } else { 0 };
for item in content.drain(start..).rev() {
if fits {
retained.push(item);
}
}
continue;
}
let mut item = content.pop()?;
match item.content_mut() {
ContentItem::InputText { text } | ContentItem::OutputText { text } => {
if remaining == 0 {
continue;
}
let token_count = approx_token_count(text);
if token_count <= remaining {
remaining -= token_count;
} else {
*text = truncate_text(text, TruncationPolicy::Tokens(remaining));
remaining = 0;
}
if !text.is_empty() {
retained.push(item);
}
}
ContentItem::InputAudio { .. } => retained.push(item),
ContentItem::InputImage { .. } => unreachable!("images are handled atomically above"),
}
}
if retained.is_empty() {
return None;
}
retained.reverse();
set_annotated_content(&mut envelope.item, retained)?;
Some(envelope)
}
+12 -6
View File
@@ -755,6 +755,16 @@ fn estimate_original_image_bytes(image_url: &str) -> Option<i64> {
})
}
/// Shared image estimate, excluding the data URL prefix and message framing.
pub(crate) fn estimate_image_bytes(image_url: &str, detail: Option<ImageDetail>) -> i64 {
match detail {
Some(ImageDetail::Original) => {
estimate_original_image_bytes(image_url).unwrap_or(RESIZED_IMAGE_BYTES_ESTIMATE)
}
_ => RESIZED_IMAGE_BYTES_ESTIMATE,
}
}
/// Scans one response item for discount-eligible inline image data URLs and
/// returns:
/// - total base64 payload bytes to subtract from raw serialized size
@@ -767,12 +777,8 @@ fn image_data_url_estimate_adjustment(item: &ResponseItem) -> (i64, i64) {
if let Some(payload_len) = parse_base64_image_data_url(image_url).map(str::len) {
payload_bytes =
payload_bytes.saturating_add(i64::try_from(payload_len).unwrap_or(i64::MAX));
replacement_bytes = replacement_bytes.saturating_add(match detail {
Some(ImageDetail::Original) => {
estimate_original_image_bytes(image_url).unwrap_or(RESIZED_IMAGE_BYTES_ESTIMATE)
}
_ => RESIZED_IMAGE_BYTES_ESTIMATE,
});
replacement_bytes =
replacement_bytes.saturating_add(estimate_image_bytes(image_url, detail));
}
};
+1
View File
@@ -3,6 +3,7 @@ mod normalize;
pub(crate) mod updates;
pub(crate) use history::ContextManager;
pub(crate) use history::estimate_image_bytes;
pub(crate) use history::estimate_item_token_count;
pub(crate) use history::is_user_turn_boundary;
pub(crate) use history::truncate_function_output_payload;
+2 -6
View File
@@ -11199,9 +11199,7 @@ async fn sample_rollout(
rollout_items.push(RolloutItem::ResponseItem(assistant1.clone().into()));
let summary1 = "summary one";
let snapshot1 = live_history
.clone()
.for_prompt(&reconstruction_turn.model_info.input_modalities);
let snapshot1 = raw_history_items(&live_history);
let user_messages1 = collect_user_messages(&snapshot1);
let rebuilt1 = compact::build_compacted_history(Vec::new(), &user_messages1, summary1);
live_history.replace_annotated(rebuilt1);
@@ -11247,9 +11245,7 @@ async fn sample_rollout(
rollout_items.push(RolloutItem::ResponseItem(assistant2.clone().into()));
let summary2 = "summary two";
let snapshot2 = live_history
.clone()
.for_prompt(&reconstruction_turn.model_info.input_modalities);
let snapshot2 = raw_history_items(&live_history);
let user_messages2 = collect_user_messages(&snapshot2);
let rebuilt2 = compact::build_compacted_history(Vec::new(), &user_messages2, summary2);
live_history.replace_annotated(rebuilt2);
+144
View File
@@ -1315,6 +1315,150 @@ async fn remote_manual_compact_chatgpt_auth_reuses_service_tier_and_prompt_cache
Ok(())
}
#[test_case(None; "default_preserves_images")]
#[test_case(Some(false); "disabled_preserves_images")]
#[test_case(Some(true); "enabled_trims_images")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_v2_charges_retained_images_to_token_budget(
image_budget_enabled: Option<bool>,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let harness = TestCodexHarness::with_auto_env_builder(
test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
let _ = config.features.enable(Feature::RemoteCompactionV2);
let _ = config.features.enable(Feature::UnifiedImageBudget);
if let Some(enabled) = image_budget_enabled {
let _ = config
.features
.set_enabled(Feature::CompactionImageBudget, enabled);
}
}),
)
.await?;
let codex = &harness.test().codex;
// Each original-detail image costs 10,000 estimated patch tokens.
let image_inputs = (1..=8)
.map(|number| {
let image = image::ImageBuffer::from_pixel(
/*width*/ 3200,
/*height*/ 3200,
image::Luma([number as u8]),
);
let mut bytes = std::io::Cursor::new(Vec::new());
image.write_to(&mut bytes, image::ImageFormat::Png)?;
Ok(UserInput::Image {
image_url: format!(
"data:image/png;base64,{}",
BASE64_STANDARD.encode(bytes.get_ref())
),
detail: Some(codex_protocol::models::ImageDetail::Original),
})
})
.collect::<Result<Vec<_>>>()?;
let mut input = image_inputs[..7].to_vec();
input.push(UserInput::Text {
text: "Compare these images".to_string(),
text_elements: Vec::new(),
});
let initial_mock = mount_sse_once(
harness.server(),
sse(vec![
responses::ev_assistant_message("initial", "done"),
responses::ev_completed("initial"),
]),
)
.await;
codex
.start_or_steer_turn(TurnInputRequest::user_input(input))
.await?;
wait_for_turn_complete(codex).await;
let initial_request = initial_mock.single_request();
let prepared_images = initial_request.message_input_image_urls("user");
assert_eq!(prepared_images.len(), 7);
for cycle in 1..=2 {
let compact_mock = mount_sse_once(
harness.server(),
sse(vec![
json!({
"type": "response.output_item.done",
"item": { "type": "compaction", "encrypted_content": "IMAGE_BUDGET_SUMMARY" },
}),
responses::ev_completed("compact-images"),
]),
)
.await;
codex.submit(Op::Compact).await?;
wait_for_turn_complete(codex).await;
let compact_request = compact_mock.single_request();
assert_eq!(compact_request.path(), "/v1/responses");
assert_eq!(
compact_request.inputs_of_type("compaction_trigger").len(),
1
);
let follow_up_mock = mount_sse_once(
harness.server(),
sse(vec![
responses::ev_assistant_message("after", "done"),
responses::ev_completed("after"),
]),
)
.await;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "after compact".to_string(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_turn_complete(codex).await;
let follow_up = follow_up_mock.single_request();
assert_eq!(
follow_up.inputs_of_type("compaction")[0]["encrypted_content"],
"IMAGE_BUDGET_SUMMARY"
);
let dropped = if image_budget_enabled == Some(true) {
cycle
} else {
0
};
let mut expected_images = prepared_images[dropped..].to_vec();
if cycle == 2 {
let UserInput::Image { image_url, .. } = &image_inputs[7] else {
unreachable!()
};
expected_images.push(image_url.clone());
}
assert_eq!(follow_up.message_input_image_urls("user"), expected_images);
assert!(
follow_up
.message_input_texts("user")
.iter()
.any(|text| text == "Compare these images")
);
if cycle == 1 {
let append_mock = mount_sse_once(
harness.server(),
sse(vec![
responses::ev_assistant_message("append", "done"),
responses::ev_completed("append"),
]),
)
.await;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![image_inputs[7].clone()]))
.await?;
wait_for_turn_complete(codex).await;
let _ = append_mock.single_request();
}
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<()> {
skip_if_no_network!(Ok(()));
+8
View File
@@ -301,6 +301,8 @@ pub enum Feature {
PreventIdleSleep,
/// Enable remote compaction v2 over the normal Responses API.
RemoteCompactionV2,
/// Include retained images in the remote compaction context budget.
CompactionImageBudget,
/// Retain client-authored developer messages across compacted context windows.
RetainClientDeveloperMessages,
/// Use Agent Identity for ChatGPT-authenticated sessions.
@@ -1529,6 +1531,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::Stable,
default_enabled: true,
},
FeatureSpec {
id: Feature::CompactionImageBudget,
key: "compaction_image_budget",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::RetainClientDeveloperMessages,
key: "retain_client_developer_messages",