From efbd12f4b66b935897474f0f8b379964460cd107 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 28 May 2026 08:41:31 -0400 Subject: [PATCH] fix(agent): preserve tool-call pairs during compaction Keep matching assistant tool calls when compaction preserves tool results so OpenAI receives valid function_call_output history. Add an OpenAI twin regression that forces compaction after tool use and validates the session continues without orphaned tool outputs. --- lib/crates/fabro-agent/src/history.rs | 86 ++++++++++++++- lib/crates/fabro-agent/tests/it/compaction.rs | 101 ++++++++++++++++++ lib/crates/fabro-agent/tests/it/main.rs | 1 + lib/crates/fabro-test/src/lib.rs | 8 ++ test/twin/openai/src/engine/scenario.rs | 16 ++- test/twin/openai/src/openai/models.rs | 29 ++++- test/twin/openai/src/sse.rs | 5 + 7 files changed, 241 insertions(+), 5 deletions(-) create mode 100644 lib/crates/fabro-agent/tests/it/compaction.rs diff --git a/lib/crates/fabro-agent/src/history.rs b/lib/crates/fabro-agent/src/history.rs index e9dd10476..94e4c91eb 100644 --- a/lib/crates/fabro-agent/src/history.rs +++ b/lib/crates/fabro-agent/src/history.rs @@ -41,7 +41,8 @@ impl History { if self.turns.len() <= preserve_count { return; } - let mut preserved = self.turns.split_off(self.turns.len() - preserve_count); + let preserve_start = compact_preserve_start(&self.turns, preserve_count); + let mut preserved = self.turns.split_off(preserve_start); Self::invalidate_preserved_usage(&mut preserved); let discarded = std::mem::take(&mut self.turns); let extracted_user_messages = @@ -165,6 +166,32 @@ fn extract_recent_user_messages(discarded: Vec, token_budget: usize) -> .collect() } +fn compact_preserve_start(turns: &[Message], preserve_count: usize) -> usize { + let mut start = turns.len().saturating_sub(preserve_count); + + loop { + let mut required_call_ids = Vec::new(); + for turn in &turns[start..] { + if let Message::ToolResults { results, .. } = turn { + required_call_ids.extend(results.iter().map(|result| result.tool_call_id.as_str())); + } + } + + let Some(call_index) = turns[..start].iter().rposition(|turn| { + let Message::Assistant { tool_calls, .. } = turn else { + return false; + }; + tool_calls + .iter() + .any(|tool_call| required_call_ids.contains(&tool_call.id.as_str())) + }) else { + return start; + }; + + start = call_index; + } +} + #[cfg(test)] mod tests { use std::time::SystemTime; @@ -223,6 +250,63 @@ mod tests { assert!(matches!(&turns[8], Message::User { content, .. } if content == "msg 7")); } + #[test] + fn compact_preserves_matching_tool_calls_for_preserved_tool_results() { + let mut history = History::default(); + history.push(Message::User { + content: "old msg".into(), + timestamp: SystemTime::now(), + }); + for index in 0..3 { + let call_id = format!("call_{index}"); + history.push(Message::Assistant { + content: String::new(), + tool_calls: vec![ToolCall::new( + &call_id, + "read_file", + serde_json::json!({ "file_path": format!("{index}.txt") }), + )], + provider_parts: vec![], + usage: Box::new(TokenCounts::default()), + response_id: format!("resp_{index}"), + timestamp: SystemTime::now(), + }); + history.push(Message::ToolResults { + results: vec![ToolResult::success(&call_id, serde_json::json!("ok"))], + timestamp: SystemTime::now(), + }); + } + history.push(Message::Assistant { + content: String::new(), + tool_calls: vec![ToolCall::new( + "call_3", + "read_file", + serde_json::json!({ "file_path": "3.txt" }), + )], + provider_parts: vec![], + usage: Box::new(TokenCounts::default()), + response_id: "resp_3".into(), + timestamp: SystemTime::now(), + }); + + history.compact(6, "Summary".into()); + let messages = history.convert_to_messages(); + let mut seen_tool_calls = Vec::new(); + for message in messages { + for part in message.content { + match part { + ContentPart::ToolCall(tool_call) => seen_tool_calls.push(tool_call.id), + ContentPart::ToolResult(result) => assert!( + seen_tool_calls.contains(&result.tool_call_id), + "tool result {} should have a matching preserved tool call", + result.tool_call_id + ), + _ => {} + } + } + } + } + #[test] fn compact_summary_maps_to_system_message() { let mut history = History::default(); diff --git a/lib/crates/fabro-agent/tests/it/compaction.rs b/lib/crates/fabro-agent/tests/it/compaction.rs new file mode 100644 index 000000000..7eb9c5128 --- /dev/null +++ b/lib/crates/fabro-agent/tests/it/compaction.rs @@ -0,0 +1,101 @@ +use std::collections::HashMap; +use std::path::Path; +use std::sync::Arc; + +use fabro_agent::{AgentProfile, LocalSandbox, OpenAiProfile, Session, SessionOptions}; +use fabro_llm::client::Client; +use fabro_llm::provider::ProviderAdapter; +use fabro_llm::providers::OpenAiAdapter; +use fabro_model::ProviderId; +use fabro_test::{TwinScenario, TwinScenarios, TwinToolCall, twin_openai}; +use tokio::fs::read_to_string; + +const MODEL: &str = "gpt-5.4-mini"; + +#[expect( + clippy::disallowed_methods, + reason = "e2e_openai! intentionally reads test credentials from the environment" +)] +#[fabro_macros::e2e_test(twin, live("OPENAI_API_KEY"))] +async fn openai_twin_compaction_preserves_tool_call_pairs() { + let tmp = tempfile::tempdir().expect("failed to create tempdir"); + let (base_url, api_key) = fabro_test::e2e_openai!(); + + if fabro_test::TestMode::from_env().is_twin() { + load_compaction_scenarios(&api_key).await; + } + + let mut session = make_openai_session(tmp.path(), base_url, api_key); + session.initialize().await.unwrap(); + + let result = session + .process_input( + "Trigger the compaction regression by writing four small files, then say done.", + ) + .await; + + assert!( + result.is_ok(), + "session should complete without sending an orphaned function_call_output: {result:?}" + ); + assert_eq!( + read_to_string(tmp.path().join("four.txt")) + .await + .expect("four.txt should be written"), + "four" + ); +} + +fn make_openai_session(cwd: &Path, base_url: String, api_key: String) -> Session { + let adapter: Arc = + Arc::new(OpenAiAdapter::new(api_key).with_base_url(base_url)); + let mut providers = HashMap::new(); + providers.insert(ProviderId::OPENAI.to_string(), adapter); + let client = Client::new(providers, Some(ProviderId::OPENAI.to_string()), Vec::new()); + let profile: Arc = Arc::new(OpenAiProfile::new(MODEL)); + let sandbox = Arc::new(LocalSandbox::new(cwd.to_path_buf())); + let options = SessionOptions { + max_turns: 20, + enable_context_compaction: true, + compaction_threshold_percent: 80, + compaction_preserve_turns: 6, + ..SessionOptions::default() + }; + + Session::new(client, profile, sandbox, options, None) +} + +async fn load_compaction_scenarios(namespace: &str) { + TwinScenarios::new(namespace.to_string()) + .scenario( + TwinScenario::responses(MODEL) + .stream(true) + .input_contains("Trigger the compaction regression") + .tool_call(TwinToolCall::write_file("one.txt", "one")), + ) + .scenario( + TwinScenario::responses(MODEL) + .stream(true) + .tool_call(TwinToolCall::write_file("two.txt", "two")), + ) + .scenario( + TwinScenario::responses(MODEL) + .stream(true) + .tool_call(TwinToolCall::write_file("three.txt", "three")), + ) + .scenario( + TwinScenario::responses(MODEL) + .stream(true) + .tool_call(TwinToolCall::write_file("four.txt", "four")) + .usage(180_000, 5), + ) + .scenario( + TwinScenario::responses(MODEL) + .stream(false) + .input_contains("Here is the conversation to summarize") + .text("short summary"), + ) + .scenario(TwinScenario::responses(MODEL).stream(true).text("Done.")) + .load(twin_openai().await) + .await; +} diff --git a/lib/crates/fabro-agent/tests/it/main.rs b/lib/crates/fabro-agent/tests/it/main.rs index 3ffb1e7e2..efb7dbdce 100644 --- a/lib/crates/fabro-agent/tests/it/main.rs +++ b/lib/crates/fabro-agent/tests/it/main.rs @@ -1,2 +1,3 @@ +mod compaction; mod guardrails; mod parity_matrix; diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index cd8b041f0..6d3dd64a6 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -2205,6 +2205,14 @@ impl TwinScenario { self } + #[must_use] + pub fn usage(mut self, input_tokens: u64, output_tokens: u64) -> Self { + self.assert_script_kind("success", "usage"); + self.script["input_tokens"] = Value::Number(input_tokens.into()); + self.script["output_tokens"] = Value::Number(output_tokens.into()); + self + } + #[must_use] pub fn input_contains(mut self, needle: impl Into) -> Self { self.matcher diff --git a/test/twin/openai/src/engine/scenario.rs b/test/twin/openai/src/engine/scenario.rs index ee7ff7377..46b980404 100644 --- a/test/twin/openai/src/engine/scenario.rs +++ b/test/twin/openai/src/engine/scenario.rs @@ -35,6 +35,8 @@ pub enum ScenarioScript { reasoning: Option>, structured_output: Option, tool_calls: Option>, + input_tokens: Option, + output_tokens: Option, delay_before_headers_ms: Option, inter_event_delay_ms: Option, close_after_chunks: Option, @@ -123,6 +125,8 @@ impl Scenario { reasoning, structured_output, tool_calls, + input_tokens, + output_tokens, delay_before_headers_ms, inter_event_delay_ms, close_after_chunks, @@ -136,6 +140,8 @@ impl Scenario { reasoning.clone().unwrap_or_default(), structured_output.clone(), tool_calls.clone().unwrap_or_default(), + *input_tokens, + *output_tokens, ), transport: TransportOptions { delay_before_headers_ms: delay_before_headers_ms.unwrap_or_default(), @@ -178,6 +184,8 @@ impl Scenario { reasoning, structured_output, tool_calls, + input_tokens, + output_tokens, delay_before_headers_ms, inter_event_delay_ms, close_after_chunks, @@ -191,6 +199,8 @@ impl Scenario { reasoning.clone().unwrap_or_default(), structured_output.clone(), tool_calls.clone().unwrap_or_default(), + *input_tokens, + *output_tokens, ), transport: TransportOptions { delay_before_headers_ms: delay_before_headers_ms.unwrap_or_default(), @@ -231,6 +241,8 @@ fn build_plan_from_script( reasoning: Vec, structured_output: Option, tool_calls: Vec, + input_tokens: Option, + output_tokens: Option, ) -> ResponsePlan { let output_text = match response_text { Some(response_text) => response_text, @@ -257,7 +269,7 @@ fn build_plan_from_script( arguments: tool_call.arguments, }) .collect(), - input_tokens: 1, - output_tokens: 5, + input_tokens: input_tokens.unwrap_or(1), + output_tokens: output_tokens.unwrap_or(5), } } diff --git a/test/twin/openai/src/openai/models.rs b/test/twin/openai/src/openai/models.rs index bd7996891..79799b6d3 100644 --- a/test/twin/openai/src/openai/models.rs +++ b/test/twin/openai/src/openai/models.rs @@ -1,3 +1,5 @@ +use std::collections::HashSet; + use axum::Json; use axum::extract::rejection::JsonRejection; use axum::http::StatusCode; @@ -99,7 +101,7 @@ impl ResponsesRequest { )?; validate_tool_choice_requires_tools(self.tool_choice.as_ref(), self.tools.as_ref())?; validate_stop(self.stop.as_ref(), "stop")?; - validate_response_input(&self.input)?; + validate_response_input(&self.input, self.previous_response_id.as_deref())?; Ok(()) } @@ -223,13 +225,36 @@ impl ContentPart { } } -fn validate_response_input(input: &ResponseInput) -> Result<(), OpenAiError> { +fn validate_response_input( + input: &ResponseInput, + previous_response_id: Option<&str>, +) -> Result<(), OpenAiError> { let ResponseInput::Items(items) = input else { return Ok(()); }; + let mut function_call_ids = HashSet::new(); for item in items { validate_input_item(item)?; + match item.item_type.as_deref() { + Some("function_call" | "custom_tool_call") => { + if let Some(call_id) = item.call_id.as_deref().filter(|id| !id.is_empty()) { + function_call_ids.insert(call_id); + } + } + Some("function_call_output") if previous_response_id.is_none() => { + let call_id = item.call_id.as_deref().unwrap_or_default(); + if !function_call_ids.contains(call_id) { + return Err(OpenAiError::invalid_request( + "input", + &format!( + "No tool call found for function call output with call_id {call_id}." + ), + )); + } + } + _ => {} + } } Ok(()) diff --git a/test/twin/openai/src/sse.rs b/test/twin/openai/src/sse.rs index 1649dae32..7934c9782 100644 --- a/test/twin/openai/src/sse.rs +++ b/test/twin/openai/src/sse.rs @@ -227,6 +227,11 @@ pub fn responses_sse_response(plan: &ResponsePlan, transport: TransportOptions) "created": plan.created, "model": plan.model, "status": "completed", + "usage": { + "input_tokens": plan.input_tokens, + "output_tokens": plan.output_tokens, + "total_tokens": plan.input_tokens + plan.output_tokens, + }, }, }), ));