From a37f260dc6534813827f7abc8634788b08d29458 Mon Sep 17 00:00:00 2001 From: ningyi-oai Date: Tue, 22 Sep 2026 02:48:50 +0000 Subject: [PATCH] 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 --- .../core/src/tools/executed_tool_calls.rs | 13 +- .../request_metadata_tests.rs | 37 ++++- codex-rs/core/tests/suite/code_mode.rs | 140 +++++++++++++++++- 3 files changed, 179 insertions(+), 11 deletions(-) diff --git a/codex-rs/core/src/tools/executed_tool_calls.rs b/codex-rs/core/src/tools/executed_tool_calls.rs index aec694bd24..db8f8e6aa6 100644 --- a/codex-rs/core/src/tools/executed_tool_calls.rs +++ b/codex-rs/core/src/tools/executed_tool_calls.rs @@ -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::>(); 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 diff --git a/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs b/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs index bb617a1b7d..7f79e1d227 100644 --- a/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs +++ b/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs @@ -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(), diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index a2bebfe6d0..67c77fb907 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/codex-rs/core/tests/suite/code_mode.rs @@ -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>>, } -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::::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::::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::(request)) + .collect::>>()?; + 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(()));