fix(agent): align compaction preserve boundary (#449)

## 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

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

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

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

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
     }
 

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

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<Value>,
     pub reasoning:         Vec<String>,
     pub tool_calls:        Vec<ToolCallPlan>,
-    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::<Vec<_>>(),
                 }
             }],
-            "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()
         })
     }
 }

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<Vec<String>>,
         structured_output:       Option<Value>,
         tool_calls:              Option<Vec<ToolCallTemplate>>,
-        input_tokens:            Option<u64>,
-        output_tokens:           Option<u64>,
+        usage:                   Option<TokenUsage>,
         delay_before_headers_ms: Option<u64>,
         inter_event_delay_ms:    Option<u64>,
         close_after_chunks:      Option<usize>,
@@ -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<String>,
     structured_output: Option<Value>,
     tool_calls: Vec<ToolCallTemplate>,
-    input_tokens: Option<u64>,
-    output_tokens: Option<u64>,
+    usage: Option<TokenUsage>,
 ) -> 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(),
     }
 }

test/twin/openai/src/openai/models.rs

@@ -162,7 +162,7 @@ pub struct InputItem {
 
 impl InputItem {
     fn extract_texts_for_fallback(&self) -> Vec<String> {
-        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(())
+        }
     }
 }
 

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