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

@@ -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<Message>, 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();

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<dyn ProviderAdapter> =
+        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<dyn AgentProfile> = 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;
+}

lib/crates/fabro-agent/tests/it/main.rs

@@ -1,2 +1,3 @@
+mod compaction;
 mod guardrails;
 mod parity_matrix;

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<String>) -> Self {
         self.matcher

test/twin/openai/src/engine/scenario.rs

@@ -35,6 +35,8 @@ pub enum ScenarioScript {
         reasoning:               Option<Vec<String>>,
         structured_output:       Option<Value>,
         tool_calls:              Option<Vec<ToolCallTemplate>>,
+        input_tokens:            Option<u64>,
+        output_tokens:           Option<u64>,
         delay_before_headers_ms: Option<u64>,
         inter_event_delay_ms:    Option<u64>,
         close_after_chunks:      Option<usize>,
@@ -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<String>,
     structured_output: Option<Value>,
     tool_calls: Vec<ToolCallTemplate>,
+    input_tokens: Option<u64>,
+    output_tokens: Option<u64>,
 ) -> 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),
     }
 }

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(())

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,
+                    },
                 },
             }),
         ));