use async_trait::async_trait;
use harness_core::{
Block, Budget, CompactError, CompactionStage, Compactor, Context, Model, Policy, Task, Turn,
TurnRole,
};
use std::sync::Arc;
pub const CALIBRATION_KEY: &str = "compactor.correction";
fn calibration(ctx: &Context) -> f64 {
ctx.metadata
.get(CALIBRATION_KEY)
.and_then(|v| v.as_f64())
.filter(|f| f.is_finite() && *f > 0.0)
.unwrap_or(1.0)
}
pub struct DefaultCompactor {
pub tokens_per_char: f32,
}
impl Default for DefaultCompactor {
fn default() -> Self {
Self {
tokens_per_char: 0.30,
}
}
}
impl DefaultCompactor {
pub fn new() -> Self {
Self::default()
}
fn estimate_tokens(&self, ctx: &Context) -> u32 {
let mut chars: usize = 0;
for b in ctx.system.iter().chain(ctx.guides.iter()) {
chars += block_chars(b);
}
for turn in &ctx.history {
for b in &turn.blocks {
chars += block_chars(b);
}
}
chars += ctx.task.description.len();
(chars as f32 * self.tokens_per_char) as u32
}
}
#[async_trait]
impl Compactor for DefaultCompactor {
fn budget(&self, ctx: &Context) -> Budget {
let raw = self.estimate_tokens(ctx);
let used = (raw as f64 * calibration(ctx)).round() as u32;
Budget {
used,
window: ctx.policy.max_input_tokens,
}
}
async fn compact(&self, stage: CompactionStage, ctx: &mut Context) -> Result<(), CompactError> {
tracing::debug!(?stage, "compaction stage running");
match stage {
CompactionStage::BudgetReduce => budget_reduce(ctx),
CompactionStage::Snip => snip_file_reads(ctx),
CompactionStage::Microcompact => microcompact_old(ctx),
CompactionStage::ContextCollapse => context_collapse(ctx),
CompactionStage::AutoCompact => auto_compact(ctx),
_ => tracing::warn!(?stage, "unknown compaction stage — ignoring"),
}
Ok(())
}
}
pub struct ModelBackedCompactor {
pub model: Arc<dyn Model>,
pub tokens_per_char: f32,
pub keep_recent: usize,
pub summary_max_tokens: u32,
}
fn safe_split(history: &[Turn], want: usize) -> Option<usize> {
if want == 0 || want >= history.len() {
return None;
}
(0..=want)
.rev()
.chain(want + 1..history.len())
.filter(|&i| i > 0)
.find(|&i| starts_an_exchange(&history[i]))
}
fn starts_an_exchange(turn: &Turn) -> bool {
turn.role != TurnRole::Tool
}
impl ModelBackedCompactor {
pub fn new(model: Arc<dyn Model>) -> Self {
Self {
model,
tokens_per_char: 0.30,
keep_recent: 6,
summary_max_tokens: 600,
}
}
}
#[async_trait]
impl Compactor for ModelBackedCompactor {
fn budget(&self, ctx: &Context) -> Budget {
DefaultCompactor {
tokens_per_char: self.tokens_per_char,
}
.budget(ctx)
}
async fn compact(&self, stage: CompactionStage, ctx: &mut Context) -> Result<(), CompactError> {
match stage {
CompactionStage::BudgetReduce => {
budget_reduce(ctx);
Ok(())
}
CompactionStage::Snip => {
snip_file_reads(ctx);
Ok(())
}
CompactionStage::ContextCollapse => {
context_collapse(ctx);
Ok(())
}
CompactionStage::Microcompact => {
self.model_summarise(ctx, "microcompact-summary").await
}
CompactionStage::AutoCompact => self.model_summarise(ctx, "auto-compact-summary").await,
_ => Ok(()),
}
}
}
impl ModelBackedCompactor {
async fn model_summarise(&self, ctx: &mut Context, tag: &str) -> Result<(), CompactError> {
if ctx.history.len() <= self.keep_recent {
return Ok(());
}
let Some(split) = safe_split(&ctx.history, ctx.history.len() - self.keep_recent) else {
return Ok(());
};
let mut dump = String::new();
for turn in ctx.history.iter().take(split) {
dump.push_str(&format_turn_for_summary(turn));
}
if dump.trim().is_empty() {
return Ok(());
}
let prompt = format!(
"You are compacting an in-progress agent conversation for downstream replay. \
Produce a terse summary (≤ 200 words) of the conversation below. Preserve: \
concrete file paths, decisions made, sensor outcomes, and the current goal. \
Drop: chit-chat, redundant tool reads, verbose stack traces.\n\n\
---- TRANSCRIPT ----\n{dump}\n---- END ----\n\n\
Reply with the summary text only, no preamble."
);
let mut summary_ctx = Context::new(Task {
description: prompt,
source: None,
deadline: None,
});
summary_ctx.policy = Policy {
max_iters: 1,
max_input_tokens: 100_000,
max_output_tokens: self.summary_max_tokens,
self_correct_rounds: 0,
};
summary_ctx.history.push(Turn {
role: TurnRole::User,
blocks: vec![Block::Text(summary_ctx.task.description.clone())],
});
let out = self
.model
.complete(&summary_ctx)
.await
.map_err(|e| CompactError::Failed {
stage: tag.into(),
reason: format!("model: {e}"),
})?;
let summary = out.text.unwrap_or_else(|| "(empty summary)".into());
let mut new_history = vec![Turn {
role: TurnRole::User,
blocks: vec![Block::Text(format!(
"{}{summary}",
with_task(ctx, &format!("[{tag}]\n"))
))],
}];
new_history.extend(ctx.history.drain(split..));
ctx.history = new_history;
Ok(())
}
}
fn format_turn_for_summary(turn: &Turn) -> String {
let role = match turn.role {
TurnRole::User => "user",
TurnRole::Assistant => "assistant",
TurnRole::Tool => "tool",
TurnRole::System => "system",
_ => "unknown",
};
let mut s = format!("[{role}]\n");
for b in &turn.blocks {
match b {
Block::Text(t) => {
s.push_str(t);
s.push('\n');
}
Block::ToolCall { name, args, .. } => {
s.push_str(&format!("(tool-call {name} {args})\n"));
}
Block::ToolResult { call_id, content } => {
let preview = content.to_string();
let preview = preview.chars().take(160).collect::<String>();
s.push_str(&format!("(tool-result {call_id}: {preview}…)\n"));
}
Block::FileRef { path, .. } => {
s.push_str(&format!("(file-ref {path})\n"));
}
_ => {}
}
}
s.push('\n');
s
}
fn block_chars(b: &Block) -> usize {
match b {
Block::Text(s) => s.len(),
Block::FileRef {
path,
hash: _,
excerpt,
} => path.len() + excerpt.as_ref().map_or(0, String::len),
Block::Skill { name, body } => name.len() + body.len(),
Block::ToolCall {
call_id,
name,
args,
} => call_id.len() + name.len() + args.to_string().len(),
Block::ToolResult { call_id, content } => call_id.len() + content.to_string().len(),
Block::Feedback(signals) => signals
.iter()
.map(|s| s.message.len() + s.agent_hint.as_ref().map_or(0, String::len))
.sum(),
Block::Reasoning(s) => s.len(),
Block::Image { base64, .. } => base64.len(),
_ => 0,
}
}
fn budget_reduce(ctx: &mut Context) {
const OLD_TRIM_BYTES: usize = 800;
const RECENT_TRIM_BYTES: usize = 16 * 1024;
let keep_recent = 8;
let split = ctx.history.len().saturating_sub(keep_recent);
for (i, turn) in ctx.history.iter_mut().enumerate() {
let limit = if i < split {
OLD_TRIM_BYTES
} else {
RECENT_TRIM_BYTES
};
for b in turn.blocks.iter_mut() {
if let Block::ToolResult { call_id, content } = b {
let body = content.to_string();
if body.len() > limit {
let keep = if i < split { 200 } else { 2_000 };
let preview = body.chars().take(keep).collect::<String>();
*b = Block::Text(format!(
"[tool-result:{call_id} (trimmed from {} bytes)] {preview}…",
body.len()
));
}
}
}
}
}
fn snip_file_reads(ctx: &mut Context) {
let keep_recent = 4;
if ctx.history.len() <= keep_recent {
return;
}
let split = ctx.history.len() - keep_recent;
for turn in ctx.history.iter_mut().take(split) {
for b in turn.blocks.iter_mut() {
if let Block::FileRef {
path,
hash,
excerpt,
} = b
&& excerpt.is_some()
{
*b = Block::FileRef {
path: path.clone(),
hash: hash.clone(),
excerpt: None,
};
}
}
}
}
fn microcompact_old(ctx: &mut Context) {
if ctx.history.len() < 12 {
return;
}
let keep_recent = 6;
let Some(split) = safe_split(&ctx.history, ctx.history.len() - keep_recent) else {
return;
};
let mut summary = with_task(ctx, "[microcompact-summary]\n");
for turn in ctx.history.iter().take(split) {
let role = match turn.role {
TurnRole::User => "user",
TurnRole::Assistant => "assistant",
TurnRole::Tool => "tool",
TurnRole::System => "system",
_ => "unknown",
};
summary.push_str(&format!("- {role}: "));
for b in &turn.blocks {
match b {
Block::Text(t) => {
summary.push_str(&t.chars().take(80).collect::<String>());
summary.push(' ');
}
Block::ToolCall { name, .. } => summary.push_str(&format!("(call:{name}) ")),
Block::ToolResult { call_id, .. } => {
summary.push_str(&format!("(result:{call_id}) "))
}
Block::FileRef { path, .. } => summary.push_str(&format!("(file:{path}) ")),
_ => {}
}
}
summary.push('\n');
}
let mut new_history = vec![Turn {
role: TurnRole::User,
blocks: vec![Block::Text(summary)],
}];
new_history.extend(ctx.history.drain(split..));
ctx.history = new_history;
}
fn context_collapse(ctx: &mut Context) {
let mut files = std::collections::BTreeSet::new();
for turn in &ctx.history {
for b in &turn.blocks {
if let Block::FileRef { path, .. } = b {
files.insert(path.clone());
}
}
}
if files.is_empty() {
return;
}
let mut inv = String::from("[file-inventory]\n");
for f in &files {
inv.push_str(&format!("- {f}\n"));
}
for turn in ctx.history.iter_mut() {
turn.blocks.retain(|b| !matches!(b, Block::FileRef { .. }));
}
ctx.history.insert(
0,
Turn {
role: TurnRole::System,
blocks: vec![Block::Text(inv)],
},
);
}
fn with_task(ctx: &Context, header: &str) -> String {
let task = ctx.task.description.trim();
if task.is_empty() {
return header.to_string();
}
let (tag, rest) = header.split_once('\n').unwrap_or((header, ""));
format!("{tag}\nThe task (verbatim):\n{task}\n\n{rest}")
}
fn auto_compact(ctx: &mut Context) {
let keep_recent = 2;
if ctx.history.len() <= keep_recent {
return;
}
let Some(split) = safe_split(&ctx.history, ctx.history.len() - keep_recent) else {
return;
};
let mut combined = with_task(
ctx,
"[auto-compact-summary]\nCondensed history of earlier turns:\n",
);
let mut counts = std::collections::BTreeMap::new();
for turn in ctx.history.iter().take(split) {
for b in &turn.blocks {
let key = match b {
Block::Text(_) => "text",
Block::ToolCall { .. } => "tool_call",
Block::ToolResult { .. } => "tool_result",
Block::FileRef { .. } => "file_ref",
Block::Skill { .. } => "skill",
Block::Feedback(_) => "feedback",
Block::Reasoning(_) => "reasoning",
_ => "unknown",
};
*counts.entry(key).or_insert(0u32) += 1;
}
}
for (k, v) in counts {
combined.push_str(&format!("- {v} × {k} block(s)\n"));
}
let mut new_history = vec![Turn {
role: TurnRole::User,
blocks: vec![Block::Text(combined)],
}];
new_history.extend(ctx.history.drain(split..));
ctx.history = new_history;
}
#[cfg(test)]
mod tests {
use super::*;
use harness_core::{Block, Policy, Task, Turn, TurnRole};
use std::collections::BTreeMap;
fn tool_using_history(exchanges: usize) -> Context {
let mut ctx = mk_ctx(0);
for i in 0..exchanges {
ctx.history.push(Turn {
role: TurnRole::User,
blocks: vec![Block::Text(format!("do step {i}"))],
});
ctx.history.push(Turn {
role: TurnRole::Assistant,
blocks: vec![Block::ToolCall {
call_id: format!("c{i}"),
name: "write_file".into(),
args: serde_json::json!({ "path": format!("{i}.txt") }),
}],
});
ctx.history.push(Turn {
role: TurnRole::Tool,
blocks: vec![Block::ToolResult {
call_id: format!("c{i}"),
content: serde_json::json!({ "ok": true }),
}],
});
}
ctx
}
fn assert_history_is_legal(history: &[Turn], what: &str) {
if let Some(first) = history.first() {
assert!(
first.role != TurnRole::Tool,
"{what}: history opens with a tool result whose request was dropped"
);
}
for (i, turn) in history.iter().enumerate() {
if turn.role == TurnRole::Tool {
let prev_is_call = i > 0
&& history[i - 1]
.blocks
.iter()
.any(|b| matches!(b, Block::ToolCall { .. }));
assert!(
prev_is_call,
"{what}: tool result at {i} lost the call that made it"
);
}
}
}
#[test]
fn compaction_never_splits_a_tool_call_from_its_result() {
for exchanges in 4..14 {
for (name, stage) in [
("microcompact_old", microcompact_old as fn(&mut Context)),
("context_collapse", context_collapse as fn(&mut Context)),
("auto_compact", auto_compact as fn(&mut Context)),
] {
let mut ctx = tool_using_history(exchanges);
stage(&mut ctx);
assert_history_is_legal(&ctx.history, &format!("{name} @ {exchanges} exchanges"));
}
}
}
#[test]
fn a_split_with_nowhere_legal_to_land_is_declined() {
let history = vec![
Turn {
role: TurnRole::Assistant,
blocks: vec![Block::ToolCall {
call_id: "c0".into(),
name: "t".into(),
args: serde_json::json!({}),
}],
},
Turn {
role: TurnRole::Tool,
blocks: vec![Block::ToolResult {
call_id: "c0".into(),
content: serde_json::json!({}),
}],
},
];
assert_eq!(safe_split(&history, 1), None);
}
#[test]
fn a_bare_alternation_of_tool_calls_can_still_be_compacted() {
let mut ctx = mk_ctx(0);
ctx.history.push(Turn {
role: TurnRole::User,
blocks: vec![Block::Text("the task".into())],
});
for i in 0..10 {
ctx.history.push(Turn {
role: TurnRole::Assistant,
blocks: vec![Block::ToolCall {
call_id: format!("c{i}"),
name: "read_file".into(),
args: serde_json::json!({ "path": format!("{i}.txt") }),
}],
});
ctx.history.push(Turn {
role: TurnRole::Tool,
blocks: vec![Block::ToolResult {
call_id: format!("c{i}"),
content: serde_json::json!({ "ok": true }),
}],
});
}
let before = ctx.history.len();
microcompact_old(&mut ctx);
assert!(
ctx.history.len() < before,
"compaction did nothing on a bare tool alternation: {before} turns in, {} out",
ctx.history.len()
);
assert_history_is_legal(&ctx.history, "bare tool alternation");
}
#[test]
fn the_task_survives_every_rewriting_stage() {
let task = "Apply policy/fraud_policy.md to every transaction and write out/scored.csv";
for stage in ["micro", "auto"] {
let mut ctx = mk_ctx(0);
ctx.task.description = task.into();
ctx.history.push(Turn {
role: TurnRole::User,
blocks: vec![Block::Text(task.into())],
});
for i in 0..10 {
ctx.history.push(Turn {
role: TurnRole::Assistant,
blocks: vec![Block::Text(format!("step {i}"))],
});
}
match stage {
"micro" => microcompact_old(&mut ctx),
_ => auto_compact(&mut ctx),
}
let head = match &ctx.history[0].blocks[0] {
Block::Text(t) => t.clone(),
other => panic!("{stage}: head is {other:?}"),
};
assert!(head.contains(task), "{stage} dropped the task: {head}");
}
}
#[test]
fn a_run_with_one_user_turn_can_still_be_compacted() {
let mut ctx = mk_ctx(0);
ctx.history.push(Turn {
role: TurnRole::User,
blocks: vec![Block::Text("the task".into())],
});
for i in 0..8 {
ctx.history.push(Turn {
role: TurnRole::Assistant,
blocks: vec![Block::ToolCall {
call_id: format!("c{i}"),
name: "read_file".into(),
args: serde_json::json!({}),
}],
});
ctx.history.push(Turn {
role: TurnRole::Tool,
blocks: vec![Block::ToolResult {
call_id: format!("c{i}"),
content: serde_json::json!({ "ok": true }),
}],
});
ctx.history.push(Turn {
role: TurnRole::Assistant,
blocks: vec![Block::Text(format!("step {i} done"))],
});
}
let before = ctx.history.len();
microcompact_old(&mut ctx);
assert!(
ctx.history.len() < before,
"compaction did nothing: {before} turns in, {} out",
ctx.history.len()
);
assert_history_is_legal(&ctx.history, "single-user-turn run");
}
fn mk_ctx(turns: usize) -> Context {
let mut ctx = Context {
system: vec![],
guides: vec![],
history: Vec::new(),
task: Task {
description: "t".into(),
source: None,
deadline: None,
},
policy: Policy::default(),
metadata: BTreeMap::new(),
tools: Vec::new(),
response_format: harness_core::ResponseFormat::Free,
};
for i in 0..turns {
ctx.history.push(Turn {
role: if i % 2 == 0 {
TurnRole::User
} else {
TurnRole::Assistant
},
blocks: vec![Block::Text(format!("turn {i}: {}", "x".repeat(50)))],
});
}
ctx
}
#[tokio::test]
async fn budget_reduce_keeps_recent() {
let c = DefaultCompactor::new();
let mut ctx = mk_ctx(20);
ctx.history[0].blocks.push(Block::ToolResult {
call_id: "c1".into(),
content: serde_json::Value::String("y".repeat(2000)),
});
c.compact(CompactionStage::BudgetReduce, &mut ctx)
.await
.unwrap();
let has_trim = ctx.history[0]
.blocks
.iter()
.any(|b| matches!(b, Block::Text(t) if t.contains("trimmed")));
assert!(has_trim);
}
#[tokio::test]
async fn microcompact_collapses_old_turns() {
let c = DefaultCompactor::new();
let mut ctx = mk_ctx(20);
c.compact(CompactionStage::Microcompact, &mut ctx)
.await
.unwrap();
assert!(matches!(ctx.history[0].role, TurnRole::User));
let first_text = match &ctx.history[0].blocks[0] {
Block::Text(t) => t.clone(),
_ => String::new(),
};
assert!(first_text.starts_with("[microcompact-summary]"));
}
#[tokio::test]
async fn model_backed_compactor_replaces_old_turns_with_summary() {
use harness_models::{MockModel, MockResponse};
let model = Arc::new(MockModel::new().script(MockResponse::text("CONCISE-SUMMARY-OK")))
as Arc<dyn Model>;
let c = ModelBackedCompactor::new(model);
let mut ctx = mk_ctx(20);
let original_len = ctx.history.len();
c.compact(CompactionStage::Microcompact, &mut ctx)
.await
.unwrap();
assert_eq!(ctx.history.len(), c.keep_recent + 1);
assert!(original_len > ctx.history.len());
let first = match &ctx.history[0].blocks[0] {
Block::Text(t) => t.clone(),
_ => String::new(),
};
assert!(first.starts_with("[microcompact-summary]"));
assert!(first.contains("CONCISE-SUMMARY-OK"));
}
#[tokio::test]
async fn model_backed_compactor_noop_when_history_short() {
use harness_models::{MockModel, MockResponse};
let mock = Arc::new(MockModel::new().script(MockResponse::text("never called")));
let c = ModelBackedCompactor::new(mock.clone() as Arc<dyn Model>);
let mut ctx = mk_ctx(4); c.compact(CompactionStage::Microcompact, &mut ctx)
.await
.unwrap();
assert_eq!(ctx.history.len(), 4);
assert_eq!(
mock.call_count(),
0,
"model must not be called when history is short"
);
}
#[test]
fn budget_applies_calibration_factor() {
let c = DefaultCompactor::new();
let mut ctx = mk_ctx(10);
let raw = c.budget(&ctx).used;
assert!(raw > 0);
ctx.metadata
.insert(CALIBRATION_KEY.into(), serde_json::json!(2.0));
let calibrated = c.budget(&ctx).used;
assert_eq!(calibrated, (raw as f64 * 2.0).round() as u32);
ctx.metadata
.insert(CALIBRATION_KEY.into(), serde_json::json!(-1.0));
assert_eq!(c.budget(&ctx).used, raw);
}
#[tokio::test]
async fn budget_required_stages_escalates() {
let b = Budget {
used: 95,
window: 100,
};
assert_eq!(b.required_stages().len(), 4);
let b = Budget {
used: 99,
window: 100,
};
assert_eq!(b.required_stages().len(), 5);
}
}