| 1 | //! Core engine for `DeepSeek` CLI. |
| 2 | //! |
| 3 | //! The engine handles all AI interactions in a background task, |
| 4 | //! communicating with the UI via channels. This enables: |
| 5 | //! - Non-blocking UI during API calls |
| 6 | //! - Real-time streaming updates |
| 7 | //! - Proper cancellation support |
| 8 | //! - Tool execution orchestration |
| 9 | |
| 10 | use std::collections::HashMap; |
| 11 | use std::collections::hash_map::DefaultHasher; |
| 12 | use std::hash::{Hash, Hasher}; |
| 13 | use std::path::PathBuf; |
| 14 | use std::sync::{Arc, Mutex as StdMutex}; |
| 15 | use std::time::{Duration, Instant}; |
| 16 | |
| 17 | use anyhow::Result; |
| 18 | use futures_util::StreamExt; |
| 19 | use futures_util::stream::FuturesUnordered; |
| 20 | use serde_json::json; |
| 21 | use tokio::sync::{Mutex as AsyncMutex, RwLock, mpsc}; |
| 22 | use tokio_util::sync::CancellationToken; |
| 23 | |
| 24 | use crate::client::DeepSeekClient; |
| 25 | use crate::compaction::{ |
| 26 | CompactionConfig, compact_messages_safe, merge_system_prompts, should_compact, |
| 27 | }; |
| 28 | use crate::config::{ApiProvider, Config, DEFAULT_MAX_SUBAGENTS, DEFAULT_TEXT_MODEL}; |
| 29 | use crate::cycle_manager::{ |
| 30 | CycleBriefing, CycleConfig, StructuredState, archive_cycle, build_seed_messages, |
| 31 | estimate_briefing_tokens, produce_briefing, should_advance_cycle, |
| 32 | }; |
| 33 | use crate::error_taxonomy::{ErrorCategory, ErrorEnvelope, StreamError}; |
| 34 | use crate::features::{Feature, Features}; |
| 35 | use crate::llm_client::LlmClient; |
| 36 | use crate::mcp::McpPool; |
| 37 | #[cfg(test)] |
| 38 | use crate::models::ToolCaller; |
| 39 | use crate::models::{ |
| 40 | ContentBlock, ContentBlockStart, Delta, LEGACY_DEEPSEEK_CONTEXT_WINDOW_TOKENS, Message, |
| 41 | MessageRequest, StreamEvent, SystemPrompt, Tool, Usage, |
| 42 | }; |
| 43 | use crate::prompts; |
| 44 | use crate::seam_manager::{SeamConfig, SeamManager}; |
| 45 | use crate::tools::plan::{SharedPlanState, new_shared_plan_state}; |
| 46 | use crate::tools::shell::{SharedShellManager, new_shared_shell_manager}; |
| 47 | use crate::tools::spec::RuntimeToolServices; |
| 48 | use crate::tools::spec::{ApprovalRequirement, ToolError, ToolResult}; |
| 49 | use crate::tools::subagent::{ |
| 50 | Mailbox, SharedSubAgentManager, SubAgentCompletion, SubAgentRuntime, SubAgentType, |
| 51 | new_shared_subagent_manager, resolve_subagent_assignment_route, |
| 52 | }; |
| 53 | use crate::tools::todo::{SharedTodoList, new_shared_todo_list}; |
| 54 | use crate::tools::user_input::{UserInputRequest, UserInputResponse}; |
| 55 | use crate::tools::{ToolContext, ToolRegistryBuilder}; |
| 56 | use crate::tui::app::AppMode; |
| 57 | use crate::utils::spawn_supervised; |
| 58 | |
| 59 | use super::capacity::{ |
| 60 | CapacityController, CapacityControllerConfig, CapacityDecision, CapacityObservationInput, |
| 61 | CapacitySnapshot, GuardrailAction, RiskBand, |
| 62 | }; |
| 63 | use super::capacity_memory::{ |
| 64 | CanonicalState, CapacityMemoryRecord, ReplayInfo, append_capacity_record, |
| 65 | load_last_k_capacity_records, new_record_id, now_rfc3339, |
| 66 | }; |
| 67 | use super::coherence::{CoherenceSignal, CoherenceState, next_coherence_state}; |
| 68 | use super::events::{Event, TurnOutcomeStatus}; |
| 69 | use super::ops::Op; |
| 70 | use super::session::Session; |
| 71 | use super::tool_parser; |
| 72 | use super::turn::{TurnContext, TurnToolCall, post_turn_snapshot, pre_turn_snapshot}; |
| 73 | |
| 74 | // === Types === |
| 75 | |
| 76 | /// Configuration for the engine |
| 77 | #[derive(Debug, Clone)] |
| 78 | pub struct EngineConfig { |
| 79 | /// Model identifier to use for responses. |
| 80 | pub model: String, |
| 81 | /// Workspace root for tool execution and file operations. |
| 82 | pub workspace: PathBuf, |
| 83 | /// Allow shell tool execution when true. |
| 84 | pub allow_shell: bool, |
| 85 | /// Enable trust mode (skip approvals) when true. |
| 86 | pub trust_mode: bool, |
| 87 | /// Path to the notes file used by the notes tool. |
| 88 | pub notes_path: PathBuf, |
| 89 | /// Path to the MCP configuration file. |
| 90 | pub mcp_config_path: PathBuf, |
| 91 | /// Directory containing discoverable skills. |
| 92 | pub skills_dir: PathBuf, |
| 93 | /// Additional instruction files concatenated into the system |
| 94 | /// prompt (#454). Loaded in declared order from the user's |
| 95 | /// `instructions = [...]` config (or the per-project override). |
| 96 | /// Resolved via `expand_path` so `~` works. |
| 97 | pub instructions: Vec<PathBuf>, |
| 98 | /// Maximum number of assistant steps before stopping. |
| 99 | pub max_steps: u32, |
| 100 | /// Maximum number of concurrently active subagents. |
| 101 | pub max_subagents: usize, |
| 102 | /// Feature flags controlling tool availability. |
| 103 | pub features: Features, |
| 104 | /// Auto-compaction settings for long conversations. |
| 105 | /// |
| 106 | /// As of v0.6.6 the high-level summarization compaction (`compact_messages_safe`) |
| 107 | /// is **disabled by default**; the checkpoint-restart cycle architecture |
| 108 | /// (`cycle_manager`) replaces it. The compaction config is still wired through |
| 109 | /// for the per-tool-result truncation path (`compact_tool_result_for_context`) |
| 110 | /// and for users who explicitly opt back in through the `auto_compact` |
| 111 | /// setting or a direct engine config. |
| 112 | pub compaction: CompactionConfig, |
| 113 | /// Checkpoint-restart cycle settings (issue #124). |
| 114 | pub cycle: CycleConfig, |
| 115 | /// Capacity-controller settings. |
| 116 | pub capacity: CapacityControllerConfig, |
| 117 | /// Shared Todo list state. |
| 118 | pub todos: SharedTodoList, |
| 119 | /// Shared Plan state. |
| 120 | pub plan_state: SharedPlanState, |
| 121 | /// Maximum sub-agent recursion depth (default 3). See |
| 122 | /// `SubAgentRuntime::max_spawn_depth`. Override via |
| 123 | /// `[runtime] max_spawn_depth = N` in `~/.deepseek/config.toml`. |
| 124 | pub max_spawn_depth: u32, |
| 125 | /// Per-domain network policy decider (#135). Shared across the session so |
| 126 | /// session-scoped approvals (`/network allow <host>`) persist for the |
| 127 | /// remainder of the run. |
| 128 | pub network_policy: Option<crate::network_policy::NetworkPolicyDecider>, |
| 129 | /// Whether to take side-git workspace snapshots before/after each turn. |
| 130 | pub snapshots_enabled: bool, |
| 131 | /// Post-edit LSP diagnostics injection (#136). When `None`, the engine |
| 132 | /// constructs a disabled manager so the field is always present. |
| 133 | pub lsp_config: Option<crate::lsp::LspConfig>, |
| 134 | /// Durable runtime services exposed to model-visible tools. |
| 135 | pub runtime_services: RuntimeToolServices, |
| 136 | /// Per-role/type sub-agent model overrides already resolved from config. |
| 137 | pub subagent_model_overrides: HashMap<String, String>, |
| 138 | /// Whether the user-memory feature is enabled (#489). When `true` the |
| 139 | /// engine reads `memory_path` on each prompt assembly and prepends a |
| 140 | /// `<user_memory>` block to the system prompt. |
| 141 | pub memory_enabled: bool, |
| 142 | /// Path to the user memory file (#489). Always populated; only |
| 143 | /// consulted when `memory_enabled` is `true`. |
| 144 | pub memory_path: PathBuf, |
| 145 | pub goal_objective: Option<String>, |
| 146 | /// Resolved BCP-47 locale tag (e.g. `"en"`, `"zh-Hans"`, `"ja"`) |
| 147 | /// for the `## Environment` block in the system prompt. The |
| 148 | /// caller resolves this from `Settings` once at engine |
| 149 | /// construction; the engine never touches disk for it. |
| 150 | pub locale_tag: String, |
| 151 | /// When true, force `tool_choice: "required"` so the model always calls |
| 152 | /// a tool on every turn step (V4 strict tool-following mode). |
| 153 | pub strict_tool_mode: bool, |
| 154 | /// Workshop / large-tool-output routing (#548). `None` disables routing. |
| 155 | pub workshop: Option<crate::tools::large_output_router::WorkshopConfig>, |
| 156 | } |
| 157 | |
| 158 | impl Default for EngineConfig { |
| 159 | fn default() -> Self { |
| 160 | Self { |
| 161 | model: DEFAULT_TEXT_MODEL.to_string(), |
| 162 | workspace: PathBuf::from("."), |
| 163 | allow_shell: true, |
| 164 | trust_mode: false, |
| 165 | notes_path: PathBuf::from("notes.txt"), |
| 166 | mcp_config_path: PathBuf::from("mcp.json"), |
| 167 | skills_dir: crate::skills::default_skills_dir(), |
| 168 | instructions: Vec::new(), |
| 169 | max_steps: 100, |
| 170 | max_subagents: DEFAULT_MAX_SUBAGENTS, |
| 171 | features: Features::with_defaults(), |
| 172 | compaction: CompactionConfig::default(), |
| 173 | cycle: CycleConfig::default(), |
| 174 | capacity: CapacityControllerConfig::default(), |
| 175 | todos: new_shared_todo_list(), |
| 176 | plan_state: new_shared_plan_state(), |
| 177 | max_spawn_depth: crate::tools::subagent::DEFAULT_MAX_SPAWN_DEPTH, |
| 178 | network_policy: None, |
| 179 | snapshots_enabled: true, |
| 180 | lsp_config: None, |
| 181 | runtime_services: RuntimeToolServices::default(), |
| 182 | subagent_model_overrides: HashMap::new(), |
| 183 | memory_enabled: false, |
| 184 | memory_path: PathBuf::from("./memory.md"), |
| 185 | strict_tool_mode: false, |
| 186 | goal_objective: None, |
| 187 | locale_tag: "en".to_string(), |
| 188 | workshop: None, |
| 189 | } |
| 190 | } |
| 191 | } |
| 192 | |
| 193 | /// Handle to communicate with the engine |
| 194 | #[derive(Clone)] |
| 195 | pub struct EngineHandle { |
| 196 | /// Send operations to the engine |
| 197 | pub tx_op: mpsc::Sender<Op>, |
| 198 | /// Receive events from the engine |
| 199 | pub rx_event: Arc<RwLock<mpsc::Receiver<Event>>>, |
| 200 | /// Shared pointer to the cancellation token for the current request. |
| 201 | cancel_token: Arc<StdMutex<CancellationToken>>, |
| 202 | /// Send approval decisions to the engine |
| 203 | tx_approval: mpsc::Sender<ApprovalDecision>, |
| 204 | /// Send user input responses to the engine |
| 205 | tx_user_input: mpsc::Sender<UserInputDecision>, |
| 206 | /// Send steer input for an in-flight turn. |
| 207 | tx_steer: mpsc::Sender<String>, |
| 208 | } |
| 209 | |
| 210 | impl EngineHandle { |
| 211 | /// Send an operation to the engine |
| 212 | pub async fn send(&self, op: Op) -> Result<()> { |
| 213 | self.tx_op.send(op).await?; |
| 214 | Ok(()) |
| 215 | } |
| 216 | |
| 217 | /// Cancel the current request |
| 218 | pub fn cancel(&self) { |
| 219 | match self.cancel_token.lock() { |
| 220 | Ok(token) => token.cancel(), |
| 221 | Err(poisoned) => poisoned.into_inner().cancel(), |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | /// Check if a request is currently cancelled |
| 226 | #[must_use] |
| 227 | #[allow(dead_code)] |
| 228 | pub fn is_cancelled(&self) -> bool { |
| 229 | match self.cancel_token.lock() { |
| 230 | Ok(token) => token.is_cancelled(), |
| 231 | Err(poisoned) => poisoned.into_inner().is_cancelled(), |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | /// Approve a pending tool call |
| 236 | pub async fn approve_tool_call(&self, id: impl Into<String>) -> Result<()> { |
| 237 | self.tx_approval |
| 238 | .send(ApprovalDecision::Approved { id: id.into() }) |
| 239 | .await?; |
| 240 | Ok(()) |
| 241 | } |
| 242 | |
| 243 | /// Deny a pending tool call |
| 244 | pub async fn deny_tool_call(&self, id: impl Into<String>) -> Result<()> { |
| 245 | self.tx_approval |
| 246 | .send(ApprovalDecision::Denied { id: id.into() }) |
| 247 | .await?; |
| 248 | Ok(()) |
| 249 | } |
| 250 | |
| 251 | /// Retry a tool call with an elevated sandbox policy. |
| 252 | pub async fn retry_tool_with_policy( |
| 253 | &self, |
| 254 | id: impl Into<String>, |
| 255 | policy: crate::sandbox::SandboxPolicy, |
| 256 | ) -> Result<()> { |
| 257 | self.tx_approval |
| 258 | .send(ApprovalDecision::RetryWithPolicy { |
| 259 | id: id.into(), |
| 260 | policy, |
| 261 | }) |
| 262 | .await?; |
| 263 | Ok(()) |
| 264 | } |
| 265 | |
| 266 | /// Submit a response for request_user_input. |
| 267 | pub async fn submit_user_input( |
| 268 | &self, |
| 269 | id: impl Into<String>, |
| 270 | response: UserInputResponse, |
| 271 | ) -> Result<()> { |
| 272 | self.tx_user_input |
| 273 | .send(UserInputDecision::Submitted { |
| 274 | id: id.into(), |
| 275 | response, |
| 276 | }) |
| 277 | .await?; |
| 278 | Ok(()) |
| 279 | } |
| 280 | |
| 281 | /// Cancel a request_user_input prompt. |
| 282 | pub async fn cancel_user_input(&self, id: impl Into<String>) -> Result<()> { |
| 283 | self.tx_user_input |
| 284 | .send(UserInputDecision::Cancelled { id: id.into() }) |
| 285 | .await?; |
| 286 | Ok(()) |
| 287 | } |
| 288 | |
| 289 | /// Steer an in-flight turn with additional user input. |
| 290 | pub async fn steer(&self, content: impl Into<String>) -> Result<()> { |
| 291 | self.tx_steer.send(content.into()).await?; |
| 292 | Ok(()) |
| 293 | } |
| 294 | } |
| 295 | |
| 296 | // === Engine === |
| 297 | |
| 298 | /// The core engine that processes operations and emits events |
| 299 | pub struct Engine { |
| 300 | config: EngineConfig, |
| 301 | deepseek_client: Option<DeepSeekClient>, |
| 302 | deepseek_client_error: Option<String>, |
| 303 | api_key_env_only_recovery: Option<String>, |
| 304 | session: Session, |
| 305 | subagent_manager: SharedSubAgentManager, |
| 306 | shell_manager: SharedShellManager, |
| 307 | mcp_pool: Option<Arc<AsyncMutex<McpPool>>>, |
| 308 | rx_op: mpsc::Receiver<Op>, |
| 309 | rx_approval: mpsc::Receiver<ApprovalDecision>, |
| 310 | rx_user_input: mpsc::Receiver<UserInputDecision>, |
| 311 | rx_steer: mpsc::Receiver<String>, |
| 312 | tx_event: mpsc::Sender<Event>, |
| 313 | /// Wakeup channel for the parent turn loop when a direct child sub-agent |
| 314 | /// terminates (issue #756). Cloned into `SubAgentRuntime` so the runtime |
| 315 | /// can fan completion events back into the engine. |
| 316 | tx_subagent_completion: mpsc::UnboundedSender<SubAgentCompletion>, |
| 317 | /// Receiver paired with `tx_subagent_completion`. Drained at the |
| 318 | /// turn-loop's empty-tool_uses branch to surface `<deepseek:subagent.done>` |
| 319 | /// sentinels into the parent's transcript before deciding to end the turn. |
| 320 | pub(super) rx_subagent_completion: mpsc::UnboundedReceiver<SubAgentCompletion>, |
| 321 | cancel_token: CancellationToken, |
| 322 | shared_cancel_token: Arc<StdMutex<CancellationToken>>, |
| 323 | tool_exec_lock: Arc<RwLock<()>>, |
| 324 | capacity_controller: CapacityController, |
| 325 | /// Append-only layered context manager (#159). Opt-in for v0.7.5 while |
| 326 | /// cache-hit behavior is audited. |
| 327 | seam_manager: Option<SeamManager>, |
| 328 | coherence_state: CoherenceState, |
| 329 | turn_counter: u64, |
| 330 | /// Post-edit LSP diagnostics injection (#136). Populated unconditionally |
| 331 | /// — when LSP is disabled in config, this is an inert manager that |
| 332 | /// always returns `None` from `diagnostics_for`. |
| 333 | lsp_manager: Arc<crate::lsp::LspManager>, |
| 334 | /// Session-scoped workshop variable store (#548). Shared across all tool |
| 335 | /// calls so `last_tool_result` persists within the session and can be |
| 336 | /// promoted to the parent context via `promote_to_context`. |
| 337 | workshop_vars: Option< |
| 338 | std::sync::Arc<tokio::sync::Mutex<crate::tools::large_output_router::WorkshopVariables>>, |
| 339 | >, |
| 340 | /// External sandbox backend (#516). When `Some`, exec_shell routes commands |
| 341 | /// through this instead of spawning a local process. |
| 342 | sandbox_backend: Option<std::sync::Arc<dyn crate::sandbox::backend::SandboxBackend>>, |
| 343 | /// Diagnostics collected during the current step's tool calls. Drained |
| 344 | /// and forwarded as a synthetic user message before the next API call. |
| 345 | pending_lsp_blocks: Vec<crate::lsp::DiagnosticBlock>, |
| 346 | } |
| 347 | |
| 348 | // === Internal tool helpers === |
| 349 | |
| 350 | impl Engine { |
| 351 | fn reset_cancel_token(&mut self) { |
| 352 | let token = CancellationToken::new(); |
| 353 | self.cancel_token = token.clone(); |
| 354 | match self.shared_cancel_token.lock() { |
| 355 | Ok(mut shared) => { |
| 356 | *shared = token; |
| 357 | } |
| 358 | Err(poisoned) => { |
| 359 | *poisoned.into_inner() = token; |
| 360 | } |
| 361 | } |
| 362 | } |
| 363 | |
| 364 | fn env_only_api_key_recovery_hint(api_config: &Config) -> Option<String> { |
| 365 | if !crate::config::active_provider_uses_env_only_api_key(api_config) { |
| 366 | return None; |
| 367 | } |
| 368 | |
| 369 | let provider = api_config.api_provider(); |
| 370 | let env_var = match provider { |
| 371 | ApiProvider::Deepseek | ApiProvider::DeepseekCN => "DEEPSEEK_API_KEY", |
| 372 | ApiProvider::NvidiaNim => "NVIDIA_API_KEY/NVIDIA_NIM_API_KEY", |
| 373 | ApiProvider::Openrouter => "OPENROUTER_API_KEY", |
| 374 | ApiProvider::Novita => "NOVITA_API_KEY", |
| 375 | ApiProvider::Fireworks => "FIREWORKS_API_KEY", |
| 376 | ApiProvider::Sglang => "SGLANG_API_KEY", |
| 377 | ApiProvider::Vllm => "VLLM_API_KEY", |
| 378 | }; |
| 379 | |
| 380 | Some(format!( |
| 381 | "The rejected key came from {env_var}; no saved config key is present.\n\ |
| 382 | Run `deepseek auth set --provider {provider}` to save a valid key in ~/.deepseek/config.toml, \ |
| 383 | or remove the stale export and open a fresh shell.", |
| 384 | provider = provider.as_str() |
| 385 | )) |
| 386 | } |
| 387 | |
| 388 | pub(super) fn decorate_auth_error_message(&self, message: String) -> String { |
| 389 | let Some(hint) = self.api_key_env_only_recovery.as_ref() else { |
| 390 | return message; |
| 391 | }; |
| 392 | if crate::error_taxonomy::classify_error_message(&message) != ErrorCategory::Authentication |
| 393 | || message.contains("no saved config key is present") |
| 394 | { |
| 395 | return message; |
| 396 | } |
| 397 | format!("{message}\n\n{hint}") |
| 398 | } |
| 399 | |
| 400 | /// Create a new engine with the given configuration |
| 401 | pub fn new(config: EngineConfig, api_config: &Config) -> (Self, EngineHandle) { |
| 402 | let (tx_op, rx_op) = mpsc::channel(32); |
| 403 | let (tx_event, rx_event) = mpsc::channel(256); |
| 404 | let (tx_approval, rx_approval) = mpsc::channel(64); |
| 405 | let (tx_user_input, rx_user_input) = mpsc::channel(32); |
| 406 | let (tx_steer, rx_steer) = mpsc::channel(64); |
| 407 | let (tx_subagent_completion, rx_subagent_completion) = mpsc::unbounded_channel(); |
| 408 | let cancel_token = CancellationToken::new(); |
| 409 | let shared_cancel_token = Arc::new(StdMutex::new(cancel_token.clone())); |
| 410 | let tool_exec_lock = Arc::new(RwLock::new(())); |
| 411 | |
| 412 | // Create clients for both providers |
| 413 | let (deepseek_client, deepseek_client_error) = match DeepSeekClient::new(api_config) { |
| 414 | Ok(client) => (Some(client), None), |
| 415 | Err(err) => (None, Some(err.to_string())), |
| 416 | }; |
| 417 | let api_key_env_only_recovery = Self::env_only_api_key_recovery_hint(api_config); |
| 418 | |
| 419 | let mut session = Session::new( |
| 420 | config.model.clone(), |
| 421 | config.workspace.clone(), |
| 422 | config.allow_shell, |
| 423 | config.trust_mode, |
| 424 | config.notes_path.clone(), |
| 425 | config.mcp_config_path.clone(), |
| 426 | ); |
| 427 | // Set up stable system prompt with project context (default to agent mode). |
| 428 | // Per-turn working-set metadata is injected into the latest user |
| 429 | // message at request time so file churn does not rewrite this prefix. |
| 430 | let user_memory_block = |
| 431 | crate::memory::compose_block(config.memory_enabled, &config.memory_path); |
| 432 | let system_prompt = |
| 433 | prompts::system_prompt_for_mode_with_context_skills_session_and_approval( |
| 434 | AppMode::Agent, |
| 435 | &config.workspace, |
| 436 | None, |
| 437 | Some(&config.skills_dir), |
| 438 | Some(&config.instructions), |
| 439 | prompts::PromptSessionContext { |
| 440 | user_memory_block: user_memory_block.as_deref(), |
| 441 | goal_objective: config.goal_objective.as_deref(), |
| 442 | locale_tag: &config.locale_tag, |
| 443 | }, |
| 444 | session.approval_mode, |
| 445 | ); |
| 446 | let stable_prompt = Some(system_prompt); |
| 447 | session.last_system_prompt_hash = Some(system_prompt_hash(stable_prompt.as_ref())); |
| 448 | session.system_prompt = stable_prompt; |
| 449 | |
| 450 | let subagent_manager = |
| 451 | new_shared_subagent_manager(config.workspace.clone(), config.max_subagents); |
| 452 | let shell_manager = config |
| 453 | .runtime_services |
| 454 | .shell_manager |
| 455 | .clone() |
| 456 | .unwrap_or_else(|| new_shared_shell_manager(config.workspace.clone())); |
| 457 | let capacity_controller = CapacityController::new(config.capacity.clone()); |
| 458 | |
| 459 | // Create Flash seam manager for layered context (#159). v0.7.5 keeps |
| 460 | // this opt-in until the prefix-cache audit proves when seam production |
| 461 | // is worth the extra request and transcript mutation. |
| 462 | let seam_manager = deepseek_client.as_ref().map(|main_client| { |
| 463 | let seam_config = SeamConfig { |
| 464 | enabled: api_config.context.enabled.unwrap_or(false), |
| 465 | verbatim_window_turns: api_config |
| 466 | .context |
| 467 | .verbatim_window_turns |
| 468 | .unwrap_or(crate::seam_manager::VERBATIM_WINDOW_TURNS), |
| 469 | l1_threshold: api_config |
| 470 | .context |
| 471 | .l1_threshold |
| 472 | .unwrap_or(crate::seam_manager::DEFAULT_L1_THRESHOLD), |
| 473 | l2_threshold: api_config |
| 474 | .context |
| 475 | .l2_threshold |
| 476 | .unwrap_or(crate::seam_manager::DEFAULT_L2_THRESHOLD), |
| 477 | l3_threshold: api_config |
| 478 | .context |
| 479 | .l3_threshold |
| 480 | .unwrap_or(crate::seam_manager::DEFAULT_L3_THRESHOLD), |
| 481 | cycle_threshold: api_config |
| 482 | .context |
| 483 | .cycle_threshold |
| 484 | .unwrap_or(crate::seam_manager::DEFAULT_CYCLE_THRESHOLD), |
| 485 | seam_model: api_config |
| 486 | .context |
| 487 | .seam_model |
| 488 | .clone() |
| 489 | .unwrap_or_else(|| crate::seam_manager::DEFAULT_SEAM_MODEL.to_string()), |
| 490 | }; |
| 491 | SeamManager::new(main_client.clone(), seam_config) |
| 492 | }); |
| 493 | |
| 494 | let lsp_manager = Arc::new(match config.lsp_config.clone() { |
| 495 | Some(cfg) => crate::lsp::LspManager::new(cfg, config.workspace.clone()), |
| 496 | None => crate::lsp::LspManager::disabled(), |
| 497 | }); |
| 498 | |
| 499 | // Workshop variable store (#548). Created unconditionally so the Arc |
| 500 | // can be handed to every ToolContext; routing is gated on the router |
| 501 | // field being Some rather than on the vars Arc being present. |
| 502 | let workshop_vars: Option< |
| 503 | std::sync::Arc< |
| 504 | tokio::sync::Mutex<crate::tools::large_output_router::WorkshopVariables>, |
| 505 | >, |
| 506 | > = if config.workshop.is_some() { |
| 507 | Some(std::sync::Arc::new(tokio::sync::Mutex::new( |
| 508 | crate::tools::large_output_router::WorkshopVariables::default(), |
| 509 | ))) |
| 510 | } else { |
| 511 | None |
| 512 | }; |
| 513 | |
| 514 | // External sandbox backend (#516). Logged but non-fatal: if the |
| 515 | // backend fails to construct, the engine continues with local |
| 516 | // execution as the fallback. |
| 517 | let sandbox_backend = crate::sandbox::backend::create_backend(api_config) |
| 518 | .unwrap_or_else(|e| { |
| 519 | tracing::warn!("Failed to create sandbox backend: {e}"); |
| 520 | None |
| 521 | }) |
| 522 | .map(std::sync::Arc::from); |
| 523 | |
| 524 | let mut engine = Engine { |
| 525 | config, |
| 526 | deepseek_client, |
| 527 | deepseek_client_error, |
| 528 | api_key_env_only_recovery, |
| 529 | session, |
| 530 | subagent_manager, |
| 531 | shell_manager, |
| 532 | mcp_pool: None, |
| 533 | rx_op, |
| 534 | rx_approval, |
| 535 | rx_user_input, |
| 536 | rx_steer, |
| 537 | tx_event, |
| 538 | tx_subagent_completion, |
| 539 | rx_subagent_completion, |
| 540 | cancel_token: cancel_token.clone(), |
| 541 | shared_cancel_token: shared_cancel_token.clone(), |
| 542 | tool_exec_lock, |
| 543 | capacity_controller, |
| 544 | seam_manager, |
| 545 | coherence_state: CoherenceState::default(), |
| 546 | turn_counter: 0, |
| 547 | lsp_manager, |
| 548 | pending_lsp_blocks: Vec::new(), |
| 549 | workshop_vars, |
| 550 | sandbox_backend, |
| 551 | }; |
| 552 | engine.rehydrate_latest_canonical_state(); |
| 553 | |
| 554 | let handle = EngineHandle { |
| 555 | tx_op, |
| 556 | rx_event: Arc::new(RwLock::new(rx_event)), |
| 557 | cancel_token: shared_cancel_token, |
| 558 | tx_approval, |
| 559 | tx_user_input, |
| 560 | tx_steer, |
| 561 | }; |
| 562 | |
| 563 | (engine, handle) |
| 564 | } |
| 565 | |
| 566 | /// Run the engine event loop |
| 567 | #[allow(clippy::too_many_lines)] |
| 568 | pub async fn run(mut self) { |
| 569 | while let Some(op) = self.rx_op.recv().await { |
| 570 | match op { |
| 571 | Op::SendMessage { |
| 572 | content, |
| 573 | mode, |
| 574 | model, |
| 575 | goal_objective, |
| 576 | reasoning_effort, |
| 577 | reasoning_effort_auto, |
| 578 | auto_model, |
| 579 | allow_shell, |
| 580 | trust_mode, |
| 581 | auto_approve, |
| 582 | approval_mode, |
| 583 | } => { |
| 584 | self.handle_send_message( |
| 585 | content, |
| 586 | mode, |
| 587 | model, |
| 588 | goal_objective, |
| 589 | reasoning_effort, |
| 590 | reasoning_effort_auto, |
| 591 | auto_model, |
| 592 | allow_shell, |
| 593 | trust_mode, |
| 594 | auto_approve, |
| 595 | approval_mode, |
| 596 | ) |
| 597 | .await; |
| 598 | } |
| 599 | Op::CancelRequest => { |
| 600 | self.cancel_token.cancel(); |
| 601 | self.reset_cancel_token(); |
| 602 | } |
| 603 | Op::ApproveToolCall { id } => { |
| 604 | // Tool approval handling will be implemented in tools module |
| 605 | let _ = self |
| 606 | .tx_event |
| 607 | .send(Event::status(format!("Approved tool call: {id}"))) |
| 608 | .await; |
| 609 | } |
| 610 | Op::DenyToolCall { id } => { |
| 611 | let _ = self |
| 612 | .tx_event |
| 613 | .send(Event::status(format!("Denied tool call: {id}"))) |
| 614 | .await; |
| 615 | } |
| 616 | Op::SpawnSubAgent { prompt } => { |
| 617 | let Some(client) = self.deepseek_client.clone() else { |
| 618 | let message = self |
| 619 | .deepseek_client_error |
| 620 | .as_deref() |
| 621 | .map(|err| format!("Failed to spawn sub-agent: {err}")) |
| 622 | .unwrap_or_else(|| { |
| 623 | "Failed to spawn sub-agent: API client not configured".to_string() |
| 624 | }); |
| 625 | let _ = self |
| 626 | .tx_event |
| 627 | .send(Event::error(ErrorEnvelope::fatal(message))) |
| 628 | .await; |
| 629 | continue; |
| 630 | }; |
| 631 | |
| 632 | let mut runtime = SubAgentRuntime::new( |
| 633 | client, |
| 634 | self.session.model.clone(), |
| 635 | // Sub-agents don't inherit YOLO mode - use Agent mode defaults |
| 636 | self.build_tool_context(AppMode::Agent, self.session.auto_approve), |
| 637 | self.session.allow_shell, |
| 638 | Some(self.tx_event.clone()), |
| 639 | Arc::clone(&self.subagent_manager), |
| 640 | ) |
| 641 | .with_role_models(self.config.subagent_model_overrides.clone()) |
| 642 | .with_auto_model(self.session.auto_model) |
| 643 | .with_reasoning_effort( |
| 644 | self.session.reasoning_effort.clone(), |
| 645 | self.session.reasoning_effort_auto, |
| 646 | ) |
| 647 | .with_max_spawn_depth(self.config.max_spawn_depth) |
| 648 | .background_runtime(); |
| 649 | let route = resolve_subagent_assignment_route(&runtime, None, &prompt).await; |
| 650 | runtime.model = route.model; |
| 651 | runtime.reasoning_effort = route.reasoning_effort; |
| 652 | runtime.reasoning_effort_auto = false; |
| 653 | |
| 654 | let result = { |
| 655 | let mut manager = self.subagent_manager.write().await; |
| 656 | manager.spawn_background( |
| 657 | Arc::clone(&self.subagent_manager), |
| 658 | runtime, |
| 659 | SubAgentType::General, |
| 660 | prompt.clone(), |
| 661 | None, |
| 662 | ) |
| 663 | }; |
| 664 | |
| 665 | match result { |
| 666 | Ok(snapshot) => { |
| 667 | let _ = self |
| 668 | .tx_event |
| 669 | .send(Event::status(format!( |
| 670 | "Spawned sub-agent {}", |
| 671 | snapshot.agent_id |
| 672 | ))) |
| 673 | .await; |
| 674 | } |
| 675 | Err(err) => { |
| 676 | let _ = self |
| 677 | .tx_event |
| 678 | .send(Event::error(ErrorEnvelope::fatal(format!( |
| 679 | "Failed to spawn sub-agent: {err}" |
| 680 | )))) |
| 681 | .await; |
| 682 | } |
| 683 | } |
| 684 | } |
| 685 | Op::ListSubAgents => { |
| 686 | let agents = { |
| 687 | let mut manager = self.subagent_manager.write().await; |
| 688 | manager.cleanup(Duration::from_secs(60 * 60)); |
| 689 | manager.list() |
| 690 | }; |
| 691 | let _ = self.tx_event.send(Event::AgentList { agents }).await; |
| 692 | } |
| 693 | Op::ChangeMode { mode } => { |
| 694 | let _ = self |
| 695 | .tx_event |
| 696 | .send(Event::status(format!("Mode changed to: {mode:?}"))) |
| 697 | .await; |
| 698 | } |
| 699 | Op::SetModel { model } => { |
| 700 | self.session.auto_model = model.trim().eq_ignore_ascii_case("auto"); |
| 701 | self.session.model = model; |
| 702 | self.config.model.clone_from(&self.session.model); |
| 703 | let _ = self |
| 704 | .tx_event |
| 705 | .send(Event::status(format!( |
| 706 | "Model set to: {}", |
| 707 | self.session.model |
| 708 | ))) |
| 709 | .await; |
| 710 | } |
| 711 | Op::SetCompaction { config } => { |
| 712 | let enabled = config.enabled; |
| 713 | self.config.compaction = config; |
| 714 | let _ = self |
| 715 | .tx_event |
| 716 | .send(Event::status(format!( |
| 717 | "Auto-compaction {}", |
| 718 | if enabled { "enabled" } else { "disabled" } |
| 719 | ))) |
| 720 | .await; |
| 721 | } |
| 722 | Op::SyncSession { |
| 723 | messages, |
| 724 | system_prompt, |
| 725 | model, |
| 726 | workspace, |
| 727 | } => { |
| 728 | self.session.messages = messages; |
| 729 | self.session.compaction_summary_prompt = |
| 730 | extract_compaction_summary_prompt(system_prompt.clone()); |
| 731 | self.session.system_prompt = system_prompt; |
| 732 | self.session.auto_model = model.trim().eq_ignore_ascii_case("auto"); |
| 733 | self.session.model = model; |
| 734 | self.session.workspace = workspace.clone(); |
| 735 | self.config.model.clone_from(&self.session.model); |
| 736 | self.config.workspace = workspace.clone(); |
| 737 | let ctx = crate::project_context::load_project_context_with_parents(&workspace); |
| 738 | self.session.project_context = if ctx.has_instructions() { |
| 739 | Some(ctx) |
| 740 | } else { |
| 741 | None |
| 742 | }; |
| 743 | self.session.rebuild_working_set(); |
| 744 | self.rehydrate_latest_canonical_state(); |
| 745 | self.emit_session_updated().await; |
| 746 | let _ = self |
| 747 | .tx_event |
| 748 | .send(Event::status("Session context synced".to_string())) |
| 749 | .await; |
| 750 | } |
| 751 | Op::CompactContext => { |
| 752 | self.handle_manual_compaction().await; |
| 753 | } |
| 754 | Op::Rlm { |
| 755 | content, |
| 756 | model, |
| 757 | child_model, |
| 758 | max_depth, |
| 759 | } => { |
| 760 | self.handle_rlm(content, model, child_model, max_depth) |
| 761 | .await; |
| 762 | } |
| 763 | Op::EditLastTurn { new_message } => { |
| 764 | // #383: /edit — remove the last user+assistant exchange |
| 765 | // from the session, then re-send with the new content. |
| 766 | // Pop messages from the tail until we've removed the |
| 767 | // most recent user message and everything after it. |
| 768 | // First, find the last user message index. |
| 769 | let mut cut = None; |
| 770 | for (idx, msg) in self.session.messages.iter().enumerate().rev() { |
| 771 | if msg.role == "user" { |
| 772 | cut = Some(idx); |
| 773 | break; |
| 774 | } |
| 775 | } |
| 776 | if let Some(idx) = cut { |
| 777 | self.session.messages.truncate(idx); |
| 778 | } |
| 779 | // Now dispatch the new message as a normal send, |
| 780 | // reusing the engine's stored mode/model config. |
| 781 | let mode = AppMode::Agent; // default fallback |
| 782 | self.handle_send_message( |
| 783 | new_message, |
| 784 | mode, |
| 785 | self.session.model.clone(), |
| 786 | self.config.goal_objective.clone(), |
| 787 | self.session.reasoning_effort.clone(), |
| 788 | self.session.reasoning_effort_auto, |
| 789 | self.session.auto_model, |
| 790 | self.session.allow_shell, |
| 791 | self.session.trust_mode, |
| 792 | self.session.auto_approve, |
| 793 | self.session.approval_mode, |
| 794 | ) |
| 795 | .await; |
| 796 | } |
| 797 | Op::Shutdown => { |
| 798 | break; |
| 799 | } |
| 800 | } |
| 801 | } |
| 802 | |
| 803 | // #420: graceful MCP shutdown — send SIGTERM and give stdio servers |
| 804 | // a brief window to exit before drop fires SIGKILL via kill_on_drop. |
| 805 | // Best-effort: pool may not exist (no MCP configured) and the lock |
| 806 | // can fail under contention; either way the kill_on_drop fallback |
| 807 | // still reaps the children. |
| 808 | if let Some(pool) = self.mcp_pool.as_ref() { |
| 809 | let mut guard = pool.lock().await; |
| 810 | guard.shutdown_all().await; |
| 811 | } |
| 812 | } |
| 813 | |
| 814 | async fn emit_session_updated(&self) { |
| 815 | let _ = self |
| 816 | .tx_event |
| 817 | .send(Event::SessionUpdated { |
| 818 | messages: self.session.messages.clone(), |
| 819 | system_prompt: self.session.system_prompt.clone(), |
| 820 | model: self.session.model.clone(), |
| 821 | workspace: self.session.workspace.clone(), |
| 822 | }) |
| 823 | .await; |
| 824 | } |
| 825 | |
| 826 | async fn add_session_message(&mut self, message: Message) { |
| 827 | self.session.add_message(message); |
| 828 | self.emit_session_updated().await; |
| 829 | } |
| 830 | |
| 831 | /// Handle a send message operation |
| 832 | #[allow(clippy::too_many_arguments)] |
| 833 | async fn handle_send_message( |
| 834 | &mut self, |
| 835 | content: String, |
| 836 | mode: AppMode, |
| 837 | model: String, |
| 838 | goal_objective: Option<String>, |
| 839 | reasoning_effort: Option<String>, |
| 840 | reasoning_effort_auto: bool, |
| 841 | auto_model: bool, |
| 842 | allow_shell: bool, |
| 843 | trust_mode: bool, |
| 844 | auto_approve: bool, |
| 845 | approval_mode: crate::tui::approval::ApprovalMode, |
| 846 | ) { |
| 847 | // Reset cancel token for fresh turn (in case previous was cancelled) |
| 848 | self.reset_cancel_token(); |
| 849 | |
| 850 | // Drain stale steer messages from previous turns. |
| 851 | while self.rx_steer.try_recv().is_ok() {} |
| 852 | |
| 853 | // Create turn context first so start event includes a stable turn id. |
| 854 | let mut turn = TurnContext::new(self.config.max_steps); |
| 855 | self.turn_counter = self.turn_counter.saturating_add(1); |
| 856 | self.capacity_controller.mark_turn_start(self.turn_counter); |
| 857 | |
| 858 | // Snapshot the workspace BEFORE we touch a single tool. Run the git |
| 859 | // work on the blocking pool so the async runtime stays responsive; |
| 860 | // failure is non-fatal (the helper logs at WARN). |
| 861 | if self.config.snapshots_enabled { |
| 862 | let pre_workspace = self.session.workspace.clone(); |
| 863 | let pre_seq = self.turn_counter; |
| 864 | let _ = tokio::task::spawn_blocking(move || pre_turn_snapshot(&pre_workspace, pre_seq)) |
| 865 | .await; |
| 866 | } |
| 867 | |
| 868 | // Emit turn started event |
| 869 | let _ = self |
| 870 | .tx_event |
| 871 | .send(Event::TurnStarted { |
| 872 | turn_id: turn.id.clone(), |
| 873 | }) |
| 874 | .await; |
| 875 | |
| 876 | // A new turn means any leftover retry banner (success cleared |
| 877 | // it, failure pinned it) is no longer relevant — reset to idle |
| 878 | // so the footer doesn't display a stale failure row across |
| 879 | // turns (#499). |
| 880 | crate::retry_status::clear(); |
| 881 | |
| 882 | // Check if we have the appropriate client |
| 883 | if self.deepseek_client.is_none() { |
| 884 | let message = self |
| 885 | .deepseek_client_error |
| 886 | .as_deref() |
| 887 | .map(|err| format!("Failed to send message: {err}")) |
| 888 | .unwrap_or_else(|| "Failed to send message: API client not configured".to_string()); |
| 889 | let _ = self |
| 890 | .tx_event |
| 891 | .send(Event::error(ErrorEnvelope::fatal_auth(message.clone()))) |
| 892 | .await; |
| 893 | let _ = self |
| 894 | .tx_event |
| 895 | .send(Event::TurnComplete { |
| 896 | usage: turn.usage.clone(), |
| 897 | status: TurnOutcomeStatus::Failed, |
| 898 | error: Some(message), |
| 899 | }) |
| 900 | .await; |
| 901 | return; |
| 902 | } |
| 903 | |
| 904 | self.session |
| 905 | .working_set |
| 906 | .observe_user_message(&content, &self.session.workspace); |
| 907 | let force_update_plan_first = should_force_update_plan_first(mode, &content); |
| 908 | |
| 909 | // Add user message to session |
| 910 | let user_msg = Message { |
| 911 | role: "user".to_string(), |
| 912 | content: vec![ContentBlock::Text { |
| 913 | text: content, |
| 914 | cache_control: None, |
| 915 | }], |
| 916 | }; |
| 917 | self.session.add_message(user_msg); |
| 918 | |
| 919 | self.session.model = model; |
| 920 | self.config.model.clone_from(&self.session.model); |
| 921 | self.config.goal_objective = goal_objective; |
| 922 | self.session.reasoning_effort = reasoning_effort; |
| 923 | self.session.reasoning_effort_auto = reasoning_effort_auto; |
| 924 | self.session.auto_model = auto_model; |
| 925 | self.session.allow_shell = allow_shell; |
| 926 | self.config.allow_shell = allow_shell; |
| 927 | self.session.trust_mode = trust_mode; |
| 928 | self.config.trust_mode = trust_mode; |
| 929 | self.session.auto_approve = auto_approve; |
| 930 | self.session.approval_mode = if auto_approve { |
| 931 | crate::tui::approval::ApprovalMode::Auto |
| 932 | } else { |
| 933 | approval_mode |
| 934 | }; |
| 935 | |
| 936 | // Update system prompt to match current mode and include persisted compaction context. |
| 937 | self.refresh_system_prompt(mode); |
| 938 | self.emit_session_updated().await; |
| 939 | |
| 940 | // Build tool registry and tool list for the current mode |
| 941 | let todo_list = self.config.todos.clone(); |
| 942 | let plan_state = self.config.plan_state.clone(); |
| 943 | |
| 944 | let tool_context = self.build_tool_context(mode, auto_approve); |
| 945 | let builder = self.build_turn_tool_registry_builder(mode, todo_list, plan_state); |
| 946 | |
| 947 | // Mailbox for structured sub-agent envelopes (#128/#130). One per |
| 948 | // turn: the receiver is drained by a short-lived task that converts |
| 949 | // envelopes into `Event::SubAgentMailbox` so the UI can route them |
| 950 | // to the matching in-transcript card. The drainer exits naturally |
| 951 | // when every cloned sender is dropped at turn-end. |
| 952 | let mailbox_for_runtime = if self.config.features.enabled(Feature::Subagents) { |
| 953 | let cancel_token = self.cancel_token.child_token(); |
| 954 | let (mailbox, mut receiver) = Mailbox::new(cancel_token.clone()); |
| 955 | let tx_event_clone = self.tx_event.clone(); |
| 956 | spawn_supervised( |
| 957 | "subagent-mailbox-drainer", |
| 958 | std::panic::Location::caller(), |
| 959 | async move { |
| 960 | while let Some(envelope) = receiver.recv().await { |
| 961 | if tx_event_clone |
| 962 | .send(Event::SubAgentMailbox { |
| 963 | seq: envelope.seq, |
| 964 | message: envelope.message, |
| 965 | }) |
| 966 | .await |
| 967 | .is_err() |
| 968 | { |
| 969 | break; |
| 970 | } |
| 971 | } |
| 972 | }, |
| 973 | ); |
| 974 | Some((mailbox, cancel_token)) |
| 975 | } else { |
| 976 | None |
| 977 | }; |
| 978 | |
| 979 | let tool_registry = match mode { |
| 980 | AppMode::Agent | AppMode::Yolo => { |
| 981 | if self.config.features.enabled(Feature::Subagents) { |
| 982 | let runtime = if let Some(client) = self.deepseek_client.clone() { |
| 983 | let mut rt = SubAgentRuntime::new( |
| 984 | client, |
| 985 | self.session.model.clone(), |
| 986 | tool_context.clone(), |
| 987 | self.session.allow_shell, |
| 988 | Some(self.tx_event.clone()), |
| 989 | Arc::clone(&self.subagent_manager), |
| 990 | ) |
| 991 | .with_role_models(self.config.subagent_model_overrides.clone()) |
| 992 | .with_auto_model(self.session.auto_model) |
| 993 | .with_reasoning_effort( |
| 994 | self.session.reasoning_effort.clone(), |
| 995 | self.session.reasoning_effort_auto, |
| 996 | ) |
| 997 | .with_max_spawn_depth(self.config.max_spawn_depth) |
| 998 | .with_parent_completion_tx(self.tx_subagent_completion.clone()); |
| 999 | if let Some((mailbox, cancel_token)) = mailbox_for_runtime.as_ref() { |
| 1000 | rt = rt |
| 1001 | .with_mailbox(mailbox.clone()) |
| 1002 | .with_cancel_token(cancel_token.clone()); |
| 1003 | } |
| 1004 | Some(rt) |
| 1005 | } else { |
| 1006 | None |
| 1007 | }; |
| 1008 | Some( |
| 1009 | builder |
| 1010 | .with_subagent_tools( |
| 1011 | self.subagent_manager.clone(), |
| 1012 | runtime.expect("sub-agent runtime should exist with active client"), |
| 1013 | ) |
| 1014 | .build(tool_context), |
| 1015 | ) |
| 1016 | } else { |
| 1017 | Some(builder.build(tool_context)) |
| 1018 | } |
| 1019 | } |
| 1020 | _ => Some(builder.build(tool_context)), |
| 1021 | }; |
| 1022 | |
| 1023 | let mcp_tools = if self.config.features.enabled(Feature::Mcp) { |
| 1024 | self.mcp_tools().await |
| 1025 | } else { |
| 1026 | Vec::new() |
| 1027 | }; |
| 1028 | let tools = tool_registry.as_ref().map(|registry| { |
| 1029 | build_model_tool_catalog(registry.to_api_tools_with_cache(true), mcp_tools, mode) |
| 1030 | }); |
| 1031 | |
| 1032 | // Main turn loop |
| 1033 | let (status, error) = self |
| 1034 | .handle_deepseek_turn( |
| 1035 | &mut turn, |
| 1036 | tool_registry.as_ref(), |
| 1037 | tools, |
| 1038 | mode, |
| 1039 | force_update_plan_first, |
| 1040 | ) |
| 1041 | .await; |
| 1042 | |
| 1043 | // Checkpoint-restart cycle boundary (issue #124). Run BEFORE |
| 1044 | // TurnComplete so the engine loop doesn't block the terminal after |
| 1045 | // the turn signal (#234). The status chip ("↻ context refreshing...") |
| 1046 | // is visible during the wait, and once TurnComplete fires the |
| 1047 | // terminal is immediately responsive. No-op unless the estimated |
| 1048 | // input tokens have crossed the per-cycle threshold. |
| 1049 | if matches!(status, TurnOutcomeStatus::Completed) { |
| 1050 | self.maybe_advance_cycle(mode).await; |
| 1051 | } |
| 1052 | |
| 1053 | // Update session usage |
| 1054 | self.session.total_usage.add(&turn.usage); |
| 1055 | |
| 1056 | // Emit turn complete event — after all post-turn bookkeeping so |
| 1057 | // the terminal is immediately responsive when the UI receives it. |
| 1058 | let _ = self |
| 1059 | .tx_event |
| 1060 | .send(Event::TurnComplete { |
| 1061 | usage: turn.usage, |
| 1062 | status, |
| 1063 | error, |
| 1064 | }) |
| 1065 | .await; |
| 1066 | |
| 1067 | // Post-turn snapshot. Fire-and-forget: TurnComplete is already |
| 1068 | // emitted, so the UI is unblocked and the user can type / select / |
| 1069 | // paste immediately (#234). The git work proceeds on the blocking |
| 1070 | // pool without forcing the engine loop to await it. |
| 1071 | if self.config.snapshots_enabled { |
| 1072 | let post_workspace = self.session.workspace.clone(); |
| 1073 | let post_seq = self.turn_counter; |
| 1074 | crate::utils::spawn_blocking_supervised("post-turn-snapshot", move || { |
| 1075 | post_turn_snapshot(&post_workspace, post_seq); |
| 1076 | }); |
| 1077 | } |
| 1078 | } |
| 1079 | |
| 1080 | async fn handle_manual_compaction(&mut self) { |
| 1081 | let id = format!("compact_{}", &uuid::Uuid::new_v4().to_string()[..8]); |
| 1082 | let zero_usage = Usage { |
| 1083 | input_tokens: 0, |
| 1084 | output_tokens: 0, |
| 1085 | ..Usage::default() |
| 1086 | }; |
| 1087 | let Some(client) = self.deepseek_client.clone() else { |
| 1088 | let message = "Manual compaction unavailable: API client not configured".to_string(); |
| 1089 | self.emit_compaction_failed(id, false, message.clone()) |
| 1090 | .await; |
| 1091 | let _ = self |
| 1092 | .tx_event |
| 1093 | .send(Event::error(ErrorEnvelope::fatal_auth(message.clone()))) |
| 1094 | .await; |
| 1095 | let _ = self |
| 1096 | .tx_event |
| 1097 | .send(Event::TurnComplete { |
| 1098 | usage: zero_usage, |
| 1099 | status: TurnOutcomeStatus::Failed, |
| 1100 | error: Some(message), |
| 1101 | }) |
| 1102 | .await; |
| 1103 | return; |
| 1104 | }; |
| 1105 | |
| 1106 | let start_message = "Manual context compaction started".to_string(); |
| 1107 | self.emit_compaction_started(id.clone(), false, start_message) |
| 1108 | .await; |
| 1109 | |
| 1110 | let compaction_pins = self |
| 1111 | .session |
| 1112 | .working_set |
| 1113 | .pinned_message_indices(&self.session.messages, &self.session.workspace); |
| 1114 | let compaction_paths = self.session.working_set.top_paths(24); |
| 1115 | let messages_before = self.session.messages.len(); |
| 1116 | let mut turn_status = TurnOutcomeStatus::Completed; |
| 1117 | let mut turn_error = None; |
| 1118 | |
| 1119 | match compact_messages_safe( |
| 1120 | &client, |
| 1121 | &self.session.messages, |
| 1122 | &self.config.compaction, |
| 1123 | Some(&self.session.workspace), |
| 1124 | Some(&compaction_pins), |
| 1125 | Some(&compaction_paths), |
| 1126 | ) |
| 1127 | .await |
| 1128 | { |
| 1129 | Ok(result) => { |
| 1130 | if !result.messages.is_empty() || self.session.messages.is_empty() { |
| 1131 | let messages_after = result.messages.len(); |
| 1132 | self.session.messages = result.messages; |
| 1133 | self.merge_compaction_summary(result.summary_prompt); |
| 1134 | self.emit_session_updated().await; |
| 1135 | let removed = messages_before.saturating_sub(messages_after); |
| 1136 | let message = if result.retries_used > 0 { |
| 1137 | format!( |
| 1138 | "Compaction complete: {messages_before} → {messages_after} messages ({removed} removed, {} retries)", |
| 1139 | result.retries_used |
| 1140 | ) |
| 1141 | } else { |
| 1142 | format!( |
| 1143 | "Compaction complete: {messages_before} → {messages_after} messages ({removed} removed)" |
| 1144 | ) |
| 1145 | }; |
| 1146 | self.emit_compaction_completed( |
| 1147 | id, |
| 1148 | false, |
| 1149 | message, |
| 1150 | Some(messages_before), |
| 1151 | Some(messages_after), |
| 1152 | ) |
| 1153 | .await; |
| 1154 | } else { |
| 1155 | let message = "Compaction skipped: produced empty result".to_string(); |
| 1156 | self.emit_compaction_failed(id, false, message.clone()) |
| 1157 | .await; |
| 1158 | turn_status = TurnOutcomeStatus::Failed; |
| 1159 | turn_error = Some(message); |
| 1160 | } |
| 1161 | } |
| 1162 | Err(err) => { |
| 1163 | let message = format!("Manual context compaction failed: {err}"); |
| 1164 | self.emit_compaction_failed(id, false, message.clone()) |
| 1165 | .await; |
| 1166 | let _ = self.tx_event.send(Event::status(message.clone())).await; |
| 1167 | turn_status = TurnOutcomeStatus::Failed; |
| 1168 | turn_error = Some(message); |
| 1169 | } |
| 1170 | } |
| 1171 | |
| 1172 | let _ = self |
| 1173 | .tx_event |
| 1174 | .send(Event::TurnComplete { |
| 1175 | usage: zero_usage, |
| 1176 | status: turn_status, |
| 1177 | error: turn_error, |
| 1178 | }) |
| 1179 | .await; |
| 1180 | } |
| 1181 | |
| 1182 | /// Handle a Recursive Language Model (RLM) query — Algorithm 1 from |
| 1183 | /// Zhang et al. (arXiv:2512.24601). |
| 1184 | /// |
| 1185 | /// The prompt is stored as PROMPT in a REPL variable. The root LLM |
| 1186 | /// only sees metadata about the REPL state, never the prompt text |
| 1187 | /// directly. The model generates Python code, which is executed by |
| 1188 | /// the REPL. When FINAL() is called, the loop ends. |
| 1189 | async fn handle_rlm( |
| 1190 | &mut self, |
| 1191 | content: String, |
| 1192 | model: String, |
| 1193 | child_model: String, |
| 1194 | max_depth: u32, |
| 1195 | ) { |
| 1196 | use crate::rlm::turn::run_rlm_turn; |
| 1197 | |
| 1198 | let Some(ref client) = self.deepseek_client else { |
| 1199 | let err = self |
| 1200 | .deepseek_client_error |
| 1201 | .as_deref() |
| 1202 | .map(|s| s.to_string()) |
| 1203 | .unwrap_or_else(|| "API client not configured".to_string()); |
| 1204 | let _ = self |
| 1205 | .tx_event |
| 1206 | .send(Event::error(ErrorEnvelope::fatal_auth(format!( |
| 1207 | "RLM error: {err}" |
| 1208 | )))) |
| 1209 | .await; |
| 1210 | return; |
| 1211 | }; |
| 1212 | |
| 1213 | let _ = self |
| 1214 | .tx_event |
| 1215 | .send(Event::status("RLM turn started".to_string())) |
| 1216 | .await; |
| 1217 | |
| 1218 | let result = run_rlm_turn( |
| 1219 | client, |
| 1220 | model, |
| 1221 | content, |
| 1222 | child_model, |
| 1223 | self.tx_event.clone(), |
| 1224 | max_depth, |
| 1225 | ) |
| 1226 | .await; |
| 1227 | |
| 1228 | let has_error = result.error.is_some(); |
| 1229 | if let Some(ref err) = result.error { |
| 1230 | let _ = self |
| 1231 | .tx_event |
| 1232 | .send(Event::error(ErrorEnvelope::tool(format!( |
| 1233 | "RLM error: {err}" |
| 1234 | )))) |
| 1235 | .await; |
| 1236 | } |
| 1237 | |
| 1238 | if !result.answer.is_empty() { |
| 1239 | // Add the final answer as an assistant message in the session. |
| 1240 | self.add_session_message(crate::models::Message { |
| 1241 | role: "assistant".to_string(), |
| 1242 | content: vec![crate::models::ContentBlock::Text { |
| 1243 | text: result.answer.clone(), |
| 1244 | cache_control: None, |
| 1245 | }], |
| 1246 | }) |
| 1247 | .await; |
| 1248 | |
| 1249 | let _ = self |
| 1250 | .tx_event |
| 1251 | .send(Event::MessageDelta { |
| 1252 | index: 0, |
| 1253 | content: result.answer.clone(), |
| 1254 | }) |
| 1255 | .await; |
| 1256 | let _ = self |
| 1257 | .tx_event |
| 1258 | .send(Event::MessageComplete { index: 0 }) |
| 1259 | .await; |
| 1260 | } |
| 1261 | |
| 1262 | let _ = self |
| 1263 | .tx_event |
| 1264 | .send(Event::TurnComplete { |
| 1265 | usage: result.usage, |
| 1266 | status: if has_error { |
| 1267 | crate::core::events::TurnOutcomeStatus::Failed |
| 1268 | } else { |
| 1269 | crate::core::events::TurnOutcomeStatus::Completed |
| 1270 | }, |
| 1271 | error: result.error, |
| 1272 | }) |
| 1273 | .await; |
| 1274 | } |
| 1275 | |
| 1276 | fn estimated_input_tokens(&self) -> usize { |
| 1277 | estimate_input_tokens_conservative( |
| 1278 | &self.session.messages, |
| 1279 | self.session.system_prompt.as_ref(), |
| 1280 | ) |
| 1281 | } |
| 1282 | |
| 1283 | fn trim_oldest_messages_to_budget(&mut self, target_input_budget: usize) -> usize { |
| 1284 | let mut removed = 0usize; |
| 1285 | while self.session.messages.len() > MIN_RECENT_MESSAGES_TO_KEEP |
| 1286 | && self.estimated_input_tokens() > target_input_budget |
| 1287 | { |
| 1288 | self.session.messages.remove(0); |
| 1289 | removed = removed.saturating_add(1); |
| 1290 | } |
| 1291 | removed |
| 1292 | } |
| 1293 | |
| 1294 | async fn recover_context_overflow( |
| 1295 | &mut self, |
| 1296 | client: &DeepSeekClient, |
| 1297 | reason: &str, |
| 1298 | requested_output_tokens: u32, |
| 1299 | ) -> bool { |
| 1300 | let Some(target_budget) = |
| 1301 | context_input_budget(&self.session.model, requested_output_tokens) |
| 1302 | else { |
| 1303 | return false; |
| 1304 | }; |
| 1305 | |
| 1306 | let id = format!("compact_{}", &uuid::Uuid::new_v4().to_string()[..8]); |
| 1307 | let start_message = format!("Emergency context compaction started ({reason})"); |
| 1308 | self.emit_compaction_started(id.clone(), true, start_message) |
| 1309 | .await; |
| 1310 | |
| 1311 | let before_tokens = self.estimated_input_tokens(); |
| 1312 | let before_count = self.session.messages.len(); |
| 1313 | |
| 1314 | let mut retries_used = 0u32; |
| 1315 | let mut summary_prompt = None; |
| 1316 | let mut compacted_messages = self.session.messages.clone(); |
| 1317 | |
| 1318 | let mut forced_config = self.config.compaction.clone(); |
| 1319 | forced_config.enabled = true; |
| 1320 | forced_config.token_threshold = forced_config |
| 1321 | .token_threshold |
| 1322 | .min(target_budget.saturating_sub(1)) |
| 1323 | .max(1); |
| 1324 | // v0.8.11: forced compaction (capacity guardrail) bypasses the floor |
| 1325 | // because we're at a hard ceiling and have to free budget regardless |
| 1326 | // of cache cost. |
| 1327 | forced_config.auto_floor_tokens = 0; |
| 1328 | |
| 1329 | match compact_messages_safe( |
| 1330 | client, |
| 1331 | &self.session.messages, |
| 1332 | &forced_config, |
| 1333 | Some(&self.session.workspace), |
| 1334 | None, |
| 1335 | None, |
| 1336 | ) |
| 1337 | .await |
| 1338 | { |
| 1339 | Ok(result) => { |
| 1340 | retries_used = result.retries_used; |
| 1341 | compacted_messages = result.messages; |
| 1342 | summary_prompt = result.summary_prompt; |
| 1343 | } |
| 1344 | Err(err) => { |
| 1345 | let _ = self |
| 1346 | .tx_event |
| 1347 | .send(Event::status(format!( |
| 1348 | "Emergency compaction API pass failed: {err}. Falling back to local trim." |
| 1349 | ))) |
| 1350 | .await; |
| 1351 | } |
| 1352 | } |
| 1353 | |
| 1354 | if !compacted_messages.is_empty() || self.session.messages.is_empty() { |
| 1355 | self.session.messages = compacted_messages; |
| 1356 | } |
| 1357 | self.merge_compaction_summary(summary_prompt); |
| 1358 | |
| 1359 | let trimmed = self.trim_oldest_messages_to_budget(target_budget); |
| 1360 | self.emit_session_updated().await; |
| 1361 | let after_tokens = self.estimated_input_tokens(); |
| 1362 | let after_count = self.session.messages.len(); |
| 1363 | let recovered = after_tokens <= target_budget |
| 1364 | && (after_tokens < before_tokens || after_count < before_count || trimmed > 0); |
| 1365 | |
| 1366 | if recovered { |
| 1367 | let removed = before_count.saturating_sub(after_count); |
| 1368 | let mut details = format!( |
| 1369 | "Emergency compaction complete: {before_count} → {after_count} messages ({removed} removed), ~{before_tokens} → ~{after_tokens} tokens" |
| 1370 | ); |
| 1371 | if retries_used > 0 { |
| 1372 | details.push_str(&format!(" ({} retries)", retries_used)); |
| 1373 | } |
| 1374 | if trimmed > 0 { |
| 1375 | details.push_str(&format!(", trimmed {trimmed} oldest")); |
| 1376 | } |
| 1377 | self.emit_compaction_completed( |
| 1378 | id, |
| 1379 | true, |
| 1380 | details.clone(), |
| 1381 | Some(before_count), |
| 1382 | Some(after_count), |
| 1383 | ) |
| 1384 | .await; |
| 1385 | let _ = self.tx_event.send(Event::status(details)).await; |
| 1386 | return true; |
| 1387 | } |
| 1388 | |
| 1389 | let message = format!( |
| 1390 | "Emergency context compaction failed to reduce request below model limit \ |
| 1391 | (estimate ~{} tokens, budget ~{}).", |
| 1392 | after_tokens, target_budget |
| 1393 | ); |
| 1394 | self.emit_compaction_failed(id, true, message.clone()).await; |
| 1395 | let _ = self.tx_event.send(Event::status(message)).await; |
| 1396 | false |
| 1397 | } |
| 1398 | |
| 1399 | fn build_tool_context(&self, mode: AppMode, auto_approve: bool) -> ToolContext { |
| 1400 | // Load the per-workspace trusted-paths list (#29) on every tool-context |
| 1401 | // build. Cheap (a small JSON file) and always reflects the latest |
| 1402 | // `/trust add` / `/trust remove` mutations without an explicit cache |
| 1403 | // refresh hook. |
| 1404 | let trusted = crate::workspace_trust::WorkspaceTrust::load_for(&self.session.workspace); |
| 1405 | let mut ctx = ToolContext::with_auto_approve( |
| 1406 | self.session.workspace.clone(), |
| 1407 | self.session.trust_mode, |
| 1408 | self.session.notes_path.clone(), |
| 1409 | self.session.mcp_config_path.clone(), |
| 1410 | mode == AppMode::Yolo || auto_approve, |
| 1411 | ) |
| 1412 | .with_state_namespace(self.session.id.clone()) |
| 1413 | .with_features(self.config.features.clone()) |
| 1414 | .with_shell_manager(self.shell_manager.clone()) |
| 1415 | .with_runtime_services(self.config.runtime_services.clone()) |
| 1416 | .with_cancel_token(self.cancel_token.clone()) |
| 1417 | .with_trusted_external_paths(trusted.paths().to_vec()); |
| 1418 | |
| 1419 | // Hand the user-memory path to tools so the model-callable |
| 1420 | // `remember` tool can append entries (#489). `None` when the |
| 1421 | // feature is disabled — tools short-circuit on that. |
| 1422 | if self.config.memory_enabled { |
| 1423 | ctx.memory_path = Some(self.config.memory_path.clone()); |
| 1424 | } |
| 1425 | |
| 1426 | if let Some(decider) = self.config.network_policy.as_ref() { |
| 1427 | ctx = ctx.with_network_policy(decider.clone()); |
| 1428 | } |
| 1429 | |
| 1430 | // Wire the large-output router (#548). Only attaches when the |
| 1431 | // [workshop] config table is present; sub-agents don't inherit the |
| 1432 | // router (their ToolContext is built separately) to prevent recursive |
| 1433 | // routing of the synthesis call itself. |
| 1434 | if let Some(workshop_cfg) = self.config.workshop.as_ref() |
| 1435 | && let Some(vars_arc) = self.workshop_vars.as_ref() |
| 1436 | { |
| 1437 | let router = |
| 1438 | crate::tools::large_output_router::LargeOutputRouter::new(workshop_cfg.clone()); |
| 1439 | ctx = ctx.with_large_output_router(router, vars_arc.clone()); |
| 1440 | } |
| 1441 | |
| 1442 | // Wire the external sandbox backend (#516). exec_shell checks this |
| 1443 | // field and routes commands through the backend instead of spawning |
| 1444 | // a local process when it's set. |
| 1445 | if let Some(backend) = self.sandbox_backend.as_ref() { |
| 1446 | ctx = ctx.with_sandbox_backend(std::sync::Arc::clone(backend)); |
| 1447 | } |
| 1448 | |
| 1449 | match mode { |
| 1450 | // Plan mode is read-only investigation; the shell tool is not |
| 1451 | // registered, so leaving the sandbox policy at the seatbelt-strict |
| 1452 | // default is fine. |
| 1453 | AppMode::Plan => ctx, |
| 1454 | // Agent registers the shell tool and runs each command through |
| 1455 | // the per-mode sandbox + per-tool approval flow. The sandbox |
| 1456 | // default would deny all outbound network — including DNS — |
| 1457 | // which breaks ordinary developer commands (cargo fetch, npm |
| 1458 | // install, curl, yt-dlp, …) without buying the user any safety |
| 1459 | // the approval flow doesn't already provide. Elevate to |
| 1460 | // workspace-write + network. (#273) |
| 1461 | AppMode::Agent => { |
| 1462 | ctx.with_elevated_sandbox_policy(crate::sandbox::SandboxPolicy::WorkspaceWrite { |
| 1463 | writable_roots: vec![self.session.workspace.clone()], |
| 1464 | network_access: true, |
| 1465 | exclude_tmpdir: false, |
| 1466 | exclude_slash_tmp: false, |
| 1467 | }) |
| 1468 | } |
| 1469 | // YOLO is the explicit "no guardrails" mode — auto-approve all |
| 1470 | // tools, trust mode on, no sandbox. Workspace-write was still |
| 1471 | // intercepting commands that wanted to write outside the |
| 1472 | // workspace (rare but legitimate: pipx install, npm install |
| 1473 | // -g, brew, package-manager state under ~/.cache, sub-agent |
| 1474 | // workspaces, …) which forced approval round-trips and |
| 1475 | // contradicts the YOLO contract. The user opted into YOLO |
| 1476 | // deliberately; trust them. |
| 1477 | AppMode::Yolo => { |
| 1478 | ctx.with_elevated_sandbox_policy(crate::sandbox::SandboxPolicy::DangerFullAccess) |
| 1479 | } |
| 1480 | } |
| 1481 | } |
| 1482 | |
| 1483 | async fn ensure_mcp_pool(&mut self) -> Result<Arc<AsyncMutex<McpPool>>, ToolError> { |
| 1484 | if let Some(pool) = self.mcp_pool.as_ref() { |
| 1485 | return Ok(Arc::clone(pool)); |
| 1486 | } |
| 1487 | let mut pool = McpPool::from_config_path(&self.session.mcp_config_path) |
| 1488 | .map_err(|e| ToolError::execution_failed(format!("Failed to load MCP config: {e}")))?; |
| 1489 | if let Some(decider) = self.config.network_policy.as_ref() { |
| 1490 | pool = pool.with_network_policy(decider.clone()); |
| 1491 | } |
| 1492 | let pool = Arc::new(AsyncMutex::new(pool)); |
| 1493 | self.mcp_pool = Some(Arc::clone(&pool)); |
| 1494 | Ok(pool) |
| 1495 | } |
| 1496 | |
| 1497 | async fn mcp_tools(&mut self) -> Vec<Tool> { |
| 1498 | let pool = match self.ensure_mcp_pool().await { |
| 1499 | Ok(pool) => pool, |
| 1500 | Err(err) => { |
| 1501 | let _ = self.tx_event.send(Event::status(err.to_string())).await; |
| 1502 | return Vec::new(); |
| 1503 | } |
| 1504 | }; |
| 1505 | |
| 1506 | let mut pool = pool.lock().await; |
| 1507 | let errors = pool.connect_all().await; |
| 1508 | for (server, err) in errors { |
| 1509 | let _ = self |
| 1510 | .tx_event |
| 1511 | .send(Event::status(format!( |
| 1512 | "Failed to connect MCP server '{server}': {err}" |
| 1513 | ))) |
| 1514 | .await; |
| 1515 | } |
| 1516 | |
| 1517 | pool.to_api_tools() |
| 1518 | } |
| 1519 | |
| 1520 | /// Handle a turn using the DeepSeek API. |
| 1521 | #[allow(clippy::too_many_lines)] |
| 1522 | /// Run the pre-request layered-context checkpoint (#159). Checks whether |
| 1523 | /// the active input estimate has crossed a soft-seam threshold and, if so, |
| 1524 | /// produces an `<archived_context>` block via Flash and appends it as an |
| 1525 | /// assistant message. Called from `handle_deepseek_turn` before each API |
| 1526 | /// request so the model always has the latest navigation aids. |
| 1527 | async fn layered_context_checkpoint(&mut self) { |
| 1528 | let Some(ref seam_mgr) = self.seam_manager else { |
| 1529 | return; |
| 1530 | }; |
| 1531 | if !seam_mgr.config().enabled { |
| 1532 | return; |
| 1533 | } |
| 1534 | |
| 1535 | let highest = seam_mgr.highest_level().await; |
| 1536 | let Some(level) = seam_mgr.seam_level_for(self.estimated_input_tokens(), highest) else { |
| 1537 | return; |
| 1538 | }; |
| 1539 | |
| 1540 | // Determine the message range to summarize: everything before the |
| 1541 | // verbatim window. The verbatim window (last ~16 turns) stays |
| 1542 | // untouched so the model always has ground-truth recent context. |
| 1543 | let msg_count = self.session.messages.len(); |
| 1544 | let verbatim_start = seam_mgr.verbatim_window_start(msg_count); |
| 1545 | if verbatim_start == 0 { |
| 1546 | return; // Not enough messages to summarize. |
| 1547 | } |
| 1548 | |
| 1549 | let msg_range_end = verbatim_start; |
| 1550 | let pinned = self |
| 1551 | .session |
| 1552 | .working_set |
| 1553 | .pinned_message_indices(&self.session.messages, &self.session.workspace); |
| 1554 | |
| 1555 | let _ = self |
| 1556 | .tx_event |
| 1557 | .send(Event::status(format!( |
| 1558 | "⏻ producing L{level} context seam ({msg_range_end} messages)…" |
| 1559 | ))) |
| 1560 | .await; |
| 1561 | |
| 1562 | // If we have existing seams, recompact; otherwise produce fresh. |
| 1563 | let existing_seams = seam_mgr.collect_seam_texts(&self.session.messages).await; |
| 1564 | let seam_text = if existing_seams.is_empty() { |
| 1565 | match seam_mgr |
| 1566 | .produce_soft_seam( |
| 1567 | &self.session.messages, |
| 1568 | level, |
| 1569 | 0, |
| 1570 | msg_range_end, |
| 1571 | Some(&self.session.workspace), |
| 1572 | &pinned, |
| 1573 | ) |
| 1574 | .await |
| 1575 | { |
| 1576 | Ok(text) => text, |
| 1577 | Err(err) => { |
| 1578 | crate::logging::warn(format!("L{level} soft seam failed: {err}")); |
| 1579 | return; |
| 1580 | } |
| 1581 | } |
| 1582 | } else { |
| 1583 | let recent: Vec<&Message> = (0..msg_range_end) |
| 1584 | .filter_map(|i| self.session.messages.get(i)) |
| 1585 | .collect(); |
| 1586 | match seam_mgr |
| 1587 | .recompact(&existing_seams, &recent, level, 0, msg_range_end) |
| 1588 | .await |
| 1589 | { |
| 1590 | Ok(text) => text, |
| 1591 | Err(err) => { |
| 1592 | crate::logging::warn(format!("L{level} recompact failed: {err}")); |
| 1593 | return; |
| 1594 | } |
| 1595 | } |
| 1596 | }; |
| 1597 | |
| 1598 | if seam_text.is_empty() { |
| 1599 | return; |
| 1600 | } |
| 1601 | |
| 1602 | // Capture seam count before the mutable borrow below. |
| 1603 | let seam_count = seam_mgr.seam_count().await; |
| 1604 | |
| 1605 | // Append the seam as an assistant message. This is an append-only |
| 1606 | // operation — no messages are deleted. The prefix cache stays hot. |
| 1607 | self.add_session_message(Message { |
| 1608 | role: "assistant".to_string(), |
| 1609 | content: vec![ContentBlock::Text { |
| 1610 | text: seam_text, |
| 1611 | cache_control: None, |
| 1612 | }], |
| 1613 | }) |
| 1614 | .await; |
| 1615 | |
| 1616 | let _ = self |
| 1617 | .tx_event |
| 1618 | .send(Event::status(format!( |
| 1619 | "⏻ L{level} seam complete ({seam_count} total, {msg_range_end} messages covered)" |
| 1620 | ))) |
| 1621 | .await; |
| 1622 | } |
| 1623 | /// its token threshold (issue #124). No-op in the common case. |
| 1624 | /// |
| 1625 | /// Caller must invoke this only at a clean turn boundary (no in-flight |
| 1626 | /// tool, no open stream, no pending approval modal). The phase guard |
| 1627 | /// inside `should_advance_cycle` is a defence-in-depth check; the |
| 1628 | /// engine's wider state machine is the primary enforcement layer. |
| 1629 | /// |
| 1630 | /// Sub-agents are intentionally NOT awaited: each sub-agent has its own |
| 1631 | /// context, the parent's reset doesn't invalidate them. Their handles |
| 1632 | /// are captured in the structured-state block so the next cycle can see |
| 1633 | /// they're still running. |
| 1634 | async fn maybe_advance_cycle(&mut self, mode: AppMode) { |
| 1635 | if !should_advance_cycle( |
| 1636 | self.estimated_input_tokens() as u64, |
| 1637 | turn_response_headroom_tokens(), |
| 1638 | &self.session.model, |
| 1639 | &self.config.cycle, |
| 1640 | false, |
| 1641 | ) { |
| 1642 | return; |
| 1643 | } |
| 1644 | |
| 1645 | let Some(client) = self.deepseek_client.clone() else { |
| 1646 | crate::logging::warn( |
| 1647 | "Cycle boundary skipped: API client not configured for briefing turn", |
| 1648 | ); |
| 1649 | return; |
| 1650 | }; |
| 1651 | |
| 1652 | let from = self.session.cycle_count; |
| 1653 | let to = from.saturating_add(1); |
| 1654 | let archive_started = self.session.current_cycle_started; |
| 1655 | let max_briefing_tokens = self.config.cycle.briefing_max_for(&self.session.model); |
| 1656 | |
| 1657 | let _ = self |
| 1658 | .tx_event |
| 1659 | .send(Event::status(format!( |
| 1660 | "↻ context refreshing (cycle {from} → {to}, generating briefing…)" |
| 1661 | ))) |
| 1662 | .await; |
| 1663 | |
| 1664 | // 1. Generate the model-curated briefing. Prefer the Flash seam |
| 1665 | // manager (#159) for cost and speed; fall back to the main model |
| 1666 | // (legacy produce_briefing) when the seam manager isn't available. |
| 1667 | let briefing_text = if let Some(ref seam_mgr) = self.seam_manager { |
| 1668 | let seams = seam_mgr.collect_seam_texts(&self.session.messages).await; |
| 1669 | let state_text = { |
| 1670 | let s = StructuredState::capture( |
| 1671 | mode.label(), |
| 1672 | self.config.workspace.clone(), |
| 1673 | std::env::current_dir().ok(), |
| 1674 | &self.session.working_set, |
| 1675 | &self.config.todos, |
| 1676 | &self.config.plan_state, |
| 1677 | Some(&self.subagent_manager), |
| 1678 | ) |
| 1679 | .await; |
| 1680 | s.to_system_block() |
| 1681 | }; |
| 1682 | match seam_mgr |
| 1683 | .produce_flash_briefing(&seams, state_text.as_deref()) |
| 1684 | .await |
| 1685 | { |
| 1686 | Ok(text) => text, |
| 1687 | Err(err) => { |
| 1688 | crate::logging::warn(format!( |
| 1689 | "Flash briefing failed, falling back to main model: {err}" |
| 1690 | )); |
| 1691 | match produce_briefing( |
| 1692 | &client, |
| 1693 | &self.session.model, |
| 1694 | &self.session.messages, |
| 1695 | max_briefing_tokens, |
| 1696 | ) |
| 1697 | .await |
| 1698 | { |
| 1699 | Ok(text) => text, |
| 1700 | Err(err2) => { |
| 1701 | crate::logging::warn(format!( |
| 1702 | "Cycle briefing turn failed; skipping cycle advance: {err2}" |
| 1703 | )); |
| 1704 | let _ = self |
| 1705 | .tx_event |
| 1706 | .send(Event::status(format!( |
| 1707 | "↻ cycle handoff failed (continuing in cycle {from}): {err2}" |
| 1708 | ))) |
| 1709 | .await; |
| 1710 | return; |
| 1711 | } |
| 1712 | } |
| 1713 | } |
| 1714 | } |
| 1715 | } else { |
| 1716 | match produce_briefing( |
| 1717 | &client, |
| 1718 | &self.session.model, |
| 1719 | &self.session.messages, |
| 1720 | max_briefing_tokens, |
| 1721 | ) |
| 1722 | .await |
| 1723 | { |
| 1724 | Ok(text) => text, |
| 1725 | Err(err) => { |
| 1726 | crate::logging::warn(format!( |
| 1727 | "Cycle briefing turn failed; skipping cycle advance: {err}" |
| 1728 | )); |
| 1729 | let _ = self |
| 1730 | .tx_event |
| 1731 | .send(Event::status(format!( |
| 1732 | "↻ cycle handoff failed (continuing in cycle {from}): {err}" |
| 1733 | ))) |
| 1734 | .await; |
| 1735 | return; |
| 1736 | } |
| 1737 | } |
| 1738 | }; |
| 1739 | |
| 1740 | let briefing_tokens = estimate_briefing_tokens(&briefing_text); |
| 1741 | let now = chrono::Utc::now(); |
| 1742 | let briefing = CycleBriefing { |
| 1743 | cycle: to, |
| 1744 | timestamp: now, |
| 1745 | briefing_text: briefing_text.clone(), |
| 1746 | token_estimate: briefing_tokens, |
| 1747 | }; |
| 1748 | |
| 1749 | // 2. Archive the cycle to disk. If the archive write fails we still |
| 1750 | // proceed with the swap — the briefing alone preserves enough |
| 1751 | // state to continue, and the user can recover the lost archive |
| 1752 | // from their session log if needed. |
| 1753 | match archive_cycle( |
| 1754 | &self.session.id, |
| 1755 | to, |
| 1756 | &self.session.messages, |
| 1757 | &self.session.model, |
| 1758 | archive_started, |
| 1759 | ) { |
| 1760 | Ok(path) => { |
| 1761 | crate::logging::info(format!("Cycle {to} archived to {}", path.display())); |
| 1762 | } |
| 1763 | Err(err) => { |
| 1764 | crate::logging::warn(format!( |
| 1765 | "Failed to archive cycle {to}; continuing with swap: {err}" |
| 1766 | )); |
| 1767 | } |
| 1768 | } |
| 1769 | |
| 1770 | // 3. Capture structured state. Locks are held only for the snapshot. |
| 1771 | let state = StructuredState::capture( |
| 1772 | mode.label(), |
| 1773 | self.config.workspace.clone(), |
| 1774 | std::env::current_dir().ok(), |
| 1775 | &self.session.working_set, |
| 1776 | &self.config.todos, |
| 1777 | &self.config.plan_state, |
| 1778 | Some(&self.subagent_manager), |
| 1779 | ) |
| 1780 | .await; |
| 1781 | let state_block = state.to_system_block(); |
| 1782 | |
| 1783 | // 4. Build the seed messages. The next cycle starts with the |
| 1784 | // base system prompt (refreshed below) and these seeds. |
| 1785 | let seed_messages = build_seed_messages( |
| 1786 | state_block.as_deref(), |
| 1787 | Some(&briefing), |
| 1788 | None, // pending_user_message — pulled from steer/queue elsewhere |
| 1789 | ); |
| 1790 | |
| 1791 | // 5. Atomic swap. |
| 1792 | self.session.messages = seed_messages; |
| 1793 | self.session.cycle_count = to; |
| 1794 | self.session.current_cycle_started = now; |
| 1795 | self.session.cycle_briefings.push(briefing.clone()); |
| 1796 | // Reset seam tracking for the new cycle. |
| 1797 | if let Some(ref seam_mgr) = self.seam_manager { |
| 1798 | seam_mgr.reset().await; |
| 1799 | } |
| 1800 | // Drop any compaction summary — that path is incompatible with the |
| 1801 | // fresh-context model and would Frankenstein-merge with the briefing. |
| 1802 | self.session.compaction_summary_prompt = None; |
| 1803 | self.refresh_system_prompt(mode); |
| 1804 | self.emit_session_updated().await; |
| 1805 | |
| 1806 | let _ = self |
| 1807 | .tx_event |
| 1808 | .send(Event::CycleAdvanced { |
| 1809 | from, |
| 1810 | to, |
| 1811 | briefing: briefing.clone(), |
| 1812 | }) |
| 1813 | .await; |
| 1814 | let _ = self |
| 1815 | .tx_event |
| 1816 | .send(Event::status(format!( |
| 1817 | "↻ context refreshed (cycle {from} → {to}, briefing: {briefing_tokens} tokens carried)" |
| 1818 | ))) |
| 1819 | .await; |
| 1820 | } |
| 1821 | |
| 1822 | /// Refresh the system prompt based on current mode and context. |
| 1823 | fn refresh_system_prompt(&mut self, mode: AppMode) { |
| 1824 | let user_memory_block = |
| 1825 | crate::memory::compose_block(self.config.memory_enabled, &self.config.memory_path); |
| 1826 | let base = prompts::system_prompt_for_mode_with_context_skills_session_and_approval( |
| 1827 | mode, |
| 1828 | &self.config.workspace, |
| 1829 | None, |
| 1830 | Some(&self.config.skills_dir), |
| 1831 | Some(&self.config.instructions), |
| 1832 | prompts::PromptSessionContext { |
| 1833 | user_memory_block: user_memory_block.as_deref(), |
| 1834 | goal_objective: self.config.goal_objective.as_deref(), |
| 1835 | locale_tag: &self.config.locale_tag, |
| 1836 | }, |
| 1837 | self.session.approval_mode, |
| 1838 | ); |
| 1839 | let stable_prompt = |
| 1840 | merge_system_prompts(Some(&base), self.session.compaction_summary_prompt.clone()); |
| 1841 | let stable_hash = system_prompt_hash(stable_prompt.as_ref()); |
| 1842 | if self.session.last_system_prompt_hash != Some(stable_hash) { |
| 1843 | self.session.system_prompt = stable_prompt; |
| 1844 | self.session.last_system_prompt_hash = Some(stable_hash); |
| 1845 | } |
| 1846 | } |
| 1847 | |
| 1848 | fn merge_compaction_summary(&mut self, summary_prompt: Option<SystemPrompt>) { |
| 1849 | if summary_prompt.is_none() { |
| 1850 | return; |
| 1851 | } |
| 1852 | self.session.compaction_summary_prompt = merge_system_prompts( |
| 1853 | self.session.compaction_summary_prompt.as_ref(), |
| 1854 | summary_prompt.clone(), |
| 1855 | ); |
| 1856 | let merged = merge_system_prompts(self.session.system_prompt.as_ref(), summary_prompt); |
| 1857 | self.session.last_system_prompt_hash = Some(system_prompt_hash(merged.as_ref())); |
| 1858 | self.session.system_prompt = merged; |
| 1859 | } |
| 1860 | } |
| 1861 | |
| 1862 | fn system_prompt_hash(prompt: Option<&SystemPrompt>) -> u64 { |
| 1863 | let mut hasher = DefaultHasher::new(); |
| 1864 | match prompt { |
| 1865 | Some(SystemPrompt::Text(text)) => { |
| 1866 | 0u8.hash(&mut hasher); |
| 1867 | text.hash(&mut hasher); |
| 1868 | } |
| 1869 | Some(SystemPrompt::Blocks(blocks)) => { |
| 1870 | 1u8.hash(&mut hasher); |
| 1871 | for block in blocks { |
| 1872 | block.block_type.hash(&mut hasher); |
| 1873 | block.text.hash(&mut hasher); |
| 1874 | if let Some(cache_control) = &block.cache_control { |
| 1875 | cache_control.cache_type.hash(&mut hasher); |
| 1876 | } |
| 1877 | } |
| 1878 | } |
| 1879 | None => { |
| 1880 | 2u8.hash(&mut hasher); |
| 1881 | } |
| 1882 | } |
| 1883 | hasher.finish() |
| 1884 | } |
| 1885 | |
| 1886 | /// Spawn the engine in a background task |
| 1887 | pub fn spawn_engine(config: EngineConfig, api_config: &Config) -> EngineHandle { |
| 1888 | let (engine, handle) = Engine::new(config, api_config); |
| 1889 | |
| 1890 | spawn_supervised( |
| 1891 | "engine-event-loop", |
| 1892 | std::panic::Location::caller(), |
| 1893 | async move { |
| 1894 | engine.run().await; |
| 1895 | }, |
| 1896 | ); |
| 1897 | |
| 1898 | handle |
| 1899 | } |
| 1900 | |
| 1901 | #[cfg(test)] |
| 1902 | pub(crate) struct MockEngineHandle { |
| 1903 | pub handle: EngineHandle, |
| 1904 | pub rx_op: mpsc::Receiver<Op>, |
| 1905 | rx_approval: mpsc::Receiver<ApprovalDecision>, |
| 1906 | pub rx_steer: mpsc::Receiver<String>, |
| 1907 | pub tx_event: mpsc::Sender<Event>, |
| 1908 | pub cancel_token: CancellationToken, |
| 1909 | } |
| 1910 | |
| 1911 | #[cfg(test)] |
| 1912 | #[derive(Debug, Clone, PartialEq, Eq)] |
| 1913 | pub(crate) enum MockApprovalEvent { |
| 1914 | Approved { |
| 1915 | id: String, |
| 1916 | }, |
| 1917 | Denied { |
| 1918 | id: String, |
| 1919 | }, |
| 1920 | RetryWithPolicy { |
| 1921 | id: String, |
| 1922 | policy: crate::sandbox::SandboxPolicy, |
| 1923 | }, |
| 1924 | } |
| 1925 | |
| 1926 | #[cfg(test)] |
| 1927 | impl MockEngineHandle { |
| 1928 | pub(crate) async fn recv_approval_event(&mut self) -> Option<MockApprovalEvent> { |
| 1929 | match self.rx_approval.recv().await? { |
| 1930 | ApprovalDecision::Approved { id } => Some(MockApprovalEvent::Approved { id }), |
| 1931 | ApprovalDecision::Denied { id } => Some(MockApprovalEvent::Denied { id }), |
| 1932 | ApprovalDecision::RetryWithPolicy { id, policy } => { |
| 1933 | Some(MockApprovalEvent::RetryWithPolicy { id, policy }) |
| 1934 | } |
| 1935 | } |
| 1936 | } |
| 1937 | } |
| 1938 | |
| 1939 | #[cfg(test)] |
| 1940 | pub(crate) fn mock_engine_handle() -> MockEngineHandle { |
| 1941 | let (tx_op, rx_op) = mpsc::channel(32); |
| 1942 | let (tx_event, rx_event) = mpsc::channel(256); |
| 1943 | let (tx_approval, rx_approval) = mpsc::channel(64); |
| 1944 | let (tx_user_input, _rx_user_input) = mpsc::channel(32); |
| 1945 | let (tx_steer, rx_steer) = mpsc::channel(64); |
| 1946 | let cancel_token = CancellationToken::new(); |
| 1947 | let shared_cancel_token = Arc::new(StdMutex::new(cancel_token.clone())); |
| 1948 | let handle = EngineHandle { |
| 1949 | tx_op, |
| 1950 | rx_event: Arc::new(RwLock::new(rx_event)), |
| 1951 | cancel_token: shared_cancel_token, |
| 1952 | tx_approval, |
| 1953 | tx_user_input, |
| 1954 | tx_steer, |
| 1955 | }; |
| 1956 | |
| 1957 | MockEngineHandle { |
| 1958 | handle, |
| 1959 | rx_op, |
| 1960 | rx_approval, |
| 1961 | rx_steer, |
| 1962 | tx_event, |
| 1963 | cancel_token, |
| 1964 | } |
| 1965 | } |
| 1966 | |
| 1967 | mod approval; |
| 1968 | mod capacity_flow; |
| 1969 | mod context; |
| 1970 | pub(crate) use context::compact_tool_result_for_context; |
| 1971 | use context::{ |
| 1972 | COMPACTION_SUMMARY_MARKER, MAX_CONTEXT_RECOVERY_ATTEMPTS, MIN_RECENT_MESSAGES_TO_KEEP, |
| 1973 | TURN_MAX_OUTPUT_TOKENS, context_input_budget, effective_max_output_tokens, |
| 1974 | estimate_input_tokens_conservative, extract_compaction_summary_prompt, |
| 1975 | is_context_length_error_message, summarize_text, turn_response_headroom_tokens, |
| 1976 | }; |
| 1977 | mod dispatch; |
| 1978 | mod loop_guard; |
| 1979 | mod lsp_hooks; |
| 1980 | mod streaming; |
| 1981 | mod tool_catalog; |
| 1982 | mod tool_execution; |
| 1983 | mod tool_setup; |
| 1984 | mod turn_loop; |
| 1985 | |
| 1986 | use self::approval::{ApprovalDecision, ApprovalResult, UserInputDecision}; |
| 1987 | use self::dispatch::{ |
| 1988 | ParallelToolResult, ParallelToolResultEntry, ToolExecGuard, ToolExecOutcome, ToolExecutionPlan, |
| 1989 | caller_allowed_for_tool, caller_type_for_tool_use, final_tool_input, format_tool_error, |
| 1990 | mcp_tool_approval_description, mcp_tool_is_parallel_safe, mcp_tool_is_read_only, |
| 1991 | parse_parallel_tool_calls, parse_tool_input, should_force_update_plan_first, |
| 1992 | should_parallelize_tool_batch, should_stop_after_plan_tool, |
| 1993 | }; |
| 1994 | use self::loop_guard::{AttemptDecision, LoopGuard, OutcomeDecision}; |
| 1995 | #[cfg(test)] |
| 1996 | use self::lsp_hooks::{edited_paths_for_tool, parse_patch_paths}; |
| 1997 | #[cfg(test)] |
| 1998 | use self::streaming::TOOL_CALL_START_MARKERS; |
| 1999 | use self::streaming::{ |
| 2000 | ContentBlockKind, FAKE_WRAPPER_NOTICE, MAX_STREAM_ERRORS_BEFORE_FAIL, |
| 2001 | MAX_TRANSPARENT_STREAM_RETRIES, STREAM_CHUNK_TIMEOUT_SECS, STREAM_MAX_CONTENT_BYTES, |
| 2002 | STREAM_MAX_DURATION_SECS, ToolUseState, contains_fake_tool_wrapper, filter_tool_call_delta, |
| 2003 | should_transparently_retry_stream, |
| 2004 | }; |
| 2005 | use self::tool_catalog::{ |
| 2006 | CODE_EXECUTION_TOOL_NAME, MULTI_TOOL_PARALLEL_NAME, REQUEST_USER_INPUT_NAME, |
| 2007 | active_tools_for_step, build_model_tool_catalog, ensure_advanced_tooling, |
| 2008 | execute_code_execution_tool, execute_tool_search, initial_active_tools, is_tool_search_tool, |
| 2009 | maybe_activate_requested_deferred_tool, missing_tool_error_message, |
| 2010 | }; |
| 2011 | #[cfg(test)] |
| 2012 | use self::tool_catalog::{TOOL_SEARCH_BM25_NAME, should_default_defer_tool}; |
| 2013 | use self::tool_execution::emit_tool_audit; |
| 2014 | |
| 2015 | #[cfg(test)] |
| 2016 | mod tests; |
| 2017 |