From 2e39dfc70eb90e794216f7e66f8369daa0e25600 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp <19+brynary@users.noreply.github.com> Date: Thu, 28 May 2026 11:26:52 -0400 Subject: [PATCH] fix(agent): align compaction preserve boundary (#449) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Follow-up to fabro-sh/fabro#447. This keeps context compaction's effective preserve boundary consistent between summary generation, history mutation, and emitted telemetry so tool-call/result pairs that remain in raw history are not also summarized. The branch also tightens the OpenAI twin support added for this regression: scripted usage is modeled as a single `TokenUsage`, SSE completion payloads reuse the canonical Responses JSON shape, and request validation now treats custom tool-call outputs as tool outputs instead of spreading raw item-type string checks. ## Verification - `cargo +nightly-2026-04-14 fmt --check --all` - `cargo nextest run -p fabro-agent compaction` - `cargo nextest run -p twin-openai` - `FABRO_TEST_MODE=twin cargo nextest run -p fabro-agent --profile e2e --run-ignored only --test it openai_twin_compaction_preserves_tool_call_pairs` - `git diff --check origin/main...HEAD` --- [![Compound Engineering](https://img.shields.io/badge/Compound_Engineering-6366f1)](https://github.com/EveryInc/compound-engineering-plugin) 🤖 Generated with GPT-5 via [Codex](https://openai.com/codex) --- lib/crates/fabro-agent/src/compaction.rs | 15 ++-- lib/crates/fabro-agent/src/history.rs | 58 +++++++++++--- lib/crates/fabro-agent/tests/it/compaction.rs | 8 +- lib/crates/fabro-test/src/lib.rs | 6 +- test/twin/openai/src/engine/defaults.rs | 8 +- test/twin/openai/src/engine/plan.rs | 61 ++++++++++++--- test/twin/openai/src/engine/scenario.rs | 23 ++---- test/twin/openai/src/openai/models.rs | 76 ++++++++++++++----- test/twin/openai/src/sse.rs | 13 +--- 9 files changed, 181 insertions(+), 87 deletions(-) diff --git a/lib/crates/fabro-agent/src/compaction.rs b/lib/crates/fabro-agent/src/compaction.rs index 03d55638b..c4395a2e3 100644 --- a/lib/crates/fabro-agent/src/compaction.rs +++ b/lib/crates/fabro-agent/src/compaction.rs @@ -78,19 +78,22 @@ pub(crate) async fn compact_context( session_id: &str, ) -> Result<(), Error> { let original_turn_count = history.turns().len(); + let preserve_start = history.compact_preserve_start(preserve_count); - // Determine turns to summarize. If there are not enough turns to compact, - // do not emit a started event without a matching completion. - if original_turn_count <= preserve_count { + // If preserving tool call/result pairs leaves no prefix to summarize, do + // not spend a summarization call or emit a started event without a + // matching completion. + if preserve_start == 0 { return Ok(()); } + let preserved_turn_count = original_turn_count - preserve_start; emitter.emit(session_id.to_owned(), AgentEvent::CompactionStarted { estimated_tokens: estimate.tokens, context_window_size: provider_profile.context_window_size(), }); - let turns_to_summarize = &history.turns()[..original_turn_count - preserve_count]; + let turns_to_summarize = &history.turns()[..preserve_start]; let rendered = render_turns_for_summary(turns_to_summarize); // Build structured summarization prompt @@ -157,11 +160,11 @@ Build on their progress — do not repeat completed steps.\n\n{summary_text}" ); let summary_token_estimate = estimate_chars_local_tokens(summary_content.len()); - history.compact(preserve_count, summary_content); + history.compact_from(preserve_start, summary_content); emitter.emit(session_id.to_owned(), AgentEvent::CompactionCompleted { original_turn_count, - preserved_turn_count: preserve_count, + preserved_turn_count, summary_token_estimate, tracked_file_count: file_tracker.file_count(), }); diff --git a/lib/crates/fabro-agent/src/history.rs b/lib/crates/fabro-agent/src/history.rs index 94e4c91eb..b42cf1993 100644 --- a/lib/crates/fabro-agent/src/history.rs +++ b/lib/crates/fabro-agent/src/history.rs @@ -1,3 +1,5 @@ +use std::collections::HashSet; + use fabro_llm::types::{ContentPart, Message as LlmMessage, Role, TokenCounts}; use fabro_types::SessionMessage; @@ -41,7 +43,19 @@ impl History { if self.turns.len() <= preserve_count { return; } - let preserve_start = compact_preserve_start(&self.turns, preserve_count); + let preserve_start = self.compact_preserve_start(preserve_count); + self.compact_from(preserve_start, summary); + } + + #[must_use] + pub(crate) fn compact_preserve_start(&self, preserve_count: usize) -> usize { + compact_preserve_start(&self.turns, preserve_count) + } + + pub(crate) fn compact_from(&mut self, preserve_start: usize, summary: String) { + if preserve_start == 0 || preserve_start > self.turns.len() { + return; + } let mut preserved = self.turns.split_off(preserve_start); Self::invalidate_preserved_usage(&mut preserved); let discarded = std::mem::take(&mut self.turns); @@ -168,30 +182,34 @@ fn extract_recent_user_messages(discarded: Vec, token_budget: usize) -> fn compact_preserve_start(turns: &[Message], preserve_count: usize) -> usize { let mut start = turns.len().saturating_sub(preserve_count); + let mut required_call_ids = HashSet::new(); + add_tool_result_call_ids(&turns[start..], &mut required_call_ids); 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())) + .any(|tool_call| required_call_ids.contains(tool_call.id.as_str())) }) else { return start; }; + add_tool_result_call_ids(&turns[call_index..start], &mut required_call_ids); start = call_index; } } +fn add_tool_result_call_ids<'a>(turns: &'a [Message], call_ids: &mut HashSet<&'a str>) { + for turn in turns { + if let Message::ToolResults { results, .. } = turn { + call_ids.extend(results.iter().map(|result| result.tool_call_id.as_str())); + } + } +} + #[cfg(test)] mod tests { use std::time::SystemTime; @@ -307,6 +325,28 @@ mod tests { } } + #[test] + fn compact_noops_when_preserved_tool_result_requires_first_turn() { + let mut history = History::default(); + history.push(Message::Assistant { + content: String::new(), + tool_calls: vec![ToolCall::new("call_1", "read_file", serde_json::json!({}))], + provider_parts: vec![], + usage: Box::new(TokenCounts::default()), + response_id: "resp_1".into(), + timestamp: SystemTime::now(), + }); + history.push(Message::ToolResults { + results: vec![ToolResult::success("call_1", serde_json::json!("ok"))], + timestamp: SystemTime::now(), + }); + + history.compact(1, "Summary".into()); + + assert_eq!(history.turns().len(), 2); + assert!(!matches!(history.turns()[0], Message::System { .. })); + } + #[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 index 7eb9c5128..084deae6c 100644 --- a/lib/crates/fabro-agent/tests/it/compaction.rs +++ b/lib/crates/fabro-agent/tests/it/compaction.rs @@ -14,16 +14,14 @@ const MODEL: &str = "gpt-5.4-mini"; #[expect( clippy::disallowed_methods, - reason = "e2e_openai! intentionally reads test credentials from the environment" + reason = "e2e_openai! expands live-mode environment lookups even for twin-only tests" )] -#[fabro_macros::e2e_test(twin, live("OPENAI_API_KEY"))] +#[fabro_macros::e2e_test(twin)] 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; - } + load_compaction_scenarios(&api_key).await; let mut session = make_openai_session(tmp.path(), base_url, api_key); session.initialize().await.unwrap(); diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index 6d3dd64a6..77512bfe6 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -2208,8 +2208,10 @@ impl TwinScenario { #[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.script["usage"] = json!({ + "input_tokens": input_tokens, + "output_tokens": output_tokens, + }); self } diff --git a/test/twin/openai/src/engine/defaults.rs b/test/twin/openai/src/engine/defaults.rs index d259759e7..169caf4da 100644 --- a/test/twin/openai/src/engine/defaults.rs +++ b/test/twin/openai/src/engine/defaults.rs @@ -1,6 +1,6 @@ use serde_json::{Value, json}; -use super::plan::ResponsePlan; +use super::plan::{ResponsePlan, TokenUsage}; use crate::openai::models::{ResponseFormat, ResponsesRequest, normalize_whitespace}; pub fn build_default_response_plan( @@ -35,8 +35,7 @@ pub fn build_default_response_plan( structured_output, reasoning, tool_calls: Vec::new(), - input_tokens, - output_tokens: 5, + usage: TokenUsage::new(input_tokens, 5), } } @@ -74,8 +73,7 @@ pub fn build_default_chat_plan( structured_output, reasoning, tool_calls: Vec::new(), - input_tokens, - output_tokens: 5, + usage: TokenUsage::new(input_tokens, 5), } } diff --git a/test/twin/openai/src/engine/plan.rs b/test/twin/openai/src/engine/plan.rs index 054078ebe..6440d28e1 100644 --- a/test/twin/openai/src/engine/plan.rs +++ b/test/twin/openai/src/engine/plan.rs @@ -1,5 +1,51 @@ +use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct TokenUsage { + pub input_tokens: u64, + pub output_tokens: u64, +} + +impl TokenUsage { + #[must_use] + pub const fn new(input_tokens: u64, output_tokens: u64) -> Self { + Self { + input_tokens, + output_tokens, + } + } + + #[must_use] + pub const fn total_tokens(self) -> u64 { + self.input_tokens + self.output_tokens + } + + #[must_use] + pub fn responses_json(self) -> Value { + json!({ + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "total_tokens": self.total_tokens(), + }) + } + + #[must_use] + pub fn chat_completions_json(self) -> Value { + json!({ + "prompt_tokens": self.input_tokens, + "completion_tokens": self.output_tokens, + "total_tokens": self.total_tokens(), + }) + } +} + +impl Default for TokenUsage { + fn default() -> Self { + Self::new(1, 5) + } +} + #[derive(Clone, Debug)] pub struct ResponsePlan { pub id: String, @@ -9,8 +55,7 @@ pub struct ResponsePlan { pub structured_output: Option, pub reasoning: Vec, pub tool_calls: Vec, - pub input_tokens: u64, - pub output_tokens: u64, + pub usage: TokenUsage, } #[derive(Clone, Debug)] @@ -77,11 +122,7 @@ impl ResponsePlan { "status": "completed", "reasoning": self.reasoning, "output": output, - "usage": { - "input_tokens": self.input_tokens, - "output_tokens": self.output_tokens, - "total_tokens": self.input_tokens + self.output_tokens, - } + "usage": self.usage.responses_json() }) } @@ -108,11 +149,7 @@ impl ResponsePlan { })).collect::>(), } }], - "usage": { - "prompt_tokens": self.input_tokens, - "completion_tokens": self.output_tokens, - "total_tokens": self.input_tokens + self.output_tokens, - } + "usage": self.usage.chat_completions_json() }) } } diff --git a/test/twin/openai/src/engine/scenario.rs b/test/twin/openai/src/engine/scenario.rs index 46b980404..e88715808 100644 --- a/test/twin/openai/src/engine/scenario.rs +++ b/test/twin/openai/src/engine/scenario.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; use super::failures::{ErrorOutcome, ExecutionOutcome, SuccessOutcome, TransportOptions}; -use super::plan::{ResponsePlan, ToolCallPlan}; +use super::plan::{ResponsePlan, TokenUsage, ToolCallPlan}; use crate::openai::models::{ChatCompletionsRequest, ResponsesRequest}; #[derive(Clone, Debug, Deserialize)] @@ -35,8 +35,7 @@ pub enum ScenarioScript { reasoning: Option>, structured_output: Option, tool_calls: Option>, - input_tokens: Option, - output_tokens: Option, + usage: Option, delay_before_headers_ms: Option, inter_event_delay_ms: Option, close_after_chunks: Option, @@ -125,8 +124,7 @@ impl Scenario { reasoning, structured_output, tool_calls, - input_tokens, - output_tokens, + usage, delay_before_headers_ms, inter_event_delay_ms, close_after_chunks, @@ -140,8 +138,7 @@ impl Scenario { reasoning.clone().unwrap_or_default(), structured_output.clone(), tool_calls.clone().unwrap_or_default(), - *input_tokens, - *output_tokens, + *usage, ), transport: TransportOptions { delay_before_headers_ms: delay_before_headers_ms.unwrap_or_default(), @@ -184,8 +181,7 @@ impl Scenario { reasoning, structured_output, tool_calls, - input_tokens, - output_tokens, + usage, delay_before_headers_ms, inter_event_delay_ms, close_after_chunks, @@ -199,8 +195,7 @@ impl Scenario { reasoning.clone().unwrap_or_default(), structured_output.clone(), tool_calls.clone().unwrap_or_default(), - *input_tokens, - *output_tokens, + *usage, ), transport: TransportOptions { delay_before_headers_ms: delay_before_headers_ms.unwrap_or_default(), @@ -241,8 +236,7 @@ fn build_plan_from_script( reasoning: Vec, structured_output: Option, tool_calls: Vec, - input_tokens: Option, - output_tokens: Option, + usage: Option, ) -> ResponsePlan { let output_text = match response_text { Some(response_text) => response_text, @@ -269,7 +263,6 @@ fn build_plan_from_script( arguments: tool_call.arguments, }) .collect(), - input_tokens: input_tokens.unwrap_or(1), - output_tokens: output_tokens.unwrap_or(5), + usage: usage.unwrap_or_default(), } } diff --git a/test/twin/openai/src/openai/models.rs b/test/twin/openai/src/openai/models.rs index 79799b6d3..01d8fa4cc 100644 --- a/test/twin/openai/src/openai/models.rs +++ b/test/twin/openai/src/openai/models.rs @@ -162,7 +162,7 @@ pub struct InputItem { impl InputItem { fn extract_texts_for_fallback(&self) -> Vec { - if self.item_type.as_deref() == Some("function_call_output") { + if self.kind().is_tool_output() { return self .output .as_ref() @@ -176,6 +176,41 @@ impl InputItem { Vec::new() } + + fn kind(&self) -> InputItemKind { + InputItemKind::from_wire(self.item_type.as_deref()) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum InputItemKind { + Message, + FunctionCall, + CustomToolCall, + FunctionCallOutput, + CustomToolCallOutput, + Other, +} + +impl InputItemKind { + fn from_wire(item_type: Option<&str>) -> Self { + match item_type { + None | Some("message") => Self::Message, + Some("function_call") => Self::FunctionCall, + Some("custom_tool_call") => Self::CustomToolCall, + Some("function_call_output") => Self::FunctionCallOutput, + Some("custom_tool_call_output") => Self::CustomToolCallOutput, + Some(_) => Self::Other, + } + } + + fn is_tool_call(self) -> bool { + matches!(self, Self::FunctionCall | Self::CustomToolCall) + } + + fn is_tool_output(self) -> bool { + matches!(self, Self::FunctionCallOutput | Self::CustomToolCallOutput) + } } #[derive(Clone, Debug, Deserialize, Default)] @@ -236,24 +271,19 @@ fn validate_response_input( 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); - } + let kind = item.kind(); + if kind.is_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}." - ), - )); - } + } else if kind.is_tool_output() && 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 tool call output with call_id {call_id}."), + )); } - _ => {} } } @@ -261,12 +291,16 @@ fn validate_response_input( } fn validate_input_item(item: &InputItem) -> Result<(), OpenAiError> { - match item.item_type.as_deref() { - Some("function_call_output") => validate_function_call_output_item(item), - None => validate_message_input_item(item), + match item.kind() { + InputItemKind::FunctionCallOutput | InputItemKind::CustomToolCallOutput => { + validate_function_call_output_item(item) + } + InputItemKind::Message => validate_message_input_item(item), // Accept any other item type — the twin extracts user text for fallback // responses and ignores items it doesn't understand. - Some(_) => Ok(()), + InputItemKind::FunctionCall | InputItemKind::CustomToolCall | InputItemKind::Other => { + Ok(()) + } } } diff --git a/test/twin/openai/src/sse.rs b/test/twin/openai/src/sse.rs index 7934c9782..728fde5ac 100644 --- a/test/twin/openai/src/sse.rs +++ b/test/twin/openai/src/sse.rs @@ -221,18 +221,7 @@ pub fn responses_sse_response(plan: &ResponsePlan, transport: TransportOptions) "response.completed", &json!({ "type": "response.completed", - "response": { - "id": plan.id, - "object": "response", - "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, - }, - }, + "response": plan.responses_json(), }), )); }