Preserve tool output truncation budgets across resume and fork (#44248)

## Why

Replaying tool outputs under a different model can expand or shrink the history shown to the model if truncation uses the new model's budget.

## What changed

Save the originating history truncation budget on function and custom tool outputs and reuse it during replay, preserving existing tool-specific overrides. Include the existing 20% serialization allowance once, before converting byte budgets to tokens.

Rename the metadata field to `history_truncation_token_limit` while retaining `fallback_token_limit_override` as its serialized name for compatibility.

## Testing

Add regression coverage for resume and fork with different model budgets, including custom tool outputs. Extend unit coverage for existing overrides and byte-budget conversion.

GitOrigin-RevId: a62c8fd3e43198e2252b150fba72f022939c1070
This commit is contained in:
Abhinav
2026-09-09 17:55:00 +00:00
committed by copyberry
parent b64de2f3ad
commit aa88a0333c
15 changed files with 287 additions and 47 deletions
@@ -152,7 +152,7 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history(
envelope
.metadata
.as_ref()
.map(|metadata| metadata.client_authored),
.is_some_and(|metadata| metadata.client_authored),
)),
_ => None,
})
@@ -166,10 +166,10 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history(
assert_eq!(
persisted_injected_items,
vec![
(injected_item.clone(), None),
(injected_developer_item.clone(), Some(true)),
(marker_shaped_developer_item.clone(), Some(true)),
(named_tool_item, None),
(injected_item.clone(), false),
(injected_developer_item.clone(), true),
(marker_shaped_developer_item.clone(), true),
(named_tool_item, false),
]
);
+3 -2
View File
@@ -57,6 +57,7 @@ use codex_utils_output_truncation::approx_bytes_for_tokens;
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_function_output_payload;
use codex_utils_output_truncation::with_serialization_allowance;
use std::num::NonZeroUsize;
use std::ops::Deref;
use std::sync::Arc;
@@ -366,9 +367,9 @@ impl ContextManager {
{
// The override already includes the tool's serialization allowance.
let policy = metadata
.and_then(|metadata| metadata.fallback_token_limit_override)
.and_then(|metadata| metadata.history_truncation_token_limit)
.map(TruncationPolicy::Tokens)
.unwrap_or(policy * 1.2);
.unwrap_or_else(|| with_serialization_allowance(policy));
truncate_function_output_payload(output, policy, estimate_audio_token_count);
}
if let Some(review_history) = &mut self.review_history
@@ -659,7 +659,7 @@ fn annotated_history_apis_preserve_envelopes() {
#[test_case(Some(100), 85, 100, true; "saved limit has no additional allowance")]
#[test_case(Some(30_000), 20_000, 30_000, false; "large explicit budget")]
fn record_annotated_items_preserves_metadata_while_processing_item(
fallback_token_limit_override: Option<usize>,
history_truncation_token_limit: Option<usize>,
repeat_count: usize,
expected_token_limit: usize,
expected_truncation: bool,
@@ -677,7 +677,7 @@ fn record_annotated_items_preserves_metadata_while_processing_item(
internal_chat_message_metadata_passthrough: None,
},
metadata: Some(CodexHarnessMetadata {
fallback_token_limit_override,
history_truncation_token_limit,
..Default::default()
}),
};
+18 -1
View File
@@ -172,6 +172,7 @@ use codex_thread_store::ThreadPersistenceMetadata;
use codex_thread_store::ThreadStore;
use codex_utils_audio::prepare_response_items as prepare_audio_response_items;
use codex_utils_git_discovery::GitRootDiscovery;
use codex_utils_output_truncation::with_serialization_allowance;
use codex_utils_path_uri::PathUri;
use futures::future::BoxFuture;
use futures::future::Shared;
@@ -3429,9 +3430,25 @@ impl Session {
&self,
turn_context: &TurnContext,
model_info: &ModelInfo,
items: Vec<ResponseItemEnvelope>,
mut items: Vec<ResponseItemEnvelope>,
image_preparations: Vec<ImagePreparationMetadata>,
) {
// Save the originating history budget for replay.
// Preserve any existing tool-specific override.
let policy: codex_utils_output_truncation::TruncationPolicy =
model_info.truncation_policy.into();
for envelope in &mut items {
if matches!(
envelope.item,
ResponseItem::FunctionCallOutput { .. } | ResponseItem::CustomToolCallOutput { .. }
) {
envelope
.metadata
.get_or_insert_default()
.history_truncation_token_limit
.get_or_insert_with(|| with_serialization_allowance(policy).token_budget());
}
}
let response_items = items
.iter()
.map(|envelope| envelope.item.clone())
+53 -15
View File
@@ -2532,10 +2532,17 @@ async fn record_inter_agent_communication_preserves_item_id_in_rollout_and_resum
);
}
#[test_case::test_case(false; "plain items")]
#[test_case::test_case(true; "annotated items")]
#[test_case::test_case(false, None, false, 5; "plain items")]
#[test_case::test_case(true, None, false, 5; "annotated items")]
#[test_case::test_case(true, Some(100), false, 100; "existing tool budget")]
#[test_case::test_case(false, None, true, 31; "byte policy includes allowance before conversion")]
#[tokio::test]
async fn annotated_history_uses_explicit_model_without_a_step(retain_metadata: bool) {
async fn annotated_history_uses_explicit_model_without_a_step(
retain_metadata: bool,
saved_budget: Option<usize>,
byte_policy: bool,
expected_budget: usize,
) {
let (session, turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx(
CodexAuth::from_api_key("Test API Key"),
Vec::new(),
@@ -2548,8 +2555,11 @@ async fn annotated_history_uses_explicit_model_without_a_step(retain_metadata: b
)
.await;
let mut model_info = turn_context.model_info().as_ref().clone();
model_info.truncation_policy =
codex_protocol::openai_models::TruncationPolicyConfig::tokens(/*limit*/ 4);
model_info.truncation_policy = if byte_policy {
codex_protocol::openai_models::TruncationPolicyConfig::bytes(/*limit*/ 101)
} else {
codex_protocol::openai_models::TruncationPolicyConfig::tokens(/*limit*/ 4)
};
let text = "diagnostic line\n".repeat(50);
let mut item = ResponseItem::FunctionCallOutput {
id: Some(ResponseItemId::with_suffix("fco", "existing")),
@@ -2564,24 +2574,52 @@ async fn annotated_history_uses_explicit_model_without_a_step(retain_metadata: b
};
item.set_turn_id_if_missing("existing-turn");
item.set_create_time_if_missing(123.into());
let mut expected = ResponseItemEnvelope {
let expected = ResponseItemEnvelope {
item,
metadata: retain_metadata.then_some(CodexHarnessMetadata {
client_authored: true,
history_truncation_token_limit: saved_budget,
..Default::default()
}),
};
let mut expected = vec![
expected,
ResponseItemEnvelope {
item: ResponseItem::CustomToolCallOutput {
id: Some(ResponseItemId::with_suffix("ctco", "existing")),
call_id: "call-custom".to_string(),
name: None,
output: FunctionCallOutputPayload::from_text(text.clone()),
internal_chat_message_metadata_passthrough: None,
},
metadata: retain_metadata.then_some(CodexHarnessMetadata {
client_authored: true,
history_truncation_token_limit: saved_budget,
..Default::default()
}),
},
];
expected[1].item.set_turn_id_if_missing("existing-turn");
expected[1].item.set_create_time_if_missing(123.into());
session
.record_annotated_conversation_items(&turn_context, &model_info, vec![expected.clone()])
.record_annotated_conversation_items(&turn_context, &model_info, expected.clone())
.await;
let ResponseItem::FunctionCallOutput { output, .. } = &mut expected.item else {
unreachable!("fixture is a tool output");
};
output.body = FunctionCallOutputBody::Text(codex_utils_output_truncation::truncate_text(
&text,
codex_utils_output_truncation::TruncationPolicy::Tokens(4) * 1.2,
));
assert_eq!(session.clone_history().await.annotated_items(), &[expected]);
for envelope in &mut expected {
envelope
.metadata
.get_or_insert_default()
.history_truncation_token_limit = Some(expected_budget);
let (ResponseItem::FunctionCallOutput { output, .. }
| ResponseItem::CustomToolCallOutput { output, .. }) = &mut envelope.item
else {
unreachable!("fixture is a tool output");
};
output.body = FunctionCallOutputBody::Text(codex_utils_output_truncation::truncate_text(
&text,
codex_utils_output_truncation::TruncationPolicy::Tokens(expected_budget),
));
}
assert_eq!(session.clone_history().await.annotated_items(), &expected);
}
#[tokio::test]
+4 -3
View File
@@ -21,6 +21,7 @@ use codex_utils_output_truncation::approx_token_count;
use codex_utils_output_truncation::formatted_truncate_text;
use codex_utils_output_truncation::truncate_function_output_payload;
use codex_utils_output_truncation::truncate_text;
use codex_utils_output_truncation::with_serialization_allowance;
use serde::Serialize;
use serde_json::Value as JsonValue;
use std::num::NonZeroUsize;
@@ -124,7 +125,7 @@ impl ToolOutput for McpToolOutput {
}
fn fallback_token_limit_override(&self) -> Option<usize> {
Some((self.truncation_policy * 1.2).token_budget())
Some(with_serialization_allowance(self.truncation_policy).token_budget())
}
fn to_response_item(&self, call_id: &str, _payload: &ToolPayload) -> ResponseInputItem {
@@ -173,7 +174,7 @@ impl McpToolOutput {
// History receives this budget in tokens. Code Mode keeps the raw result.
truncate_function_output_payload(
&mut payload,
self.truncation_policy * 1.2,
with_serialization_allowance(self.truncation_policy),
estimate_audio_token_count,
);
payload
@@ -515,7 +516,7 @@ impl ExecCommandToolOutput {
fn response_text(&self) -> String {
let header = self.response_header();
let output_budget = (self.truncation_policy * 1.2)
let output_budget = with_serialization_allowance(self.truncation_policy)
.byte_budget()
.saturating_sub(header.len().saturating_add(/*rhs*/ 1));
let mut policy = self.model_output_policy();
@@ -13,6 +13,7 @@ use codex_protocol::mcp::CallToolResult;
use codex_protocol::models::function_call_output_content_items_to_text;
use codex_protocol::protocol::TruncationPolicy;
use codex_utils_output_truncation::truncate_text;
use codex_utils_output_truncation::with_serialization_allowance;
use rmcp::model::ListResourceTemplatesResult;
use rmcp::model::ListResourcesResult;
use rmcp::model::PaginatedRequestParams;
@@ -362,7 +363,7 @@ where
})?;
// Match regular MCP tool outputs by bounding the copy persisted to the
// rollout and injected into model context.
let content = truncate_text(&content, truncation_policy * 1.2);
let content = truncate_text(&content, with_serialization_allowance(truncation_policy));
Ok(FunctionToolOutput::from_text(content, Some(true)))
}
+1 -1
View File
@@ -203,7 +203,7 @@ impl AnyToolResult {
metadata: result
.fallback_token_limit_override()
.map(|limit| CodexHarnessMetadata {
fallback_token_limit_override: Some(limit),
history_truncation_token_limit: Some(limit),
..Default::default()
}),
}
+1 -1
View File
@@ -629,7 +629,7 @@ fn post_tool_use_feedback_output_preserves_fallback_token_limit_override(
output: FunctionCallOutputPayload::from_text("hook feedback".to_string()),
}),
metadata: Some(CodexHarnessMetadata {
fallback_token_limit_override: Some(expected_token_limit),
history_truncation_token_limit: Some(expected_token_limit),
..Default::default()
}),
}
+179 -7
View File
@@ -3,6 +3,7 @@ use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_config::McpServerConfig;
use codex_core::CodexThread;
use codex_core::ForkSnapshot;
use codex_core::StartThreadOptions;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
@@ -29,6 +30,7 @@ use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem;
use codex_protocol::dynamic_tools::DynamicToolFunctionSpec;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ApplyPatchToolType;
@@ -83,6 +85,7 @@ use core_test_support::responses::ev_function_call_with_namespace;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_models_once;
use core_test_support::responses::mount_response_sequence;
use core_test_support::responses::mount_sse_once;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::sse;
use core_test_support::responses::sse_completed;
@@ -286,7 +289,7 @@ fn request_turn_id(request: &ResponsesRequest) -> String {
// Dynamic tools return the original payload, so handler truncation cannot hide a recorder bug.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn tool_result_history_keeps_originating_model_across_switch() -> Result<()> {
async fn tool_result_history_keeps_originating_model_across_switch_and_replay() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let mut test = step_settings_test()
@@ -438,10 +441,8 @@ async fn tool_result_history_keeps_originating_model_across_switch() -> Result<(
// Persistence and raw notifications retain the prepared, untruncated payload in append order.
test.codex.shutdown_and_wait().await?;
let history = codex_rollout::RolloutRecorder::get_rollout_history(
&test.codex.rollout_path().expect("rollout path"),
)
.await?;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let history = codex_rollout::RolloutRecorder::get_rollout_history(&rollout_path).await?;
let saved_outputs = history
.get_rollout_items()
.iter()
@@ -449,12 +450,183 @@ async fn tool_result_history_keeps_originating_model_across_switch() -> Result<(
RolloutItem::ResponseItem(envelope)
if matches!(&envelope.item, ResponseItem::FunctionCallOutput { .. }) =>
{
Some(serde_json::to_value(&envelope.item).expect("saved output"))
Some((
serde_json::to_value(&envelope.item).expect("saved output"),
envelope
.metadata
.as_ref()
.and_then(|metadata| metadata.history_truncation_token_limit),
))
}
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(saved_outputs, raw_outputs);
assert_eq!(
saved_outputs,
raw_outputs
.into_iter()
.zip([Some(120), Some(480)])
.collect::<Vec<_>>()
);
// Resume and fork under both limits. A's output must not grow under B, and
// B's output must not shrink under A; image preparation and item IDs also survive.
for model in [MODEL_A, MODEL_B] {
let mut replay_config = test.config.clone();
replay_config.model = Some(model.to_string());
let resumed = test
.thread_manager
.resume_thread_from_rollout(
replay_config.clone(),
rollout_path.clone(),
codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
/*parent_trace*/ None,
ClientMcpExtensions::default(),
)
.await?
.thread;
let forked = test
.thread_manager
.fork_thread(
ForkSnapshot::Interrupted,
StartThreadOptions::new(replay_config),
rollout_path.clone(),
)
.await?
.thread;
for thread in [resumed, forked] {
let replay = mount_sse_once(&server, sse_completed("resp-replay")).await;
thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "review saved diagnostics".to_string(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let request = replay.single_request();
assert_eq!(request.body_json()["model"], model);
for call_id in ["call-a", "call-b"] {
assert_eq!(
request.function_call_output(call_id),
requests[2].function_call_output(call_id)
);
}
thread.shutdown_and_wait().await?;
}
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn custom_tool_output_replay_preserves_originating_budget() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let test = step_settings_test()
.with_model(MODEL_B)
.with_config(|config| {
config.features.enable(Feature::CodeMode).unwrap();
for model in &mut config.model_catalog.as_mut().expect("models").models {
model.truncation_policy =
TruncationPolicyConfig::tokens(if model.slug == MODEL_B { 400 } else { 100 });
model.tool_mode = Some(ToolMode::CodeModeOnly);
model.use_responses_lite = false;
}
})
.build_with_auto_env(&server)
.await?;
let responses = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-custom"),
ev_custom_tool_call(
"call-custom",
"exec",
"text('diagnostic line\\n'.repeat(500));",
),
ev_completed("resp-custom"),
]),
sse_completed("resp-done"),
],
)
.await;
test.submit_text_turn("collect diagnostics").await?;
let live_output = responses.requests()[1].custom_tool_call_output("call-custom");
let text = "diagnostic line\n".repeat(500);
let live_text = live_output["output"][1]["text"]
.as_str()
.expect("bounded custom output");
assert!(live_text.len() < text.len());
// A's smaller history budget would truncate this result again without the saved budget.
assert_ne!(
truncate_text(live_text, TruncationPolicy::Tokens(120)),
live_text
);
test.codex.shutdown_and_wait().await?;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let history = codex_rollout::RolloutRecorder::get_rollout_history(&rollout_path).await?;
let saved = history
.get_rollout_items()
.iter()
.find_map(|item| match item {
RolloutItem::ResponseItem(envelope)
if matches!(&envelope.item, ResponseItem::CustomToolCallOutput { call_id, .. } if call_id == "call-custom") =>
{
Some(envelope)
}
_ => None,
})
.expect("saved custom output");
assert_eq!(
saved
.metadata
.as_ref()
.and_then(|metadata| metadata.history_truncation_token_limit),
Some(480)
);
assert_eq!(
serde_json::to_value(&saved.item)?["output"][1]["text"],
text
);
let mut replay_config = test.config.clone();
replay_config.model = Some(MODEL_A.to_string());
let resumed = test
.thread_manager
.resume_thread_from_rollout(
replay_config.clone(),
rollout_path.clone(),
codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
/*parent_trace*/ None,
ClientMcpExtensions::default(),
)
.await?
.thread;
let forked = test
.thread_manager
.fork_thread(
ForkSnapshot::Interrupted,
StartThreadOptions::new(replay_config),
rollout_path,
)
.await?
.thread;
for thread in [resumed, forked] {
let replay = mount_sse_once(&server, sse_completed("resp-replay")).await;
thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "review saved diagnostics".to_string(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let request = replay.single_request();
assert_eq!(request.body_json()["model"], MODEL_A);
assert_eq!(request.custom_tool_call_output("call-custom"), live_output);
thread.shutdown_and_wait().await?;
}
Ok(())
}
+8 -4
View File
@@ -50,10 +50,14 @@ pub struct CodexHarnessMetadata {
#[serde(default)]
pub client_authored: bool,
/// Overrides history's fallback truncation budget, including on resume.
/// Measured in tokens, with any tool-specific allowance already included.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub fallback_token_limit_override: Option<usize>,
/// The originating history budget, including any tool-specific allowance.
/// Measured in tokens and reused when replaying persisted history.
#[serde(
default,
rename = "fallback_token_limit_override",
skip_serializing_if = "Option::is_none"
)]
pub history_truncation_token_limit: Option<usize>,
/// Whether a response configuration update was created by the Codex harness itself.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
+2 -2
View File
@@ -78,7 +78,7 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()>
item: response_item.clone(),
metadata: Some(CodexHarnessMetadata {
client_authored: true,
fallback_token_limit_override: Some(20_000),
history_truncation_token_limit: Some(20_000),
inherited_user_message: true,
..Default::default()
}),
@@ -110,7 +110,7 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()>
envelope.metadata,
Some(CodexHarnessMetadata {
client_authored: true,
fallback_token_limit_override: Some(20_000),
history_truncation_token_limit: Some(20_000),
inherited_user_message: true,
..Default::default()
})
+3 -3
View File
@@ -8,6 +8,7 @@ use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_output_truncation::TruncationPolicy;
use codex_utils_output_truncation::with_serialization_allowance;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
@@ -147,9 +148,8 @@ impl ToolCall<'_> {
/// Callers must include serialization overhead when fitting a response to this budget.
pub fn response_byte_budget(&self, max_response_bytes: usize) -> usize {
match &self.source {
ToolCallSource::Direct => {
max_response_bytes.min((self.truncation_policy * 1.2).byte_budget())
}
ToolCallSource::Direct => max_response_bytes
.min(with_serialization_allowance(self.truncation_policy).byte_budget()),
ToolCallSource::CodeMode {
cell_id: _,
runtime_tool_call_id: _,
@@ -11,6 +11,12 @@ use codex_utils_string::truncate_middle_with_token_budget;
pub use codex_protocol::protocol::TruncationPolicy;
/// Adds the existing 20% allowance for serialization and headers.
/// Saved history budgets already include this allowance.
pub fn with_serialization_allowance(policy: TruncationPolicy) -> TruncationPolicy {
policy * 1.2
}
pub fn formatted_truncate_text(content: &str, policy: TruncationPolicy) -> String {
if content.len() <= policy.byte_budget() {
return content.to_string();