Preserve empty cell metadata under pending tool-call pressure (#47174)

## Why

Reaching the pending tool-call limit could evict finished empty cells even though removing them frees no pending-call capacity. This loses the metadata needed to certify an empty tool-call inventory on a later `wait`.

## What changed

Only evict cells with pending calls when reclaiming pending-call capacity. Keep empty cells eligible for eviction when a cell slot is needed.

## Testing

Extend recorder coverage to check empty inventories under pending-call pressure in fresh and resumed sessions. Add a code-mode integration test verifying that `wait` serializes an empty inventory with `tool_calls_complete: true` while another cell has filled the pending-call budget.

GitOrigin-RevId: 559d0b4fdc9864c42a90b7d087eb2a3327272d5c
This commit is contained in:
ningyi-oai
2026-09-22 03:03:52 +00:00
committed by copyberry
parent f4389de279
commit a37f260dc6
3 changed files with 179 additions and 11 deletions
@@ -167,15 +167,20 @@ impl ExecutedToolCallRecorderState {
self.output_cells
.retain(|_, cell_id| self.cells.contains_key(cell_id));
}
while (self.cells.len() >= MAX_PENDING_EXECUTED_TOOL_CALLS
&& !self.cells.contains_key(cell_id))
|| self.pending_nested_calls >= MAX_PENDING_EXECUTED_TOOL_CALLS
{
loop {
let needs_cell_slot = self.cells.len() >= MAX_PENDING_EXECUTED_TOOL_CALLS
&& !self.cells.contains_key(cell_id);
if !needs_cell_slot && self.pending_nested_calls < MAX_PENDING_EXECUTED_TOOL_CALLS {
break;
}
let output_cells = self.output_cells.values().collect::<HashSet<_>>();
let finished_cell = self.cells.iter().find_map(|(id, cell)| {
// Preserve late records until pressure, and never discard the cell
// whose output is about to make those records attachable again.
// Empty cells can free cell slots, but not pending-call slots.
(id != cell_id
&& (needs_cell_slot || !cell.pending_calls.is_empty())
&& matches!(
cell.completion,
CellCompletion::Complete | CellCompletion::Incomplete
@@ -714,7 +714,7 @@ fn empty_inventory_revalidates_history_on_retry() {
#[test]
fn empty_inventory_wait_requires_a_fresh_session() {
for fresh in [true, false] {
for (fresh, pending_pressure) in [(true, false), (false, false), (true, true), (false, true)] {
let history = if fresh {
InitialHistory::New
} else {
@@ -727,12 +727,45 @@ fn empty_inventory_wait_requires_a_fresh_session() {
let mut retry_cache = HashMap::new();
recorder.attach_to_prompt(&mut prompt, &mut retry_cache);
assert!(prompt[1].executed_tool_call_metadata().is_none());
if pending_pressure {
// Model a finished empty cell whose original output mapping was consumed.
// Register another cell under pending-call pressure before its next wait.
recorder.finish_cell_recording(&cell);
let busy_cell = CellId::new("busy-cell".to_string());
recorder.start_cell(&busy_cell, "busy-exec");
for index in 0..MAX_PENDING_EXECUTED_TOOL_CALLS {
record_nested_call(&recorder, &busy_cell, &format!("busy-{index}"));
}
{
let state = recorder.lock_state();
let state = state.as_ref().unwrap();
assert_eq!(state.pending_nested_calls, MAX_PENDING_EXECUTED_TOOL_CALLS);
assert!(state.cells.len() < MAX_PENDING_EXECUTED_TOOL_CALLS);
assert!(!state.output_cells.values().any(|id| id == &cell));
let empty = state.cells.get(&cell).unwrap();
assert!(matches!(empty.completion, CellCompletion::Complete));
assert!(empty.pending_calls.is_empty());
}
recorder.register_cell(&busy_cell, "busy-wait");
recorder.finish_cell_recording(&busy_cell);
let mut busy_prompt = vec![
exec_input("busy-exec"),
exec_output("busy-exec"),
wait_input("busy-wait", &busy_cell),
output("busy-wait"),
];
recorder.attach_to_prompt(&mut busy_prompt, &mut HashMap::new());
}
recorder.register_cell(&cell, "wait");
recorder.finish_cell_recording(&cell);
prompt.extend([wait_input("wait", &cell), output("wait")]);
recorder.attach_to_prompt(&mut prompt, &mut retry_cache);
assert!(!has_direct_call_metadata(&prompt[3]));
assert_eq!(tool_calls_complete(&prompt[3]), fresh.then_some(true));
assert_eq!(
tool_calls_complete(&prompt[3]),
fresh.then_some(true),
"fresh={fresh}, pending_pressure={pending_pressure}",
);
if fresh {
assert_eq!(
serde_json::to_value(prompt[3].executed_tool_call_metadata()).unwrap(),
+135 -5
View File
@@ -2552,17 +2552,19 @@ async fn code_mode_resumed_wait_does_not_certify_a_reused_runtime_cell() -> Resu
Ok(())
}
struct MappingPressureObserver {
struct ExecCompletionObserver {
call_id_prefix: &'static str,
expected_count: usize,
finished: AtomicUsize,
release: Mutex<Option<oneshot::Sender<()>>>,
}
impl ToolLifecycleContributor for MappingPressureObserver {
impl ToolLifecycleContributor for ExecCompletionObserver {
fn on_tool_finish<'a>(&'a self, input: ToolFinishInput<'a>) -> ToolLifecycleFuture<'a> {
Box::pin(async move {
if input.tool_name.name == "exec" && input.call_id.starts_with("seed-") {
if input.tool_name.name == "exec" && input.call_id.starts_with(self.call_id_prefix) {
assert_eq!(input.outcome, ToolCallOutcome::Completed { success: true });
if self.finished.fetch_add(1, Ordering::SeqCst) == 511 {
if self.finished.fetch_add(1, Ordering::SeqCst) + 1 == self.expected_count {
self.release
.lock()
.unwrap()
@@ -2620,7 +2622,9 @@ async fn code_mode_recovers_complete_inventory_after_orphaned_mapping_pressure()
])
.await;
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
extensions.tool_lifecycle_contributor(Arc::new(MappingPressureObserver {
extensions.tool_lifecycle_contributor(Arc::new(ExecCompletionObserver {
call_id_prefix: "seed-",
expected_count: 512,
finished: AtomicUsize::new(0),
release: Mutex::new(Some(release)),
}));
@@ -2670,6 +2674,132 @@ async fn code_mode_recovers_complete_inventory_after_orphaned_mapping_pressure()
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn code_mode_wait_serializes_empty_inventory_under_pending_call_pressure() -> Result<()> {
skip_if_no_network!(Ok(()));
// The recorder unit test covers eviction ordering. This host test covers wait
// serialization under pressure; the empty cell is still recording until wait.
let (release, pressure_reached) = oneshot::channel();
let (server, _) = start_streaming_sse_server(vec![
vec![StreamingSseChunk {
gate: None,
body: sse(vec![
ev_custom_tool_call(
"call-empty",
"exec",
"yield_control(); await new Promise(() => {});",
),
ev_completed("resp-empty"),
]),
}],
vec![
StreamingSseChunk {
gate: None,
body: sse(vec![ev_custom_tool_call(
"call-pressure",
"exec",
r#"// @exec: {"yield_time_ms": 60000}
for (let index = 0; index < 256; index++) await tools.test_sync_tool({});
text("pressure ready");"#,
)]),
},
StreamingSseChunk {
// Finish the pressure cell before waiting, but keep this response
// open so no follow-up request can drain its pending inventory.
gate: Some(pressure_reached),
body: sse(vec![
responses::ev_function_call(
"call-wait",
"wait",
r#"{"cell_id":"1","terminate":true}"#,
),
ev_completed("resp-pressure"),
]),
},
],
vec![StreamingSseChunk {
gate: None,
body: sse(vec![
ev_assistant_message("msg-done", "done"),
ev_completed("resp-done"),
]),
}],
])
.await;
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
extensions.tool_lifecycle_contributor(Arc::new(ExecCompletionObserver {
call_id_prefix: "call-pressure",
expected_count: 1,
finished: AtomicUsize::new(0),
release: Mutex::new(Some(release)),
}));
let base_url = format!("{}/v1", server.uri());
let config_server = responses::start_mock_server().await;
let test = test_codex()
.with_model("test-gpt-5.1-codex")
.with_extensions(Arc::new(extensions.build()))
.with_config(move |config| {
config.model_provider.base_url = Some(base_url);
let _ = config.features.enable(Feature::CodeMode);
let _ = config.features.enable(Feature::ExecutedToolCallMetadata);
let _ = config.features.disable(Feature::RemoteCompactionV2);
})
.build_with_auto_env(&config_server)
.await?;
test.submit_turn("Wait for an empty cell while the pending-call budget is full")
.await?;
test.codex.shutdown_and_wait().await?;
let requests = server
.requests()
.await
.iter()
.map(|request| serde_json::from_slice::<Value>(request))
.collect::<serde_json::Result<Vec<_>>>()?;
assert_eq!(requests.len(), 3);
let output_for = |request_index: usize, call_id: &str| {
requests[request_index]["input"]
.as_array()
.unwrap()
.iter()
.find(|item| {
item["call_id"] == call_id
&& matches!(
item["type"].as_str(),
Some("custom_tool_call_output" | "function_call_output")
)
})
.unwrap()
};
let initial = output_for(/*request_index*/ 1, "call-empty");
assert_eq!(
extract_running_cell_id(initial["output"].as_str().unwrap()),
"1",
);
assert!(
initial["internal_chat_message_metadata_passthrough"]
.get("tool_calls_complete")
.is_none()
);
let pressure = &output_for(/*request_index*/ 2, "call-pressure")["internal_chat_message_metadata_passthrough"];
assert_eq!(
pressure["executed_tool_calls"],
serde_json::json!(vec![
serde_json::json!({"name": "test_sync_tool", "arguments": {}});
256
]),
);
assert_eq!(pressure["tool_calls_complete"], true);
let metadata =
&output_for(/*request_index*/ 2, "call-wait")["internal_chat_message_metadata_passthrough"];
assert_eq!(metadata["cell_id"], "call-empty");
assert_eq!(metadata["executed_tool_calls"], serde_json::json!([]));
assert_eq!(metadata["tool_calls_complete"], true);
server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn code_mode_complete_call_survives_unrelated_truncation() -> Result<()> {
skip_if_no_network!(Ok(()));