| 1 | //! Engine-owned compaction lifecycle, recovery and checkpoint installation. |
| 2 | //! Automatic, manual and emergency paths share the existing session and event authority. |
| 3 | |
| 4 | use super::*; |
| 5 | |
| 6 | pub(super) struct CompactionPass { |
| 7 | pub trigger: &'static str, |
| 8 | pub path: crate::compaction::CompactionPath, |
| 9 | pub tokens_before: usize, |
| 10 | pub threshold_tokens: usize, |
| 11 | pub usage: Usage, |
| 12 | } |
| 13 | |
| 14 | impl Engine { |
| 15 | pub(super) async fn emit_compaction_started( |
| 16 | &mut self, |
| 17 | id: String, |
| 18 | auto: bool, |
| 19 | message: String, |
| 20 | ) { |
| 21 | let _ = self |
| 22 | .tx_event |
| 23 | .send(Event::CompactionStarted { id, auto, message }) |
| 24 | .await; |
| 25 | } |
| 26 | |
| 27 | pub(super) async fn emit_compaction_completed( |
| 28 | &mut self, |
| 29 | id: String, |
| 30 | auto: bool, |
| 31 | message: String, |
| 32 | messages_before: Option<usize>, |
| 33 | messages_after: Option<usize>, |
| 34 | pass: CompactionPass, |
| 35 | ) { |
| 36 | let summary_prompt = self.rendered_compaction_summary(); |
| 37 | // Every call site runs after message replacement and checkpoint |
| 38 | // commit. Reuse the same complete estimate as context pressure. |
| 39 | let post_input_tokens = Some(self.estimated_input_tokens() as u64); |
| 40 | let reduction_ratio = messages_before |
| 41 | .zip(messages_after) |
| 42 | .filter(|(before, _)| *before > 0) |
| 43 | .map(|(before, after)| 1.0 - after as f64 / before as f64); |
| 44 | self.record_compaction_event( |
| 45 | "compaction.completed", |
| 46 | serde_json::json!({ |
| 47 | "compaction_id": id, |
| 48 | "trigger": pass.trigger, |
| 49 | "path": match pass.path { |
| 50 | crate::compaction::CompactionPath::Summary => "summary", |
| 51 | crate::compaction::CompactionPath::PruneOnly => "pruning_only", |
| 52 | }, |
| 53 | "messages_before": messages_before, |
| 54 | "messages_after": messages_after, |
| 55 | "estimated_tokens_before": pass.tokens_before, |
| 56 | "estimated_tokens_after": post_input_tokens, |
| 57 | "threshold_tokens": pass.threshold_tokens, |
| 58 | "summarizer_usage": pass.usage, |
| 59 | "reduction_ratio": reduction_ratio, |
| 60 | }), |
| 61 | ) |
| 62 | .await; |
| 63 | let _ = self |
| 64 | .tx_event |
| 65 | .send(Event::CompactionCompleted { |
| 66 | id, |
| 67 | auto, |
| 68 | message, |
| 69 | messages_before, |
| 70 | messages_after, |
| 71 | summary_prompt, |
| 72 | post_input_tokens, |
| 73 | }) |
| 74 | .await; |
| 75 | } |
| 76 | |
| 77 | /// One audit producer shared by interactive, headless and Runtime hosts. |
| 78 | /// No transcript content or credentials enter this diagnostic record. |
| 79 | pub(super) async fn record_compaction_event( |
| 80 | &self, |
| 81 | event: &'static str, |
| 82 | mut details: serde_json::Value, |
| 83 | ) { |
| 84 | details["session_id"] = serde_json::json!(self.session.id); |
| 85 | details["thread_id"] = serde_json::json!(self.config.runtime_services.active_thread_id); |
| 86 | details["model"] = serde_json::json!(self.config.model); |
| 87 | if let Err(error) = tokio::task::spawn_blocking(move || { |
| 88 | crate::audit::log_sensitive_event(event, details); |
| 89 | }) |
| 90 | .await |
| 91 | { |
| 92 | tracing::warn!(%error, "compaction audit writer failed"); |
| 93 | } |
| 94 | } |
| 95 | |
| 96 | pub(super) async fn emit_compaction_cancelled( |
| 97 | &mut self, |
| 98 | id: String, |
| 99 | auto: bool, |
| 100 | message: String, |
| 101 | ) { |
| 102 | let _ = self |
| 103 | .tx_event |
| 104 | .send(Event::CompactionCancelled { id, auto, message }) |
| 105 | .await; |
| 106 | } |
| 107 | |
| 108 | /// Render the accumulated compaction summary prompt to plain text so it |
| 109 | /// can travel in events and be persisted by host layers. All emit sites |
| 110 | /// run after `commit_compaction_checkpoint`, so this reflects the checkpoint |
| 111 | /// state the engine will use for subsequent requests. |
| 112 | pub(super) fn rendered_compaction_summary(&self) -> Option<String> { |
| 113 | self.session |
| 114 | .compaction_summary_prompt |
| 115 | .as_ref() |
| 116 | .map(|prompt| match prompt { |
| 117 | SystemPrompt::Text(text) => text.clone(), |
| 118 | SystemPrompt::Blocks(blocks) => blocks |
| 119 | .iter() |
| 120 | .map(|block| block.text.as_str()) |
| 121 | .collect::<Vec<_>>() |
| 122 | .join("\n\n"), |
| 123 | }) |
| 124 | .filter(|text| !text.trim().is_empty()) |
| 125 | } |
| 126 | |
| 127 | pub(super) async fn emit_compaction_failed(&mut self, id: String, auto: bool, message: String) { |
| 128 | let _ = self |
| 129 | .tx_event |
| 130 | .send(Event::CompactionFailed { id, auto, message }) |
| 131 | .await; |
| 132 | } |
| 133 | |
| 134 | pub(super) fn claim_compaction(&self, id: &str) -> Option<CancellationToken> { |
| 135 | self.compaction_cancellation |
| 136 | .lock() |
| 137 | .unwrap_or_else(std::sync::PoisonError::into_inner) |
| 138 | .claim(id) |
| 139 | } |
| 140 | |
| 141 | pub(super) fn finish_compaction(&self, id: &str) { |
| 142 | self.compaction_cancellation |
| 143 | .lock() |
| 144 | .unwrap_or_else(std::sync::PoisonError::into_inner) |
| 145 | .finish(id); |
| 146 | } |
| 147 | |
| 148 | /// Pressure and effective trigger in append-only turn metadata. Numeric |
| 149 | /// estimates never modify the session-pinned system/tool prefix. |
| 150 | pub(super) fn context_pressure_line( |
| 151 | &self, |
| 152 | current_text: &str, |
| 153 | prompt_context: &NextTurnPromptContext, |
| 154 | system_prompt: Option<&SystemPrompt>, |
| 155 | ) -> Option<String> { |
| 156 | // The engine owns automatic compaction. Asking the model to warn the |
| 157 | // user here created a competing save/compact ceremony before the |
| 158 | // automatic request-boundary guard could do its work (#5620). |
| 159 | if self.config.compaction.enabled { |
| 160 | return None; |
| 161 | } |
| 162 | let input_tokens = self.active_input_tokens_with_current_text(current_text, system_prompt); |
| 163 | let budget = route_context_budget_for_route( |
| 164 | prompt_context.provider, |
| 165 | &prompt_context.model, |
| 166 | prompt_context.route_limits, |
| 167 | input_tokens, |
| 168 | )?; |
| 169 | context_pressure_message(budget.usage_percent()).map(|warning| format!( |
| 170 | "{warning}. Estimated input: {input_tokens} tokens ({:.1}% of route budget). Automatic compaction is explicitly disabled for this session. A manual /compact saves the original conversation and its model-written handoff before replacing context.", |
| 171 | budget.usage_percent(), |
| 172 | )) |
| 173 | } |
| 174 | |
| 175 | pub(super) fn prepare_compaction_envelope( |
| 176 | &self, |
| 177 | mut config: CompactionConfig, |
| 178 | ) -> PreparedCompactionEnvelope { |
| 179 | // Host-supplied configs may not carry the workspace; compaction needs |
| 180 | // it only to re-state the user's `/anchor` file after the summary. |
| 181 | config |
| 182 | .workspace |
| 183 | .get_or_insert_with(|| self.config.workspace.clone()); |
| 184 | let mut prepared = PreparedCompactionEnvelope::new(config); |
| 185 | prepared.session_id = Some(self.session.id.clone()); |
| 186 | prepared |
| 187 | } |
| 188 | |
| 189 | pub(super) async fn handle_manual_compaction_op( |
| 190 | &mut self, |
| 191 | id: String, |
| 192 | route: ResolvedRuntimeRoute, |
| 193 | compaction: CompactionConfig, |
| 194 | ) { |
| 195 | self.emit_compaction_started( |
| 196 | id.clone(), |
| 197 | false, |
| 198 | "Manual context compaction started".to_string(), |
| 199 | ) |
| 200 | .await; |
| 201 | let Some(cancel_token) = self.claim_compaction(&id) else { |
| 202 | let message = "Context compaction canceled before it started".to_string(); |
| 203 | self.emit_compaction_cancelled(id, false, message).await; |
| 204 | let _ = self |
| 205 | .tx_event |
| 206 | .send(Event::TurnComplete { |
| 207 | usage: Usage::default(), |
| 208 | parent_route_usage: Usage::default(), |
| 209 | routed_usage_dropped_records: 0, |
| 210 | status: TurnOutcomeStatus::Interrupted, |
| 211 | error: None, |
| 212 | tool_catalog: None, |
| 213 | base_url: None, |
| 214 | }) |
| 215 | .await; |
| 216 | return; |
| 217 | }; |
| 218 | if let Err(err) = self.install_resolved_runtime_route(route) { |
| 219 | let message = |
| 220 | format!("Cannot compact context because its provider route is not ready: {err}"); |
| 221 | self.finish_compaction(&id); |
| 222 | self.emit_compaction_failed(id, false, message.clone()) |
| 223 | .await; |
| 224 | let _ = self |
| 225 | .tx_event |
| 226 | .send(Event::error(ErrorEnvelope::fatal_auth(message))) |
| 227 | .await; |
| 228 | return; |
| 229 | } |
| 230 | self.config.compaction = compaction; |
| 231 | self.handle_manual_compaction(id, cancel_token).await; |
| 232 | } |
| 233 | |
| 234 | pub(super) async fn emit_compaction_usage(&self, usage: &Usage, elapsed: Duration) { |
| 235 | if *usage == Usage::default() { |
| 236 | return; |
| 237 | } |
| 238 | let _ = self |
| 239 | .tx_event |
| 240 | .send(Event::RoutedTurnUsage { |
| 241 | usage: usage.clone(), |
| 242 | duration_ms: u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX), |
| 243 | first_token_ms: None, |
| 244 | request_ms: None, |
| 245 | }) |
| 246 | .await; |
| 247 | } |
| 248 | |
| 249 | pub(super) async fn handle_manual_compaction( |
| 250 | &mut self, |
| 251 | id: String, |
| 252 | cancel_token: CancellationToken, |
| 253 | ) { |
| 254 | let zero_usage = Usage { |
| 255 | input_tokens: 0, |
| 256 | output_tokens: 0, |
| 257 | ..Usage::default() |
| 258 | }; |
| 259 | let Some(client) = self.codewhale_client.clone() else { |
| 260 | let message = "Manual compaction unavailable: API client not configured".to_string(); |
| 261 | self.finish_compaction(&id); |
| 262 | self.emit_compaction_failed(id, false, message.clone()) |
| 263 | .await; |
| 264 | let _ = self |
| 265 | .tx_event |
| 266 | .send(Event::error(ErrorEnvelope::fatal_auth(message.clone()))) |
| 267 | .await; |
| 268 | let _ = self |
| 269 | .tx_event |
| 270 | .send(Event::TurnComplete { |
| 271 | usage: zero_usage, |
| 272 | parent_route_usage: Usage::default(), |
| 273 | routed_usage_dropped_records: 0, |
| 274 | status: TurnOutcomeStatus::Failed, |
| 275 | error: Some(message), |
| 276 | tool_catalog: None, |
| 277 | base_url: None, |
| 278 | }) |
| 279 | .await; |
| 280 | return; |
| 281 | }; |
| 282 | |
| 283 | let messages_before = self.session.messages.len(); |
| 284 | // Message counts alone do not show the win the user cares about: a |
| 285 | // compaction that drops few but enormous messages reads as a no-op. |
| 286 | // The emergency path already reports tokens; manual and auto now match. |
| 287 | let tokens_before = self.estimated_input_tokens(); |
| 288 | let mut turn_status = TurnOutcomeStatus::Completed; |
| 289 | let mut turn_error = None; |
| 290 | |
| 291 | let prepared = self.prepare_compaction_envelope(self.config.compaction.clone()); |
| 292 | |
| 293 | let started = Instant::now(); |
| 294 | let mut compaction_usage = Usage::default(); |
| 295 | let compaction_result = tokio::select! { |
| 296 | biased; |
| 297 | _ = cancel_token.cancelled() => None, |
| 298 | result = compact_messages_safe( |
| 299 | &client, |
| 300 | &self.session.messages, |
| 301 | self.session.system_prompt.as_ref(), |
| 302 | &prepared, |
| 303 | &mut compaction_usage, |
| 304 | ) => Some(result), |
| 305 | }; |
| 306 | self.session.total_usage.add(&compaction_usage); |
| 307 | self.record_goal_usage_for_turn(&compaction_usage, started.elapsed()); |
| 308 | self.emit_compaction_usage(&compaction_usage, started.elapsed()) |
| 309 | .await; |
| 310 | |
| 311 | let Some(compaction_result) = compaction_result else { |
| 312 | self.finish_compaction(&id); |
| 313 | self.emit_compaction_cancelled( |
| 314 | id, |
| 315 | false, |
| 316 | "Context compaction canceled; conversation context was not changed".to_string(), |
| 317 | ) |
| 318 | .await; |
| 319 | let _ = self |
| 320 | .tx_event |
| 321 | .send(Event::TurnComplete { |
| 322 | usage: compaction_usage, |
| 323 | parent_route_usage: Usage::default(), |
| 324 | routed_usage_dropped_records: 0, |
| 325 | status: TurnOutcomeStatus::Interrupted, |
| 326 | error: None, |
| 327 | tool_catalog: None, |
| 328 | base_url: None, |
| 329 | }) |
| 330 | .await; |
| 331 | return; |
| 332 | }; |
| 333 | |
| 334 | match compaction_result { |
| 335 | Ok(mut result) => { |
| 336 | if !result.messages.is_empty() || self.session.messages.is_empty() { |
| 337 | self.append_compaction_agent_topology(&mut result.messages) |
| 338 | .await; |
| 339 | if cancel_token.is_cancelled() { |
| 340 | self.finish_compaction(&id); |
| 341 | self.emit_compaction_cancelled( |
| 342 | id, |
| 343 | false, |
| 344 | "Context compaction canceled; conversation context was not changed" |
| 345 | .to_string(), |
| 346 | ) |
| 347 | .await; |
| 348 | let _ = self |
| 349 | .tx_event |
| 350 | .send(Event::TurnComplete { |
| 351 | usage: compaction_usage, |
| 352 | parent_route_usage: Usage::default(), |
| 353 | routed_usage_dropped_records: 0, |
| 354 | status: TurnOutcomeStatus::Interrupted, |
| 355 | error: None, |
| 356 | tool_catalog: None, |
| 357 | base_url: None, |
| 358 | }) |
| 359 | .await; |
| 360 | return; |
| 361 | } |
| 362 | let messages_after = result.messages.len(); |
| 363 | let retries_used = result.retries_used; |
| 364 | let coverage_clause = result.coverage.receipt_clause(); |
| 365 | let path = result.coverage.path; |
| 366 | self.session.replace_messages(result.messages); |
| 367 | if let Some(pm) = self.session.prefix_stability.as_mut() { |
| 368 | pm.note_history_reset("compaction"); |
| 369 | } |
| 370 | self.commit_compaction_checkpoint(result.summary_prompt); |
| 371 | self.emit_session_updated().await; |
| 372 | let removed = messages_before.saturating_sub(messages_after); |
| 373 | let tokens_after = self.estimated_input_tokens(); |
| 374 | let message = if retries_used > 0 { |
| 375 | format!( |
| 376 | "Compaction complete: {messages_before} → {messages_after} messages ({removed} removed, {retries_used} retries), ~{tokens_before} → ~{tokens_after} tokens ({coverage_clause})" |
| 377 | ) |
| 378 | } else { |
| 379 | format!( |
| 380 | "Compaction complete: {messages_before} → {messages_after} messages ({removed} removed), ~{tokens_before} → ~{tokens_after} tokens ({coverage_clause})" |
| 381 | ) |
| 382 | }; |
| 383 | self.emit_compaction_completed( |
| 384 | id.clone(), |
| 385 | false, |
| 386 | message, |
| 387 | Some(messages_before), |
| 388 | Some(messages_after), |
| 389 | CompactionPass { |
| 390 | trigger: "manual", |
| 391 | path, |
| 392 | tokens_before, |
| 393 | threshold_tokens: prepared.config.token_threshold, |
| 394 | usage: compaction_usage.clone(), |
| 395 | }, |
| 396 | ) |
| 397 | .await; |
| 398 | } else { |
| 399 | let message = "Compaction skipped: produced empty result".to_string(); |
| 400 | self.emit_compaction_failed(id.clone(), false, message.clone()) |
| 401 | .await; |
| 402 | turn_status = TurnOutcomeStatus::Failed; |
| 403 | turn_error = Some(message); |
| 404 | } |
| 405 | } |
| 406 | Err(err) => { |
| 407 | let message = crate::compaction::report_compaction_failure( |
| 408 | "Manual context compaction failed", |
| 409 | &id, |
| 410 | false, |
| 411 | &err, |
| 412 | ); |
| 413 | self.emit_compaction_failed(id.clone(), false, message.clone()) |
| 414 | .await; |
| 415 | let _ = self.tx_event.send(Event::status(message.clone())).await; |
| 416 | turn_status = TurnOutcomeStatus::Failed; |
| 417 | turn_error = Some(message); |
| 418 | } |
| 419 | } |
| 420 | |
| 421 | self.finish_compaction(&id); |
| 422 | |
| 423 | let _ = self |
| 424 | .tx_event |
| 425 | .send(Event::TurnComplete { |
| 426 | usage: compaction_usage, |
| 427 | parent_route_usage: Usage::default(), |
| 428 | routed_usage_dropped_records: 0, |
| 429 | status: turn_status, |
| 430 | error: turn_error, |
| 431 | tool_catalog: None, |
| 432 | base_url: None, |
| 433 | }) |
| 434 | .await; |
| 435 | } |
| 436 | |
| 437 | pub(super) async fn recover_context_overflow( |
| 438 | &mut self, |
| 439 | client: &dyn crate::core::model_client::ModelClient, |
| 440 | tools: Option<&[Tool]>, |
| 441 | reason: &str, |
| 442 | turn: &mut TurnContext, |
| 443 | ) -> bool { |
| 444 | let Some(target_budget) = context_input_budget_for_route( |
| 445 | self.api_provider, |
| 446 | &self.session.model, |
| 447 | self.active_route_limits, |
| 448 | 0, |
| 449 | ) else { |
| 450 | return false; |
| 451 | }; |
| 452 | |
| 453 | let id = format!("compact_{}", &uuid::Uuid::new_v4().to_string()[..8]); |
| 454 | turn.stop_diagnostics.emergency_compaction_attempts = turn |
| 455 | .stop_diagnostics |
| 456 | .emergency_compaction_attempts |
| 457 | .saturating_add(1); |
| 458 | let start_message = format!("Emergency context compaction started ({reason})"); |
| 459 | self.emit_compaction_started(id.clone(), true, start_message) |
| 460 | .await; |
| 461 | let Some(compaction_cancel) = self.claim_compaction(&id) else { |
| 462 | self.emit_compaction_cancelled( |
| 463 | id, |
| 464 | true, |
| 465 | "Emergency context compaction canceled before it started; conversation context was not changed" |
| 466 | .to_string(), |
| 467 | ) |
| 468 | .await; |
| 469 | return false; |
| 470 | }; |
| 471 | let turn_cancel = self.cancel_token.clone(); |
| 472 | |
| 473 | let before_tokens = self.estimated_input_tokens(); |
| 474 | let before_count = self.session.messages.len(); |
| 475 | |
| 476 | let mut forced_config = self.config.compaction.clone(); |
| 477 | forced_config.enabled = true; |
| 478 | forced_config.token_threshold = forced_config |
| 479 | .token_threshold |
| 480 | .min(target_budget.saturating_sub(1)) |
| 481 | .max(1); |
| 482 | let mut prepared = self.prepare_compaction_envelope(forced_config); |
| 483 | prepared.tools = tools.map(<[Tool]>::to_vec); |
| 484 | |
| 485 | let started = Instant::now(); |
| 486 | let mut compaction_usage = Usage::default(); |
| 487 | let (compaction_result, turn_was_canceled) = tokio::select! { |
| 488 | biased; |
| 489 | _ = turn_cancel.cancelled() => (None, true), |
| 490 | _ = compaction_cancel.cancelled() => (None, false), |
| 491 | result = compact_messages_safe( |
| 492 | client, |
| 493 | &self.session.messages, |
| 494 | self.session.system_prompt.as_ref(), |
| 495 | &prepared, |
| 496 | &mut compaction_usage, |
| 497 | ) => (Some(result), false), |
| 498 | }; |
| 499 | turn.add_usage(&compaction_usage); |
| 500 | self.emit_compaction_usage(&compaction_usage, started.elapsed()) |
| 501 | .await; |
| 502 | let Some(compaction_result) = compaction_result else { |
| 503 | self.finish_compaction(&id); |
| 504 | let message = if turn_was_canceled { |
| 505 | "Emergency context compaction canceled with the active turn; conversation context was not changed" |
| 506 | } else { |
| 507 | "Emergency context compaction canceled; conversation context was not changed" |
| 508 | } |
| 509 | .to_string(); |
| 510 | self.emit_compaction_cancelled(id, true, message).await; |
| 511 | return false; |
| 512 | }; |
| 513 | |
| 514 | let result = match compaction_result { |
| 515 | Ok(result) => result, |
| 516 | Err(err) => { |
| 517 | let message = |
| 518 | format!("Context recovery failed: {err}. Original conversation was preserved."); |
| 519 | self.emit_compaction_failed(id.clone(), true, message).await; |
| 520 | self.finish_compaction(&id); |
| 521 | return false; |
| 522 | } |
| 523 | }; |
| 524 | let retries_used = result.retries_used; |
| 525 | let summary_prompt = result.summary_prompt; |
| 526 | let path = result.coverage.path; |
| 527 | let mut compacted_messages = result.messages; |
| 528 | |
| 529 | let turn_was_canceled = turn_cancel.is_cancelled(); |
| 530 | if turn_was_canceled || compaction_cancel.is_cancelled() { |
| 531 | self.finish_compaction(&id); |
| 532 | let message = if turn_was_canceled { |
| 533 | "Emergency context compaction canceled with the active turn; conversation context was not changed" |
| 534 | } else { |
| 535 | "Emergency context compaction canceled; conversation context was not changed" |
| 536 | } |
| 537 | .to_string(); |
| 538 | self.emit_compaction_cancelled(id, true, message).await; |
| 539 | return false; |
| 540 | } |
| 541 | |
| 542 | if !compacted_messages.is_empty() || self.session.messages.is_empty() { |
| 543 | self.append_compaction_agent_topology(&mut compacted_messages) |
| 544 | .await; |
| 545 | let turn_was_canceled = turn_cancel.is_cancelled(); |
| 546 | if turn_was_canceled || compaction_cancel.is_cancelled() { |
| 547 | self.finish_compaction(&id); |
| 548 | let message = if turn_was_canceled { |
| 549 | "Emergency context compaction canceled with the active turn; conversation context was not changed" |
| 550 | } else { |
| 551 | "Emergency context compaction canceled; conversation context was not changed" |
| 552 | } |
| 553 | .to_string(); |
| 554 | self.emit_compaction_cancelled(id, true, message).await; |
| 555 | return false; |
| 556 | } |
| 557 | } |
| 558 | // Validate the complete candidate before the only history swap. Bare |
| 559 | // front-trimming after a failed summary silently lost user state and |
| 560 | // could leave orphan tool results in an apparently recovered session. |
| 561 | let after_tokens = crate::compaction::estimate_input_tokens_for_pressure( |
| 562 | &compacted_messages, |
| 563 | self.session.system_prompt.as_ref(), |
| 564 | ); |
| 565 | let after_count = compacted_messages.len(); |
| 566 | let recovered = after_tokens <= target_budget && after_tokens < before_tokens; |
| 567 | |
| 568 | if recovered { |
| 569 | self.session.replace_messages(compacted_messages); |
| 570 | turn.clear_parent_input_tokens(); |
| 571 | if let Some(pm) = self.session.prefix_stability.as_mut() { |
| 572 | pm.note_history_reset("compaction"); |
| 573 | } |
| 574 | self.commit_compaction_checkpoint(summary_prompt); |
| 575 | self.emit_session_updated().await; |
| 576 | let removed = before_count.saturating_sub(after_count); |
| 577 | let mut details = format!( |
| 578 | "Emergency compaction complete: {before_count} → {after_count} messages ({removed} removed), ~{before_tokens} → ~{after_tokens} tokens" |
| 579 | ); |
| 580 | if retries_used > 0 { |
| 581 | details.push_str(&format!(" ({retries_used} retries)")); |
| 582 | } |
| 583 | self.emit_compaction_completed( |
| 584 | id.clone(), |
| 585 | true, |
| 586 | details.clone(), |
| 587 | Some(before_count), |
| 588 | Some(after_count), |
| 589 | CompactionPass { |
| 590 | trigger: "emergency", |
| 591 | path, |
| 592 | tokens_before: before_tokens, |
| 593 | threshold_tokens: prepared.config.token_threshold, |
| 594 | usage: compaction_usage.clone(), |
| 595 | }, |
| 596 | ) |
| 597 | .await; |
| 598 | let _ = self.tx_event.send(Event::status(details)).await; |
| 599 | self.finish_compaction(&id); |
| 600 | return true; |
| 601 | } |
| 602 | |
| 603 | // Two distinct failures were previously conflated into one banner. |
| 604 | // When the provider rejected the request (its bill counts framing we |
| 605 | // cannot see), our estimate may already sit within the budget while |
| 606 | // the pass removed nothing — reporting that as "failed to reduce |
| 607 | // below model limit" with an estimate printed *under* the budget |
| 608 | // reads as self-contradictory. Name the actual outcome instead. |
| 609 | let message = if after_tokens > target_budget { |
| 610 | format!( |
| 611 | "Emergency context compaction failed to reduce request below model limit \ |
| 612 | (estimate ~{after_tokens} tokens, budget ~{target_budget}). Original conversation was preserved." |
| 613 | ) |
| 614 | } else { |
| 615 | format!( |
| 616 | "Emergency context compaction made no progress (estimate ~{after_tokens} tokens \ |
| 617 | is already within the ~{target_budget} budget; the provider may count the \ |
| 618 | request differently). Original conversation was preserved." |
| 619 | ) |
| 620 | }; |
| 621 | self.emit_compaction_failed(id.clone(), true, message.clone()) |
| 622 | .await; |
| 623 | let _ = self.tx_event.send(Event::status(message)).await; |
| 624 | self.finish_compaction(&id); |
| 625 | false |
| 626 | } |
| 627 | |
| 628 | /// Keep the rendered checkpoint for host persistence and repeat-compaction |
| 629 | /// metadata. The model sees the checkpoint exactly once through ordinary |
| 630 | /// conversation history; the stable system prefix never carries it. |
| 631 | pub(super) fn commit_compaction_checkpoint(&mut self, summary_prompt: Option<SystemPrompt>) { |
| 632 | let Some(summary_prompt) = summary_prompt else { |
| 633 | return; |
| 634 | }; |
| 635 | self.session.compaction_summary_prompt = Some(summary_prompt); |
| 636 | } |
| 637 | |
| 638 | /// Capture the current session-owned Agent topology at the replacement |
| 639 | /// history boundary. This is the Codewhale equivalent of Codex clearing |
| 640 | /// its world-state reference after standalone compaction so the next turn |
| 641 | /// receives fresh environment/subagent context instead of trusting the |
| 642 | /// narrative summary as live process state. |
| 643 | pub(super) async fn append_compaction_agent_topology(&self, messages: &mut Vec<Message>) { |
| 644 | let snapshots = { |
| 645 | let manager = self.subagent_manager.read().await; |
| 646 | manager.list_for_session(&self.session.id) |
| 647 | }; |
| 648 | crate::runtime_handoff::replace_agent_topology_checkpoint(messages, &snapshots); |
| 649 | } |
| 650 | } |
| 651 |