Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: reconcile Ticket tags with current develop contracts and Runtime auth

Browse Source
This commit is contained in:
Hare
2026-10-11 02:03:44 +09:00
parent e9397fa0dd 71ab4c06b9
commit b1fe562e15
67 changed files with 5629 additions and 4519 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/compact/worker.rs
+124 -6
View File
@@ -379,6 +379,7 @@ pub(crate) struct CompactWorkerInterceptor {
pub on_warning: Option<Arc<dyn Fn(String) + Send + Sync>>,
warning_sent: AtomicBool,
last_remaining_tokens: AtomicU64,
request_shape: Option<(String, agen::tool_server::ToolServerHandle)>,
}
impl CompactWorkerInterceptor {
@@ -397,9 +398,21 @@ impl CompactWorkerInterceptor {
on_warning,
warning_sent: AtomicBool::new(false),
last_remaining_tokens: AtomicU64::new(max_input_tokens),
request_shape: None,
}
}
pub(crate) fn with_request_shape(
mut self,
system_prompt: String,
tools: agen::tool_server::ToolServerHandle,
) -> Self {
// Features install after engine configuration. This shared handle sees
// the final installed tool definitions at the actual request boundary.
self.request_shape = Some((system_prompt, tools));
self
}
fn maybe_emit_warning(&self, remaining: u64) -> Option<Item> {
let warning_threshold = self.finish_warning_remaining_tokens;
let reserve_threshold = self.final_reserve_tokens;
@@ -412,7 +425,12 @@ impl CompactWorkerInterceptor {
let message = format!(
"compact worker context budget is low ({remaining}/{} tokens remaining). \
Stop broad exploration now, read only if absolutely necessary, then call \
`write_summary` with the final structured summary.",
`write_summary` with the final structured summary. Preserve ongoing goals and \
user directives from the full initial summary and segment. Compress duplicate \
explanations and old work details first, not scope, prohibitions, completion \
conditions or exception-reporting duties. Apply explicit later changes/retractions \
and retain uncertainty; recovery procedures and recent progress do not replace \
the original request.",
self.max_input_tokens
);
if let Some(cb) = self.on_warning.as_ref() {
@@ -433,17 +451,38 @@ impl<A: Send + Sync> Interceptor<A> for CompactWorkerInterceptor {
let context = context.items;
let records = self.usage_tracker.records();
let estimate = agen::token_counter::total_tokens(context, &records);
if estimate.tokens > self.max_input_tokens {
// Before the first provider measurement, include the system and tool
// shape as well as *all* mandatory summary/input bytes. Later exact
// input measurements already include that shape; do not double charge.
let shape_tokens = if records.is_empty() {
self.request_shape.as_ref().map_or(0, |(system, tools)| {
let tool_bytes = serde_json::to_vec(&tools.tool_definitions_sorted())
.map(|bytes| bytes.len())
.unwrap_or(0);
estimate_tokens(system.len().saturating_add(tool_bytes))
})
} else {
0
};
let input_tokens = estimate.tokens.saturating_add(shape_tokens);
if input_tokens > self.max_input_tokens {
return Ok(PreRequestAction::Cancel(format!(
"compact worker input occupancy exceeded {} tokens",
"compact worker input occupancy (including mandatory full initial summary) exceeded {} tokens; no summary was truncated",
self.max_input_tokens
)));
}
let remaining = self.max_input_tokens.saturating_sub(estimate.tokens);
let remaining = self.max_input_tokens.saturating_sub(input_tokens);
self.last_remaining_tokens
.store(remaining, Ordering::Release);
if let Some(item) = self.maybe_emit_warning(remaining) {
let warning_tokens =
agen::token_counter::total_tokens(std::slice::from_ref(&item), &[]).tokens;
if input_tokens.saturating_add(warning_tokens) > self.max_input_tokens {
return Ok(PreRequestAction::Cancel(
"compact worker input occupancy cannot fit budget warning alongside mandatory full input; no summary was truncated".into(),
));
}
self.usage_tracker.note_request(context.len() + 1);
return Ok(PreRequestAction::ContinueWith(vec![item]));
}
@@ -542,6 +581,85 @@ mod tests {
));
}
#[tokio::test]
async fn compact_worker_warning_policy_keeps_user_intent_and_fails_if_warning_cannot_fit() {
let interceptor =
CompactWorkerInterceptor::new(Arc::new(UsageTracker::new()), 100, 100, 0, None);
let mut context = vec![Item::user_message("full input")];
let result = interceptor
.pre_llm_request(PreLlmRequestContext::<()> {
invocation: Default::default(),
items: &mut context,
history: &[],
})
.await
.unwrap();
assert!(
matches!(result, PreRequestAction::Cancel(message) if message.contains("no summary was truncated"))
);
let interceptor =
CompactWorkerInterceptor::new(Arc::new(UsageTracker::new()), 1000, 1000, 0, None);
let result = interceptor
.pre_llm_request(PreLlmRequestContext::<()> {
invocation: Default::default(),
items: &mut context,
history: &[],
})
.await
.unwrap();
let PreRequestAction::ContinueWith(items) = result else {
panic!("expected persisted warning")
};
let warning = items[0].as_text().unwrap();
for policy in [
"full initial summary and segment",
"prohibitions",
"completion conditions",
"exception-reporting duties",
"changes/retractions",
"uncertainty",
"Compress duplicate",
] {
assert!(warning.contains(policy), "missing warning policy: {policy}");
}
}
#[tokio::test]
async fn compact_worker_initial_request_counts_system_shape_but_measurements_do_not_double_count()
{
let tracker = Arc::new(UsageTracker::new());
let interceptor = CompactWorkerInterceptor::new(tracker.clone(), 100, 0, 0, None)
.with_request_shape(
"system".repeat(100),
agen::tool_server::ToolServerHandle::default(),
);
let mut context = vec![Item::user_message("full input")];
assert!(matches!(
interceptor
.pre_llm_request(PreLlmRequestContext::<()> {
invocation: Default::default(),
items: &mut context,
history: &[],
})
.await
.unwrap(),
PreRequestAction::Cancel(_)
));
tracker.note_request(context.len());
tracker.record_usage(&make_usage(50));
assert!(matches!(
interceptor
.pre_llm_request(PreLlmRequestContext::<()> {
invocation: Default::default(),
items: &mut context,
history: &[],
})
.await
.unwrap(),
PreRequestAction::Continue
));
}
#[tokio::test]
async fn compact_worker_interceptor_warns_before_hard_cap() {
let tracker = Arc::new(UsageTracker::new());
@@ -549,8 +667,8 @@ mod tests {
let captured = warnings.clone();
let interceptor = CompactWorkerInterceptor::new(
tracker.clone(),
150,
60,
500,
410,
20,
Some(Arc::new(move |message| {
captured.lock().unwrap().push(message);
crates/worker/src/feature/builtin/flow_transition.rs
+1
View File
@@ -560,6 +560,7 @@ where
attempt.instance_id, attempt.attempt_id
)),
max_turns: Some(12),
compaction_budget: None,
engine_configurator: None,
features,
required_tools: &[
crates/worker/src/feature/builtin/memory_lifecycle.rs
+1
View File
@@ -560,6 +560,7 @@ impl MemoryLifecycleTask {
.extraction
.worker_max_turns
.or(manifest::defaults::MEMORY_EXTRACT_WORKER_MAX_TURNS),
compaction_budget: None,
engine_configurator: None,
features,
required_tools: &[
crates/worker/src/feature/builtin/memory_surface_lifecycle.rs
+1
View File
@@ -376,6 +376,7 @@ impl SubjektivSurfaceLifecycleTask {
input,
cache_key: None,
max_turns: Some(EDITOR_MAX_TURNS),
compaction_budget: None,
engine_configurator: None,
features,
required_tools: &["SubmitMemorySurface"],
crates/worker/src/internal_worker.rs
+8
View File
@@ -68,6 +68,7 @@ pub(crate) struct InternalWorkerSpec {
) + Send,
>,
>,
pub compaction_budget: Option<crate::compact::worker::CompactWorkerInterceptor>,
pub features: FeatureRegistryBuilder,
pub required_tools: &'static [&'static str],
pub authority: InternalWorkerAuthority,
@@ -118,6 +119,7 @@ where
cache_key,
max_turns,
engine_configurator,
compaction_budget,
features,
required_tools,
authority,
@@ -171,6 +173,7 @@ where
identity: identity.clone(),
history_entries: 0,
})?;
worker.install_compaction_budget(compaction_budget);
worker.disable_manifest_lifecycle_features();
if let Some(session) = inherited_workdir_session {
worker.bind_single_workdir_session(Some(session));
@@ -666,6 +669,7 @@ pub(crate) async fn spawn_internal_worker_session(
cache_key,
max_turns,
engine_configurator,
compaction_budget,
features,
required_tools,
authority,
@@ -701,6 +705,7 @@ pub(crate) async fn spawn_internal_worker_session(
.map_err(|source| InternalWorkerSessionError::Build {
message: source.to_string(),
})?;
worker.install_compaction_budget(compaction_budget);
worker.disable_manifest_lifecycle_features();
if let Some(session) = inherited_workdir_session {
worker.bind_single_workdir_session(Some(session));
@@ -762,6 +767,7 @@ pub(crate) fn prepare_internal_worker_from_spec(
cache_key,
max_turns,
engine_configurator,
compaction_budget,
features,
required_tools,
authority,
@@ -794,6 +800,7 @@ pub(crate) fn prepare_internal_worker_from_spec(
.map_err(|source| InternalWorkerSessionError::Build {
message: source.to_string(),
})?;
worker.install_compaction_budget(compaction_budget);
worker.disable_manifest_lifecycle_features();
if let Some(session) = inherited_workdir_session {
worker.bind_single_workdir_session(Some(session));
@@ -1442,6 +1449,7 @@ permission = "write"
input: "input".to_string(),
cache_key: Some("internal-test".to_string()),
max_turns: Some(1),
compaction_budget: None,
engine_configurator: None,
features: FeatureRegistryBuilder::new(),
required_tools,
crates/worker/src/ipc/interceptor.rs
+40 -4
View File
@@ -74,6 +74,7 @@ async fn call_hook_before_deadline<E: HookEventKind>(
}
pub(crate) struct WorkerInterceptor {
compaction_budget: Option<Arc<crate::compact::worker::CompactWorkerInterceptor>>,
registry: Arc<HookRegistry>,
compact_state: Option<Arc<CompactState>>,
/// Shared view of the cumulative UsageRecord timeline. Used with the
@@ -157,6 +158,7 @@ impl WorkerInterceptor {
pending_committed_history: Arc<Mutex<VecDeque<HistoryEntry<SessionHistoryMetadata>>>>,
) -> Self {
Self {
compaction_budget: None,
registry,
compact_state,
usage_history,
@@ -175,6 +177,14 @@ impl WorkerInterceptor {
}
}
pub(crate) fn with_compaction_budget(
mut self,
budget: Option<Arc<crate::compact::worker::CompactWorkerInterceptor>>,
) -> Self {
self.compaction_budget = budget;
self
}
pub(crate) fn with_notification_committer(
mut self,
committer: Option<Arc<dyn DurableNotificationCommitter>>,
@@ -552,6 +562,8 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
&self,
context: PreLlmRequestContext<'_, SessionHistoryMetadata>,
) -> InterceptorResult<PreRequestAction> {
let context_invocation = context.invocation;
let context_history = context.history;
let context = context.items;
let initial_tokens = self.estimated_tokens(context);
if let Some(action) =
@@ -616,7 +628,7 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
.expect("pending hook system-item queue poisoned"),
);
self.attach_prompt_provenance(&mut system_items);
let appended_items: Vec<Item> = system_items
let mut appended_items: Vec<Item> = system_items
.iter()
.map(SystemItem::to_history_item)
.collect();
@@ -652,10 +664,28 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
});
}
if let Some(usage_tracker) = self.usage_tracker.as_ref() {
usage_tracker.note_request(effective_context.len());
// Internal compaction safety must survive the normal Worker interceptor
// installation. Run it on the complete request after durable hook additions;
// Engine persists any returned warning through its normal append path.
if let Some(budget) = self.compaction_budget.as_ref() {
let mut budget_context = effective_context.into_owned();
match budget
.pre_llm_request(PreLlmRequestContext {
invocation: context_invocation,
items: &mut budget_context,
history: context_history,
})
.await?
{
PreRequestAction::Continue => {}
PreRequestAction::ContinueWith(warnings) => appended_items.extend(warnings),
action => return Ok(action),
}
}
if system_items.is_empty() {
if let Some(usage_tracker) = self.usage_tracker.as_ref() {
usage_tracker.note_request(context.len() + appended_items.len());
}
if appended_items.is_empty() {
return Ok(PreRequestAction::Continue);
}
Ok(match self.commit_system_items(&system_items) {
@@ -700,6 +730,12 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
if let Some(reason) = denials.into_iter().next() {
return Ok(HookPreToolAction::Deny(reason).into_worker_action(summary.call_id.clone()));
}
if let Some(budget) = self.compaction_budget.as_ref() {
match budget.pre_tool_call(info).await? {
PreToolAction::Continue => {}
action => return Ok(action),
}
}
self.tool_calls_this_turn.fetch_add(1, Ordering::Relaxed);
Ok(PreToolAction::Continue)
}
crates/worker/src/job.rs
+1
View File
@@ -221,6 +221,7 @@ impl PreparedInternalJob {
run_id: uuid::Uuid::now_v7(),
},
max_turns: engine_policy.max_turns.map(|n| n.get()),
compaction_budget: None,
engine_configurator: Some(Box::new(move |engine| {
crate::apply_worker_manifest(engine, &engine_policy);
})),
crates/worker/src/session_capture.rs
+57 -1
View File
@@ -725,7 +725,9 @@ fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance
}
WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
WorkerHistoryProvenance::DerivedSummary => SessionEntryProvenance::DerivedSummary,
WorkerHistoryProvenance::CompactionSummary | WorkerHistoryProvenance::DerivedSummary => {
SessionEntryProvenance::DerivedSummary
}
WorkerHistoryProvenance::LegacyUnknown => SessionEntryProvenance::LegacyUnknown,
}
}
@@ -878,6 +880,60 @@ mod tests {
);
}
#[test]
fn host_summary_remains_private_to_compaction_not_observation_or_memory_evidence() {
let entries = [
(
Item::system_message("host-summary-secret"),
WorkerHistoryProvenance::CompactionSummary,
),
(
Item::system_message("other-derived-secret"),
WorkerHistoryProvenance::DerivedSummary,
),
(
Item::system_message("raw-system-secret"),
WorkerHistoryProvenance::LegacyUnknown,
),
(
Item::reasoning("reasoning-secret"),
WorkerHistoryProvenance::LegacyUnknown,
),
]
.into_iter()
.map(|(item, origin)| crate::session_history::history_entry(item, origin))
.collect::<Vec<_>>();
let refs = entries
.iter()
.map(|entry| SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id))
.collect::<Vec<_>>();
let capture = SessionCapture::from_history_entries("segment", entries);
assert!(capture.overview().is_empty());
assert!(
capture
.search(&SearchOptions {
query: "secret".into(),
..SearchOptions::default()
})
.is_empty()
);
assert!(
capture
.read(ReadSelector::EntryRange([0, 3]), ReadOptions::default())
.entries
.is_empty()
);
for reference in refs {
assert!(capture.evidence_for(reference.as_str()).is_none());
assert!(
capture
.read(ReadSelector::Id(reference.as_str()), ReadOptions::default())
.entries
.is_empty()
);
}
}
#[test]
fn overview_contains_user_and_assistant_only() {
let view = SessionCapture::new(
crates/worker/src/session_history.rs
+100 -1
View File
@@ -17,6 +17,8 @@ pub type WorkerHistoryProvenance = LoggedSessionHistoryOrigin;
pub type SessionHistoryDerivation = LoggedHistoryDerivation;
pub type WorkerSubjectSnapshot = LoggedWorkerSubject;
pub(crate) const COMPACTION_SUMMARY_HEADER: &str = "[Compacted context summary]\n\n";
#[cfg(test)]
pub(crate) fn test_logged_history_entry(item: impl Into<Item>) -> LoggedHistoryEntry {
LoggedHistoryEntry {
@@ -108,8 +110,32 @@ pub(crate) fn restore_history_entries(
let mut history = Vec::new();
for entry in entries {
match entry {
LogEntry::AnnotatedSegmentStart { history: seed, .. } => {
LogEntry::AnnotatedSegmentStart {
history: seed,
compacted_from,
..
} => {
history = seed.iter().map(from_logged).collect();
// Older Host compactions always put this DerivedSummary system item
// first in the replacement seed. Only that trusted construction boundary
// may interpret the marker; never infer origin from appended/pasted text.
if compacted_from.is_some()
&& let Some(first) = history.first_mut()
&& first.annotation.origin == WorkerHistoryProvenance::DerivedSummary
&& matches!(
&first.item,
Item::Message {
role: agen::Role::System,
..
}
)
&& first
.item
.as_text()
.is_some_and(|text| text.starts_with(COMPACTION_SUMMARY_HEADER))
{
first.annotation.origin = WorkerHistoryProvenance::CompactionSummary;
}
}
LogEntry::AnnotatedUserInput { history: input, .. } => {
history.extend(input.iter().map(from_logged))
@@ -144,6 +170,79 @@ mod tests {
use protocol::Segment;
use session_store::LogEntry;
#[test]
fn restore_recognizes_only_legacy_host_compaction_seed_summary() {
let summary = history_entry(
Item::system_message(format!("{COMPACTION_SUMMARY_HEADER}host summary")),
WorkerHistoryProvenance::DerivedSummary,
);
for compacted in [false, true] {
let seed = LogEntry::AnnotatedSegmentStart {
ts: 0,
session_id: SessionId::now_v7(),
system_prompt: None,
config: RequestConfig::default(),
history: vec![to_logged_history_entry(&summary)],
forked_from: None,
compacted_from: compacted.then(|| session_store::SegmentOrigin {
segment_id: SegmentId::now_v7(),
at_turn_index: 0,
}),
};
let appended = LogEntry::AnnotatedUserInput {
ts: 1,
segments: vec![],
extensions: vec![],
history: vec![to_logged_history_entry(&summary)],
};
let restored = restore_history_entries(
SessionId::now_v7(),
SegmentId::now_v7(),
&[seed, appended],
)
.unwrap();
assert_eq!(
restored[0].annotation.origin,
if compacted {
WorkerHistoryProvenance::CompactionSummary
} else {
WorkerHistoryProvenance::DerivedSummary
}
);
assert_eq!(
restored[1].annotation.origin,
WorkerHistoryProvenance::DerivedSummary
);
assert_eq!(restored[0].annotation.entry_id, summary.annotation.entry_id);
}
for item in [
Item::user_message(format!("{COMPACTION_SUMMARY_HEADER}pasted summary")),
Item::system_message("[Session TaskStore snapshot] tasks"),
] {
let seed = LogEntry::AnnotatedSegmentStart {
ts: 0,
session_id: SessionId::now_v7(),
system_prompt: None,
config: RequestConfig::default(),
history: vec![to_logged_history_entry(&history_entry(
item,
WorkerHistoryProvenance::DerivedSummary,
))],
forked_from: None,
compacted_from: Some(session_store::SegmentOrigin {
segment_id: SegmentId::now_v7(),
at_turn_index: 0,
}),
};
let restored =
restore_history_entries(SessionId::now_v7(), SegmentId::now_v7(), &[seed]).unwrap();
assert_eq!(
restored[0].annotation.origin,
WorkerHistoryProvenance::DerivedSummary
);
}
}
#[test]
fn typed_flow_and_unknown_caller_input_round_trip_without_role_inference() {
let session_id = SessionId::now_v7();
crates/worker/src/worker.rs
+205 -20
View File
@@ -2801,6 +2801,7 @@ pub struct Worker<C: LlmClient, St: Store> {
/// manifest-derived lifecycle Features before controller startup.
manifest_lifecycle_features_enabled: bool,
interceptor_installed: bool,
compaction_service_budget: Option<Arc<crate::compact::worker::CompactWorkerInterceptor>>,
/// Shared compaction state (present when threshold is configured).
compact_state: Option<Arc<CompactState>>,
/// Per-LLM-request Usage tracker. Always present after construction.
@@ -3145,6 +3146,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
feature_invocations: FeatureInvocationRegistry::default(),
manifest_lifecycle_features_enabled: true,
interceptor_installed: false,
compaction_service_budget: None,
compact_state: None,
usage_tracker: Arc::new(UsageTracker::new()),
metrics_tracker: Arc::new(crate::compact::metrics_tracker::MetricsTracker::new()),
@@ -4221,6 +4223,24 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
self.callback_socket.as_ref()
}
pub(crate) fn install_compaction_budget(
&mut self,
budget: Option<crate::compact::worker::CompactWorkerInterceptor>,
) {
self.assert_hooks_open();
self.compaction_service_budget = budget.map(|budget| {
Arc::new(
budget.with_request_shape(
self.engine()
.get_system_prompt()
.unwrap_or_default()
.to_owned(),
self.engine().tool_server_handle(),
),
)
});
}
// --- Hook registration ---
fn assert_hooks_open(&self) {
@@ -4316,6 +4336,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
self.log_writer.clone(),
self.pending_committed_history.clone(),
)
.with_compaction_budget(self.compaction_service_budget.clone())
.with_notification_committer(self.notification_committer.clone())
.with_usage_tracker(self.usage_tracker.clone())
.with_metrics_tracker(self.metrics_tracker.clone())
@@ -6610,7 +6631,15 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
let history_entries = self.session.history().entries();
let retain_from = cut.index.min(history_entries.len());
let retained_history_entries = history_entries[retain_from..].to_vec();
let initial_summary = initial_compaction_summary(history_entries)?;
// The old summary is consumed in full by the compaction service even when
// the cut falls before it. The replacement summary supersedes it: retaining
// it as well would duplicate inherited context on every compact.
let retained_history_entries = history_entries[retain_from..]
.iter()
.filter(|entry| entry.annotation.origin != WorkerHistoryProvenance::CompactionSummary)
.cloned()
.collect::<Vec<_>>();
let mut retained_items = retained_history_entries
.iter()
.map(|entry| entry.item.clone())
@@ -6623,7 +6652,11 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
project_compaction_attachments(&mut retained_items, default_workdir_alias.as_deref());
// The compactor explores summaries, text details, and ViewImage call arguments. It
// never needs durable image bodies; relevant images are nominated with add_reference.
let mut entries_to_summarise = history_entries[..retain_from].to_vec();
let mut entries_to_summarise = history_entries[..retain_from]
.iter()
.filter(|entry| entry.annotation.origin != WorkerHistoryProvenance::CompactionSummary)
.cloned()
.collect::<Vec<_>>();
project_compactor_capture_attachments(&mut entries_to_summarise);
let items_to_summarise = entries_to_summarise
.iter()
@@ -6688,11 +6721,12 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
Vec::new()
};
// Input text fed to the compact worker. Includes the default
// references, current TaskStore snapshot, current TaskStore snapshot, and the (pruned) conversation text.
// The full Host summary is mandatory input, separate from the bounded
// prefix overview and current TaskStore snapshot.
let task_snapshot_text = self.task_feature.snapshot_text();
let summary_input = build_summary_input(
&items_to_summarise,
initial_summary.as_deref(),
&default_refs,
Some(task_snapshot_text.as_str()),
SummaryInputOptions {
@@ -6804,12 +6838,12 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
input: summary_input.text.clone(),
cache_key: Some(self.segment_id().to_string()),
max_turns: worker_max_turns,
compaction_budget: Some(interceptor),
engine_configurator: Some(Box::new(move |engine| {
let tracker = tracker_for_engine;
engine.on_usage(move |event| {
tracker.record_usage(event);
});
engine.set_interceptor(interceptor);
})),
features,
required_tools,
@@ -6869,7 +6903,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
Some(
"You have not called `write_summary` yet. Deliver the structured \
summary now (Completed Tasks / Active Task / Key Decisions / \
User Directives / Current Work) and nominate any files the next \
User Directives / Current Work), preserving ongoing goals and user \
directives from the initial summary and segment. Nominate any files the next \
session needs with `mark_read_required` for UTF-8 text or \
`add_reference` for images, binary files, and on-demand reads."
.to_string(),
@@ -6901,11 +6936,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.ok_or(WorkerError::CompactSummaryMissing)?;
let mut summary_tokens = estimate_text_tokens(summary_text.len());
if summary_max_tokens > 0 && summary_tokens > summary_max_tokens {
let prompt = format!(
"Your `write_summary` output is too large (≈{summary_tokens} tokens; max \
{summary_max_tokens}). Rewrite it now with `write_summary`, preserving the \
same five sections but making it concise. Target ≈{summary_target_tokens} tokens."
);
let prompt =
summary_shrink_prompt(summary_tokens, summary_max_tokens, summary_target_tokens);
handle
.send(prompt)
.await
@@ -7014,8 +7046,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
);
let mut compact_introduced_system_messages =
Vec::with_capacity(2 + auto_read_messages.len() + reference_message.is_some() as usize);
let summary_message =
Item::system_message(format!("[Compacted context summary]\n\n{summary_text}"));
let summary_message = Item::system_message(format!(
"{}{summary_text}",
crate::session_history::COMPACTION_SUMMARY_HEADER
));
compact_introduced_system_messages.push(summary_message.clone());
compact_introduced_system_messages.extend(auto_read_messages.iter().cloned());
if let Some(msg) = reference_message.as_ref() {
@@ -7072,8 +7106,19 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
let compacted_history_entries = new_history
.iter()
.cloned()
.map(|item| {
if let Some((offset, original)) = original_entries[original_cursor..]
.enumerate()
.map(|(index, item)| {
if index == 0 {
HistoryEntry::new(
item,
new_history_metadata(
WorkerHistoryProvenance::CompactionSummary,
Some(SessionHistoryDerivation {
sources: derived_sources.clone(),
}),
),
)
} else if let Some((offset, original)) = original_entries[original_cursor..]
.iter()
.enumerate()
.find(|(_, original)| original.item == item)
@@ -7489,6 +7534,7 @@ where
feature_invocations: FeatureInvocationRegistry::default(),
manifest_lifecycle_features_enabled: true,
interceptor_installed: false,
compaction_service_budget: None,
compact_state: None,
usage_tracker: Arc::new(UsageTracker::new()),
metrics_tracker: Arc::new(crate::compact::metrics_tracker::MetricsTracker::new()),
@@ -7828,6 +7874,7 @@ where
feature_invocations: FeatureInvocationRegistry::default(),
manifest_lifecycle_features_enabled: false,
interceptor_installed: false,
compaction_service_budget: None,
compact_state: None,
usage_tracker: Arc::new(UsageTracker::new()),
metrics_tracker: Arc::new(crate::compact::metrics_tracker::MetricsTracker::new()),
@@ -7957,6 +8004,7 @@ where
feature_invocations: FeatureInvocationRegistry::default(),
manifest_lifecycle_features_enabled: true,
interceptor_installed: false,
compaction_service_budget: None,
compact_state: None,
usage_tracker: Arc::new(UsageTracker::new()),
metrics_tracker: Arc::new(crate::compact::metrics_tracker::MetricsTracker::new()),
@@ -8342,6 +8390,7 @@ where
feature_invocations: FeatureInvocationRegistry::default(),
manifest_lifecycle_features_enabled: true,
interceptor_installed: false,
compaction_service_budget: None,
compact_state: None,
usage_tracker: Arc::new(UsageTracker::new()),
metrics_tracker: Arc::new(crate::compact::metrics_tracker::MetricsTracker::new()),
@@ -8740,6 +8789,61 @@ fn build_related_file_message(
)))
}
fn summary_shrink_prompt(
summary_tokens: u64,
summary_max_tokens: u64,
summary_target_tokens: u64,
) -> String {
format!(
"Your `write_summary` output is too large (≈{summary_tokens} tokens; max \
{summary_max_tokens}). Rewrite it now with `write_summary`, preserving the \
same five sections. Compress duplicate explanations and old work details \
first; preserve ongoing goals and user directives from the initial summary \
and segment, including scope, prohibitions, completion conditions and \
exception-reporting duties. Apply explicit later changes/retractions and \
retain uncertainty; do not replace the original request with recovery \
procedures or recent progress. Target ≈{summary_target_tokens} tokens."
)
}
/// Select only the Host's typed summary, never a marker in ordinary content.
/// Invalid/duplicate Host summaries are a broken seed contract, not permission
/// to silently omit inherited instructions.
fn initial_compaction_summary(
entries: &[HistoryEntry<SessionHistoryMetadata>],
) -> Result<Option<String>, WorkerError> {
let mut summaries = entries
.iter()
.filter(|entry| entry.annotation.origin == WorkerHistoryProvenance::CompactionSummary);
let Some(entry) = summaries.next() else {
return Ok(None);
};
if summaries.next().is_some() {
return Err(WorkerError::InvalidState(
"multiple Host compaction summaries in active history".into(),
));
}
if !matches!(
&entry.item,
Item::Message {
role: agen::Role::System,
..
}
) {
return Err(WorkerError::InvalidState(
"Host compaction summary must be a system message".into(),
));
}
let body = entry
.item
.as_text()
.and_then(|text| text.strip_prefix(crate::session_history::COMPACTION_SUMMARY_HEADER))
.ok_or_else(|| {
WorkerError::InvalidState("Host compaction summary header is missing".into())
})?;
Ok(Some(body.to_string()))
}
#[derive(Debug, Clone, Copy)]
struct SummaryInputOptions {
overview_target_tokens: u64,
@@ -8756,11 +8860,12 @@ struct SummaryInputBuild {
deadline_fallback_used: bool,
}
/// Build the compact worker's input: default-reference instructions,
/// the list of recently-touched files, task snapshot,
/// and a bounded overview rather than a prefix-wide transcript.
/// Build mandatory full Host-summary input plus a bounded prefix overview,
/// recently-touched files and current tasks. Overview budgets never apply to
/// the summary; the complete request is checked by the compaction interceptor.
fn build_summary_input(
items: &[Item],
initial_summary: Option<&str>,
default_refs: &[PathBuf],
task_snapshot: Option<&str>,
options: SummaryInputOptions,
@@ -8786,10 +8891,20 @@ fn build_summary_input(
out.push_str(&format!(
"Summarise this session into a structured summary of about {} tokens and \
nominate files the next session needs. The conversation below is a \
bounded overview/index, not the full transcript. Use tools to inspect \
bounded overview/index, not the full transcript. A separately supplied initial \
compaction summary is complete, mandatory source material. Preserve ongoing \
user goals and directives from it and from this segment's user messages; \
confirm important wording with exploration rather than reconstructing the \
request from recent progress. Retained tail entries remain verbatim in the \
next session and are outside this prefix overview. Use tools to inspect \
current files when deciding auto-read/reference output.\n\n",
options.summary_target_tokens
));
if let Some(summary) = initial_summary {
out.push_str("## Initial compaction summary (full text)\n");
out.push_str(summary);
out.push_str("\n\n## End initial compaction summary\n\n");
}
if !default_refs.is_empty() {
out.push_str(
"These files were touched recently in this session. Use `read_file` \
@@ -8808,7 +8923,8 @@ fn build_summary_input(
out.push_str(
"## Current Session TaskStore\n\
This is the full current task list. Use it as source material for the \
summary, especially active (pending/inprogress) tasks, but do not edit tasks \
summary, especially active (pending/inprogress) tasks, but not as a \
replacement for user goals, directives or completion conditions. Do not edit tasks \
from the compact worker.\n",
);
out.push_str(task_snapshot);
@@ -10103,6 +10219,7 @@ mod build_summary_prompt_tests {
fn test_summary_input(items: &[Item]) -> String {
build_summary_input(
items,
None,
&[],
None,
SummaryInputOptions {
@@ -10115,6 +10232,72 @@ mod build_summary_prompt_tests {
.text
}
#[test]
fn compact_prompt_and_shrink_policy_prioritize_continuing_user_intent() {
// The builtin resource is the default model-visible compaction contract.
let prompt = include_str!("../../../resources/prompts/internal/compact_system.md");
for policy in [
"initial compaction summary in full",
"segment's user messages",
"prohibitions",
"completion conditions",
"explicit changes or retractions",
"Preserve uncertainty",
"retained tail stays verbatim",
"TaskStore",
] {
assert!(prompt.contains(policy), "missing prompt policy: {policy}");
}
let shrink = summary_shrink_prompt(2_000, 1_000, 500);
for policy in [
"initial summary",
"and segment",
"Compress duplicate",
"prohibitions",
"completion conditions",
"exception-reporting duties",
"changes/retractions",
"uncertainty",
"recovery",
] {
assert!(shrink.contains(policy), "missing shrink policy: {policy}");
}
let input = test_summary_input(&[Item::user_message("keep original scope")]);
assert!(input.contains("Preserve ongoing user goals and directives"));
assert!(input.contains("Retained tail entries remain verbatim"));
}
#[test]
fn malformed_or_duplicate_host_summaries_fail_instead_of_omitting_context() {
use crate::session_history::history_entry;
let valid = history_entry(
Item::system_message(format!(
"{}body",
crate::session_history::COMPACTION_SUMMARY_HEADER
)),
WorkerHistoryProvenance::CompactionSummary,
);
assert_eq!(
initial_compaction_summary(std::slice::from_ref(&valid))
.unwrap()
.as_deref(),
Some("body")
);
assert!(initial_compaction_summary(&[valid.clone(), valid]).is_err());
for item in [
Item::user_message("pasted"),
Item::system_message("missing marker"),
] {
assert!(
initial_compaction_summary(&[history_entry(
item,
WorkerHistoryProvenance::CompactionSummary
)])
.is_err()
);
}
}
#[test]
fn retained_images_become_on_demand_related_files() {
let mut items = vec![
@@ -10215,6 +10398,7 @@ mod build_summary_prompt_tests {
let items = vec![Item::user_message("x".repeat(4_000))];
let built = build_summary_input(
&items,
None,
&[],
None,
SummaryInputOptions {
@@ -10234,6 +10418,7 @@ mod build_summary_prompt_tests {
let items = vec![Item::user_message("x".repeat(4_000))];
let built = build_summary_input(
&items,
None,
&[],
None,
SummaryInputOptions {
crates/worker/tests/compact_events_test.rs
+726 -1
View File
@@ -133,6 +133,7 @@ struct MockClient {
responses: Arc<Vec<Vec<LlmEvent>>>,
call_count: Arc<AtomicUsize>,
observer: Arc<Mutex<Option<CallObserver>>>,
requests: Arc<Mutex<Vec<Request>>>,
}
impl MockClient {
@@ -141,9 +142,14 @@ impl MockClient {
responses: Arc::new(responses),
call_count: Arc::new(AtomicUsize::new(0)),
observer: Arc::new(Mutex::new(None)),
requests: Arc::new(Mutex::new(Vec::new())),
}
}
fn requests(&self) -> Vec<Request> {
self.requests.lock().unwrap().clone()
}
fn set_observer(&self, observer: impl Fn(usize) + Send + Sync + 'static) {
*self.observer.lock().unwrap() = Some(Arc::new(observer));
}
@@ -157,9 +163,10 @@ impl LlmClient for MockClient {
async fn stream(
&self,
_request: Request,
request: Request,
) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
{
self.requests.lock().unwrap().push(request);
let count = self.call_count.fetch_add(1, Ordering::SeqCst);
if let Some(observer) = self.observer.lock().unwrap().clone() {
observer(count);
@@ -2065,3 +2072,721 @@ async fn controller_compact_method_publishes_progress_and_clear() {
let command = protocol::WorkerCommandEnvelope::new(2);
let _ = handle.send(Method::Shutdown { command }).await;
}
const INITIAL_SUMMARY_HEADING: &str = "## Initial compaction summary (full text)\n";
const OVERVIEW_HEADING: &str = "## Conversation overview/index\n";
const SUMMARY_MARKER: &str = "[Compacted context summary]\n\n";
// These tests prove transport and durable replacement policy, not whether a live
// model preserves meaning. All write_summary outputs below are predetermined.
struct SummaryInputFixture {
worker: Worker<MockClient, TestStore>,
_store_dir: tempfile::TempDir,
}
impl SummaryInputFixture {
async fn new(manifest_toml: &str, client: MockClient) -> Self {
let store_dir = tempfile::tempdir().unwrap();
let store = CombinedStore::new(
FsStore::new(store_dir.path()).unwrap(),
FsWorkerStore::new(store_dir.path().join("pods")).unwrap(),
);
let engine =
Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
client,
);
let mut worker = Worker::new(
worker::WorkerManifest::from_toml(manifest_toml).unwrap(),
engine,
store,
worker::WorkerWorkspaceContext::no_workspace(),
worker::WorkerFilesystemAuthority::None,
worker::Scope::empty(),
)
.await
.unwrap();
worker.enable_worker_metadata_write_through().unwrap();
worker.attach_log_writer(Arc::new(worker.log_writer_handle()));
Self {
worker,
_store_dir: store_dir,
}
}
// Seed the persisted boundary, then exercise the public rewind API's shared
// typed history restore path. No production client or global allocation is
// needed. This is restoration coverage, not a cold Runtime restart test.
async fn restore_seed(
&mut self,
history: Vec<session_store::LoggedHistoryEntry>,
compacted: bool,
) {
self.worker
.run_text("fixture rewind boundary")
.await
.unwrap();
let session_id = self.worker.session_id();
let segment_id = self.worker.segment_id();
let mut records = self
.worker
.store()
.read_all(session_id, segment_id)
.unwrap();
let start = records
.iter_mut()
.find(|entry| matches!(entry, LogEntry::AnnotatedSegmentStart { .. }))
.unwrap();
match start {
LogEntry::AnnotatedSegmentStart {
history: seed,
compacted_from,
..
} => {
*seed = history;
*compacted_from = compacted.then(|| session_store::SegmentOrigin {
segment_id: session_store::new_segment_id(),
at_turn_index: 0,
});
}
_ => unreachable!(),
}
self.worker
.store()
.create_segment(session_id, segment_id, &records)
.unwrap();
let (head, targets) = self.worker.list_rewind_targets().unwrap();
self.worker
.rewind_to(targets[0].id.clone(), head)
.await
.unwrap();
}
}
fn summary_entry(
item: Item,
origin: session_store::LoggedSessionHistoryOrigin,
) -> session_store::LoggedHistoryEntry {
let mut entry = annotated(item);
entry.metadata.origin = origin;
entry
}
fn long_summary(label: &str) -> String {
format!(
"## Completed Tasks\n{label}: investigated the transport boundary.\n\n\
## Active Task\nKeep the original user's completion conditions.\n{}\n\
## Key Decisions\nDo not infer provenance from pasted markers.\n\n\
## User Directives\n{label}_LOAD_BEARING_SENTINEL_AT_END\n\n\
## Current Work\nContinue the requested implementation.",
"Historical constraints must remain explicit, including the original acceptance criteria.\n"
.repeat(70)
)
}
fn recent_history(count: usize) -> Vec<session_store::LoggedHistoryEntry> {
(0..count)
.flat_map(|index| {
[
annotated(Item::user_message(format!(
"recent-user-{index}: {}",
"recent progress is not a replacement for the original directive. ".repeat(40)
))),
annotated(Item::assistant_message(format!("recent-assistant-{index}"))),
]
})
.collect()
}
fn first_compactor_input(request: &Request) -> String {
request
.items
.iter()
.find_map(|item| match item {
Item::Message {
role: agen::Role::User,
..
} => item
.as_text()
.filter(|text| text.contains(OVERVIEW_HEADING)),
_ => None,
})
.expect("the first compactor request must carry its overview input")
.to_string()
}
fn assert_full_initial_summary(input: &str, summary: &str) {
let heading_at = input
.find(INITIAL_SUMMARY_HEADING)
.expect("full summary heading");
let overview_at = input.find(OVERVIEW_HEADING).expect("overview heading");
let summary_at = heading_at + INITIAL_SUMMARY_HEADING.len();
assert!(summary_at < overview_at);
assert!(
input[summary_at..].starts_with(summary),
"summary must be verbatim"
);
assert_eq!(
input.matches(summary).count(),
1,
"summary must not be duplicated"
);
assert!(!input[overview_at..].contains(summary));
let opening = summary.chars().take(100).collect::<String>();
assert!(
!input[overview_at..].contains(&opening),
"overview must not carry even a clipped summary"
);
assert!(!input[overview_at..].contains("LOAD_BEARING_SENTINEL_AT_END"));
}
fn replacement_seed(
worker: &Worker<MockClient, TestStore>,
) -> Vec<session_store::LoggedHistoryEntry> {
worker
.store()
.read_all(worker.session_id(), worker.segment_id())
.unwrap()
.into_iter()
.find_map(|entry| match entry {
LogEntry::AnnotatedSegmentStart { history, .. } => Some(history),
_ => None,
})
.expect("replacement SegmentStart")
}
#[tokio::test]
async fn full_initial_summary_survives_small_overview_budgets_and_both_retention_extremes() {
use session_store::LoggedSessionHistoryOrigin::CompactionSummary;
let inherited = long_summary("original");
assert!(inherited.chars().count() > 4_000);
for retained in [0, 1_000_000] {
for (warning, deadline) in [(0, 0), (1, 0), (1, 1)] {
let manifest = MANUAL_ONLY_MANIFEST_TOML.replace(
"[compaction]",
&format!(
"[compaction]\noverview_target_tokens = 1\noverview_warning_tokens = {warning}\noverview_deadline_tokens = {deadline}"
),
);
let client = MockClient::new(vec![
single_text_events("fixture"),
write_summary_tool_use_events("replace", "new summary"),
single_text_events("done"),
]);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(&manifest, client).await;
let mut seed = vec![summary_entry(
Item::system_message(format!("{SUMMARY_MARKER}{inherited}")),
CompactionSummary,
)];
let recent = recent_history(40);
seed.extend(recent.clone());
fixture.restore_seed(seed, true).await;
let (tx, mut rx) = broadcast::channel::<Event>(64);
fixture.worker.attach_alerter(worker::Alerter::new(tx));
fixture.worker.compact(retained).await.unwrap();
let requests = observed.requests();
assert_eq!(requests.len(), 3);
let input = first_compactor_input(&requests[1]);
assert_full_initial_summary(&input, &inherited);
let alerts = drain(&mut rx);
let has_alert = |snippet: &str| {
alerts.iter().any(|event| {
matches!(event, Event::Alert(alert)
if alert.source == protocol::AlertSource::Compactor
&& alert.level == protocol::AlertLevel::Warn
&& alert.message.contains(snippet))
})
};
assert_eq!(has_alert("larger than expected"), warning != 0);
assert_eq!(has_alert("using coarse fallback"), deadline != 0);
let replacement = replacement_seed(&fixture.worker);
assert_eq!(
replacement
.iter()
.filter(|entry| entry.metadata.origin == CompactionSummary)
.count(),
1
);
assert!(replacement.iter().all(|entry| {
!Item::from(entry.item.clone())
.as_text()
.is_some_and(|text| text.contains(&inherited))
}));
for entry in &recent {
assert_eq!(
replacement
.iter()
.any(|retained_entry| retained_entry == entry),
retained != 0,
"ordinary retained entries must keep their item and metadata"
);
}
let metrics = session_metrics::read_session_metrics(
fixture.worker.store(),
fixture.worker.session_id(),
)
.unwrap();
let overview_tokens = metrics
.iter()
.find(|record| record.metric.name == "compact.overview_tokens")
.and_then(|record| record.metric.value)
.unwrap();
assert!(
overview_tokens < inherited.len().div_ceil(4) as f64,
"overview accounting must exclude the inherited full summary"
);
}
}
}
#[tokio::test]
async fn pasted_markers_and_other_derived_context_remain_ordinary_overview_material() {
use session_store::LoggedSessionHistoryOrigin::{
CompactionSummary, DerivedSummary, LegacyUnknown,
};
for retained in [0, 1_000_000] {
let client = MockClient::new(vec![
single_text_events("fixture"),
write_summary_tool_use_events("replace", "replacement"),
single_text_events("done"),
]);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(MANUAL_ONLY_MANIFEST_TOML, client).await;
let ordinary = vec![
summary_entry(
Item::system_message("[Session TaskStore snapshot]\nactive task"),
DerivedSummary,
),
summary_entry(
Item::system_message("auto-read file source material"),
DerivedSummary,
),
summary_entry(
Item::system_message(format!("{SUMMARY_MARKER}pasted system marker")),
LegacyUnknown,
),
summary_entry(
Item::user_message(format!("{SUMMARY_MARKER}pasted user marker")),
LegacyUnknown,
),
summary_entry(
Item::system_message(format!("{SUMMARY_MARKER}later derived marker")),
DerivedSummary,
),
];
fixture.restore_seed(ordinary.clone(), true).await;
assert!(
fixture
.worker
.session_history()
.entries()
.iter()
.all(|entry| entry.annotation.origin != CompactionSummary)
);
fixture.worker.compact(retained).await.unwrap();
let input = first_compactor_input(&observed.requests()[1]);
assert!(!input.contains(INITIAL_SUMMARY_HEADING));
let overview = input.split_once(OVERVIEW_HEADING).unwrap().1;
let replacement = replacement_seed(&fixture.worker);
for entry in &ordinary {
let text = Item::from(entry.item.clone())
.as_text()
.unwrap()
.to_string();
assert_eq!(
overview.contains(&text),
retained == 0,
"ordinary overview source: {text}"
);
assert_eq!(
replacement
.iter()
.any(|retained_entry| retained_entry == entry),
retained != 0,
"other derived context and pasted markers must follow ordinary retention"
);
}
}
}
#[tokio::test]
async fn restored_legacy_summary_requires_compacted_lineage_first_position_origin_role_and_marker()
{
use session_store::LoggedSessionHistoryOrigin::{
CompactionSummary, DerivedSummary, LegacyUnknown,
};
let inherited = long_summary("legacy");
// Each false case lacks one trusted construction condition; marker text
// alone must never promote ordinary context into inherited Host material.
for case in [
"trusted",
"no-lineage",
"not-first",
"unknown-origin",
"user-role",
"no-marker",
] {
let client = MockClient::new(vec![
single_text_events("fixture"),
write_summary_tool_use_events("replace", "replacement"),
single_text_events("done"),
]);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(MANUAL_ONLY_MANIFEST_TOML, client).await;
let text = if case == "no-marker" {
inherited.clone()
} else {
format!("{SUMMARY_MARKER}{inherited}")
};
let item = if case == "user-role" {
Item::user_message(text)
} else {
Item::system_message(text)
};
let origin = if case == "unknown-origin" {
LegacyUnknown
} else {
DerivedSummary
};
let candidate = summary_entry(item, origin);
let candidate_id = candidate.metadata.entry_id.clone();
let mut history = vec![candidate];
if case == "not-first" {
history.insert(
0,
annotated(Item::system_message("unrelated first system context")),
);
}
history.extend(recent_history(20));
fixture.restore_seed(history, case != "no-lineage").await;
let restored = fixture
.worker
.session_history()
.entries()
.iter()
.find(|entry| entry.annotation.entry_id == candidate_id)
.unwrap();
assert_eq!(
restored.annotation.origin == CompactionSummary,
case == "trusted",
"{case}"
);
// A restore-time upgrade is an in-memory interpretation of the trusted
// seed; it must not silently rewrite the historical durable record.
assert!(
replacement_seed(&fixture.worker)
.iter()
.all(|entry| entry.metadata.origin != CompactionSummary)
);
fixture.worker.compact(0).await.unwrap();
let input = first_compactor_input(&observed.requests()[1]);
if case == "trusted" {
assert_full_initial_summary(&input, &inherited);
} else {
assert!(!input.contains(INITIAL_SUMMARY_HEADING), "{case}");
}
}
}
#[tokio::test]
async fn three_chained_compactions_transport_each_complete_predecessor_without_retaining_old_summaries()
{
use session_store::LoggedSessionHistoryOrigin::CompactionSummary;
let original_goal =
"CHAIN_ORIGINAL_USER_GOAL: finish only after all explicit acceptance conditions hold";
// Every predetermined output keeps the original directive. The assertions
// prove that complete predecessors reach later requests, not that a model
// chose to preserve this directive on its own.
let summaries = (0..3)
.map(|index| {
format!(
"{}\n{original_goal}",
long_summary(&format!("generation-{index}"))
)
})
.collect::<Vec<_>>();
let initial = format!("{}\n{original_goal}", long_summary("initial"));
let mut responses = vec![single_text_events("seed response")];
for (index, summary) in summaries.iter().enumerate() {
responses.push(single_text_events(
"acknowledge new directive and temporary recovery",
));
responses.push(write_summary_tool_use_events(
&format!("summary-{index}"),
summary,
));
responses.push(single_text_events("done"));
}
let client = MockClient::new(responses);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(MANUAL_ONLY_MANIFEST_TOML, client).await;
fixture
.restore_seed(
vec![summary_entry(
Item::system_message(format!("{SUMMARY_MARKER}{initial}")),
CompactionSummary,
)],
true,
)
.await;
let original_session = fixture.worker.session_id();
let mut predecessor = fixture.worker.segment_id();
for (index, summary) in summaries.iter().enumerate() {
let new_directive = format!(
"NEW_USER_DIRECTIVE_{index}: do not merge. Temporarily recover the interrupted work and collect SubWorkers; this does not retract the original goal."
);
fixture.worker.run_text(&new_directive).await.unwrap();
let first_request = observed.requests().len();
fixture.worker.compact(0).await.unwrap();
let input = first_compactor_input(&observed.requests()[first_request]);
let prior = if index == 0 {
&initial
} else {
&summaries[index - 1]
};
assert_full_initial_summary(&input, prior);
assert!(input.contains(original_goal));
assert!(input.contains(&new_directive));
let seed = replacement_seed(&fixture.worker);
let inherited = seed
.iter()
.filter(|entry| entry.metadata.origin == CompactionSummary)
.collect::<Vec<_>>();
assert_eq!(inherited.len(), 1);
assert_eq!(
Item::from(inherited[0].item.clone()).as_text().unwrap(),
format!("{SUMMARY_MARKER}{summary}")
);
for old in &summaries[..index] {
assert!(
fixture
.worker
.history()
.iter()
.all(|item| !item.as_text().is_some_and(|text| text.contains(old)))
);
}
let records = fixture
.worker
.store()
.read_all(original_session, fixture.worker.segment_id())
.unwrap();
assert!(
records
.iter()
.any(|entry| matches!(entry, LogEntry::AnnotatedSegmentStart {
compacted_from: Some(origin), ..
} if origin.segment_id == predecessor))
);
let restored = session_store::restore(
fixture.worker.store(),
original_session,
fixture.worker.segment_id(),
)
.unwrap();
assert!(restored.history.iter().any(|item| {
item.as_text()
.is_some_and(|text| text == format!("{SUMMARY_MARKER}{summary}"))
}));
predecessor = fixture.worker.segment_id();
}
assert_eq!(observed.requests().len(), 10);
}
#[tokio::test]
async fn inherited_summary_counts_against_hard_input_limit_and_failure_does_not_activate() {
use session_store::LoggedSessionHistoryOrigin::CompactionSummary;
let manifest = MANUAL_ONLY_MANIFEST_TOML.replace(
"[compaction]",
"[compaction]\nworker_context_max_tokens = 10000\nfinish_warning_remaining_tokens = 0\nfinal_reserve_tokens = 0\noverview_target_tokens = 1",
);
for include_summary in [false, true] {
let client = MockClient::new(vec![
single_text_events("fixture"),
write_summary_tool_use_events("replace", "replacement"),
single_text_events("done"),
]);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(&manifest, client).await;
let summary =
"Preserve the user's goal and every explicit completion condition.\n".repeat(3_000);
assert!(
agen::token_counter::total_tokens(&[Item::user_message(&summary)], &[]).tokens > 10_000
);
let mut history = Vec::new();
if include_summary {
history.push(summary_entry(
Item::system_message(format!("{SUMMARY_MARKER}{summary}")),
CompactionSummary,
));
}
history.extend(recent_history(20));
fixture.restore_seed(history, true).await;
let source = fixture.worker.segment_id();
let source_history = fixture.worker.history();
let result = fixture.worker.compact(0).await;
if include_summary {
let error = result.expect_err("mandatory input over hard cap must fail, not truncate");
assert!(
matches!(error, WorkerError::CompactCancelled)
|| error.to_string().contains("input occupancy"),
"unexpected hard-cap failure: {error:?}"
);
assert_eq!(
observed.requests().len(),
1,
"no compactor request may cross the hard cap"
);
assert_eq!(fixture.worker.segment_id(), source);
assert_eq!(fixture.worker.history(), source_history);
assert_eq!(
fixture
.worker
.store()
.read_by_name("test-worker")
.unwrap()
.unwrap()
.active
.unwrap()
.segment_id,
Some(source)
);
assert_eq!(
fixture
.worker
.store()
.list_segments(fixture.worker.session_id())
.unwrap()
.len(),
1
);
} else {
result.expect("the same bounded overview without inherited input fits the cap");
assert_eq!(observed.requests().len(), 3);
assert_ne!(fixture.worker.segment_id(), source);
}
}
}
#[tokio::test]
async fn summary_output_still_over_hard_max_after_rewrite_fails_without_activation() {
let oversized = long_summary("oversized");
let manifest = MANUAL_ONLY_MANIFEST_TOML.replace(
"[compaction]",
"[compaction]\nsummary_target_tokens = 50\nsummary_max_tokens = 100",
);
let client = MockClient::new(vec![
single_text_events("seed"),
write_summary_tool_use_events("first-output", &oversized),
single_text_events("done"),
write_summary_tool_use_events("rewrite-output", &oversized),
single_text_events("done"),
]);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(&manifest, client).await;
fixture.worker.run_text("user goal").await.unwrap();
let source = fixture.worker.segment_id();
let history = fixture.worker.history();
let error = fixture.worker.compact(0).await.unwrap_err();
assert!(matches!(
error,
WorkerError::CompactSummaryTooLarge { max: 100, .. }
));
assert_eq!(fixture.worker.segment_id(), source);
assert_eq!(fixture.worker.history(), history);
assert_eq!(
fixture
.worker
.store()
.read_by_name("test-worker")
.unwrap()
.unwrap()
.active
.unwrap()
.segment_id,
Some(source)
);
assert_eq!(
fixture
.worker
.store()
.list_segments(fixture.worker.session_id())
.unwrap()
.len(),
1
);
assert!(
observed
.requests()
.iter()
.any(|request| request.items.iter().any(|item| item
.as_text()
.is_some_and(|text| text.contains("Rewrite it now with `write_summary`"))))
);
}
#[tokio::test]
async fn budget_warning_is_durable_and_keeps_full_summary_in_first_model_request() {
use session_store::LoggedSessionHistoryOrigin::CompactionSummary;
let manifest = MANUAL_ONLY_MANIFEST_TOML.replace(
"[compaction]",
"[compaction]\nworker_context_max_tokens = 10000\nfinish_warning_remaining_tokens = 9999\nfinal_reserve_tokens = 9999",
);
let client = MockClient::new(vec![
single_text_events("fixture"),
write_summary_tool_use_events("summary", "replacement"),
single_text_events("done"),
]);
let observed = client.clone();
let mut fixture = SummaryInputFixture::new(&manifest, client).await;
let inherited = long_summary("budget-warning");
fixture
.restore_seed(
vec![summary_entry(
Item::system_message(format!("{SUMMARY_MARKER}{inherited}")),
CompactionSummary,
)],
true,
)
.await;
let (tx, mut rx) = broadcast::channel::<Event>(128);
fixture.worker.attach_alerter(worker::Alerter::new(tx));
fixture.worker.compact(0).await.unwrap();
let requests = observed.requests();
assert_full_initial_summary(&first_compactor_input(&requests[1]), &inherited);
let warning = requests[1]
.items
.iter()
.filter_map(Item::as_text)
.find(|text| text.starts_with("[Compact worker budget warning]"))
.expect("normal internal Worker must preserve the budget interceptor warning");
assert!(warning.contains("full initial summary and segment"));
assert!(warning.contains("changes/retractions"));
assert!(
requests[2]
.items
.iter()
.filter_map(Item::as_text)
.any(|text| text == warning),
"warning must remain in history, not be a transient request insertion"
);
assert!(
drain(&mut rx)
.iter()
.any(|event| matches!(event, Event::Alert(alert)
if alert.message.contains("compact worker context budget is low")))
);
}