use crate::counter::{TiktokenTokenCounter, TokenCounter};
use crate::limits::{create_budget_for_model, ModelLimitsRegistry};
use crate::{BudgetStrategy, TokenBudget};
use bamboo_domain::MessagePhase;
use bamboo_domain::{
CompressionEvent, CompressionTriggerType, ConversationSummary, Message, Session,
};
/// Checks if a message is part of a skill tool chain (load_skill / read_skill_resource).
fn is_skill_tool_chain_message(message: &Message) -> bool {
message.tool_calls.as_ref().is_some_and(|calls| {
calls.iter().any(|call| {
matches!(
call.function.name.as_str(),
"load_skill" | "read_skill_resource"
)
})
})
}
/// The `tool_call_id`s a message participates in: the ids of the tool calls an
/// assistant message initiates, plus the id of the call a `tool` result answers.
/// Two messages belong to the same tool chain iff these overlap.
fn tool_chain_call_ids(message: &Message) -> impl Iterator<Item = String> + '_ {
message
.tool_calls
.iter()
.flatten()
.map(|call| call.id.clone())
.chain(message.tool_call_id.clone())
}
/// Close the compressed (`messages_to_summarize`) set over tool chains: repeatedly
/// move any message still in `messages_to_keep` that shares a `tool_call_id` with
/// an already-compressed message into the summarize set. This keeps an assistant
/// `tool_calls` message and its matching `tool` result(s) on the same side — a
/// split leaves an orphan `tool_result` (or a `tool_use` with no result) in the
/// active set, which providers reject with a 400 that then poisons every
/// subsequent request in the session (#340). Protected messages are left in place
/// (skill chains are already fully protected upstream, so are never partially
/// compressed; protected user messages carry no `tool_call_id`).
fn close_compressed_set_over_tool_chains(
messages_to_keep: &mut Vec<Message>,
messages_to_summarize: &mut Vec<Message>,
protected_user_ids: &HashSet<String>,
never_compress_ids: &[String],
) {
loop {
let compressed_call_ids: HashSet<String> = messages_to_summarize
.iter()
.flat_map(tool_chain_call_ids)
.collect();
let split_index = messages_to_keep.iter().position(|message| {
!protected_user_ids.contains(message.id.as_str())
&& !never_compress_ids.contains(&message.id)
&& tool_chain_call_ids(message).any(|id| compressed_call_ids.contains(&id))
});
match split_index {
Some(index) => messages_to_summarize.push(messages_to_keep.remove(index)),
None => break,
}
}
}
use chrono::Utc;
use std::collections::HashSet;
/// Structured reason why a compression plan could not be built.
#[derive(Debug, Clone)]
pub enum CompressionPlanError {
/// The exposure gate (threshold not reached) prevented building.
ExposureGateNotMet {
usage_percent: f64,
trigger_percent: u8,
},
/// No active messages in the session.
NoActiveMessages,
/// Not enough non-system messages to compress (need >=3).
NotEnoughMessages { non_system_count: usize },
/// Nothing to compress after anchor/keep splitting.
NothingToCompress {
anchor_index: usize,
non_system_count: usize,
},
/// Eligible history was exhausted while protected/recent content still kept
/// the active prompt above the configured post-compression target.
ProtectedContentExceedsTarget {
projected_tokens: u32,
target_tokens: u32,
},
/// The session changed after candidate selection and before finalization.
CandidateSetChanged,
/// The real generated summary was larger than the source-derived reserve,
/// so applying the candidate set would miss the post-compression target.
SummaryExceedsTarget {
projected_tokens: u32,
target_tokens: u32,
},
}
impl std::fmt::Display for CompressionPlanError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::ExposureGateNotMet {
usage_percent,
trigger_percent,
} => write!(
f,
"compression threshold not reached (usage={:.1}%, trigger={}%)",
usage_percent, trigger_percent
),
Self::NoActiveMessages => write!(f, "no active messages to compress"),
Self::NotEnoughMessages { non_system_count } => write!(
f,
"not enough non-system messages to compress ({}, need >=3)",
non_system_count
),
Self::NothingToCompress {
anchor_index,
non_system_count,
} => write!(
f,
"nothing to compress after anchor/keep splitting (anchor_index={}, non_system={})",
anchor_index, non_system_count
),
Self::ProtectedContentExceedsTarget {
projected_tokens,
target_tokens,
} => write!(
f,
"protected active content prevents compression target (projected={}, target={})",
projected_tokens, target_tokens
),
Self::CandidateSetChanged => {
write!(f, "compression candidate set changed before finalization")
}
Self::SummaryExceedsTarget {
projected_tokens,
target_tokens,
} => write!(
f,
"actual summary misses compression target (projected={}, target={})",
projected_tokens, target_tokens
),
}
}
}
/// Metadata about current context pressure, used to decide when compression
/// should be requested by host-side control flow.
#[derive(Debug, Clone)]
pub struct ContextCompressionExposure {
pub budget: TokenBudget,
pub active_tokens: u32,
pub active_usage_percent: f64,
pub active_usage_percent_rounded: u8,
pub should_expose_tool: bool,
}
/// A compression plan describing which active historical messages should be
/// archived and summarized.
#[derive(Debug, Clone)]
pub struct CompressionPlan {
/// Stable identifier for all requests and the persisted event belonging to
/// one logical compression pass.
pub logical_pass_id: Option<String>,
/// Tokens from active prompt blocks rendered outside `Session.messages`.
pub fixed_prompt_tokens: u32,
pub compressed_message_ids: Vec<String>,
pub messages_to_summarize: Vec<Message>,
pub summary_tokens: u32,
pub summary_content: String,
pub active_usage_before_percent: f64,
pub active_usage_after_percent: f64,
pub trigger_percent: u8,
pub target_percent: u8,
pub segments_removed: usize,
pub trigger_type: CompressionTriggerType,
pub compression_ratio: f64,
pub model_used: Option<String>,
pub latency_ms: u64,
pub source_tokens: u32,
pub represented_source_tokens: u32,
pub target_summary_tokens: u32,
pub actual_summary_content_tokens: u32,
pub summary_target_ratio: f64,
pub summary_budget_clamped: bool,
pub summary_budget_clamp_reason: Option<String>,
pub summarization_map_calls: u32,
pub summarization_reduce_calls: u32,
pub summarization_fallback_used: bool,
}
/// Immutable archive selection produced before any summarization request is
/// sent. The final summary must represent exactly this set.
#[derive(Debug, Clone)]
pub struct CompressionCandidatePlan {
pub compressed_message_ids: Vec<String>,
pub messages_to_summarize: Vec<Message>,
pub source_tokens: u32,
pub previous_represented_source_tokens: u32,
pub represented_source_tokens: u32,
pub target_summary_tokens: u32,
pub summary_target_ratio: f64,
pub active_usage_before_percent: f64,
pub projected_usage_after_percent: f64,
pub trigger_percent: u8,
pub target_percent: u8,
pub segments_removed: usize,
pub trigger_type: CompressionTriggerType,
additional_fixed_tokens: u32,
context_window: u32,
target_limit: u32,
}
pub const DEFAULT_SUMMARY_TARGET_RATIO: f64 = 0.20;
pub fn normalized_summary_target_ratio(value: f64) -> f64 {
if value.is_finite() && value > 0.0 {
value.clamp(0.01, 0.50)
} else {
DEFAULT_SUMMARY_TARGET_RATIO
}
}
fn target_summary_content_tokens(
session: &Session,
counter: &impl TokenCounter,
newly_represented_tokens: u32,
target_ratio: f64,
) -> (u32, u32) {
let target_ratio = normalized_summary_target_ratio(target_ratio);
match session.conversation_summary.as_ref() {
Some(summary) if summary.represented_source_tokens > 0 => {
let represented = summary
.represented_source_tokens
.saturating_add(newly_represented_tokens);
(
((represented as f64) * target_ratio).ceil() as u32,
summary.represented_source_tokens,
)
}
Some(summary) => {
let existing_tokens = counter.count_text(&summary.content);
let inferred_previous_source = ((existing_tokens as f64) / target_ratio).ceil() as u32;
(
existing_tokens.saturating_add(
((newly_represented_tokens as f64) * target_ratio).ceil() as u32,
),
inferred_previous_source,
)
}
None => (
((newly_represented_tokens as f64) * target_ratio).ceil() as u32,
0,
),
}
}
fn protected_compression_message_ids(non_system: &[Message]) -> HashSet<String> {
let user_indexes = non_system
.iter()
.enumerate()
.filter_map(|(index, message)| {
matches!(message.role, bamboo_domain::Role::User).then_some(index)
})
.collect::<Vec<_>>();
let keep_user_count = user_indexes.len().min(3);
let mut protected = user_indexes[user_indexes.len().saturating_sub(keep_user_count)..]
.iter()
.filter_map(|index| non_system.get(*index))
.map(|message| message.id.clone())
.collect::<HashSet<_>>();
let skill_call_ids = non_system
.iter()
.filter(|message| is_skill_tool_chain_message(message))
.flat_map(|message| {
message
.tool_calls
.iter()
.flatten()
.map(|call| call.id.clone())
})
.collect::<HashSet<_>>();
for message in non_system {
if message.never_compress
|| is_skill_tool_chain_message(message)
|| message
.tool_call_id
.as_ref()
.is_some_and(|id| skill_call_ids.contains(id))
{
protected.insert(message.id.clone());
}
}
protected
}
fn post_compaction_recovery_tokens(
compressed_messages: &[Message],
session: &Session,
counter: &impl TokenCounter,
) -> u32 {
build_post_compaction_recovery_message(compressed_messages, session)
.as_ref()
.map(|message| counter.count_message(message))
.unwrap_or(0)
}
/// Select the exact archive candidates before invoking the summarization model.
///
/// Candidate segments are considered oldest-first. Generic tool chains remain
/// atomic because selection operates on [`crate::MessageSegmenter`] output;
/// newest user turns, `never_compress`, and skill chains remain active.
pub fn build_forced_compression_candidate_plan(
session: &Session,
model_name: &str,
configured_budget: Option<&TokenBudget>,
summary_target_ratio: f64,
trigger_type: CompressionTriggerType,
) -> Result<CompressionCandidatePlan, CompressionPlanError> {
build_forced_compression_candidate_plan_with_fixed_tokens(
session,
model_name,
configured_budget,
summary_target_ratio,
trigger_type,
0,
)
}
/// Variant of [`build_forced_compression_candidate_plan`] that accounts for
/// fixed prompt blocks rendered outside `Session.messages` (for example task,
/// plan, workflow, project-resource, and external-memory context blocks).
pub fn build_forced_compression_candidate_plan_with_fixed_tokens(
session: &Session,
model_name: &str,
configured_budget: Option<&TokenBudget>,
summary_target_ratio: f64,
trigger_type: CompressionTriggerType,
additional_fixed_tokens: u32,
) -> Result<CompressionCandidatePlan, CompressionPlanError> {
let exposure = estimate_context_compression_exposure(session, model_name, configured_budget);
let budget = &exposure.budget;
let counter = TiktokenTokenCounter::default();
let active_messages = active_messages_for_budget(session);
if active_messages.is_empty() {
return Err(CompressionPlanError::NoActiveMessages);
}
let system_messages = active_messages
.iter()
.filter(|message| matches!(message.role, bamboo_domain::Role::System))
.cloned()
.collect::<Vec<_>>();
let non_system = active_messages
.into_iter()
.filter(|message| !matches!(message.role, bamboo_domain::Role::System))
.collect::<Vec<_>>();
if non_system.len() < 3 {
return Err(CompressionPlanError::NotEnoughMessages {
non_system_count: non_system.len(),
});
}
let context_window = budget.max_context_tokens;
let target_limit = budget.compression_target_context_tokens();
let system_tokens = counter.count_messages(&system_messages);
let protected_ids = protected_compression_message_ids(&non_system);
let segments = crate::segmenter::MessageSegmenter::new().segment(non_system.clone());
let mut remaining_tokens = counter.count_messages(&non_system);
let mut source_tokens = 0u32;
let mut selected_messages = Vec::new();
let mut selected_segment_count = 0usize;
let ratio = normalized_summary_target_ratio(summary_target_ratio);
let summary_envelope_tokens = counter.count_messages(&[compression_summary_message("")]);
let mut projected_tokens = system_tokens
.saturating_add(remaining_tokens)
.saturating_add(additional_fixed_tokens);
let mut target_summary_tokens = 0u32;
let mut previous_represented_source_tokens = session
.conversation_summary
.as_ref()
.map(|summary| summary.represented_source_tokens)
.unwrap_or(0);
for segment in segments {
if segment
.messages
.iter()
.any(|message| protected_ids.contains(&message.id))
{
continue;
}
let segment_tokens = counter.count_messages(&segment.messages);
source_tokens = source_tokens.saturating_add(segment_tokens);
remaining_tokens = remaining_tokens.saturating_sub(segment_tokens);
selected_messages.extend(segment.messages);
selected_segment_count += 1;
let (desired, previous_represented) =
target_summary_content_tokens(session, &counter, source_tokens, ratio);
target_summary_tokens = desired;
previous_represented_source_tokens = previous_represented;
let recovery_tokens =
post_compaction_recovery_tokens(&selected_messages, session, &counter);
projected_tokens = system_tokens
.saturating_add(remaining_tokens)
.saturating_add(additional_fixed_tokens)
.saturating_add(summary_envelope_tokens)
.saturating_add(target_summary_tokens)
.saturating_add(recovery_tokens);
if projected_tokens <= target_limit {
break;
}
}
if selected_messages.is_empty() {
return Err(CompressionPlanError::NothingToCompress {
anchor_index: 0,
non_system_count: non_system.len(),
});
}
if projected_tokens > target_limit {
return Err(CompressionPlanError::ProtectedContentExceedsTarget {
projected_tokens,
target_tokens: target_limit,
});
}
let represented_source_tokens =
previous_represented_source_tokens.saturating_add(source_tokens);
let projected_usage_after_percent =
context_window_usage_percent(projected_tokens, context_window);
let compressed_message_ids = selected_messages
.iter()
.map(|message| message.id.clone())
.collect();
Ok(CompressionCandidatePlan {
compressed_message_ids,
messages_to_summarize: selected_messages,
source_tokens,
previous_represented_source_tokens,
represented_source_tokens,
target_summary_tokens,
summary_target_ratio: ratio,
active_usage_before_percent: exposure.active_usage_percent,
projected_usage_after_percent,
trigger_percent: budget.compression_trigger_percent,
target_percent: budget.compression_target_percent,
segments_removed: selected_segment_count,
trigger_type,
additional_fixed_tokens,
context_window,
target_limit,
})
}
/// Bind a completed summary to a previously selected candidate set and verify
/// the real post-compression prompt before any session mutation occurs.
pub fn finalize_compression_candidate_plan(
session: &Session,
candidate: CompressionCandidatePlan,
summary_content: String,
) -> Result<CompressionPlan, CompressionPlanError> {
let active_ids = session
.messages
.iter()
.filter(|message| !message.compressed)
.map(|message| message.id.as_str())
.collect::<HashSet<_>>();
if candidate
.compressed_message_ids
.iter()
.any(|id| !active_ids.contains(id.as_str()))
{
return Err(CompressionPlanError::CandidateSetChanged);
}
let candidate_ids = candidate
.compressed_message_ids
.iter()
.map(String::as_str)
.collect::<HashSet<_>>();
let counter = TiktokenTokenCounter::default();
let summary_tokens = counter.count_messages(&[compression_summary_message(&summary_content)]);
let actual_summary_content_tokens = counter.count_text(&summary_content);
let recovery_tokens =
post_compaction_recovery_tokens(&candidate.messages_to_summarize, session, &counter);
let remaining_tokens = session
.messages
.iter()
.filter(|message| !message.compressed)
.filter(|message| !candidate_ids.contains(message.id.as_str()))
.cloned()
.collect::<Vec<_>>();
let projected_tokens = counter
.count_messages(&remaining_tokens)
.saturating_add(candidate.additional_fixed_tokens)
.saturating_add(summary_tokens)
.saturating_add(recovery_tokens);
if projected_tokens > candidate.target_limit {
return Err(CompressionPlanError::SummaryExceedsTarget {
projected_tokens,
target_tokens: candidate.target_limit,
});
}
Ok(CompressionPlan {
logical_pass_id: None,
fixed_prompt_tokens: candidate.additional_fixed_tokens,
compressed_message_ids: candidate.compressed_message_ids,
messages_to_summarize: candidate.messages_to_summarize,
summary_tokens,
summary_content,
active_usage_before_percent: candidate.active_usage_before_percent,
active_usage_after_percent: context_window_usage_percent(
projected_tokens,
candidate.context_window,
),
trigger_percent: candidate.trigger_percent,
target_percent: candidate.target_percent,
segments_removed: candidate.segments_removed,
trigger_type: candidate.trigger_type,
compression_ratio: 0.0,
model_used: None,
latency_ms: 0,
source_tokens: candidate.source_tokens,
represented_source_tokens: candidate.represented_source_tokens,
target_summary_tokens: candidate.target_summary_tokens,
actual_summary_content_tokens,
summary_target_ratio: candidate.summary_target_ratio,
summary_budget_clamped: false,
summary_budget_clamp_reason: None,
summarization_map_calls: 0,
summarization_reduce_calls: 0,
summarization_fallback_used: false,
})
}
pub fn context_window_usage_percent(total_tokens: u32, context_window_tokens: u32) -> f64 {
if context_window_tokens == 0 {
return 0.0;
}
(total_tokens as f64 / context_window_tokens as f64) * 100.0
}
pub fn normalized_trigger_percent(trigger_percent: u8) -> f64 {
match trigger_percent {
0 => 100.0,
1..=100 => trigger_percent as f64,
_ => 100.0,
}
}
/// Estimate whether context pressure has crossed the configured threshold for
/// compression eligibility.
pub fn estimate_context_compression_exposure(
session: &Session,
model_name: &str,
configured_budget: Option<&TokenBudget>,
) -> ContextCompressionExposure {
// When a budget was already resolved upstream (the production path — see
// `resolve_token_budget`, which publishes the current-round snapshot in
// `session.resolved_token_budget` (#180),
// issue #20 bug 1), use it directly. Only when none is available do we fall
// back to a model-derived budget. No `model_limits.json` registry is in
// scope synchronously here, so this fallback resolves to the global default
// rather than silently fabricating an empty override registry (#20 bug 2).
let budget = configured_budget.cloned().unwrap_or_else(|| {
create_budget_for_model(
model_name,
BudgetStrategy::default(),
&ModelLimitsRegistry::new(),
)
});
let counter = TiktokenTokenCounter::default();
let active_messages = active_messages_for_budget(session);
let active_message_tokens = counter.count_messages(&active_messages);
let summary_tokens = session
.conversation_summary
.as_ref()
.map(|summary| counter.count_messages(&[compression_summary_message(&summary.content)]))
.unwrap_or(0);
let active_tokens = active_message_tokens.saturating_add(summary_tokens);
// Use context window as the denominator for a single, provider-aligned
// pressure scale across backend and frontend.
let context_window = budget.max_context_tokens;
let estimated_usage = context_window_usage_percent(active_tokens, context_window);
let usage = session
.token_usage
.as_ref()
.and_then(|token_usage| {
let denominator = if token_usage.max_context_tokens > 0 {
token_usage.max_context_tokens
} else if token_usage.budget_limit > 0 {
// Legacy payload compatibility.
token_usage.budget_limit
} else {
context_window
};
(denominator > 0).then_some(context_window_usage_percent(
token_usage.total_tokens,
denominator,
))
})
.map(|persisted_usage| persisted_usage.max(estimated_usage))
.unwrap_or(estimated_usage);
let rounded = usage.clamp(0.0, 100.0).round() as u8;
let trigger_tokens = budget.compression_trigger_context_tokens();
let trigger_percent = if budget.max_context_tokens > 0 {
(trigger_tokens as f64 / budget.max_context_tokens as f64) * 100.0
} else {
0.0
};
let threshold_reached = usage >= trigger_percent;
// Check non-system message count to stay consistent with the plan
// building requirement of >=3 non-system messages. Using
// active_messages.len() would include system messages and expose the
// tool even when plan building would immediately fail.
let non_system_count = active_messages
.iter()
.filter(|m| !matches!(m.role, bamboo_domain::Role::System))
.count();
let should_expose_tool = threshold_reached && non_system_count >= 3;
ContextCompressionExposure {
budget,
active_tokens,
active_usage_percent: usage,
active_usage_percent_rounded: rounded,
should_expose_tool,
}
}
/// Build a compression plan that archives older active messages and replaces
/// them with a caller-provided summary.
pub fn build_compression_plan_with_summary(
session: &Session,
model_name: &str,
configured_budget: Option<&TokenBudget>,
summary_content: String,
) -> Result<CompressionPlan, CompressionPlanError> {
build_compression_plan_with_summary_internal(
session,
model_name,
configured_budget,
summary_content,
true,
CompressionTriggerType::Auto,
)
}
/// Build a compression plan while bypassing "tool exposure" gating.
///
/// This is intended for host-enforced fallback paths when context pressure is
/// critically high and compression must be attempted regardless of the normal
/// trigger gate.
pub fn build_forced_compression_plan_with_summary(
session: &Session,
model_name: &str,
configured_budget: Option<&TokenBudget>,
summary_content: String,
trigger_type: CompressionTriggerType,
) -> Result<CompressionPlan, CompressionPlanError> {
build_compression_plan_with_summary_internal(
session,
model_name,
configured_budget,
summary_content,
false,
trigger_type,
)
}
fn build_compression_plan_with_summary_internal(
session: &Session,
model_name: &str,
configured_budget: Option<&TokenBudget>,
summary_content: String,
require_exposure_gate: bool,
trigger_type: CompressionTriggerType,
) -> Result<CompressionPlan, CompressionPlanError> {
let exposure = estimate_context_compression_exposure(session, model_name, configured_budget);
if require_exposure_gate && !exposure.should_expose_tool {
return Err(CompressionPlanError::ExposureGateNotMet {
usage_percent: exposure.active_usage_percent,
trigger_percent: exposure.budget.compression_trigger_percent,
});
}
let budget = &exposure.budget;
let counter = TiktokenTokenCounter::default();
let summary_message = compression_summary_message(&summary_content);
let summary_tokens = counter.count_messages(&[summary_message]);
let context_window = budget.max_context_tokens;
let target_limit = budget.compression_target_context_tokens();
let mut active_messages = active_messages_for_budget(session);
if active_messages.is_empty() {
tracing::debug!("compression plan: no active messages, cannot build plan");
return Err(CompressionPlanError::NoActiveMessages);
}
let system_messages: Vec<Message> = active_messages
.iter()
.filter(|m| matches!(m.role, bamboo_domain::Role::System))
.cloned()
.collect();
let system_tokens = counter.count_messages(&system_messages);
let reserved_non_window_tokens = system_tokens.saturating_add(summary_tokens);
let window_limit = target_limit.saturating_sub(reserved_non_window_tokens);
let non_system: Vec<Message> = active_messages
.drain(..)
.filter(|m| !matches!(m.role, bamboo_domain::Role::System))
.collect();
if non_system.len() < 3 {
tracing::debug!(
"compression plan: not enough non-system messages ({}), need at least 3",
non_system.len()
);
return Err(CompressionPlanError::NotEnoughMessages {
non_system_count: non_system.len(),
});
}
let user_indexes = non_system
.iter()
.enumerate()
.filter_map(|(index, message)| {
matches!(message.role, bamboo_domain::Role::User).then_some(index)
})
.collect::<Vec<_>>();
let keep_user_count = user_indexes.len().min(3);
let anchor_index = if keep_user_count > 0 {
user_indexes[user_indexes.len() - keep_user_count]
} else {
non_system
.iter()
.rposition(|m| matches!(m.role, bamboo_domain::Role::User))
.unwrap_or(non_system.len().saturating_sub(1))
};
let protected_user_ids: HashSet<String> = if keep_user_count > 0 {
user_indexes[user_indexes.len() - keep_user_count..]
.iter()
.filter_map(|idx| non_system.get(*idx))
.map(|message| message.id.clone())
.collect()
} else {
HashSet::new()
};
tracing::debug!(
"compression plan: context_window={}, target_limit={}, system_tokens={}, summary_tokens={}, window_limit={}, non_system_messages={}, keep_user_count={}, keep_from_index={}",
context_window, target_limit, system_tokens, summary_tokens, window_limit, non_system.len(), keep_user_count, anchor_index
);
// Keep the newest 3 user turns (or fewer if there are not enough user
// turns) as active context and summarize older history before that
// boundary. If budget is still too high, continue moving the oldest
// non-protected messages into the summarize set.
let mut messages_to_summarize = non_system[..anchor_index].to_vec();
// Protected messages must never be summarized — move them to the keep set.
let mut never_compress_ids: Vec<String> = messages_to_summarize
.iter()
.filter(|m| m.never_compress || is_skill_tool_chain_message(m))
.map(|m| m.id.clone())
.collect();
// Also protect tool result messages that correspond to skill tool calls.
let skill_call_ids: Vec<String> = messages_to_summarize
.iter()
.filter(|m| is_skill_tool_chain_message(m))
.flat_map(|m| m.tool_calls.iter().flatten().map(|c| c.id.clone()))
.collect();
if !skill_call_ids.is_empty() {
for m in &*messages_to_summarize {
if let Some(ref call_id) = m.tool_call_id {
if skill_call_ids.contains(call_id) && !never_compress_ids.contains(&m.id) {
never_compress_ids.push(m.id.clone());
}
}
}
}
if !never_compress_ids.is_empty() {
messages_to_summarize.retain(|m| !never_compress_ids.contains(&m.id));
}
let non_system_count = non_system.len();
let mut messages_to_keep = non_system[anchor_index..].to_vec();
// Add never_compress / skill messages to the keep set.
for id in &never_compress_ids {
if let Some(msg) = non_system.iter().find(|m| &m.id == id) {
if !messages_to_keep.iter().any(|m| m.id == *id) {
messages_to_keep.push(msg.clone());
}
}
}
while !messages_to_keep.is_empty() {
let keep_tokens = counter.count_messages(&messages_to_keep);
if keep_tokens <= window_limit {
break;
}
let Some(remove_index) = messages_to_keep.iter().position(|message| {
!protected_user_ids.contains(message.id.as_str())
&& !never_compress_ids.contains(&message.id)
}) else {
// Remaining messages are all protected; stop shrinking.
break;
};
let moved = messages_to_keep.remove(remove_index);
messages_to_summarize.push(moved);
}
// Tool-chain atomicity. The one-at-a-time eviction above can split a generic
// (non-skill) tool chain — compressing an assistant `tool_calls` message
// while keeping its `tool` result active, or vice versa. That leaves an
// orphan `tool_result` (or a `tool_use` with no result) in the active set,
// which providers reject with a 400 that then poisons EVERY subsequent
// request in the session. Close the compressed set over tool chains: keep
// moving any kept message that shares a `tool_call_id` with an
// already-compressed message into the summarize set until none remain.
// (Skill chains are fully protected above, so are never partially
// compressed; protected user messages carry no `tool_call_id`.)
close_compressed_set_over_tool_chains(
&mut messages_to_keep,
&mut messages_to_summarize,
&protected_user_ids,
&never_compress_ids,
);
if messages_to_summarize.is_empty() {
tracing::debug!(
"compression plan: messages_to_summarize is empty after anchor/keep splitting"
);
return Err(CompressionPlanError::NothingToCompress {
anchor_index,
non_system_count,
});
}
let compressed_message_ids = messages_to_summarize
.iter()
.map(|message| message.id.clone())
.collect::<Vec<_>>();
let keep_tokens = counter.count_messages(&messages_to_keep);
let active_before = exposure.active_usage_percent;
// Use context_window as denominator, consistent with
// estimate_context_compression_exposure().
let active_after = if context_window == 0 {
0.0
} else {
let after_total = reserved_non_window_tokens.saturating_add(keep_tokens);
(after_total as f64 / context_window as f64) * 100.0
};
// Count actual segments being compressed using the same segmenter that
// prepare_hybrid_context uses, so the segment count is accurate.
let segmenter = crate::segmenter::MessageSegmenter::new();
let segments_removed = segmenter.segment(messages_to_summarize.clone()).len();
let source_tokens = counter.count_messages(&messages_to_summarize);
let actual_summary_content_tokens = counter.count_text(&summary_content);
let previous_represented_source_tokens = session
.conversation_summary
.as_ref()
.map(|summary| summary.represented_source_tokens)
.unwrap_or(0);
Ok(CompressionPlan {
logical_pass_id: None,
fixed_prompt_tokens: 0,
compressed_message_ids,
messages_to_summarize,
summary_tokens,
summary_content,
active_usage_before_percent: active_before,
active_usage_after_percent: active_after,
trigger_percent: budget.compression_trigger_percent,
target_percent: budget.compression_target_percent,
segments_removed,
trigger_type,
compression_ratio: 0.0,
model_used: None,
latency_ms: 0,
source_tokens,
represented_source_tokens: previous_represented_source_tokens.saturating_add(source_tokens),
target_summary_tokens: actual_summary_content_tokens,
actual_summary_content_tokens,
summary_target_ratio: 0.0,
summary_budget_clamped: false,
summary_budget_clamp_reason: None,
summarization_map_calls: 0,
summarization_reduce_calls: 0,
summarization_fallback_used: false,
})
}
/// Apply a previously computed compression plan to the session.
/// Extract recently modified files from tool calls in the given messages.
pub(super) fn extract_recently_modified_files(messages: &[Message]) -> Vec<(String, String)> {
let mut files = Vec::new();
for message in messages {
if let Some(ref tool_calls) = message.tool_calls {
for call in tool_calls {
let tool_name = call.function.name.as_str();
if !matches!(tool_name, "Write" | "Edit" | "Bash") {
continue;
}
let args = &call.function.arguments;
if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(args) {
if let Some(path) = parsed.get("file_path").and_then(|v| v.as_str()) {
files.push((path.to_string(), tool_name.to_string()));
} else if let Some(cmd) = parsed.get("command").and_then(|v| v.as_str()) {
// Extract file paths from shell commands heuristically
for part in cmd.split_whitespace() {
if part.contains('/')
&& (part.ends_with(".rs")
|| part.ends_with(".ts")
|| part.ends_with(".js")
|| part.ends_with(".toml")
|| part.ends_with(".json")
|| part.ends_with(".md"))
{
files.push((part.to_string(), "Bash".to_string()));
}
}
}
}
}
}
}
files.truncate(10);
files
}
/// Extract key decision snippets from assistant messages.
pub(super) fn extract_key_decisions(messages: &[Message], limit: usize) -> Vec<String> {
let decision_keywords = [
"decided to",
"approach is",
"use ",
"using ",
"we'll go with",
"the plan is",
"strategy:",
"solution:",
"chose to",
"switched to",
"refactored to",
"migrated to",
"replaced with",
];
let mut decisions = Vec::new();
for message in messages {
if !matches!(message.role, bamboo_domain::Role::Assistant) {
continue;
}
let content = &message.content;
for line in content.lines() {
let line_lower = line.to_lowercase();
if decision_keywords.iter().any(|kw| line_lower.contains(kw)) {
let truncated: String = line.chars().take(200).collect();
decisions.push(truncated);
if decisions.len() >= limit {
return decisions;
}
}
}
}
decisions
}
/// Build a post-compaction recovery message that preserves critical context
/// from the compressed messages so the LLM can continue work without losing
/// track of active files, tasks, and decisions.
fn build_post_compaction_recovery_message(
compressed_messages: &[Message],
session: &Session,
) -> Option<Message> {
if compressed_messages.is_empty() {
return None;
}
let mut sections = Vec::new();
// 1. Recently modified files
let files = extract_recently_modified_files(compressed_messages);
if !files.is_empty() {
let mut section = String::from("## Recently Modified Files\n");
for (path, tool) in &files {
section.push_str(&format!("- {} ({})\n", path, tool));
}
sections.push(section);
}
// 2. Active tasks from task list
if let Some(ref task_list) = session.task_list {
let active_items: Vec<_> = task_list
.items
.iter()
.filter(|item| !matches!(item.status, bamboo_domain::TaskItemStatus::Completed))
.collect();
if !active_items.is_empty() {
let mut section = String::from("## Active Tasks\n");
for item in active_items.iter().take(10) {
section.push_str(&format!("- [{:?}] {}\n", item.status, item.description));
}
sections.push(section);
}
}
// 3. Key decisions
let decisions = extract_key_decisions(compressed_messages, 5);
if !decisions.is_empty() {
let mut section = String::from("## Key Decisions\n");
for decision in &decisions {
section.push_str(&format!("- {}\n", decision));
}
sections.push(section);
}
if sections.is_empty() {
return None;
}
let mut content = String::from("[post-compaction-recovery]\nContext extracted from compressed messages for continued work.\n\n");
content.push_str(§ions.join("\n"));
let mut message = Message::assistant(content, None);
message.never_compress = true;
Some(message)
}
struct SummaryQualityMetrics {
file_coverage: f64,
decision_coverage: f64,
}
fn validate_summary_quality(summary: &str, messages: &[Message]) -> SummaryQualityMetrics {
let files = extract_recently_modified_files(messages);
let decisions = extract_key_decisions(messages, 10);
let files_mentioned = files
.iter()
.filter(|(path, _)| summary.contains(path.as_str()))
.count();
let file_coverage = if files.is_empty() {
1.0
} else {
files_mentioned as f64 / files.len() as f64
};
let decisions_mentioned = decisions
.iter()
.filter(|d| {
let check_str: String = d.chars().take(50).collect();
summary.contains(&check_str)
})
.count();
let decision_coverage = if decisions.is_empty() {
1.0
} else {
decisions_mentioned as f64 / decisions.len() as f64
};
SummaryQualityMetrics {
file_coverage,
decision_coverage,
}
}
pub fn apply_compression_plan(session: &mut Session, plan: CompressionPlan) -> usize {
let compressed_ids: HashSet<&str> = plan
.compressed_message_ids
.iter()
.map(String::as_str)
.collect();
let mut changed_indexes = Vec::new();
for (index, message) in session.messages.iter_mut().enumerate() {
if message.compressed || !compressed_ids.contains(message.id.as_str()) {
continue;
}
message.compressed = true;
changed_indexes.push(index);
}
if changed_indexes.is_empty() {
return 0;
}
let mut event = CompressionEvent::new(
changed_indexes.len(),
plan.segments_removed,
plan.active_usage_before_percent,
plan.active_usage_after_percent,
plan.summary_tokens,
plan.trigger_type,
plan.compression_ratio,
plan.model_used.clone(),
plan.latency_ms,
);
if let Some(logical_pass_id) = plan.logical_pass_id.as_ref() {
event.id.clone_from(logical_pass_id);
}
event.source_tokens = plan.source_tokens;
event.fixed_prompt_tokens = plan.fixed_prompt_tokens;
event.actual_summary_tokens = plan.actual_summary_content_tokens;
event.target_summary_tokens = plan.target_summary_tokens;
event.summary_target_ratio = plan.summary_target_ratio;
event.actual_summary_ratio = if plan.represented_source_tokens == 0 {
0.0
} else {
plan.actual_summary_content_tokens as f64 / plan.represented_source_tokens as f64
};
event.summary_budget_clamped = plan.summary_budget_clamped;
event.summary_budget_clamp_reason = plan.summary_budget_clamp_reason.clone();
event.summarization_map_calls = plan.summarization_map_calls;
event.summarization_reduce_calls = plan.summarization_reduce_calls;
event.summarization_fallback_used = plan.summarization_fallback_used;
let event_id = event.id.clone();
for index in changed_indexes {
session.messages[index].compressed_by_event_id = Some(event_id.clone());
}
session.compression_events.push(event);
session.conversation_summary = Some(
ConversationSummary::new(
&plan.summary_content,
plan.compressed_message_ids.len(),
plan.summary_tokens,
)
.with_compression_metrics(
plan.represented_source_tokens,
plan.target_summary_tokens,
plan.summary_target_ratio,
plan.summary_budget_clamped,
plan.summary_budget_clamp_reason.clone(),
),
);
// Inject a post-compaction recovery message to preserve critical context
// from the compressed messages (files, tasks, decisions).
let compressed_messages: Vec<Message> = session
.messages
.iter()
.filter(|m| compressed_ids.contains(m.id.as_str()))
.cloned()
.collect();
if let Some(recovery) = build_post_compaction_recovery_message(&compressed_messages, session) {
// Insert just before the last user message, or at the end
let insert_pos = session
.messages
.iter()
.rposition(|m| matches!(m.role, bamboo_domain::Role::User) && !m.compressed)
.map(|pos| pos + 1)
.unwrap_or(session.messages.len());
session.messages.insert(insert_pos, recovery);
}
let quality = validate_summary_quality(&plan.summary_content, &compressed_messages);
if quality.file_coverage < 0.5 || quality.decision_coverage < 0.3 {
tracing::warn!(
"[{}] Summary quality: file_coverage={:.0}%, decision_coverage={:.0}%",
session.id,
quality.file_coverage * 100.0,
quality.decision_coverage * 100.0
);
}
// Instead of clearing token_usage entirely (which forces the next round
// to rely on heuristic estimates that don't account for tool schema
// tokens), recompute an approximate post-compression snapshot. We
// preserve both the total context-window denominator and the request input
// limit from the previous usage snapshot so their meanings stay distinct
// across rounds.
let counter = TiktokenTokenCounter::default();
let remaining_active: Vec<_> = session
.messages
.iter()
.filter(|m| !m.compressed)
.cloned()
.collect();
let system_msgs: Vec<_> = remaining_active
.iter()
.filter(|m| matches!(m.role, bamboo_domain::Role::System))
.cloned()
.collect();
let window_msgs: Vec<_> = remaining_active
.iter()
.filter(|m| !matches!(m.role, bamboo_domain::Role::System))
.cloned()
.collect();
let system_tokens = counter
.count_messages(&system_msgs)
.saturating_add(plan.fixed_prompt_tokens);
let new_summary_tokens = plan.summary_tokens;
let window_tokens = counter.count_messages(&window_msgs);
let total_tokens = system_tokens
.saturating_add(new_summary_tokens)
.saturating_add(window_tokens);
let previous_usage = session.token_usage.take();
let budget_limit = previous_usage
.as_ref()
.map(|u| {
if u.budget_limit > 0 {
u.budget_limit
} else {
// Legacy snapshots used the total context window as the only
// denominator.
u.max_context_tokens
}
})
.unwrap_or(0);
let max_context_tokens = previous_usage
.as_ref()
.map(|u| u.max_context_tokens)
.unwrap_or(0);
session.token_usage = Some(bamboo_domain::TokenBudgetUsage {
system_tokens,
summary_tokens: new_summary_tokens,
window_tokens,
total_tokens,
max_context_tokens,
budget_limit,
truncation_occurred: false,
segments_removed: 0,
prompt_cached_tool_outputs: 0,
prompt_cached_tool_tokens_saved: 0,
thinking_tokens: 0,
cache_read_input_tokens: 0,
});
session.updated_at = Utc::now();
plan.compressed_message_ids.len()
}
pub fn compression_summary_message(summary_content: &str) -> Message {
Message::system(format!(
"<!-- CONVERSATION_SUMMARY_START -->\n\
## Previous Conversation Summary\n\
The following is compressed historical context for continuity only.\n\
It is background memory, not a new user request. Follow the current task list and recent messages over this summary when they conflict.\n\n\
{}\n\
<!-- CONVERSATION_SUMMARY_END -->",
summary_content
))
}
pub fn active_messages_for_budget(session: &Session) -> Vec<Message> {
session
.messages
.iter()
.filter(|message| !message.compressed)
.cloned()
.collect()
}
pub fn summary_source_messages(session: &Session) -> Vec<Message> {
session
.messages
.iter()
.filter(|message| !message.compressed)
.filter(|message| !matches!(message.role, bamboo_domain::Role::System))
.cloned()
.collect()
}
pub fn build_summary_prompt(
session: &Session,
messages: &[Message],
existing_summary: Option<&str>,
) -> String {
let mut content = String::new();
content.push_str(
"You are compressing conversation history for continued work. Produce a compact but reliable working-memory summary.\n\n",
);
content.push_str(
"Critical requirements:\n- First capture the in-flight work right before compression (what was being done, where, and with which tool/file)\n- Distinguish clearly between ACTIVE work, COMPLETED work, and OBSOLETE or superseded work\n- Do not restate old tasks as active unless they are still unresolved\n- The current task list is the source of truth for what is actively being worked on\n- Preserve constraints, decisions, file paths, code changes, errors, tool findings, blockers, and the next step\n- If earlier plans conflict with the current task list or newer messages, treat the earlier plans as obsolete or completed\n- Explicitly evaluate each clear user requirement (e.g. requirement 1, requirement 2) with a status and evidence\n- Return only summary text in the same language as the conversation\n\n",
);
if let Some(existing) = existing_summary.map(str::trim).filter(|s| !s.is_empty()) {
content.push_str("## Existing Summary\n");
content.push_str(existing);
content.push_str("\n\n");
}
let task_list_prompt = session.format_task_list_for_prompt();
if !task_list_prompt.trim().is_empty() {
content.push_str("## Current Task List\n");
content.push_str(task_list_prompt.trim());
content.push_str("\n\n");
}
content.push_str(
"## Required Output Sections\n1. Pre-compression in-flight work (what was being done immediately before compression)\n2. Current active objective\n3. Requirement checklist (Requirement | Status: completed/in_progress/pending/blocked/obsolete | Evidence)\n4. Active tasks\n5. Completed tasks\n6. Obsolete or superseded tasks\n7. Important context and constraints\n8. Files, code, and tool findings\n9. Open issues and next step\n\n",
);
content.push_str("## Messages To Compress\n\n");
for message in messages {
let role = match message.role {
bamboo_domain::Role::System => continue,
bamboo_domain::Role::User => "User",
bamboo_domain::Role::Assistant => match message.phase {
Some(MessagePhase::Commentary) => "Assistant Commentary",
Some(MessagePhase::FinalAnswer) => "Assistant Final",
None => "Assistant",
},
bamboo_domain::Role::Tool => "Tool Result",
};
content.push_str("### ");
content.push_str(role);
content.push('\n');
if let Some(tool_calls) = &message.tool_calls {
if !tool_calls.is_empty() {
let names = tool_calls
.iter()
.map(|call| call.function.name.as_str())
.collect::<Vec<_>>()
.join(", ");
content.push_str("Called tools: ");
content.push_str(&names);
content.push('\n');
}
}
if let Some(tool_call_id) = &message.tool_call_id {
content.push_str("Tool call id: ");
content.push_str(tool_call_id);
content.push('\n');
}
// Keep this compatibility renderer lossless. Production callers bound
// the fully rendered request through the hierarchical summarizer;
// clipping each message would discard tails without bounding the total.
content.push_str(&message.content);
content.push_str("\n\n");
}
content.push_str(
"Return only the summary text. Be explicit about what is active now versus what is already done or no longer relevant.",
);
content
}
#[cfg(test)]
mod tests {
use super::*;
use bamboo_domain::TokenBudgetUsage;
use bamboo_domain::{FunctionCall, TaskItem, TaskItemStatus, TaskList, ToolCall};
use chrono::Utc;
fn make_budget() -> TokenBudget {
TokenBudget {
max_context_tokens: 1000,
max_output_tokens: 100,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 50,
compression_target_percent: 20,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
}
}
fn make_session_with_pressure() -> Session {
let mut session = Session::new("compression-hysteresis", "gpt-4o-mini");
session.token_budget = Some(make_budget());
session.add_message(Message::system("system"));
for i in 0..3 {
session.add_message(Message::user(format!(
"User message {i}: {}",
"alpha beta gamma delta epsilon ".repeat(2)
)));
session.add_message(Message::assistant(
format!(
"Assistant message {i}: {}",
"work log decisions next steps ".repeat(2)
),
None,
));
}
session
}
#[test]
fn context_window_usage_percent_uses_context_window_denominator() {
assert_eq!(context_window_usage_percent(0, 0), 0.0);
assert_eq!(context_window_usage_percent(500, 1000), 50.0);
}
#[test]
fn estimate_context_compression_exposure_crosses_trigger_when_usage_is_high_enough() {
let mut session = make_session_with_pressure();
if let Some(budget) = session.token_budget.as_mut() {
budget.compression_trigger_percent = 10;
}
let exposure = estimate_context_compression_exposure(
&session,
"gpt-4o-mini",
session.token_budget.as_ref(),
);
assert!(exposure.active_usage_percent >= 10.0);
assert!(exposure.should_expose_tool);
}
#[test]
fn estimate_context_compression_exposure_stays_below_trigger_when_usage_is_low() {
let mut session = make_session_with_pressure();
if let Some(budget) = session.token_budget.as_mut() {
budget.compression_trigger_percent = 99;
}
let exposure = estimate_context_compression_exposure(
&session,
"gpt-4o-mini",
session.token_budget.as_ref(),
);
assert!(exposure.active_usage_percent < 99.0);
assert!(!exposure.should_expose_tool);
}
#[test]
fn build_summary_prompt_includes_task_list_and_state_sections() {
let mut session = Session::new("summary-prompt", "gpt-4o-mini");
session.set_task_list(TaskList {
session_id: session.id.clone(),
title: "Task List".to_string(),
items: vec![
TaskItem {
id: "task_1".to_string(),
description: "检查 51% 又回落到 50% 的触发逻辑".to_string(),
status: TaskItemStatus::InProgress,
depends_on: Vec::new(),
notes: "避免刚压缩完又立刻再次压缩".to_string(),
..TaskItem::default()
},
TaskItem {
id: "task_2".to_string(),
description: "重写 summarizer prompt 并纳入 task list".to_string(),
status: TaskItemStatus::Pending,
depends_on: Vec::new(),
notes: String::new(),
..TaskItem::default()
},
],
created_at: Utc::now(),
updated_at: Utc::now(),
});
let prompt = build_summary_prompt(
&session,
&[
Message::user("继续修复 context compression"),
Message::assistant("先分析 trigger / target / summary", None),
],
Some("old summary"),
);
assert!(prompt.contains("## Current Task List"));
assert!(prompt.contains("Current active objective"));
assert!(prompt.contains("Requirement checklist"));
assert!(prompt.contains("Active tasks"));
assert!(prompt.contains("Completed tasks"));
assert!(prompt.contains("Obsolete or superseded tasks"));
assert!(prompt.contains("检查 51% 又回落到 50% 的触发逻辑"));
assert!(prompt.contains("old summary"));
}
#[test]
fn forced_plan_keeps_last_three_user_messages_active() {
let budget = TokenBudget {
max_context_tokens: 1200,
max_output_tokens: 100,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: 20,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
};
let mut session = Session::new("keep-last-three-user-turns", "gpt-4o-mini");
session.token_budget = Some(budget.clone());
session.add_message(Message::system("system"));
for i in 0..6 {
session.add_message(Message::user(format!(
"U{i}: {}",
"alpha beta gamma ".repeat(8)
)));
session.add_message(Message::assistant(
format!("A{i}: {}", "analysis plan steps ".repeat(8)),
None,
));
}
let plan = build_forced_compression_plan_with_summary(
&session,
"gpt-4o-mini",
Some(&budget),
"summary".to_string(),
CompressionTriggerType::CriticalOverflow,
)
.expect("forced plan should build");
let compressed_ids = plan
.compressed_message_ids
.iter()
.map(String::as_str)
.collect::<HashSet<_>>();
let kept_user_contents = session
.messages
.iter()
.filter(|message| !matches!(message.role, bamboo_domain::Role::System))
.filter(|message| !compressed_ids.contains(message.id.as_str()))
.filter(|message| matches!(message.role, bamboo_domain::Role::User))
.map(|message| message.content.clone())
.collect::<Vec<_>>();
assert!(
kept_user_contents.len() >= 3,
"expected to keep at least 3 user messages, got {}",
kept_user_contents.len()
);
assert!(kept_user_contents
.iter()
.any(|content| content.starts_with("U3:")));
assert!(kept_user_contents
.iter()
.any(|content| content.starts_with("U4:")));
assert!(kept_user_contents
.iter()
.any(|content| content.starts_with("U5:")));
}
#[test]
fn estimate_exposure_prefers_persisted_budget_usage_when_higher() {
let mut session = Session::new("persisted-usage", "gpt-4o-mini");
session.token_budget = Some(TokenBudget {
max_context_tokens: 100_000,
max_output_tokens: 1_000,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: 50,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
});
session.add_message(Message::system("system"));
session.add_message(Message::user("short"));
session.add_message(Message::assistant("short", None));
session.add_message(Message::user("follow-up"));
session.add_message(Message::assistant("reply", None));
session.token_usage = Some(TokenBudgetUsage {
system_tokens: 100,
summary_tokens: 0,
window_tokens: 95_900,
total_tokens: 96_000,
max_context_tokens: 100_000,
budget_limit: 10_000,
truncation_occurred: true,
segments_removed: 12,
prompt_cached_tool_outputs: 0,
prompt_cached_tool_tokens_saved: 0,
thinking_tokens: 0,
cache_read_input_tokens: 0,
});
let exposure = estimate_context_compression_exposure(
&session,
"gpt-4o-mini",
session.token_budget.as_ref(),
);
assert!(
exposure.active_usage_percent >= 96.0,
"expected persisted context-window usage to drive exposure, got {}",
exposure.active_usage_percent
);
assert!(exposure.should_expose_tool);
}
#[test]
fn never_compress_messages_are_excluded_from_summarize_set() {
let budget = TokenBudget {
max_context_tokens: 1200,
max_output_tokens: 100,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: 20,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
};
let mut session = Session::new("never-compress-test", "gpt-4o-mini");
session.token_budget = Some(budget.clone());
session.add_message(Message::system("system"));
// Old user message that should be summarized
session.add_message(Message::user("Old question about X"));
session.add_message(Message::assistant("Old answer about X", None));
// Protected user message (never_compress = true)
let mut protected = Message::user("Critical context that must survive");
protected.never_compress = true;
session.add_message(protected);
session.add_message(Message::assistant("Response to critical", None));
// Recent user messages that anchor the keep window
for i in 0..4 {
session.add_message(Message::user(format!(
"Recent U{i}: {}",
"padding text to fill budget ".repeat(6)
)));
session.add_message(Message::assistant(
format!("Recent A{i}: {}", "reply padding text ".repeat(6)),
None,
));
}
let plan = build_forced_compression_plan_with_summary(
&session,
"gpt-4o-mini",
Some(&budget),
"summary".to_string(),
CompressionTriggerType::Auto,
)
.expect("plan should build");
let compressed_ids: HashSet<&str> = plan
.compressed_message_ids
.iter()
.map(String::as_str)
.collect();
// Find the never_compress message
let protected_msg = session
.messages
.iter()
.find(|m| m.never_compress)
.expect("should find the protected message");
assert!(
!compressed_ids.contains(protected_msg.id.as_str()),
"never_compress message should NOT be in the compressed set"
);
}
#[test]
fn skill_tool_chain_messages_are_protected_from_compression() {
let budget = TokenBudget {
max_context_tokens: 1200,
max_output_tokens: 100,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: 20,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
};
let mut session = Session::new("skill-chain-test", "gpt-4o-mini");
session.token_budget = Some(budget.clone());
session.add_message(Message::system("system"));
// Skill tool chain (load_skill + read_skill_resource)
let mut skill_call = Message::assistant(String::new(), None);
skill_call.tool_calls = Some(vec![ToolCall {
id: "tc-skill".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "load_skill".to_string(),
arguments: r#"{"skill_id":"my-skill"}"#.to_string(),
},
}]);
session.add_message(skill_call);
let mut skill_result = Message::tool_result("tc-skill", "skill loaded");
skill_result.tool_success = Some(true);
session.add_message(skill_result);
// Regular messages to fill budget
for i in 0..6 {
session.add_message(Message::user(format!(
"U{i}: {}",
"alpha beta gamma delta ".repeat(8)
)));
session.add_message(Message::assistant(
format!("A{i}: {}", "analysis steps plan ".repeat(8)),
None,
));
}
let plan = build_forced_compression_plan_with_summary(
&session,
"gpt-4o-mini",
Some(&budget),
"summary".to_string(),
CompressionTriggerType::Auto,
)
.expect("plan should build");
let compressed_ids: HashSet<&str> = plan
.compressed_message_ids
.iter()
.map(String::as_str)
.collect();
// Skill tool chain messages should not be compressed
let skill_messages: Vec<&Message> = session
.messages
.iter()
.filter(|m| {
m.tool_calls
.as_ref()
.is_some_and(|calls| calls.iter().any(|c| c.function.name == "load_skill"))
|| m.tool_call_id.as_deref() == Some("tc-skill")
})
.collect();
for msg in &skill_messages {
assert!(
!compressed_ids.contains(msg.id.as_str()),
"skill tool chain message {} should NOT be compressed",
msg.id
);
}
}
#[test]
fn generic_tool_chain_is_never_split_by_forced_compression() {
// A generic (non-skill) tool_use and its tool_result must never be split
// across the compression boundary — a split orphans one of them in the
// active set and the provider 400s, poisoning the session. #340.
let budget = TokenBudget {
max_context_tokens: 1200,
max_output_tokens: 100,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: 20,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
};
let mut session = Session::new("generic-chain-test", "gpt-4o-mini");
session.token_budget = Some(budget.clone());
session.add_message(Message::system("system"));
// A generic tool chain (search + its result) placed early so it is an
// eviction candidate; the large result makes it a prime compression target.
let mut call = Message::assistant(String::new(), None);
call.tool_calls = Some(vec![ToolCall {
id: "tc-gen".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "search".to_string(),
arguments: r#"{"q":"rust"}"#.to_string(),
},
}]);
session.add_message(call);
let mut result = Message::tool_result("tc-gen", &"search result payload ".repeat(20));
result.tool_success = Some(true);
session.add_message(result);
// Filler to push usage over budget and force eviction.
for i in 0..8 {
session.add_message(Message::user(format!(
"U{i}: {}",
"alpha beta gamma delta ".repeat(8)
)));
session.add_message(Message::assistant(
format!("A{i}: {}", "analysis steps plan ".repeat(8)),
None,
));
}
let plan = build_forced_compression_plan_with_summary(
&session,
"gpt-4o-mini",
Some(&budget),
"summary".to_string(),
CompressionTriggerType::Auto,
)
.expect("plan should build");
let compressed: HashSet<&str> = plan
.compressed_message_ids
.iter()
.map(String::as_str)
.collect();
let assistant_id = session
.messages
.iter()
.find(|m| {
m.tool_calls
.as_ref()
.is_some_and(|c| c.iter().any(|tc| tc.id == "tc-gen"))
})
.map(|m| m.id.as_str())
.expect("assistant tool_use message present");
let result_id = session
.messages
.iter()
.find(|m| m.tool_call_id.as_deref() == Some("tc-gen"))
.map(|m| m.id.as_str())
.expect("tool_result message present");
// The tool_use and its tool_result must land on the SAME side.
assert_eq!(
compressed.contains(assistant_id),
compressed.contains(result_id),
"generic tool_use ({assistant_id}) and its tool_result ({result_id}) must not be split"
);
}
fn tool_use_message(call_id: &str) -> Message {
let mut assistant = Message::assistant(String::new(), None);
assistant.tool_calls = Some(vec![ToolCall {
id: call_id.to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "search".to_string(),
arguments: "{}".to_string(),
},
}]);
assistant
}
#[test]
fn close_compressed_set_over_tool_chains_reunites_a_split_chain() {
// Pre-split state: the assistant tool_use is compressed while its
// tool_result was left in the keep (active) set — an orphan. #340.
let assistant = tool_use_message("tc-1");
let result = Message::tool_result("tc-1", "result payload");
let assistant_id = assistant.id.clone();
let result_id = result.id.clone();
let mut messages_to_keep = vec![result];
let mut messages_to_summarize = vec![assistant];
close_compressed_set_over_tool_chains(
&mut messages_to_keep,
&mut messages_to_summarize,
&HashSet::new(),
&[],
);
// The orphan tool_result must be pulled into the compressed set.
assert!(
messages_to_keep.is_empty(),
"orphaned tool_result must be moved into the compressed set"
);
let summarized: HashSet<&str> = messages_to_summarize
.iter()
.map(|m| m.id.as_str())
.collect();
assert!(summarized.contains(assistant_id.as_str()));
assert!(summarized.contains(result_id.as_str()));
}
#[test]
fn close_compressed_set_over_tool_chains_respects_protected_messages() {
// A protected (never-compress) chain member must NOT be force-compressed.
let assistant = tool_use_message("tc-2");
let result = Message::tool_result("tc-2", "result");
let result_id = result.id.clone();
let mut messages_to_keep = vec![result];
let mut messages_to_summarize = vec![assistant];
let never_compress_ids = vec![result_id.clone()];
close_compressed_set_over_tool_chains(
&mut messages_to_keep,
&mut messages_to_summarize,
&HashSet::new(),
&never_compress_ids,
);
assert_eq!(
messages_to_keep.len(),
1,
"protected result must stay in keep"
);
assert_eq!(messages_to_keep[0].id, result_id);
}
#[test]
fn recovery_message_returns_none_for_empty_messages() {
let session = Session::new("recovery-empty", "model");
let result = build_post_compaction_recovery_message(&[], &session);
assert!(result.is_none());
}
#[test]
fn recovery_message_has_never_compress_flag() {
let mut session = Session::new("recovery-flag", "model");
let messages = vec![Message::assistant("no decisions here", None)];
session.set_task_list(TaskList {
session_id: session.id.clone(),
title: "Tasks".to_string(),
items: vec![TaskItem {
id: "t1".to_string(),
description: "Active task".to_string(),
status: TaskItemStatus::InProgress,
..TaskItem::default()
}],
created_at: Utc::now(),
updated_at: Utc::now(),
});
let recovery = build_post_compaction_recovery_message(&messages, &session)
.expect("should return recovery message");
assert!(recovery.never_compress);
assert!(recovery.content.contains("[post-compaction-recovery]"));
}
#[test]
fn recovery_message_extracts_file_paths_from_tool_calls() {
let session = Session::new("recovery-files", "model");
let mut write_call = Message::assistant("writing file", None);
write_call.tool_calls = Some(vec![ToolCall {
id: "tc1".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "Write".to_string(),
arguments: r#"{"file_path":"/src/main.rs","content":"fn main() {}"}"#.to_string(),
},
}]);
let mut edit_call = Message::assistant("editing file", None);
edit_call.tool_calls = Some(vec![ToolCall {
id: "tc2".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "Edit".to_string(),
arguments: r#"{"file_path":"/lib/utils.rs","old":"x","new":"y"}"#.to_string(),
},
}]);
let messages = vec![write_call, edit_call];
let recovery = build_post_compaction_recovery_message(&messages, &session)
.expect("should return recovery");
assert!(recovery.content.contains("/src/main.rs"));
assert!(recovery.content.contains("/lib/utils.rs"));
assert!(recovery.content.contains("Recently Modified Files"));
}
#[test]
fn recovery_message_includes_active_tasks() {
let mut session = Session::new("recovery-tasks", "model");
session.set_task_list(TaskList {
session_id: session.id.clone(),
title: "Tasks".to_string(),
items: vec![
TaskItem {
id: "t1".to_string(),
description: "Fix auth middleware".to_string(),
status: TaskItemStatus::InProgress,
..TaskItem::default()
},
TaskItem {
id: "t2".to_string(),
description: "Add tests".to_string(),
status: TaskItemStatus::Pending,
..TaskItem::default()
},
TaskItem {
id: "t3".to_string(),
description: "Done task".to_string(),
status: TaskItemStatus::Completed,
..TaskItem::default()
},
],
created_at: Utc::now(),
updated_at: Utc::now(),
});
let messages = vec![Message::assistant("some work", None)];
let recovery = build_post_compaction_recovery_message(&messages, &session)
.expect("should return recovery");
assert!(recovery.content.contains("Active Tasks"));
assert!(recovery.content.contains("Fix auth middleware"));
assert!(recovery.content.contains("Add tests"));
// Completed tasks should NOT appear in active tasks
assert!(!recovery.content.contains("Done task"));
}
#[test]
fn apply_compression_plan_injects_recovery_message() {
let budget = TokenBudget {
max_context_tokens: 1200,
max_output_tokens: 100,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: 20,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
};
let mut session = Session::new("recovery-inject", "gpt-4o-mini");
session.token_budget = Some(budget.clone());
session.add_message(Message::system("system"));
// Old messages with tool calls containing file paths
let mut write_msg = Message::assistant("writing", None);
write_msg.tool_calls = Some(vec![ToolCall {
id: "tc-w".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "Write".to_string(),
arguments: r#"{"file_path":"/src/lib.rs","content":"pub fn hello() {}"}"#
.to_string(),
},
}]);
session.add_message(Message::user("Write the file"));
session.add_message(write_msg);
// Fill with enough messages to force compression
for i in 0..6 {
session.add_message(Message::user(format!(
"U{i}: {}",
"alpha beta gamma delta ".repeat(8)
)));
session.add_message(Message::assistant(
format!("A{i}: {}", "analysis plan ".repeat(8)),
None,
));
}
let plan = build_forced_compression_plan_with_summary(
&session,
"gpt-4o-mini",
Some(&budget),
"summary text".to_string(),
CompressionTriggerType::Auto,
)
.expect("plan should build");
assert!(!plan.compressed_message_ids.is_empty());
let compressed_count = apply_compression_plan(&mut session, plan);
assert!(compressed_count > 0);
// Verify recovery message was injected
let has_recovery = session.messages.iter().any(|m| {
m.never_compress
&& m.content.contains("[post-compaction-recovery]")
&& m.content.contains("/src/lib.rs")
});
assert!(
has_recovery,
"session should contain a post-compaction recovery message with the file path"
);
}
#[test]
fn summary_quality_full_coverage_when_all_files_mentioned() {
let messages = vec![{
let mut m = Message::assistant("writing", None);
m.tool_calls = Some(vec![ToolCall {
id: "tc1".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "Write".to_string(),
arguments: r#"{"file_path":"/src/main.rs","content":"fn main() {}"}"#
.to_string(),
},
}]);
m
}];
let summary = "Modified /src/main.rs to add main function";
let quality = validate_summary_quality(summary, &messages);
assert!(
quality.file_coverage >= 0.99,
"file_coverage should be ~1.0, got {:.2}",
quality.file_coverage
);
}
#[test]
fn summary_quality_zero_coverage_when_no_files_mentioned() {
let messages = vec![{
let mut m = Message::assistant("writing", None);
m.tool_calls = Some(vec![ToolCall {
id: "tc1".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "Write".to_string(),
arguments: r#"{"file_path":"/src/main.rs","content":"fn main() {}"}"#
.to_string(),
},
}]);
m
}];
let summary = "Summary that mentions nothing about files";
let quality = validate_summary_quality(summary, &messages);
assert!(
quality.file_coverage < 0.01,
"file_coverage should be ~0.0, got {:.2}",
quality.file_coverage
);
}
#[test]
fn summary_quality_handles_empty_messages() {
let quality = validate_summary_quality("some summary", &[]);
assert_eq!(quality.file_coverage, 1.0);
assert_eq!(quality.decision_coverage, 1.0);
}
fn candidate_budget(max_context_tokens: u32, target_percent: u8) -> TokenBudget {
TokenBudget {
max_context_tokens,
max_output_tokens: max_context_tokens / 4,
strategy: BudgetStrategy::Hybrid {
window_size: 20,
enable_summarization: true,
},
safety_margin: 0,
compression_trigger_percent: 80,
compression_target_percent: target_percent,
working_reserve_tokens: 0,
fallback_trigger_percent: 75,
prompt_cache_min_tool_output_chars: 1_200,
prompt_cache_head_chars: 280,
prompt_cache_tail_chars: 180,
prompt_cache_recent_user_turns: 2,
prompt_cache_recent_tool_chains: 2,
max_tool_output_tokens: 0,
}
}
#[test]
fn candidate_plan_is_selected_before_summary_and_reserves_twenty_percent_of_source() {
let budget = candidate_budget(6_000, 40);
let mut session = Session::new("candidate-first", "main-model");
session.add_message(Message::system("system"));
let mut tool_call = Message::assistant("searching", None);
tool_call.tool_calls = Some(vec![ToolCall {
id: "candidate-chain".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "search".to_string(),
arguments: r#"{"q":"compression"}"#.to_string(),
},
}]);
let tool_call_id = tool_call.id.clone();
session.add_message(tool_call);
let tool_result = Message::tool_result(
"candidate-chain",
"result payload with decisions and paths ".repeat(80),
);
let tool_result_id = tool_result.id.clone();
session.add_message(tool_result);
let mut never = Message::assistant("protected runtime state ".repeat(80), None);
never.never_compress = true;
let never_id = never.id.clone();
session.add_message(never);
for index in 0..12 {
session.add_message(Message::user(format!(
"U{index}: {}",
"requirement context decision ".repeat(40)
)));
session.add_message(Message::assistant(
format!("A{index}: {}", "implementation evidence result ".repeat(40)),
None,
));
}
let newest_user_ids = session
.messages
.iter()
.filter(|message| matches!(message.role, bamboo_domain::Role::User))
.rev()
.take(3)
.map(|message| message.id.clone())
.collect::<HashSet<_>>();
let candidate = build_forced_compression_candidate_plan(
&session,
"main-model",
Some(&budget),
0.20,
CompressionTriggerType::Auto,
)
.expect("candidate plan should reach target");
let selected = candidate
.compressed_message_ids
.iter()
.cloned()
.collect::<HashSet<_>>();
assert_eq!(
selected,
candidate
.messages_to_summarize
.iter()
.map(|message| message.id.clone())
.collect()
);
assert_eq!(
candidate.target_summary_tokens,
((candidate.source_tokens as f64) * 0.20).ceil() as u32
);
assert!(!selected.contains(&never_id));
assert!(newest_user_ids.is_disjoint(&selected));
assert_eq!(
selected.contains(&tool_call_id),
selected.contains(&tool_result_id),
"generic tool chain must be selected atomically"
);
assert!(
candidate.projected_usage_after_percent <= budget.compression_target_percent as f64
);
}
#[test]
fn cumulative_twenty_percent_uses_represented_raw_tokens_not_previous_summary_length() {
let budget = candidate_budget(30_000, 50);
let mut session = Session::new("cumulative-ratio", "main-model");
session.add_message(Message::system("system"));
session.conversation_summary = Some(
ConversationSummary::new("existing detailed summary ".repeat(200), 40, 2_000)
.with_compression_metrics(10_000, 2_000, 0.20, false, None),
);
for index in 0..80 {
session.add_message(Message::user(format!(
"U{index}: {}",
"raw source requirement and evidence ".repeat(30)
)));
session.add_message(Message::assistant(
format!(
"A{index}: {}",
"implementation result and next step ".repeat(30)
),
None,
));
}
let candidate = build_forced_compression_candidate_plan(
&session,
"main-model",
Some(&budget),
0.20,
CompressionTriggerType::Auto,
)
.expect("cumulative candidate plan");
assert_eq!(
candidate.target_summary_tokens,
((10_000u32.saturating_add(candidate.source_tokens) as f64) * 0.20).ceil() as u32
);
assert_eq!(
candidate.represented_source_tokens,
10_000u32.saturating_add(candidate.source_tokens)
);
}
#[test]
fn legacy_summary_uses_conservative_growth_and_migrates_represented_source_metadata() {
let budget = candidate_budget(20_000, 50);
let counter = TiktokenTokenCounter::default();
let existing_content = "legacy summary fact and decision ".repeat(180);
let existing_tokens = counter.count_text(&existing_content);
let mut session = Session::new("legacy-summary-ratio", "main-model");
session.add_message(Message::system("system"));
session.conversation_summary = Some(ConversationSummary::new(
existing_content,
40,
existing_tokens,
));
for index in 0..40 {
session.add_message(Message::user(format!(
"U{index}: {}",
"raw requirement evidence ".repeat(40)
)));
session.add_message(Message::assistant(
format!("A{index}: {}", "result next step ".repeat(40)),
None,
));
}
let candidate = build_forced_compression_candidate_plan(
&session,
"main-model",
Some(&budget),
0.20,
CompressionTriggerType::Auto,
)
.expect("legacy candidate plan");
assert_eq!(
candidate.target_summary_tokens,
existing_tokens.saturating_add(((candidate.source_tokens as f64) * 0.20).ceil() as u32)
);
assert_eq!(
candidate.previous_represented_source_tokens,
((existing_tokens as f64) / 0.20).ceil() as u32
);
let expected_represented = candidate.represented_source_tokens;
let expected_target = candidate.target_summary_tokens;
let plan = finalize_compression_candidate_plan(
&session,
candidate,
"migrated legacy summary with new evidence".to_string(),
)
.expect("short real summary should satisfy target");
assert!(apply_compression_plan(&mut session, plan) > 0);
let migrated = session
.conversation_summary
.as_ref()
.expect("migrated summary");
assert_eq!(migrated.represented_source_tokens, expected_represented);
assert_eq!(migrated.target_token_count, expected_target);
assert_eq!(migrated.target_ratio, 0.20);
}
#[test]
fn final_postcondition_counts_the_recovery_message_inserted_during_apply() {
let budget = candidate_budget(8_000, 40);
let mut session = Session::new("recovery-postcondition", "main-model");
session.add_message(Message::system("system"));
for index in 0..20 {
session.add_message(Message::user(format!(
"U{index}: {}",
"requirement source content ".repeat(45)
)));
let mut assistant = Message::assistant(
format!("A{index}: {}", "implementation evidence ".repeat(45)),
None,
);
if index == 0 {
assistant.tool_calls = Some(vec![ToolCall {
id: "write-recovery-763".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "Write".to_string(),
arguments:
r#"{"file_path":"/workspace/src/recovery_763.rs","content":"fixed"}"#
.to_string(),
},
}]);
}
session.add_message(assistant);
if index == 0 {
session.add_message(Message::tool_result(
"write-recovery-763",
"write completed",
));
}
}
let candidate = build_forced_compression_candidate_plan(
&session,
"main-model",
Some(&budget),
0.20,
CompressionTriggerType::CriticalOverflow,
)
.expect("candidate should reserve recovery-message tokens");
let plan = finalize_compression_candidate_plan(
&session,
candidate,
"summary with /workspace/src/recovery_763.rs".to_string(),
)
.expect("final plan");
assert!(apply_compression_plan(&mut session, plan) > 0);
assert!(session
.messages
.iter()
.any(|message| message.content.contains("[post-compaction-recovery]")));
assert!(
session.token_usage.as_ref().is_some_and(
|usage| usage.total_tokens <= budget.compression_target_context_tokens()
),
"the real active context, including recovery, must remain at or below target"
);
}
#[test]
fn finalization_revalidates_actual_summary_without_mutating_session() {
let budget = candidate_budget(6_000, 40);
let mut session = Session::new("atomic-finalize", "main-model");
session.add_message(Message::system("system"));
for index in 0..20 {
session.add_message(Message::user(format!(
"U{index}: {}",
"source content ".repeat(50)
)));
session.add_message(Message::assistant(
format!("A{index}: {}", "response content ".repeat(50)),
None,
));
}
let candidate = build_forced_compression_candidate_plan(
&session,
"main-model",
Some(&budget),
0.20,
CompressionTriggerType::CriticalOverflow,
)
.expect("candidate plan");
let before_flags = session
.messages
.iter()
.map(|message| (message.id.clone(), message.compressed))
.collect::<Vec<_>>();
let result = finalize_compression_candidate_plan(
&session,
candidate,
"oversized summary ".repeat(20_000),
);
assert!(matches!(
result,
Err(CompressionPlanError::SummaryExceedsTarget { .. })
));
assert_eq!(
before_flags,
session
.messages
.iter()
.map(|message| (message.id.clone(), message.compressed))
.collect::<Vec<_>>()
);
assert!(session.conversation_summary.is_none());
assert!(session.compression_events.is_empty());
}
#[test]
fn candidate_plan_reports_when_protected_content_makes_target_impossible() {
let budget = candidate_budget(2_000, 20);
let mut session = Session::new("protected-capacity", "main-model");
session.add_message(Message::system("system"));
session.add_message(Message::assistant(
"one eligible old message ".repeat(200),
None,
));
for index in 0..3 {
session.add_message(Message::user(format!(
"protected recent user {index} {}",
"large active content ".repeat(300)
)));
}
let result = build_forced_compression_candidate_plan(
&session,
"main-model",
Some(&budget),
0.20,
CompressionTriggerType::CriticalOverflow,
);
assert!(matches!(
result,
Err(CompressionPlanError::ProtectedContentExceedsTarget { .. })
));
}
}