diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index c2ab78acb4..e77b2c88cf 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -268,7 +268,7 @@ async fn run_compact_task_inner_impl( .for_prompt(&turn_context.model_info().input_modalities); sess.services .executed_tool_calls - .strip_disabled_direct_metadata(&mut turn_input); + .attach_to_compaction_prompt(&mut turn_input); let turn_input_len = turn_input.len(); let prompt = Prompt { input: turn_input, diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index f770f0712b..60b5b1b772 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -73,7 +73,7 @@ pub(super) async fn run_remote_compact_v2_attempt( .unzip(); sess.services .executed_tool_calls - .strip_disabled_direct_metadata(&mut input); + .attach_to_compaction_prompt(&mut input); let tool_router = &step_context.tool_router; input.push(ResponseItem::CompactionTrigger {}); let prompt = Prompt { diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index 531d093061..532d80ed5e 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/codex-rs/core/src/compact_tests.rs @@ -1,5 +1,9 @@ use super::*; use crate::session::tests::make_session_and_context_with_auth_and_config_and_rx; +use crate::tools::context::ToolCallSource; +use crate::tools::context::ToolPayload; +use crate::tools::router::ToolCall; +use codex_code_mode::CellId; use codex_features::Feature; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; @@ -12,6 +16,7 @@ use codex_protocol::models::ExecutedToolCall; use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::ImageReference; use codex_protocol::models::InternalChatMessageMetadataPassthrough; +use codex_tools::ToolName; use core_test_support::responses; use core_test_support::skip_if_no_network; use pretty_assertions::assert_eq; @@ -77,10 +82,46 @@ async fn local_compaction_respects_tool_metadata_state( output.mark_tool_calls_complete(); items.push(output); } + let cell = CellId::new("local-compaction-cell".to_string()); + let nested_call = ExecutedToolCall::new("nested_tool".to_string(), json!({})); + let recorder = &session.services.executed_tool_calls; + recorder.start_cell(&cell, "exec"); + recorder.record_tool_call( + &ToolCall { + tool_name: ToolName::plain("nested_tool"), + call_id: "nested".to_string(), + payload: ToolPayload::Function { + arguments: "{}".to_string(), + }, + encrypted_function_args: None, + }, + &ToolCallSource::CodeMode { + cell_id: cell.as_str().to_string(), + runtime_tool_call_id: "nested".to_string(), + }, + &StepContext::for_test(Arc::clone(&turn)), + ); + recorder.finish_cell_recording(&cell); + items.push(serde_json::from_value(json!({ + "type": "custom_tool_call", "call_id": "exec", "name": "exec", "input": "", + }))?); + items.push(serde_json::from_value(json!({ + "type": "custom_tool_call_output", "call_id": "exec", "output": "done", + }))?); session .record_conversation_items(&turn, turn.model_info(), &items) .await; let live_history = session.clone_history().await; + let mut expected_code_mode_output = live_history + .raw_items() + .find(|item| matches!(item, ResponseItem::CustomToolCallOutput { .. })) + .expect("Code Mode output recorded") + .clone(); + if metadata_enabled { + expected_code_mode_output.append_executed_tool_calls(vec![nested_call]); + expected_code_mode_output.set_tool_call_cell_id("exec"); + expected_code_mode_output.mark_tool_calls_complete(); + } let outputs = live_history .raw_items() .filter_map(|item| match item { @@ -128,6 +169,10 @@ async fn local_compaction_respects_tool_metadata_state( let request = mock.single_request(); assert!(request.inputs_of_type("compaction_trigger").is_empty()); + assert_eq!( + request.custom_tool_call_output("exec"), + serde_json::to_value(expected_code_mode_output)?, + ); for mut output in outputs { let call_id = output["call_id"].as_str().expect("source call id"); let compact_output = request.function_call_output(call_id); diff --git a/codex-rs/core/src/tools/executed_tool_calls/request_metadata.rs b/codex-rs/core/src/tools/executed_tool_calls/request_metadata.rs index 23697a019e..9515c137fd 100644 --- a/codex-rs/core/src/tools/executed_tool_calls/request_metadata.rs +++ b/codex-rs/core/src/tools/executed_tool_calls/request_metadata.rs @@ -3,12 +3,26 @@ use super::*; +#[derive(Clone, Copy, PartialEq, Eq)] +enum MetadataBudgetScope { + AllCalls, + CodeModeOnly, +} + impl ExecutedToolCalls { - /// Compaction preserves captured records without applying the sampling request budget. - pub(crate) fn strip_disabled_direct_metadata(&self, items: &mut [ResponseItem]) { - if self.lock_state().is_none() { + /// Attaches Code Mode observations while preserving Direct records captured in history. + pub(crate) fn attach_to_compaction_prompt(&self, items: &mut [ResponseItem]) { + let mut state = self.lock_state(); + let Some(state) = state.as_mut() else { clear_direct_call_metadata(items); - } + return; + }; + Self::attach_pending_to_prompt_with_state( + state, + items, + &mut HashMap::new(), + MetadataBudgetScope::CodeModeOnly, + ); } /// Attaches trusted completeness; request budgeting only revokes damaged inventories. @@ -19,9 +33,14 @@ impl ExecutedToolCalls { ) { let attached = { let mut state = self.lock_state(); - state - .as_mut() - .map(|state| Self::attach_pending_to_prompt_with_state(state, items, retry_cache)) + state.as_mut().map(|state| { + Self::attach_pending_to_prompt_with_state( + state, + items, + retry_cache, + MetadataBudgetScope::AllCalls, + ) + }) }; let Some(attached) = attached else { // Direct records now live in history; disabling capture also stops replaying them. @@ -37,6 +56,7 @@ impl ExecutedToolCalls { state: &mut ExecutedToolCallRecorderState, items: &mut [ResponseItem], retry_cache: &mut ExecutedToolCallCache, + budget_scope: MetadataBudgetScope, ) -> bool { // Failed wrappers have no callback; only their bounded bitmap observation survives. state.pending_wrapper_origins.clear(); @@ -178,7 +198,9 @@ impl ExecutedToolCalls { } attached = true; } - if !pending_outputs.is_empty() { + // Compaction may retry with a shortened history without installing that history. + // Keep absent observations until a normal sampling request confirms the live window. + if budget_scope == MetadataBudgetScope::AllCalls && !pending_outputs.is_empty() { state .retained_calls .retain(|key, _| !pending_outputs.contains(key)); @@ -259,11 +281,37 @@ impl ExecutedToolCalls { } } - let metadata_bytes = items.iter().fold(0_usize, |bytes, item| { - bytes.saturating_add(executed_tool_call_metadata_bytes(item)) - }); + let metadata_bytes = items + .iter() + .filter(|item| { + budget_scope == MetadataBudgetScope::AllCalls || !has_direct_call_metadata(item) + }) + .fold(0_usize, |bytes, item| { + bytes.saturating_add(executed_tool_call_metadata_bytes(item)) + }); if metadata_bytes > MAX_EXECUTED_TOOL_CALL_FULL_ARGUMENT_BYTES_PER_OUTPUT { - bound_executed_tool_calls_for_prompt_prioritizing_recent(items); + match budget_scope { + MetadataBudgetScope::AllCalls => { + bound_executed_tool_calls_for_prompt_prioritizing_recent(items); + } + MetadataBudgetScope::CodeModeOnly => { + // Validate bindings against the complete history above, but do not make + // captured Direct records compete with Code Mode's bounded recorder. + let (indices, mut bounded): (Vec<_>, Vec<_>) = items + .iter() + .enumerate() + .filter(|(_, item)| { + !has_direct_call_metadata(item) + && item.executed_tool_call_metadata().is_some() + }) + .map(|(index, item)| (index, item.clone())) + .unzip(); + bound_executed_tool_calls_for_prompt_prioritizing_recent(&mut bounded); + for (index, item) in indices.into_iter().zip(bounded) { + items[index] = item; + } + } + } let retained_before_bounding = std::mem::take(&mut state.retained_calls); let mut bounded_outputs = HashSet::new(); for item in items.iter_mut() { @@ -315,6 +363,13 @@ impl ExecutedToolCalls { retained.result_metadata_updated = previous.result_metadata_updated; } } + if budget_scope == MetadataBudgetScope::CodeModeOnly { + state.retained_calls.extend( + retained_before_bounding + .into_iter() + .filter(|(key, _)| pending_outputs.contains(key)), + ); + } } attached 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 c7c6862ceb..ad5b8959d2 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 @@ -20,7 +20,12 @@ impl ExecutedToolCalls { let Some(state) = state.as_mut() else { return false; }; - Self::attach_pending_to_prompt_with_state(state, items, retry_cache) + Self::attach_pending_to_prompt_with_state( + state, + items, + retry_cache, + MetadataBudgetScope::AllCalls, + ) } } @@ -80,6 +85,187 @@ fn tool_calls_complete(item: &ResponseItem) -> Option { .and_then(|metadata| metadata.tool_calls_complete) } +#[test] +fn compaction_attaches_pending_and_retained_code_mode_metadata() { + for previously_sampled in [false, true] { + let recorder = new_recorder(InitialHistory::New); + let cell = CellId::new("compacting-cell".to_string()); + recorder.start_cell(&cell, "exec"); + let mut call = record_nested_call(&recorder, &cell, "nested"); + let source = ToolCallSource::CodeMode { + cell_id: cell.as_str().to_string(), + runtime_tool_call_id: "nested".to_string(), + }; + assert!(recorder.record_tool_result_metadata(&source, "nested", &json!({"id": "initial"}))); + recorder.finish_cell_recording(&cell); + let history = vec![exec_input("exec"), exec_output("exec")]; + if previously_sampled { + let mut sampled = history.clone(); + recorder.attach_to_prompt(&mut sampled, &mut HashMap::new()); + assert_eq!(tool_calls_complete(&sampled[1]), Some(true)); + } + let result_metadata = json!({"id": "latest"}); + assert!(recorder.record_tool_result_metadata(&source, "nested", &result_metadata)); + call.set_tool_result_metadata(ToolResultMetadata::new(&result_metadata)); + let mut expected = history.clone(); + expected[1].append_executed_tool_calls(vec![call]); + expected[1].set_tool_call_cell_id("exec"); + expected[1].mark_tool_calls_complete(); + + let mut compact = history.clone(); + recorder.attach_to_compaction_prompt(&mut compact); + assert_eq!(compact, expected); + let mut retry = history.clone(); + recorder.attach_to_compaction_prompt(&mut retry); + assert_eq!(retry, expected); + // A failed compaction must leave the same observations available to sampling. + let mut sampled = history; + recorder.attach_to_prompt(&mut sampled, &mut HashMap::new()); + assert_eq!(sampled, expected); + } +} + +#[test] +fn failed_compaction_preserves_observations_absent_from_trimmed_retries() { + for rebudget in [false, true] { + let recorder = new_recorder(InitialHistory::New); + let mut history = Vec::new(); + for origin in ["old", "recent"] { + let cell = CellId::new(origin.to_string()); + recorder.start_cell(&cell, origin); + for index in 0..6 { + record_nested_call(&recorder, &cell, &format!("nested-{index}")); + } + recorder.finish_cell_recording(&cell); + history.extend([exec_input(origin), exec_output(origin)]); + } + let mut full_attempt = history.clone(); + recorder.attach_to_compaction_prompt(&mut full_attempt); + assert_eq!(tool_calls_complete(&full_attempt[1]), Some(true)); + + if rebudget { + // Late results make the retained retry item exceed the budget independently of + // the older output that the compaction retry has temporarily removed. + let source = ToolCallSource::CodeMode { + cell_id: "recent".to_string(), + runtime_tool_call_id: "runtime".to_string(), + }; + for index in 0..6 { + assert!(recorder.record_tool_result_metadata( + &source, + &format!("nested-{index}"), + &json!({"provider": "x".repeat(7 * 1024)}), + )); + } + } + let mut trimmed_attempt = history[2..].to_vec(); + recorder.attach_to_compaction_prompt(&mut trimmed_attempt); + assert_eq!(tool_calls_complete(&trimmed_attempt[1]), Some(true)); + if rebudget { + let encoded = serde_json::to_value(&trimmed_attempt[1]).unwrap(); + assert_eq!( + encoded["internal_chat_message_metadata_passthrough"]["executed_tool_calls"][0]["tool_result_metadata"], + json!("omitted_due_to_size_limit"), + ); + } + + // Exhausted retries leave the session's original history intact. + let mut expected = full_attempt; + expected[2..].clone_from_slice(&trimmed_attempt); + let mut resumed = history.clone(); + recorder.attach_to_prompt(&mut resumed, &mut HashMap::new()); + assert_eq!(resumed, expected); + // Once ordinary sampling advances to a smaller window, the old record is pruned. + recorder.attach_to_prompt(&mut history[2..], &mut HashMap::new()); + let mut old = vec![exec_input("old"), exec_output("old")]; + recorder.attach_to_prompt(&mut old, &mut HashMap::new()); + assert_eq!(old, vec![exec_input("old"), exec_output("old")]); + } +} + +#[test] +fn compaction_bounds_code_mode_metadata_without_rebudgeting_direct_history() { + let recorder = new_recorder(InitialHistory::New); + let mut history = Vec::new(); + let arguments = json!({"value": "x".repeat(7 * 1024)}); + let argument_bytes = serialized_json_bytes(&arguments).unwrap(); + for index in 0..6 { + let origin = format!("exec-{index}"); + let cell = CellId::new(format!("cell-{index}")); + recorder.start_cell(&cell, &origin); + let call = ExecutedToolCall::new("nested_tool".to_string(), arguments.clone()); + recorder.record_nested_tool_call( + cell.clone(), + format!("nested-{index}"), + call, + argument_bytes, + ); + recorder.finish_cell_recording(&cell); + history.extend([exec_input(&origin), exec_output(&origin)]); + } + let direct_start = history.len(); + for index in 0..6 { + let mut item = output(&format!("direct-{index}")); + let mut call = ExecutedToolCall::new("direct_tool".to_string(), arguments.clone()); + call.set_tool_result_metadata(ToolResultMetadata::new(&json!({"id": index}))); + recorder.attach_direct_call_to_output( + &mut item, + Some((call, recorder.reserve_direct_call().unwrap())), + ); + history.push(item); + } + assert!( + history[direct_start..] + .iter() + .map(executed_tool_call_metadata_bytes) + .sum::() + > MAX_EXECUTED_TOOL_CALL_FULL_ARGUMENT_BYTES_PER_OUTPUT + ); + + let mut compact = history.clone(); + recorder.attach_to_compaction_prompt(&mut compact); + assert_eq!(&compact[direct_start..], &history[direct_start..]); + assert!( + compact[..direct_start] + .iter() + .map(executed_tool_call_metadata_bytes) + .sum::() + <= MAX_EXECUTED_TOOL_CALL_FULL_ARGUMENT_BYTES_PER_OUTPUT + ); + let mut newest = exec_output("exec-5"); + newest.append_executed_tool_calls(vec![ExecutedToolCall::new( + "nested_tool".to_string(), + arguments, + )]); + newest.set_tool_call_cell_id("exec-5"); + newest.mark_tool_calls_complete(); + assert_eq!(compact[direct_start - 1], newest); + assert_ne!(tool_calls_complete(&compact[1]), Some(true)); + let mut retry = history; + recorder.attach_to_compaction_prompt(&mut retry); + assert_eq!(retry, compact); +} + +#[test] +fn compaction_keeps_direct_outputs_in_code_mode_binding_validation() { + let recorder = new_recorder(InitialHistory::New); + let cell = CellId::new("cell".to_string()); + recorder.start_cell(&cell, "exec"); + let call = record_nested_call(&recorder, &cell, "nested"); + recorder.finish_cell_recording(&cell); + let mut direct = output("exec"); + recorder.attach_direct_call_to_output( + &mut direct, + Some((call.clone(), recorder.reserve_direct_call().unwrap())), + ); + let mut compact = vec![exec_input("exec"), exec_output("exec"), direct.clone()]; + recorder.attach_to_compaction_prompt(&mut compact); + let mut expected = exec_output("exec"); + expected.append_executed_tool_calls(vec![call]); + expected.set_tool_call_cell_id("exec"); + assert_eq!(compact, vec![exec_input("exec"), expected, direct]); +} + #[test] fn direct_retained_metadata_budget_sheds_results_then_calls_across_refresh() { let recorder = new_recorder(InitialHistory::New); diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index d003f1af72..04494892ed 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/codex-rs/core/tests/suite/code_mode.rs @@ -1315,6 +1315,10 @@ await new Promise(() => {}); .len(), 1 ); + assert_eq!( + compact.single_request().custom_tool_call_output("call-1")["internal_chat_message_metadata_passthrough"], + *metadata, + ); let wait = responses::mount_function_call_agent_response( &server, @@ -1346,6 +1350,83 @@ await new Promise(() => {}); Ok(()) } +#[test_case(true; "enabled")] +#[test_case(false; "disabled")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn code_mode_compaction_request_preserves_tool_inventory( + metadata_enabled: bool, +) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let mut builder = test_codex() + .with_model("test-gpt-5.1-codex") + .with_config(move |config| { + config.features.enable(Feature::CodeMode).unwrap(); + config.features.enable(Feature::CodeModeHost).unwrap(); + config.code_mode.disable_in_process_fallback = true; + config + .features + .set_enabled(Feature::ExecutedToolCallMetadata, metadata_enabled) + .unwrap(); + }); + let test = builder.build_with_auto_env(&server).await?; + responses::mount_sse_once( + &server, + sse(vec![ + ev_custom_tool_call( + "call-exec", + "exec", + r#"await tools.test_sync_tool({}); text("done");"#, + ), + ev_completed("resp-exec"), + ]), + ) + .await; + let follow_up = responses::mount_sse_once( + &server, + sse(vec![ + ev_assistant_message("msg-done", "done"), + ev_completed("resp-done"), + ]), + ) + .await; + test.submit_turn("Run a nested tool").await?; + let original = follow_up + .single_request() + .custom_tool_call_output("call-exec"); + assert_eq!( + original["internal_chat_message_metadata_passthrough"]["tool_calls_complete"], + if metadata_enabled { + Value::Bool(true) + } else { + Value::Null + }, + ); + if metadata_enabled { + assert_eq!( + original["internal_chat_message_metadata_passthrough"]["executed_tool_calls"], + serde_json::json!([{ "name": "test_sync_tool", "arguments": {} }]), + ); + } + + let summary = serde_json::json!({ + "type": "response.output_item.done", + "item": {"type": "compaction", "encrypted_content": "compacted history"}, + }); + let compact = + responses::mount_sse_once(&server, sse(vec![summary, ev_completed("resp-compact")])).await; + test.codex.submit(Op::Compact).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let request = compact.single_request(); + assert_eq!(request.inputs_of_type("compaction_trigger").len(), 1); + assert_eq!(request.custom_tool_call_output("call-exec"), original); + test.codex.shutdown_and_wait().await?; + Ok(()) +} + #[test_case("call-wait", None; "reused_wait_id")] #[test_case("call-fresh-wait", Some(true); "fresh_wait_id")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]