use std::collections::{BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use agent_core::config::{ContextManagementConfig, ContextManagementMode, SynapsConfig};
use agent_core::core::context_policy::ContextThresholds;
use serde_json::{json, Value};
use super::continuation::{
message_tokens, neutral_note, tool_groups, SuccessorBudget, TokenCalibration,
DROPPED_INDEX_TOKENS, PINNED_SHARE_PERCENT,
};
use crate::SharedMessage;
pub(crate) const MARKER_KEY: &str = "_synaps_worker_context";
pub(crate) const SCHEMA: &str = "synaps-worker-context/1";
const INDEX_PREVIEW_CHARS: usize = 100;
const NOTE_FRAME_TOKENS: u64 = 600;
const NOTE_LISTED_WINDOWS: usize = 12;
const MIN_SHRINK_TOKENS: u64 = 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WorkerContextSettings {
pub enabled: bool,
pub rollover_tokens: u64,
}
impl WorkerContextSettings {
pub fn from_config(config: &SynapsConfig) -> Self {
Self {
enabled: config.subagent_context_management == ContextManagementMode::Auto,
rollover_tokens: config.subagent_rollover_tokens,
}
}
pub fn policy(&self, window: u64) -> Option<WorkerPolicy> {
if !self.enabled {
return None;
}
let base = ContextManagementConfig {
mode: ContextManagementMode::Auto,
finish_rounds: 0,
..Default::default()
};
let band = ContextThresholds::resolve(&base, window).ok()?.rollover_tokens;
let config = ContextManagementConfig {
rollover_tokens: Some(self.rollover_tokens.min(band).max(2)),
..base
};
let thresholds = ContextThresholds::resolve(&config, window).ok()?;
Some(WorkerPolicy { config, thresholds })
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WorkerPolicy {
pub config: ContextManagementConfig,
pub thresholds: ContextThresholds,
}
impl WorkerPolicy {
pub fn target_tokens(&self) -> u64 {
self.thresholds.target_tokens(&self.config)
}
pub fn successor_budget(
&self,
overhead_tokens: u64,
calibration: Option<TokenCalibration>,
) -> SuccessorBudget {
super::continuation::successor_budget(
self.target_tokens(),
self.thresholds.pressure_tokens,
overhead_tokens,
calibration,
)
}
pub fn hard_capacity_has_room(&self, context_window: u64, required_next_round: u64) -> bool {
context_window.saturating_sub(required_next_round) > self.target_tokens()
}
}
pub fn hard_capacity_budget(
budget: SuccessorBudget,
context_window: u64,
required_next_round: u64,
overhead_tokens: u64,
) -> SuccessorBudget {
let room = context_window
.saturating_sub(required_next_round)
.saturating_sub(overhead_tokens);
SuccessorBudget {
history_tokens: budget.history_tokens.min(room * 2 / 3),
}
}
pub type RolloverObserver = Arc<dyn Fn(&WorkerRolloverReport) + Send + Sync>;
#[derive(Clone)]
pub struct WorkerTranscriptSink {
pub dir: PathBuf,
pub stem: String,
pub observer: Option<RolloverObserver>,
}
impl std::fmt::Debug for WorkerTranscriptSink {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WorkerTranscriptSink")
.field("dir", &self.dir)
.field("stem", &self.stem)
.field("observer", &self.observer.is_some())
.finish()
}
}
#[derive(Debug, Clone, Default)]
pub struct WorkerContext {
pub settings: Option<WorkerContextSettings>,
pub sink: Option<WorkerTranscriptSink>,
}
impl WorkerContext {
pub fn resolve(&self, window: u64) -> Option<(WorkerPolicy, WorkerTranscriptSink)> {
let policy = self.settings?.policy(window)?;
Some((policy, self.sink.clone()?))
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkerRolloverReport {
pub window: u64,
pub before_tokens: u64,
pub after_tokens: u64,
pub evicted_messages: usize,
pub transcript: PathBuf,
pub transcripts: Vec<PathBuf>,
}
impl WorkerRolloverReport {
pub fn notice(&self) -> String {
format!(
"Worker context rolled over to window {}: ~{}k → ~{}k tokens; {} earlier messages saved to {}.",
self.window,
self.before_tokens / 1000,
self.after_tokens / 1000,
self.evicted_messages,
self.transcript.display()
)
}
pub fn step(&self) -> String {
format!(
"↻ context window {}: ~{}k → ~{}k tokens",
self.window,
self.before_tokens / 1000,
self.after_tokens / 1000
)
}
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TranscriptRef {
pub window: u64,
pub path: String,
pub messages: usize,
}
pub fn window_state(messages: &[SharedMessage]) -> (u64, Vec<TranscriptRef>) {
let mut window = 1;
let mut seen = BTreeSet::new();
let mut transcripts = Vec::new();
for marker in messages
.iter()
.filter_map(|m| m.get(MARKER_KEY))
.filter(|m| m["schema"] == SCHEMA)
{
window = window.max(marker["window"].as_u64().unwrap_or(1));
for t in marker["transcripts"].as_array().into_iter().flatten() {
if let Ok(t) = serde_json::from_value::<TranscriptRef>(t.clone()) {
if seen.insert((t.window, t.path.clone())) {
transcripts.push(t);
}
}
}
}
transcripts.sort_by_key(|t| t.window);
(window, transcripts)
}
pub fn carried_marker(history: &[Value]) -> Option<Value> {
let shared: Vec<SharedMessage> = history
.iter()
.filter(|m| m.get(MARKER_KEY).is_some())
.map(|m| Arc::new(m.clone()))
.collect();
if shared.is_empty() {
return None;
}
let (window, transcripts) = window_state(&shared);
Some(json!({"schema": SCHEMA, "window": window, "transcripts": transcripts}))
}
fn is_note(message: &Value) -> bool {
message.get(MARKER_KEY).is_some()
}
fn has_tool_result(message: &Value) -> bool {
message["content"]
.as_array()
.is_some_and(|b| b.iter().any(|b| b["type"] == "tool_result"))
}
fn is_host_bookkeeping(message: &Value) -> bool {
is_note(message)
|| message.get("_synaps_context").is_some()
|| crate::engine::interrupt::is_tool_in_flight_note(message)
|| message["content"]
.as_str()
.is_some_and(crate::engine::interrupt::is_interruption_marker)
}
fn carried_instruction(message: &SharedMessage) -> Option<SharedMessage> {
if message["role"] != "user" || is_host_bookkeeping(message) {
return None;
}
match &message["content"] {
Value::String(s) if !s.trim().is_empty() => Some(Arc::clone(message)),
Value::Array(blocks) if !has_tool_result(message) => (!blocks.is_empty())
.then(|| Arc::clone(message)),
Value::Array(blocks) => {
let text: Vec<Value> = blocks
.iter()
.filter(|b| b["type"] == "text")
.filter(|b| b["text"].as_str().is_some_and(|t| !t.trim().is_empty()))
.map(|b| json!({"type": "text", "text": b["text"]}))
.collect();
(!text.is_empty()).then(|| Arc::new(json!({"role": "user", "content": text})))
}
_ => None,
}
}
#[derive(Debug)]
pub struct RolloverPlan {
pub successor: Vec<SharedMessage>,
pub transcript: String,
pub transcript_path: PathBuf,
pub window: u64,
pub evicted_messages: usize,
pub transcripts: Vec<TranscriptRef>,
evicted_indices: Vec<usize>,
}
pub fn plan_rollover(
messages: &[SharedMessage],
budget: SuccessorBudget,
before_tokens: u64,
transcript_path: impl Fn(u64) -> PathBuf,
) -> Option<RolloverPlan> {
let n = messages.len();
if n < 3 || messages[0]["role"] != "user" {
return None;
}
let last_assistant = messages.iter().rposition(|m| m["role"] == "assistant")?;
let after_notes = messages
.iter()
.enumerate()
.skip(1)
.filter(|(_, m)| is_note(m))
.map(|(i, _)| i + 1)
.max()
.unwrap_or(1);
let groups = tool_groups(messages);
let mut extent: HashMap<usize, (usize, usize)> = HashMap::new();
for (i, g) in groups.iter().enumerate() {
let e = extent.entry(*g).or_insert((i, i));
e.0 = e.0.min(i);
e.1 = e.1.max(i);
}
let mut straddled = vec![false; n + 1];
for &(lo, hi) in extent.values() {
for flag in straddled.iter_mut().take(hi + 1).skip(lo + 1) {
*flag = true;
}
}
let candidates: Vec<usize> = (after_notes.max(1)..=last_assistant)
.filter(|&k| messages[k]["role"] == "assistant" && !straddled[k])
.collect();
let &must_keep = candidates.last()?;
let costs: Vec<u64> = messages.iter().map(message_tokens).collect();
let mut suffix = vec![0u64; n + 1];
for i in (0..n).rev() {
suffix[i] = suffix[i + 1] + costs[i];
}
let total = budget.history_tokens;
let share = total.min(u64::MAX / 100) * PINNED_SHARE_PERCENT / 100;
let instructions: Vec<(usize, SharedMessage, u64)> = (1..must_keep)
.filter_map(|i| carried_instruction(&messages[i]).map(|m| (i, m)))
.map(|(i, m)| {
let cost = message_tokens(&m);
(i, m, cost)
})
.collect();
let instruction_reserve = instructions.iter().map(|(_, _, c)| c).sum::<u64>().min(share);
let (_, prior) = window_state(messages);
let note_reserve =
DROPPED_INDEX_TOKENS + NOTE_FRAME_TOKENS + 60 * (prior.len() as u64 + 1);
let available =
total.saturating_sub(costs[0] + instruction_reserve + note_reserve);
let first = candidates
.iter()
.position(|&k| suffix[k] <= available)
.unwrap_or(candidates.len() - 1);
let (window, prior_transcripts) = window_state(messages);
let path = transcript_path(window);
let mut plan = None;
for &cut in &candidates[first..] {
let built = build_successor(
messages,
cut,
after_notes,
&instructions,
share,
window,
&prior_transcripts,
&path,
before_tokens,
&costs,
)?;
let size = super::context::estimate_history(&built.successor);
plan = Some(built);
if size <= total || cut == must_keep {
if size > total && total != u64::MAX {
tracing::warn!(
successor_tokens = size,
target_tokens = total,
"worker rollover must-keep tail exceeds the successor target; keeping it whole"
);
}
break;
}
}
let mut plan = plan?;
let evicted: Vec<usize> = std::mem::take(&mut plan.evicted_indices);
plan.transcript = render_transcript(messages, &evicted, window, before_tokens);
Some(plan)
}
#[allow(clippy::too_many_arguments)]
fn build_successor(
messages: &[SharedMessage],
cut: usize,
after_notes: usize,
instructions: &[(usize, SharedMessage, u64)],
share: u64,
window: u64,
prior_transcripts: &[TranscriptRef],
path: &Path,
before_tokens: u64,
costs: &[u64],
) -> Option<RolloverPlan> {
let mut pinned = Vec::new();
let mut unpinned = Vec::new();
let mut used = 0u64;
for (i, m, cost) in instructions.iter().rev().filter(|(i, _, _)| *i < cut) {
if used + cost <= share {
used += cost;
pinned.push((*i, Arc::clone(m)));
} else {
unpinned.push(*i);
}
}
pinned.reverse();
unpinned.reverse();
let evicted: Vec<usize> = (after_notes..cut)
.filter(|&i| !is_note(&messages[i]))
.collect();
if evicted.is_empty() || evicted.iter().all(|i| pinned.iter().any(|(p, _)| p == i)) {
return None;
}
let mut transcripts = prior_transcripts.to_vec();
transcripts.push(TranscriptRef {
window,
path: path.display().to_string(),
messages: evicted.len(),
});
let evicted_tokens: u64 = evicted.iter().map(|&i| costs[i]).sum();
let kept = messages.len() - cut;
let note = note_message(NoteFacts {
window: window + 1,
before_tokens,
evicted: evicted.len(),
evicted_tokens,
kept,
carried_instructions: !pinned.is_empty(),
transcripts: &transcripts,
tool_index: tool_index(messages, &evicted),
unpinned: &unpinned
.iter()
.map(|&i| (i, preview(&messages[i])))
.collect::<Vec<_>>(),
});
let mut successor = Vec::with_capacity(2 + pinned.len() + kept);
successor.push(Arc::clone(&messages[0]));
successor.extend(pinned.into_iter().map(|(_, m)| m));
successor.push(Arc::new(note));
successor.extend(messages[cut..].iter().cloned());
super::read_dedupe::rewrite_dangling_stubs(&mut successor);
Some(RolloverPlan {
successor,
transcript: String::new(),
transcript_path: path.to_path_buf(),
window: window + 1,
evicted_messages: evicted.len(),
transcripts,
evicted_indices: evicted,
})
}
pub fn plan_is_productive(messages: &[SharedMessage], plan: &RolloverPlan) -> bool {
let before = super::context::estimate_history(messages);
let after = super::context::estimate_history(&plan.successor);
after.saturating_add(MIN_SHRINK_TOKENS) < before
&& unpaired_tool_ids(&plan.successor).is_subset(&unpaired_tool_ids(messages))
}
pub fn unpaired_tool_ids(messages: &[SharedMessage]) -> BTreeSet<String> {
let mut calls = BTreeSet::new();
let mut unpaired = BTreeSet::new();
for m in messages {
for b in m["content"].as_array().into_iter().flatten() {
match (b["type"].as_str(), m["role"].as_str()) {
(Some("tool_use"), Some("assistant")) => {
if let Some(id) = b["id"].as_str() {
calls.insert(id.to_owned());
unpaired.insert(id.to_owned());
}
}
(Some("tool_result"), Some("user")) => {
if let Some(id) = b["tool_use_id"].as_str() {
if !unpaired.remove(id) || !calls.contains(id) {
unpaired.insert(format!("result:{id}"));
}
}
}
_ => {}
}
}
}
unpaired
}
struct NoteFacts<'a> {
window: u64,
before_tokens: u64,
evicted: usize,
evicted_tokens: u64,
kept: usize,
carried_instructions: bool,
transcripts: &'a [TranscriptRef],
tool_index: Vec<String>,
unpinned: &'a [(usize, String)],
}
fn note_message(f: NoteFacts<'_>) -> Value {
let mut text = format!(
"[Worker context rollover — host note, not a new instruction]\n\
Your context reached about {} tokens, so the host started context window {}. \
{} earlier messages (about {} tokens) were moved out of your context unchanged; \
nothing was summarized. Your task (the first message){} and your most recent {} \
messages are kept verbatim.\n\
The moved messages are saved as plain text you can search with grep or read by line range:",
f.before_tokens,
f.window,
f.evicted,
f.evicted_tokens,
if f.carried_instructions {
", later orchestrator instructions"
} else {
""
},
f.kept,
);
let hidden = f.transcripts.len().saturating_sub(NOTE_LISTED_WINDOWS);
if let Some(oldest) = f.transcripts.first().filter(|_| hidden > 0) {
let dir = Path::new(&oldest.path)
.parent()
.map(|d| d.display().to_string())
.unwrap_or_default();
text.push_str(&format!(
"\n- windows {}–{}: earlier transcripts (*.window-N.md) in {}",
oldest.window,
f.transcripts[hidden - 1].window,
neutral_note(&dir)
));
}
for t in &f.transcripts[hidden..] {
text.push_str(&format!(
"\n- window {}: {} ({} messages)",
t.window,
neutral_note(&t.path),
t.messages
));
}
text.push_str(
"\nLook there only when you need a specific earlier detail (an error, a file path, a \
decision); do not re-read them wholesale.",
);
if !f.unpinned.is_empty() {
text.push_str("\nEarlier orchestrator messages not kept verbatim (they are in the saved text):");
for (i, p) in f.unpinned {
text.push_str(&format!("\n- #{i}: {p}"));
}
}
if !f.tool_index.is_empty() {
text.push_str("\nTool calls in the moved messages, most recent first:");
let mut listed = 0;
for line in &f.tool_index {
let line = format!("\n- {line}");
if super::context::conservative_token_estimate(&text)
+ super::context::conservative_token_estimate(&line)
> DROPPED_INDEX_TOKENS + NOTE_FRAME_TOKENS / 2
{
break;
}
text.push_str(&line);
listed += 1;
}
if listed < f.tool_index.len() {
text.push_str(&format!(
"\n- … and {} older not listed",
f.tool_index.len() - listed
));
}
}
text.push_str(
"\nContinue the task from where the kept messages leave off; do not redo completed \
steps.\n[End worker context rollover]",
);
let transcripts: Vec<Value> = f
.transcripts
.iter()
.map(|t| serde_json::to_value(t).unwrap_or(Value::Null))
.collect();
json!({
"role": "user",
MARKER_KEY: {"schema": SCHEMA, "window": f.window, "transcripts": transcripts},
"content": [{"type": "text", "text": text}],
})
}
fn tool_index(messages: &[SharedMessage], evicted: &[usize]) -> Vec<String> {
let mut lines = Vec::new();
for &i in evicted.iter().rev() {
let blocks: Vec<&Value> = messages[i]["content"]
.as_array()
.into_iter()
.flatten()
.filter(|b| b["type"] == "tool_use")
.collect();
for b in blocks.into_iter().rev() {
let name = neutral_note(b["name"].as_str().unwrap_or("?"));
let input = &b["input"];
let arg = ["command", "path", "file_path", "pattern", "query", "url", "handle_id"]
.iter()
.find_map(|k| input[*k].as_str().map(str::to_owned))
.unwrap_or_else(|| input.to_string());
lines.push(format!("{name}: {}", clip(&flat(&arg), INDEX_PREVIEW_CHARS)));
}
}
lines
}
fn flat(text: &str) -> String {
neutral_note(&text.split_whitespace().collect::<Vec<_>>().join(" "))
}
fn clip(text: &str, chars: usize) -> String {
let mut out: String = text.chars().take(chars).collect();
if text.chars().count() > chars {
out.push('…');
}
out
}
fn preview(message: &Value) -> String {
let text = match &message["content"] {
Value::String(s) => s.clone(),
Value::Array(blocks) => blocks
.iter()
.filter_map(|b| b["text"].as_str())
.collect::<Vec<_>>()
.join(" "),
_ => String::new(),
};
clip(&flat(&text), 80)
}
pub fn render_transcript(
messages: &[SharedMessage],
evicted: &[usize],
window: u64,
before_tokens: u64,
) -> String {
let mut out = format!(
"# Worker context window {window}: {} messages moved out of context at a rollover\n\
# (context about {before_tokens} tokens). Thinking is omitted; images are named only.\n",
evicted.len()
);
for &i in evicted {
let m = &messages[i];
let role = m["role"].as_str().unwrap_or("user");
out.push_str(&format!("\n## [{i}] {role}\n"));
match &m["content"] {
Value::String(s) => {
out.push_str(s);
out.push('\n');
}
Value::Array(blocks) => {
for b in blocks {
render_block(&mut out, b);
}
}
_ => {}
}
}
out
}
fn render_block(out: &mut String, b: &Value) {
match b["type"].as_str() {
Some("text") => {
out.push_str(b["text"].as_str().unwrap_or_default());
out.push('\n');
}
Some("tool_use") => {
let input = serde_json::to_string_pretty(&b["input"]).unwrap_or_default();
out.push_str(&format!(
"→ tool_use {} ({})\n{input}\n",
b["name"].as_str().unwrap_or("?"),
b["id"].as_str().unwrap_or("?")
));
}
Some("tool_result") => {
let error = if b["is_error"] == true { " error" } else { "" };
out.push_str(&format!(
"← tool_result{error} ({})\n",
b["tool_use_id"].as_str().unwrap_or("?")
));
match &b["content"] {
Value::String(s) => {
out.push_str(s);
out.push('\n');
}
Value::Array(inner) => {
for x in inner {
render_block(out, x);
}
}
Value::Null => {}
other => {
out.push_str(&other.to_string());
out.push('\n');
}
}
}
Some("thinking") | Some("redacted_thinking") => {}
Some(other) => out.push_str(&format!("[{other}]\n")),
None => {}
}
}
pub fn roll_over(
messages: &[SharedMessage],
budget: SuccessorBudget,
before_tokens: u64,
overhead_tokens: u64,
policy: &WorkerPolicy,
sink: &WorkerTranscriptSink,
hard_capacity: Option<u64>,
) -> std::io::Result<Option<(Vec<SharedMessage>, WorkerRolloverReport)>> {
let Some(plan) = plan_rollover(messages, budget, before_tokens, |window| {
transcript_path(&sink.dir, &sink.stem, window)
}) else {
return Ok(None);
};
if !plan_is_productive(messages, &plan) {
return Ok(None);
}
let after_tokens =
overhead_tokens.saturating_add(super::context::estimate_history(&plan.successor));
if after_tokens >= hard_capacity.unwrap_or(policy.thresholds.pressure_tokens) {
return Ok(None);
}
crate::tools::write_window_transcript(
&plan.transcript_path,
&plan.transcript,
)?;
let report = WorkerRolloverReport {
window: plan.window,
before_tokens,
after_tokens,
evicted_messages: plan.evicted_messages,
transcript: plan.transcript_path,
transcripts: plan
.transcripts
.iter()
.map(|t| PathBuf::from(&t.path))
.collect(),
};
Ok(Some((plan.successor, report)))
}
pub fn transcript_path(dir: &Path, stem: &str, window: u64) -> PathBuf {
let stem: String = stem
.chars()
.map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '_' { c } else { '-' })
.collect();
dir.join(format!("{stem}.window-{window}.md"))
}
#[cfg(test)]
mod tests {
use super::*;
fn user(text: &str) -> SharedMessage {
Arc::new(json!({"role": "user", "content": text}))
}
fn round(i: usize, pad: usize) -> [SharedMessage; 2] {
[
Arc::new(json!({"role": "assistant", "content": [
{"type": "thinking", "thinking": format!("step {i}"), "signature": format!("sig{i}")},
{"type": "tool_use", "id": format!("t{i}"), "name": "bash",
"input": {"command": format!("cargo test step{i}")}}
]})),
Arc::new(json!({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": format!("t{i}"),
"content": format!("output {i} {}", "x".repeat(pad))}
]})),
]
}
fn worker_history(rounds: usize, pad: usize) -> Vec<SharedMessage> {
let mut m = vec![user("BRIEF: fix the parser")];
for i in 0..rounds {
m.extend(round(i, pad));
}
m
}
fn path(w: u64) -> PathBuf {
PathBuf::from(format!("/tmp/x/sa_1.window-{w}.md"))
}
fn budget(tokens: u64) -> SuccessorBudget {
SuccessorBudget { history_tokens: tokens }
}
#[test]
fn policy_lowers_the_band_and_derives_pressure_and_target_like_the_foreground() {
let on = WorkerContextSettings { enabled: true, rollover_tokens: 200_000 };
let p = on.policy(1_000_000).unwrap();
assert_eq!(p.thresholds.rollover_tokens, 200_000);
assert_eq!(p.thresholds.pressure_tokens, 199_999);
assert_eq!(p.target_tokens(), 100_000);
assert_eq!(p.config.finish_rounds, 0);
let p = on.policy(200_000).unwrap();
assert_eq!(p.thresholds.rollover_tokens, 180_000);
assert_eq!(p.thresholds.pressure_tokens, 140_000);
assert_eq!(p.target_tokens(), 93_333);
assert!(p.target_tokens() < p.thresholds.pressure_tokens);
let off = WorkerContextSettings { enabled: false, ..on };
assert!(off.policy(1_000_000).is_none());
assert!(on.policy(1).is_none(), "a window that holds no policy disables it");
let ctx = WorkerContext { settings: Some(on), sink: None };
assert!(ctx.resolve(1_000_000).is_none());
}
#[test]
fn policy_engine_admits_then_rolls_at_the_worker_band() {
use agent_core::core::context_policy::{assess_context, ContextAction, ContextBudget, ContextState};
let p = WorkerContextSettings { enabled: true, rollover_tokens: 200_000 }
.policy(1_000_000)
.unwrap();
let at = |used: u64| ContextBudget {
context_window_tokens: 1_000_000,
used_tokens: used,
hard_remaining_tokens: 1_000_000 - used,
required_next_round_tokens: 150_000,
};
let state = ContextState::default();
assert_eq!(assess_context(&p.config, &state, at(150_000)).action, ContextAction::Continue);
assert_eq!(assess_context(&p.config, &state, at(200_000)).action, ContextAction::Rollover);
assert_eq!(assess_context(&p.config, &state, at(900_000)).action, ContextAction::HardStop);
}
#[test]
fn hard_capacity_aims_inside_the_room_the_window_has_left() {
let normal = budget(56_000);
assert_eq!(hard_capacity_budget(normal, 120_000, 72_000, 1_000).history_tokens, 31_333);
assert_eq!(hard_capacity_budget(normal, 1_000_000, 150_000, 1_000).history_tokens, 56_000);
assert_eq!(hard_capacity_budget(normal, 10, 72_000, 1_000).history_tokens, 0);
}
fn required(model: &str, window: u64) -> u64 {
super::super::stream::pre_request_context_budget(model, window, None, &[], &[], 0, 30_000, None)
.1
.required_next_round_tokens
}
#[test]
fn hard_capacity_has_room_only_when_the_target_fits_below_it() {
use agent_core::core::context_policy::{assess_context, ContextAction, ContextBudget, ContextState};
let on = WorkerContextSettings { enabled: true, rollover_tokens: 200_000 };
let p = on.policy(200_000).unwrap();
assert_eq!(p.target_tokens(), 93_333);
for model in ["claude-opus-5-5", "claude-sonnet-4-6"] {
let need = required(model, 200_000);
assert_eq!(need, 32_768 + 10_000 + 30_000 + 512, "{model}");
assert!(p.hard_capacity_has_room(200_000, need), "{model}");
let at = |used: u64| ContextBudget {
context_window_tokens: 200_000,
used_tokens: used,
hard_remaining_tokens: 200_000 - used,
required_next_round_tokens: need,
};
let state = ContextState::default();
assert_eq!(assess_context(&p.config, &state, at(33_000)).action, ContextAction::Continue);
assert_eq!(assess_context(&p.config, &state, at(126_720)).action, ContextAction::Continue);
assert_eq!(assess_context(&p.config, &state, at(126_721)).action, ContextAction::HardStop);
let need_1m = required(model, 1_000_000);
assert!(on.policy(1_000_000).unwrap().hard_capacity_has_room(1_000_000, need_1m));
}
assert!(!p.hard_capacity_has_room(200_000, 128_000 + 10_000 + 30_000 + 512));
let small = on.policy(100_000).unwrap();
assert_eq!(small.target_tokens(), 46_666);
assert_eq!(100_000 - required("claude-opus-5-5", 100_000), 41_720);
assert!(!small.hard_capacity_has_room(100_000, required("claude-opus-5-5", 100_000)));
assert!(!p.hard_capacity_has_room(200_000, 200_000 - 93_333));
assert!(p.hard_capacity_has_room(200_000, 200_000 - 93_334));
assert!(!p.hard_capacity_has_room(100_000, 200_000), "reserves over the window");
}
#[test]
fn a_successor_at_the_line_that_triggered_it_is_refused() {
let tmp = tempfile::tempdir().unwrap();
let sink = WorkerTranscriptSink { dir: tmp.path().into(), stem: "sa_1".into(), observer: None };
let policy = WorkerContextSettings { enabled: true, rollover_tokens: 200_000 }
.policy(200_000)
.unwrap();
let mut m = worker_history(20, 3_000);
m.extend(round(20, 60_000)); let before = super::super::context::estimate_history(&m);
let roll = |line| roll_over(&m, budget(10_000), before, 1_000, &policy, &sink, line).unwrap();
assert!(roll(Some(20_000)).is_none(), "would start at hard capacity again");
let (successor, report) = roll(Some(100_000)).expect("room below the line");
assert!(report.after_tokens < 100_000 && successor.len() < m.len());
assert!(roll(None).is_some());
}
#[test]
fn successor_keeps_brief_and_recent_rounds_within_the_target() {
let messages = worker_history(40, 4_000);
let before = super::super::context::estimate_history(&messages);
let plan = plan_rollover(&messages, budget(20_000), before, path).unwrap();
let s = &plan.successor;
assert!(Arc::ptr_eq(&s[0], &messages[0]), "the brief survives verbatim");
assert!(s[1].get(MARKER_KEY).is_some(), "then the host note");
assert_eq!(s[2]["role"], "assistant", "the suffix starts at an assistant turn");
assert!(Arc::ptr_eq(s.last().unwrap(), messages.last().unwrap()));
for m in &s[2..] {
assert!(messages.iter().any(|o| Arc::ptr_eq(o, m)));
}
assert!(s[2]["content"][0]["type"] == "thinking" && s[2]["content"][0]["signature"].is_string());
let after = super::super::context::estimate_history(s);
assert!(after <= 20_000, "target respected: {after}");
assert!(after < before / 3);
assert!(unpaired_tool_ids(s).is_empty());
assert!(plan_is_productive(&messages, &plan));
assert_eq!(plan.window, 2);
assert_eq!(plan.evicted_messages, messages.len() - 1 - (s.len() - 2));
assert!(plan.transcript.contains("cargo test step0"));
assert!(plan.transcript.contains("output 0 "));
assert!(!plan.transcript.contains("sig0") && !plan.transcript.contains("step 0\n"));
let note = s[1]["content"][0]["text"].as_str().unwrap();
assert!(note.contains("/tmp/x/sa_1.window-1.md"), "{note}");
assert!(note.contains("bash: cargo test step"), "{note}");
assert_eq!(s[1][MARKER_KEY]["window"], 2);
}
#[test]
fn a_kept_dedupe_stub_whose_read_was_evicted_is_rewritten() {
use crate::runtime::read_dedupe::{unchanged_stub, DROPPED_STUB};
let read = |id: &str, content: String| -> [SharedMessage; 2] {
[
Arc::new(json!({"role":"assistant","content":[
{"type":"tool_use","id":id,"name":"read","input":{"path":"/w/database.py"}}]})),
Arc::new(json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":id,"content":content}]})),
]
};
let full = "1\tdb = connect()\n".repeat(50);
let history = |original_late: bool| {
let mut m = vec![user("BRIEF: fix the parser")];
if !original_late {
m.extend(read("toolu_r1", full.clone()));
}
for i in 0..20 {
m.extend(round(i, 4_000));
}
if original_late {
m.extend(read("toolu_r1", full.clone()));
}
m.extend(read("toolu_r2", unchanged_stub("toolu_r1")));
m.extend(round(99, 10));
m
};
let m = history(false);
let plan = plan_rollover(&m, budget(8_000), 1, path).unwrap();
let s = &plan.successor;
assert!(!s.iter().any(|x| x.to_string().contains("toolu_r1")), "r1 evicted");
let at = s.iter().position(|x| x["content"][0]["tool_use_id"] == "toolu_r2").unwrap();
assert_eq!(s[at]["content"][0]["content"], DROPPED_STUB);
assert!(Arc::ptr_eq(&s[at - 1], &m[m.len() - 4]), "the call itself is never edited");
assert!(unpaired_tool_ids(s).is_empty() && plan_is_productive(&m, &plan));
let m = history(true);
let plan = plan_rollover(&m, budget(8_000), 1, path).unwrap();
for x in &plan.successor[2..] {
assert!(m.iter().any(|o| Arc::ptr_eq(o, x)), "nothing edited");
}
assert!(plan.successor.iter().any(|x| x["content"][0]["content"] == unchanged_stub("toolu_r1")));
}
#[test]
fn cut_never_splits_a_tool_group_and_keeps_the_pending_loop() {
let mut m = worker_history(6, 2_000);
m.push(Arc::new(json!({"role":"assistant","content":[
{"type":"tool_use","id":"p1","name":"read","input":{"path":"a"}},
{"type":"tool_use","id":"p2","name":"read","input":{"path":"b"}}]})));
m.push(Arc::new(json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":"p1","content":"A"}]})));
m.push(Arc::new(json!({"role":"assistant","content":[{"type":"text","text":"hm"}]})));
m.push(Arc::new(json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":"p2","content":"B"}]})));
m.extend(round(99, 10));
for tokens in [0, 1_000, 3_000, 6_000, 50_000] {
let Some(plan) = plan_rollover(&m, budget(tokens), 1, path) else { continue };
assert!(unpaired_tool_ids(&plan.successor).is_empty(), "budget {tokens}");
let note = plan.successor.iter().position(|x| x.get(MARKER_KEY).is_some()).unwrap();
assert_eq!(plan.successor[note + 1]["role"], "assistant");
assert!(Arc::ptr_eq(plan.successor.last().unwrap(), m.last().unwrap()));
}
let plan = plan_rollover(&m, budget(0), 1, path).unwrap();
assert_eq!(plan.successor.len(), 4, "brief, note, last call, its result");
}
#[test]
fn orchestrator_instructions_survive_and_bookkeeping_does_not() {
let mut m = worker_history(3, 3_000);
m.push(user("STEER: also update the docs"));
m.extend(round(3, 3_000));
let mut carrier = (*m.pop().unwrap()).clone();
carrier["content"].as_array_mut().unwrap().push(
json!({"type":"text","text":"[Resumed by the orchestrator.] RESUME: finish the tests"}),
);
m.push(Arc::new(carrier));
m.push(Arc::new(json!({"role":"user","content":
crate::engine::interrupt::InterruptReason::ALL[0].marker()})));
for i in 4..30 {
m.extend(round(i, 3_000));
}
let plan = plan_rollover(&m, budget(15_000), 1, path).unwrap();
let s = &plan.successor;
let texts: Vec<String> = s
.iter()
.take_while(|x| x.get(MARKER_KEY).is_none())
.map(|x| x["content"].to_string())
.collect();
assert_eq!(texts.len(), 3, "brief + two carried instructions: {texts:?}");
assert!(texts[1].contains("STEER: also update the docs"));
assert!(texts[2].contains("RESUME: finish the tests"));
assert!(!texts[2].contains("tool_result"), "only the text travels");
assert!(!s.iter().any(|x| x["content"].as_str().is_some_and(
crate::engine::interrupt::is_interruption_marker)));
assert!(unpaired_tool_ids(s).is_empty());
}
#[test]
fn second_rollover_continues_numbering_and_lists_every_window() {
let messages = worker_history(30, 3_000);
let first = plan_rollover(&messages, budget(15_000), 1, path).unwrap();
let mut next = first.successor.clone();
for i in 100..130 {
next.extend(round(i, 3_000));
}
assert_eq!(window_state(&next).0, 2);
let second = plan_rollover(&next, budget(15_000), 1, path).unwrap();
assert_eq!(second.window, 3);
let s = &second.successor;
assert!(Arc::ptr_eq(&s[0], &messages[0]), "the brief survives every rollover");
assert_eq!(s.iter().filter(|x| x.get(MARKER_KEY).is_some()).count(), 1, "one note");
let note = s[1]["content"][0]["text"].as_str().unwrap();
assert!(note.contains("window-1.md") && note.contains("window-2.md"), "{note}");
assert!(!second.transcript.contains("Worker context rollover"), "old note not re-saved");
let (window, transcripts) = window_state(s);
assert_eq!(window, 3);
assert_eq!(transcripts.iter().map(|t| t.window).collect::<Vec<_>>(), vec![1, 2]);
let history: Vec<Value> = s.iter().map(|x| (**x).clone()).collect();
let marker = carried_marker(&history).unwrap();
assert_eq!(marker["window"], 3);
assert_eq!(marker["transcripts"].as_array().unwrap().len(), 2);
assert!(carried_marker(&[json!({"role":"user","content":"x"})]).is_none());
}
#[test]
fn the_note_stays_bounded_however_many_windows_a_worker_has_had() {
let transcripts: Vec<TranscriptRef> = (1..=40)
.map(|w| TranscriptRef {
window: w,
path: format!("/d/sa_1.window-{w}.md"),
messages: 9,
})
.collect();
let note = note_message(NoteFacts {
window: 41,
before_tokens: 200_000,
evicted: 9,
evicted_tokens: 100_000,
kept: 30,
carried_instructions: false,
transcripts: &transcripts,
tool_index: (0..5_000).map(|i| format!("bash: step {i}")).collect(),
unpinned: &[],
});
let text = note["content"][0]["text"].as_str().unwrap();
assert!(text.contains("windows 1–28: earlier transcripts (*.window-N.md) in /d"), "{text}");
assert!(!text.contains("sa_1.window-28.md") && text.contains("sa_1.window-29.md"));
assert!(text.contains("sa_1.window-40.md"));
assert!(text.contains("older not listed"));
let tokens = super::super::context::conservative_token_estimate(text);
assert!(tokens <= DROPPED_INDEX_TOKENS + NOTE_FRAME_TOKENS, "{tokens}");
}
#[test]
fn nothing_to_evict_is_no_plan_and_small_wins_are_unproductive() {
assert!(plan_rollover(&[user("brief")], budget(1), 1, path).is_none());
let one = worker_history(1, 50_000);
assert!(plan_rollover(&one, budget(1), 1, path).is_none(), "only the must-keep round");
let two = worker_history(2, 100);
let plan = plan_rollover(&two, budget(0), 1, path).unwrap();
assert!(!plan_is_productive(&two, &plan), "a tiny round is not worth a rewrite");
}
#[test]
fn host_metadata_never_reaches_the_wire() {
let plan = plan_rollover(&worker_history(10, 3_000), budget(5_000), 1, path).unwrap();
let wire = serde_json::to_string(&super::super::continuation::WireMessages(&plan.successor)).unwrap();
assert!(!wire.contains(MARKER_KEY));
assert!(wire.contains("Worker context rollover"));
}
#[test]
fn transcript_path_is_confined_to_the_directory() {
let p = transcript_path(Path::new("/d"), "../sa 1", 3);
assert_eq!(p, PathBuf::from("/d/---sa-1.window-3.md"));
}
}