返回 CodeWhale
engine.rs
根目录 / crates / tui / src / core / engine.rs
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::hash_map::DefaultHasher;
11 use std::collections::{HashMap, HashSet};
12 use std::hash::{Hash, Hasher};
13 use std::path::{Path, PathBuf};
14 use std::sync::{Arc, Mutex as StdMutex};
15 use std::time::{Duration, Instant};
16
17 use anyhow::Result;
18 use codewhale_execpolicy::{AskForApproval, ExecPolicyContext};
19 use codewhale_protocol::runtime::DynamicToolSpec;
20 use futures_util::StreamExt;
21 use futures_util::stream::FuturesUnordered;
22 use serde_json::{Value, json};
23 use tokio::sync::{Mutex as AsyncMutex, RwLock, mpsc};
24 use tokio_util::sync::CancellationToken;
25
26 use crate::client::DeepSeekClient;
27 use crate::compaction::{
28 CompactionConfig, CompactionLiveState, compact_messages_safe, merge_system_prompts,
29 should_compact,
30 };
31 use crate::config::{ApiProvider, Config, DEFAULT_MAX_SUBAGENTS, DEFAULT_TEXT_MODEL};
32 use crate::core::model_client::SharedModelClient;
33 use crate::error_taxonomy::{ErrorCategory, ErrorEnvelope, StreamError};
34 use crate::features::{Feature, Features};
35 use crate::mcp::{McpConfig, McpPool};
36 #[cfg(test)]
37 use crate::models::ToolCaller;
38 use crate::models::{
39 ContentBlock, ContentBlockStart, Delta, Message, MessageRequest, StreamEvent, SystemPrompt,
40 Tool, Usage,
41 };
42 use crate::prompts;
43 use crate::purge::{emit_purge_completed, emit_purge_failed, emit_purge_started, run_purge};
44 #[cfg(test)]
45 use crate::route_runtime::resolve_runtime_route;
46 use crate::route_runtime::{
47 ResolvedRuntimeRoute, ValidatedRuntimeRoute, resolve_runtime_route_for_identity,
48 };
49 use crate::tools::goal::{
50 GoalPauseReason, GoalSnapshot, GoalStatus, SharedGoalState, new_shared_goal_state,
51 };
52 use crate::tools::plan::{SharedPlanState, new_shared_plan_state};
53 use crate::tools::shell::{SharedShellManager, new_shared_shell_manager};
54 use crate::tools::spec::{
55 ApprovalRequirement, ResourceClaim, ToolError, ToolExecutionOutcome, ToolResult,
56 };
57 use crate::tools::spec::{
58 RuntimeToolServices, SharedFileReadTracker, new_shared_file_read_tracker,
59 };
60 use crate::tools::subagent::{
61 FleetRole, Mailbox, MailboxMessage, SharedSubAgentManager, SubAgentCompletion,
62 SubAgentForkContext, SubAgentManager, SubAgentResult, SubAgentRuntime, SubAgentStatus,
63 SubAgentThinking, agent_worker_owner_snapshot, ensure_subagent_model_for_provider,
64 new_shared_subagent_manager_with_timeout, resolve_subagent_assignment_route,
65 };
66 use crate::tools::todo::{SharedTodoList, new_shared_todo_list};
67 use crate::tools::user_input::{UserInputRequest, UserInputResponse};
68 use crate::tools::{ToolContext, ToolRegistryBuilder};
69 use crate::tui::app::AppMode;
70 use crate::utils::spawn_supervised;
71 use crate::worker_profile::{ModelRoute, WorkerRuntimeProfile};
72 use crate::working_set::WorkingSet;
73
74 #[cfg(test)]
75 use super::authority::agent_approval_mode_for_turn;
76 use super::authority::{
77 PolicyNarrowingEvent, TurnAuthority, effective_input_policy, shell_policy_for_mode,
78 };
79 use super::events::{Event, TurnOutcomeStatus, TurnRoute};
80 use super::ops::{
81 Op, ProviderRuntimeStatus, SessionSnapshot, USER_SHELL_TOOL_ID_PREFIX, UserInputProvenance,
82 };
83 use super::session::Session;
84 use super::tool_parser;
85 use super::turn::{TurnContext, post_turn_snapshot, pre_turn_snapshot};
86
87 const ENGINE_OP_CHANNEL_CAPACITY: usize = 32;
88 const GOAL_CONTINUATION_FAILURE_DETAIL_MAX_BYTES: usize = 512;
89 const PLAN_SHELL_NETWORK_DENIED_HINT: &str = "Shell command blocked: Plan mode runs shell commands in a read-only sandbox — no writes, no network. Use Act mode (`/mode act`) for any command that creates or modifies files, or that needs network access.";
90
91 fn context_pressure_message(usage_percent: f64) -> Option<&'static str> {
92 if usage_percent >= crate::tui::context_inspector::CONTEXT_CRITICAL_THRESHOLD_PERCENT {
93 Some(
94 "Context pressure: critical — CRITICAL: stop expanding scope; run /compact immediately or finish the current task",
95 )
96 } else if usage_percent >= crate::tui::context_inspector::CONTEXT_WARNING_THRESHOLD_PERCENT {
97 Some(
98 "Context pressure: warning — ESCALATED: prefer /compact, narrow scope, or finish the current task",
99 )
100 } else {
101 None
102 }
103 }
104
105 fn agent_list_event(manager: &SubAgentManager) -> Event {
106 Event::AgentList {
107 agents: manager.list(),
108 coordination: manager.coordination_detail_projection(None, 24),
109 }
110 }
111
112 const MCP_REGISTRY_FIRST_INSTRUCTION_SOURCE: &str = "runtime:mcp-registry-first";
113 const MCP_REGISTRY_FIRST_INSTRUCTION: &str = "## MCP Registry-first policy\n\nFor any task centered on a specialized capability, including media or document conversion, data transformation, browser automation, database or service access, or a developer utility, you must call `registry_sync {}` before `exec_shell`, `fetch_url`, code execution, local programs, custom code, or a manual implementation. Compare the user's complete task yourself against every returned server name and description; wording need not be exact. If any server plausibly covers the core capability, you must call `start_registry_mcp_server` with its exact name and inspect its tools before considering a local alternative. An installed or familiar shell command is not a reason to skip Registry discovery. Use local tools directly only for ordinary repo-native work and simple file operations, when every Registry entry is clearly irrelevant, or after the matching server fails to start.";
114
115 /// Snapshot of parent state that can be passed to forked sub-agents without
116 /// rewriting the parent transcript.
117 ///
118 /// Deliberately **Work-free**: this is captured once at turn start, and Work
119 /// state changes during the turn. The Work section of the fork-state block is
120 /// resolved at the actual fork seam instead (see
121 /// `SubAgentForkContext::with_resolved_state_block`), so a `work_update`
122 /// followed by an `agent` spawn in the same turn hands the child the current
123 /// ledger rather than the one that existed before the turn's first tool call.
124 #[derive(Debug, Clone, Default)]
125 struct StructuredState {
126 mode_label: String,
127 workspace: PathBuf,
128 cwd: Option<PathBuf>,
129 working_set_summary: Option<String>,
130 subagent_snapshots: Vec<SubAgentResult>,
131 }
132
133 impl StructuredState {
134 async fn capture(
135 mode_label: impl Into<String>,
136 workspace: PathBuf,
137 cwd: Option<PathBuf>,
138 working_set: &WorkingSet,
139 subagents: Option<&SharedSubAgentManager>,
140 ) -> Self {
141 let working_set_summary = working_set.summary_block(&workspace);
142
143 let subagent_snapshots = if let Some(handle) = subagents {
144 let mut guard = handle.write().await;
145 guard.cleanup(Duration::from_secs(60 * 60));
146 guard
147 .list()
148 .into_iter()
149 .filter(|s| matches!(s.status, SubAgentStatus::Running))
150 .collect()
151 } else {
152 Vec::new()
153 };
154
155 Self {
156 mode_label: mode_label.into(),
157 workspace,
158 cwd,
159 working_set_summary,
160 subagent_snapshots,
161 }
162 }
163
164 #[must_use]
165 fn to_system_block(&self) -> Option<String> {
166 let mut out = String::new();
167 out.push_str("## Fork State\n\n");
168 out.push_str(&format!("- Mode: `{}`\n", self.mode_label));
169 out.push_str(&format!("- Workspace: `{}`\n", self.workspace.display()));
170 if let Some(cwd) = self.cwd.as_ref() {
171 out.push_str(&format!("- Cwd: `{}`\n", cwd.display()));
172 }
173
174 // No Work section here on purpose: it is appended at the fork seam from
175 // the authoritative projection (#3983), because this block is captured
176 // at turn start and Work moves during the turn.
177 if !self.subagent_snapshots.is_empty() {
178 out.push_str("\n### Open Sub-Agents\n");
179 for s in &self.subagent_snapshots {
180 let role = s.assignment.role.as_deref().unwrap_or("-");
181 let goal = if s.assignment.objective.is_empty() {
182 "(no objective set)"
183 } else {
184 s.assignment.objective.as_str()
185 };
186 out.push_str(&format!("- `{}` (role: {}) - {}\n", s.agent_id, role, goal));
187 }
188 }
189
190 if let Some(working_set) = self.working_set_summary.as_deref() {
191 out.push('\n');
192 out.push_str(working_set);
193 out.push('\n');
194 }
195
196 Some(out)
197 }
198 }
199
200 fn user_shell_turn_outcome(
201 result: &Result<ToolResult, ToolError>,
202 cancel_requested: bool,
203 ) -> TurnOutcomeStatus {
204 let tool_reported_cancel = result.as_ref().is_ok_and(|tool_result| {
205 tool_result
206 .metadata
207 .as_ref()
208 .and_then(|metadata| metadata.get("canceled"))
209 .and_then(Value::as_bool)
210 .unwrap_or(false)
211 });
212
213 if cancel_requested || tool_reported_cancel {
214 TurnOutcomeStatus::Interrupted
215 } else if result.as_ref().is_ok_and(|tool_result| tool_result.success) {
216 TurnOutcomeStatus::Completed
217 } else {
218 TurnOutcomeStatus::Failed
219 }
220 }
221
222 // === Types ===
223
224 /// Configuration for the engine
225 #[derive(Debug, Clone)]
226 pub struct EngineConfig {
227 /// Model identifier to use for responses.
228 pub model: String,
229 /// Route/offering limits for the active provider+model, when the runtime
230 /// route resolver had concrete catalog facts.
231 pub active_route_limits: Option<codewhale_config::route::RouteLimits>,
232 /// Workspace root for tool execution and file operations.
233 pub workspace: PathBuf,
234 /// Allow shell tool execution when true.
235 pub allow_shell: bool,
236 /// Enable trust mode (skip approvals) when true.
237 pub trust_mode: bool,
238 /// Path to the notes file used by the notes tool.
239 pub notes_path: PathBuf,
240 /// Path to the MCP configuration file.
241 pub mcp_config_path: PathBuf,
242 /// Directory containing discoverable skills.
243 pub skills_dir: PathBuf,
244 /// Restrict skill discovery to CodeWhale-owned roots plus explicit
245 /// `skills_dir` configuration.
246 pub skills_scan_codewhale_only: bool,
247 /// Immutable plugin authority snapshot scoped to `workspace`. Normal App
248 /// hosts provide this explicitly; headless/embed callers that leave it
249 /// unset receive a fresh workspace-specific snapshot in [`Engine::new`].
250 pub plugin_registry: Option<Arc<crate::plugins::PluginRegistry>>,
251 /// Sources injected as `<instructions source="…">` blocks in the system
252 /// prompt (#454). Each entry is either a disk path (read at render time)
253 /// or an inline string. Loaded in declared order from the user's
254 /// `instructions = [...]` config or constructed by embedders.
255 ///
256 /// Generalized from `Vec<PathBuf>` so embedders can inject inline content
257 /// without staging a disk file. `From<PathBuf>` impl keeps existing callers
258 /// working with `.into()` at the call site.
259 pub instructions: Vec<crate::prompts::InstructionSource>,
260 pub project_context_pack_enabled: bool,
261 /// When true, the model is instructed to respond in the current locale
262 /// and a post-hoc translation layer replaces remaining English output.
263 pub translation_enabled: bool,
264 pub verbosity: Option<String>,
265 /// Maximum number of assistant steps before stopping.
266 pub max_steps: u32,
267 /// Maximum number of concurrently active subagents.
268 pub max_subagents: usize,
269 /// Maximum queued + running sub-agents admitted for this engine session.
270 pub max_admitted_subagents: usize,
271 /// Number of direct (depth-1) sub-agents that may execute concurrently
272 /// before further launches queue for a launch slot (#3095).
273 /// Resolved from `[subagents] launch_concurrency`.
274 pub launch_concurrency: usize,
275 /// Whether the model-facing `agent` tool is available after applying
276 /// feature flags and `[subagents]` opt-out controls.
277 pub subagents_enabled: bool,
278 /// Feature flags controlling tool availability.
279 pub features: Features,
280 /// Deterministic auto-review policy for tool calls.
281 pub auto_review_policy: crate::tui::auto_review::AutoReviewPolicy,
282 /// Auto-compaction settings for long conversations.
283 pub compaction: CompactionConfig,
284 /// Shared Todo list state.
285 pub todos: SharedTodoList,
286 /// Shared Plan state.
287 pub plan_state: SharedPlanState,
288 /// Shared runtime goal state for model-visible goal tools.
289 pub goal_state: SharedGoalState,
290 /// Maximum sub-agent recursion depth (default 3). See
291 /// `SubAgentRuntime::max_spawn_depth`. Override via
292 /// `[subagents] max_depth = N` in `~/.codewhale/config.toml`.
293 pub max_spawn_depth: u32,
294 /// Optional aggregate token budget for each root sub-agent run.
295 /// Descendant agents inherit the root pool unless a child starts a new
296 /// budget scope with an explicit per-call override.
297 pub subagent_token_budget: Option<u64>,
298 /// Per-domain network policy decider (#135). Shared across the session so
299 /// session-scoped approvals (`/network allow <host>`) persist for the
300 /// remainder of the run.
301 pub network_policy: Option<crate::network_policy::NetworkPolicyDecider>,
302 /// Whether to take side-git workspace snapshots before/after each turn.
303 pub snapshots_enabled: bool,
304 /// Maximum workspace size (in bytes) before snapshots self-disable on
305 /// first init. `0` disables the cap. Resolved from
306 /// `[snapshots] max_workspace_gb` × 1 GB at engine construction.
307 pub snapshots_max_workspace_bytes: u64,
308 /// Post-edit LSP diagnostics injection (#136). When `None`, the engine
309 /// constructs a disabled manager so the field is always present.
310 pub lsp_config: Option<crate::lsp::LspConfig>,
311 /// Durable runtime services exposed to model-visible tools.
312 pub runtime_services: RuntimeToolServices,
313 /// Per-role/type sub-agent model overrides already resolved from config.
314 pub subagent_model_overrides: HashMap<String, String>,
315 /// Merged fleet roster (built-ins + config + personal/workspace agent
316 /// files) shared by model-spawned sub-agents and fleet dispatch
317 /// (#fleet-roster cutover (v0.8.67)). Defaults to built-ins only; the
318 /// engine-config construction sites load it at session start and the setup
319 /// wizard refreshes it after each successful profile save.
320 pub fleet_roster: std::sync::Arc<crate::fleet::roster::FleetRoster>,
321 /// Whether the user-memory feature is enabled (#489). When `true` the
322 /// engine reads `memory_path` on each prompt assembly and prepends a
323 /// `<user_memory>` block to the system prompt.
324 pub memory_enabled: bool,
325 /// Path to the user memory file (#489). Always populated; only
326 /// consulted when `memory_enabled` is `true`.
327 pub memory_path: PathBuf,
328 /// Default directory for Xiaomi MiMo speech/TTS tool outputs.
329 pub speech_output_dir: Option<PathBuf>,
330 pub vision_config: Option<crate::config::VisionModelConfig>,
331 pub goal_objective: Option<String>,
332 pub goal_token_budget: Option<u32>,
333 pub goal_status: GoalStatus,
334 /// Safety backstop on automatic goal continuation passes (#5052).
335 /// Resolved from `[goal] max_continuations` in config.toml; `0` disables
336 /// the backstop so only completion, blocked state, or the continuation
337 /// limit stops an operate-mode goal run.
338 pub goal_max_continuations: u32,
339 /// Tool restriction from custom slash command frontmatter.
340 /// `None` means the current turn may use the normal tool set.
341 pub allowed_tools: Option<Vec<String>>,
342 /// Tool deny-list. Deny always wins over allow (#3027).
343 /// `None` means no tools are explicitly denied.
344 pub disallowed_tools: Option<Vec<String>>,
345 /// Hard per-turn cap on admitted tool calls (#4415). `None` (the default)
346 /// means unlimited and leaves the turn admission gate inert. Task hosts
347 /// set this from the task's structured `max_tool_calls` constraint; the
348 /// per-turn counter itself lives in the turn loop, not here.
349 pub max_tool_calls: Option<u32>,
350 /// Hook executor for control-plane hooks.
351 /// `ToolCallBefore` hooks may deny a tool call with exit code 2.
352 pub hook_executor: Option<std::sync::Arc<crate::hooks::HookExecutor>>,
353 /// Resolved BCP-47 locale tag (e.g. `"en"`, `"zh-Hans"`, `"ja"`)
354 /// for the `## Environment` block in the system prompt. The
355 /// caller resolves this from `Settings` once at engine
356 /// construction; the engine never touches disk for it.
357 pub locale_tag: String,
358 /// When true, force `tool_choice: "required"` and opt compatible function
359 /// schemas into DeepSeek beta strict mode.
360 pub strict_tool_mode: bool,
361 /// Workshop / large-tool-output routing (#548). `None` disables routing.
362 pub workshop: Option<crate::tools::large_output_router::WorkshopConfig>,
363 /// Which search backend `web_search` should use. Default: DuckDuckGo.
364 pub search_provider: crate::config::SearchProvider,
365 /// API key for Tavily, Bocha, Metaso, Baidu, Volcengine, or Sofya.
366 /// `None` for Bing, DuckDuckGo, or SearXNG.
367 /// Metaso also falls back to the `METASO_API_KEY` env var.
368 /// Baidu also falls back to `BAIDU_SEARCH_API_KEY`.
369 pub search_api_key: Option<String>,
370 /// Optional DuckDuckGo-compatible HTML endpoint override.
371 pub search_base_url: Option<String>,
372 /// Per-step DeepSeek API timeout for sub-agent `create_message` requests.
373 /// Resolved from `[subagents] api_timeout_secs` (clamped to 1..=3600)
374 /// once at engine construction, then threaded onto every
375 /// `SubAgentRuntime` the engine builds (#1806, #1808).
376 pub subagent_api_timeout: Duration,
377 /// Per-SSE-chunk idle timeout for streamed model responses.
378 /// Resolved from `[tui].stream_chunk_timeout_secs` (or the legacy
379 /// `DEEPSEEK_STREAM_IDLE_TIMEOUT_SECS`) and updated live by `/config`.
380 pub stream_chunk_timeout: Duration,
381 /// No-progress heartbeat timeout for live sub-agents. Used by the manager
382 /// and parent wait loop to auto-cancel stuck children before they exhaust
383 /// the sub-agent slot pool indefinitely (#2614).
384 pub subagent_heartbeat_timeout: Duration,
385 /// Native tools that should stay in the model-visible catalog even when
386 /// they are outside the small default core surface (#2076).
387 pub tools_always_load: HashSet<String>,
388 /// When true and `/usr/bin/bwrap` is executable on Linux, route exec_shell
389 /// through bubblewrap (#2184).
390 pub prefer_bwrap: bool,
391 /// Tool override and plugin configuration (`[tools]` table in config.toml).
392 /// Applied to the per-turn tool registry after built-in tools are registered.
393 /// When `None`, no overrides or plugin loading occurs.
394 pub tools: Option<crate::config::ToolsConfig>,
395 /// Whether tools should follow symbolic links. When `true`, symlinked
396 /// directories are traversed by walk-based tools and symlinked paths
397 /// that resolve outside the workspace are still allowed (the symlink
398 /// itself must be inside the workspace). Mirrors the
399 /// `workspace_follow_symlinks` setting.
400 pub workspace_follow_symlinks: bool,
401 /// Ask-only permission rules loaded from sibling `permissions.toml`.
402 pub exec_policy_engine: codewhale_execpolicy::ExecPolicyEngine,
403 /// Whether turn startup may write terminal title/taskbar OSC sequences.
404 /// Interactive TUI sessions enable this; headless and machine-readable
405 /// hosts disable it so stdout remains protocol-clean.
406 pub terminal_chrome_enabled: bool,
407 /// Resolved advisor watcher configuration (#3982). Off by default.
408 /// Updated live by `Op::SetAdvisorEnabled`.
409 pub advisor_config: crate::tools::subagent::AdvisorConfig,
410 }
411
412 impl Default for EngineConfig {
413 fn default() -> Self {
414 Self {
415 model: DEFAULT_TEXT_MODEL.to_string(),
416 active_route_limits: None,
417 workspace: PathBuf::from("."),
418 allow_shell: true,
419 trust_mode: false,
420 notes_path: PathBuf::from("notes.txt"),
421 mcp_config_path: PathBuf::from("mcp.json"),
422 skills_dir: crate::skills::default_skills_dir(),
423 skills_scan_codewhale_only: false,
424 plugin_registry: None,
425 instructions: Vec::new(),
426 project_context_pack_enabled: false,
427 translation_enabled: false,
428 // High backstop rather than a working ceiling: the in-turn
429 // loop_guard that used to brake repetition is gone, so this only
430 // exists to terminate a pathological runaway turn via
431 // `at_max_steps()`. 1000 stays high enough to never gate real work
432 // while still guaranteeing the turn ends.
433 max_steps: 1000,
434 max_subagents: DEFAULT_MAX_SUBAGENTS,
435 max_admitted_subagents: DEFAULT_MAX_SUBAGENTS,
436 launch_concurrency: DEFAULT_MAX_SUBAGENTS,
437 subagents_enabled: true,
438 features: Features::with_defaults(),
439 auto_review_policy: crate::tui::auto_review::AutoReviewPolicy::default(),
440 compaction: CompactionConfig::default(),
441 todos: new_shared_todo_list(),
442 plan_state: new_shared_plan_state(),
443 goal_state: new_shared_goal_state(),
444 max_spawn_depth: crate::tools::subagent::DEFAULT_MAX_SPAWN_DEPTH,
445 subagent_token_budget: None,
446 network_policy: None,
447 snapshots_enabled: true,
448 snapshots_max_workspace_bytes:
449 crate::snapshot::DEFAULT_MAX_WORKSPACE_BYTES_FOR_SNAPSHOT,
450 lsp_config: None,
451 runtime_services: RuntimeToolServices::default(),
452 subagent_model_overrides: HashMap::new(),
453 fleet_roster: std::sync::Arc::new(crate::fleet::roster::FleetRoster::built_ins_only()),
454 memory_enabled: false,
455 memory_path: PathBuf::from("./memory.md"),
456 speech_output_dir: None,
457 vision_config: None,
458 strict_tool_mode: false,
459 goal_objective: None,
460 goal_token_budget: None,
461 goal_status: GoalStatus::Active,
462 goal_max_continuations: crate::goal_loop::DEFAULT_MAX_GOAL_CONTINUATIONS,
463 allowed_tools: None,
464 disallowed_tools: None,
465 max_tool_calls: None,
466 hook_executor: None,
467 locale_tag: "en".to_string(),
468 workshop: None,
469 search_provider: crate::config::SearchProvider::default(),
470 search_api_key: None,
471 search_base_url: None,
472 subagent_api_timeout: Duration::from_secs(
473 crate::config::DEFAULT_SUBAGENT_API_TIMEOUT_SECS,
474 ),
475 stream_chunk_timeout: Duration::from_secs(
476 crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS,
477 ),
478 subagent_heartbeat_timeout: Duration::from_secs(
479 crate::config::DEFAULT_SUBAGENT_HEARTBEAT_TIMEOUT_SECS,
480 ),
481 tools_always_load: HashSet::new(),
482 prefer_bwrap: false,
483 verbosity: None,
484 tools: None,
485 workspace_follow_symlinks: false,
486 exec_policy_engine: codewhale_execpolicy::ExecPolicyEngine::new(Vec::new(), Vec::new()),
487 terminal_chrome_enabled: true,
488 advisor_config: crate::tools::subagent::AdvisorConfig::disabled(),
489 }
490 }
491 }
492
493 /// Reason the active turn was cancelled. The token from `tokio_util`
494 /// does not carry a cause, so the engine keeps a sibling latch for
495 /// approval and user-input waits that need to explain cancellation.
496 ///
497 /// `External`, `Preempted`, and `Internal` are reserved for the
498 /// remaining direct cancellation paths tracked in #1541.
499 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
500 #[allow(dead_code)]
501 pub enum CancelReason {
502 /// User-initiated cancel (Esc, `/cancel`, click cancel on modal).
503 User,
504 /// External / runtime-API cancel (HTTP `DELETE /v1/threads/...`,
505 /// task manager stop, parent agent cancel).
506 External,
507 /// Cancel triggered when a new turn starts before the previous one
508 /// finished — e.g. plain Enter while busy after the queueing path
509 /// pre-empts the running turn.
510 Preempted,
511 /// Engine internals tore down the turn (drop, channel close,
512 /// shutdown). Rare — surfaced as an internal error.
513 Internal,
514 }
515
516 impl CancelReason {
517 fn describe(self) -> &'static str {
518 match self {
519 Self::User => "user cancelled the request",
520 Self::External => "request cancelled by external caller",
521 Self::Preempted => "request was preempted by a new turn",
522 Self::Internal => "engine torn down before approval resolved",
523 }
524 }
525 }
526
527 /// Handle to communicate with the engine
528 #[derive(Clone)]
529 pub struct EngineHandle {
530 /// Send operations to the engine
531 pub tx_op: mpsc::Sender<Op>,
532 /// Receive events from the engine
533 pub rx_event: Arc<RwLock<mpsc::Receiver<Event>>>,
534 /// Shared pointer to the cancellation token for the current request.
535 cancel_token: Arc<StdMutex<CancellationToken>>,
536 /// Latched reason for the most recent cancellation. Read by the
537 /// approval / user-input handlers to enrich their error strings.
538 /// Cleared by the engine when a fresh turn starts.
539 cancel_reason: Arc<StdMutex<Option<CancelReason>>>,
540 /// Send approval decisions to the engine
541 tx_approval: mpsc::Sender<ApprovalDecision>,
542 /// Send user input responses to the engine
543 tx_user_input: mpsc::Sender<UserInputDecision>,
544 /// Send steer input for an in-flight turn.
545 tx_steer: mpsc::Sender<String>,
546 /// Shared pause flag set by the TUI and read by the turn loop.
547 shared_paused: Arc<StdMutex<bool>>,
548 /// Whether the host must construct the route's concrete provider client
549 /// before it mutates turn state. Real engines own concrete provider I/O;
550 /// explicit injected/mock engines own that seam themselves.
551 client_preflight_required: bool,
552 /// Typed live permission authority shared with the running turn. A mode
553 /// change publishes here before its mailbox op is queued, so gates never
554 /// consult a stale per-turn copy.
555 live_runtime_authority: Arc<StdMutex<LiveRuntimeAuthorityState>>,
556 }
557
558 // `impl EngineHandle { ... }` moved to `engine/handle.rs` so the
559 // mailbox API can be reviewed independently of the engine internals.
560
561 // === Engine ===
562
563 /// The core engine that processes operations and emits events
564 pub struct Engine {
565 config: EngineConfig,
566 api_config: Config,
567 /// Runtime-host authority consulted only when constructing a later turn
568 /// descriptor (goal continuation, idle child completion, `/edit`). Active
569 /// turns keep their already-installed immutable descriptor.
570 authoritative_route_config: Option<Arc<parking_lot::RwLock<Config>>>,
571 deepseek_client: Option<DeepSeekClient>,
572 /// Provider-neutral client used by the canonical main turn loop. Concrete
573 /// clients remain temporarily available to provider-specific helper tools
574 /// while those boundaries migrate independently.
575 model_client: Option<SharedModelClient>,
576 /// Test/embedding seam: an explicitly injected provider-neutral client
577 /// remains the I/O authority while typed routes still validate receipts,
578 /// endpoint metadata, and budgets.
579 model_client_injected: bool,
580 deepseek_client_error: Option<String>,
581 api_key_env_only_recovery: Option<String>,
582 session: Session,
583 /// One lazy, session-scoped working kernel for inline `repl` blocks.
584 /// Its context is refreshed before each run, while user-created Python
585 /// state stays alive across model turns.
586 repl_kernel: Option<crate::repl::PythonRuntime>,
587 subagent_manager: SharedSubAgentManager,
588 shell_manager: SharedShellManager,
589 /// Read-before-edit snapshots live for the session, not for one turn's
590 /// transient `ToolContext` (#4475).
591 file_read_tracker: SharedFileReadTracker,
592 mcp_pool: Option<Arc<AsyncMutex<McpPool>>>,
593 /// Workspace-scoped immutable plugin catalogue and authority receipts.
594 plugin_registry: Arc<crate::plugins::PluginRegistry>,
595 api_provider: ApiProvider,
596 /// Exact configured route key. Named custom providers share the `Custom`
597 /// enum, so the enum alone cannot prove that the active client is current.
598 api_provider_identity: String,
599 /// Additive exact provider id. `None` preserves the legacy root-literal
600 /// custom route across snapshots and config reloads.
601 api_provider_id: Option<String>,
602 active_route_limits: Option<codewhale_config::route::RouteLimits>,
603 active_route_capabilities: codewhale_config::route::RouteCapabilities,
604 rx_op: mpsc::Receiver<Op>,
605 live_runtime_authority: Arc<StdMutex<LiveRuntimeAuthorityState>>,
606 /// Clone of the op-channel sender, so the engine can self-dispatch ops
607 /// (e.g. a goal-continuation `SendMessage` after a turn completes).
608 tx_op: mpsc::Sender<Op>,
609 /// At most one engine-owned continuation across capacity-waiting and
610 /// enqueued states. The authoritative dynamic-tool set stays here so a
611 /// later successful turn can refresh it without adding a second token.
612 scheduled_goal_continuation: Option<ScheduledGoalContinuation>,
613 goal_continuation_schedule_seq: u64,
614 rx_approval: mpsc::Receiver<ApprovalDecision>,
615 rx_user_input: mpsc::Receiver<UserInputDecision>,
616 rx_steer: mpsc::Receiver<String>,
617 tx_event: mpsc::Sender<Event>,
618 /// Wakeup channel for the parent turn loop when a direct child sub-agent
619 /// terminates (issue #756). Cloned into `SubAgentRuntime` so the runtime
620 /// can fan completion events back into the engine.
621 tx_subagent_completion: mpsc::UnboundedSender<SubAgentCompletion>,
622 /// Receiver paired with `tx_subagent_completion`. Drained at the
623 /// turn-loop's empty-tool_uses branch to surface `<codewhale:subagent.done>`
624 /// sentinels into the parent's transcript before deciding to end the turn.
625 pub(super) rx_subagent_completion: mpsc::UnboundedReceiver<SubAgentCompletion>,
626 /// Sub-agent completions already injected into the parent transcript.
627 /// Channel delivery and watchdog reconciliation both mark this set so a
628 /// dropped event can be synthesized once without duplicating a later
629 /// delivery.
630 delivered_subagent_completion_ids: HashSet<String>,
631 cancel_token: CancellationToken,
632 shared_cancel_token: Arc<StdMutex<CancellationToken>>,
633 /// Latched reason for the current cancellation, mirrored to
634 /// `EngineHandle::cancel_reason`. Read by `approval.rs` when
635 /// surfacing the "Request cancelled while awaiting …" error so the
636 /// user-facing message names a cause.
637 pub(super) cancel_reason: Arc<StdMutex<Option<CancelReason>>>,
638 tool_exec_lock: Arc<RwLock<()>>,
639 turn_counter: u64,
640 /// Post-edit LSP diagnostics injection (#136). Populated unconditionally
641 /// — when LSP is disabled in config, this is an inert manager that
642 /// always returns `None` from `diagnostics_for`.
643 lsp_manager: Arc<crate::lsp::LspManager>,
644 /// Session-scoped workshop variable store (#548). Shared across all tool
645 /// calls so `last_tool_result` persists within the session and can be
646 /// promoted to the parent context via `promote_to_context`.
647 workshop_vars: Option<
648 std::sync::Arc<tokio::sync::Mutex<crate::tools::large_output_router::WorkshopVariables>>,
649 >,
650 /// External sandbox backend (#516). When `Some`, exec_shell routes commands
651 /// through this instead of spawning a local process.
652 sandbox_backend: Option<std::sync::Arc<dyn crate::sandbox::backend::SandboxBackend>>,
653 /// Diagnostics collected during the current step's tool calls. Drained
654 /// and forwarded as a synthetic user message before the next API call.
655 pending_lsp_blocks: Vec<crate::lsp::DiagnosticBlock>,
656 /// Current operating mode. Updated on `ChangeMode` and `SendMessage`.
657 current_mode: AppMode,
658 /// The most recent authority narrowing, if any (#3947). Kept on the engine
659 /// so doctor and debug surfaces can answer "why is this tool unavailable"
660 /// with the same record the user and the model already saw.
661 last_policy_narrowing: Option<PolicyNarrowingEvent>,
662 /// The git snapshot line last emitted in a `<turn_meta>` block this
663 /// session (#5187, k3-gap F3). The snapshot re-collects branch/dirty
664 /// state every turn, so without change-detection the block's bytes drift
665 /// after every edit the model itself makes, defeating cross-turn prefix
666 /// stability. `None` until the first block is built; the line is then
667 /// emitted only when the snapshot actually changed.
668 last_turn_meta_git_snapshot: StdMutex<Option<String>>,
669 /// Process-local cache for `estimated_input_tokens`. Memoizes the most
670 /// recent token estimate keyed on `(session.messages_revision,
671 /// system_prompt_fingerprint)`. Five call sites per turn consult this
672 /// (engine capacity checkpoints, seam manager, trim budget, etc.) plus
673 /// four TUI / command consumers; the cache turns N×O(messages) walks
674 /// into a single recompute on a content change.
675 token_estimate_cache: TokenEstimateCache,
676 /// Shared pause flag set by the TUI and read before tool execution.
677 shared_paused: Arc<StdMutex<bool>>,
678 /// Rate-limit + dedup guard for the background advisor watcher (#3982).
679 /// `None` until the first turn completes with the advisor enabled, then
680 /// held for the session lifetime so state persists across turns.
681 advisor_emission_guard: Option<Arc<tokio::sync::Mutex<crate::tools::subagent::EmissionGuard>>>,
682 }
683
684 #[derive(Debug, Clone, PartialEq, Eq)]
685 struct LiveRuntimeAuthority {
686 mode: AppMode,
687 allow_shell: bool,
688 trust_mode: bool,
689 auto_approve: bool,
690 approval_mode: crate::tui::approval::ApprovalMode,
691 configured_sandbox_mode: Option<String>,
692 }
693
694 impl LiveRuntimeAuthority {
695 fn from_fields(
696 mode: AppMode,
697 allow_shell: bool,
698 trust_mode: bool,
699 auto_approve: bool,
700 approval_mode: crate::tui::approval::ApprovalMode,
701 configured_sandbox_mode: Option<String>,
702 ) -> Self {
703 let authority = TurnAuthority::from_effective_fields(
704 mode,
705 allow_shell,
706 trust_mode,
707 auto_approve,
708 approval_mode,
709 );
710 Self::from_turn_authority(&authority, configured_sandbox_mode)
711 }
712
713 fn from_turn_authority(
714 authority: &TurnAuthority,
715 configured_sandbox_mode: Option<String>,
716 ) -> Self {
717 let approval_mode = authority.approval_mode_for_session();
718 Self {
719 mode: authority.mode,
720 allow_shell: authority.allow_shell,
721 trust_mode: authority.trust_mode,
722 auto_approve: authority.auto_approve
723 || approval_mode == crate::tui::approval::ApprovalMode::Bypass,
724 approval_mode,
725 configured_sandbox_mode,
726 }
727 }
728
729 fn permission_snapshot(&self) -> RuntimePermissionAuthority {
730 RuntimePermissionAuthority {
731 auto_approve: self.auto_approve,
732 trust_mode: self.trust_mode,
733 approval_mode: self.approval_mode,
734 }
735 }
736 }
737
738 #[derive(Debug)]
739 struct LiveRuntimeAuthorityState {
740 revision: u64,
741 applied_revision: u64,
742 authority: LiveRuntimeAuthority,
743 }
744
745 impl LiveRuntimeAuthorityState {
746 fn new(authority: LiveRuntimeAuthority) -> Self {
747 Self {
748 revision: 0,
749 applied_revision: 0,
750 authority,
751 }
752 }
753 }
754
755 /// Runtime-facing view of the engine's exact live permission authority.
756 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
757 pub(crate) struct RuntimePermissionAuthority {
758 pub(crate) auto_approve: bool,
759 pub(crate) trust_mode: bool,
760 pub(crate) approval_mode: crate::tui::approval::ApprovalMode,
761 }
762
763 fn claim_subagent_completion(
764 delivered_ids: &mut HashSet<String>,
765 completion: SubAgentCompletion,
766 ) -> Option<SubAgentCompletion> {
767 delivered_ids
768 .insert(completion.agent_id.clone())
769 .then_some(completion)
770 }
771
772 #[derive(Debug)]
773 enum GoalContinuationAction {
774 Inactive,
775 Dispatch {
776 content: String,
777 snapshot: Box<GoalSnapshot>,
778 },
779 Stopped {
780 message: String,
781 reason: GoalPauseReason,
782 },
783 }
784
785 struct ScheduledGoalContinuation {
786 id: u64,
787 dynamic_tools: Vec<DynamicToolSpec>,
788 enqueued: bool,
789 }
790
791 enum SendMessageOutcome {
792 NotStarted {
793 error: Option<String>,
794 },
795 Finished {
796 status: TurnOutcomeStatus,
797 error: Option<String>,
798 },
799 }
800
801 /// Idle-poll cadence for unclaimed background shell completion while a
802 /// goal is active. Coarse on purpose: this is a liveness backstop, not an
803 /// animation loop.
804 const SHELL_WAKE_POLL_MS: u64 = 750;
805
806 enum EngineRunInput {
807 Operation(Box<Op>),
808 SubAgentCompletion(SubAgentCompletion),
809 /// A background shell job finished while the engine sat idle with an
810 /// active goal. Shell completion is pull-only (no channel), so without
811 /// this wake an active goal waiting on background work stayed inert until
812 /// the user typed something (morning-report continuation gap).
813 ShellCompletionWake,
814 }
815
816 impl SendMessageOutcome {
817 fn started(&self) -> bool {
818 matches!(self, Self::Finished { .. })
819 }
820 }
821
822 // === Internal tool helpers ===
823
824 fn subagent_mailbox_message_is_best_effort(message: &MailboxMessage) -> bool {
825 matches!(
826 message,
827 MailboxMessage::Progress { .. }
828 | MailboxMessage::ToolCallStarted { .. }
829 | MailboxMessage::ToolCallCompleted { .. }
830 )
831 }
832
833 const SUBAGENT_MAILBOX_BEST_EFFORT_MIN_INTERVAL: Duration = Duration::from_millis(100);
834
835 fn subagent_mailbox_best_effort_send_permitted(
836 last_sent_at: &mut HashMap<String, Instant>,
837 message: &MailboxMessage,
838 now: Instant,
839 ) -> bool {
840 if !subagent_mailbox_message_is_best_effort(message) {
841 return true;
842 }
843
844 let agent_id = message.agent_id().to_string();
845 if last_sent_at
846 .get(&agent_id)
847 .is_some_and(|last| now.duration_since(*last) < SUBAGENT_MAILBOX_BEST_EFFORT_MIN_INTERVAL)
848 {
849 return false;
850 }
851
852 last_sent_at.insert(agent_id, now);
853 true
854 }
855
856 /// Forward one turn-scoped mailbox envelope. Returns `false` when the engine
857 /// event channel is closed and the drainer should stop.
858 async fn forward_subagent_mailbox_message(
859 tx: &mpsc::Sender<Event>,
860 turn_id: &str,
861 seq: u64,
862 message: MailboxMessage,
863 best_effort_sent_at: &mut HashMap<String, Instant>,
864 ) -> bool {
865 let event = Event::SubAgentMailbox {
866 turn_id: turn_id.to_string(),
867 seq,
868 message,
869 };
870 if let Event::SubAgentMailbox { message, .. } = &event
871 && subagent_mailbox_message_is_best_effort(message)
872 {
873 if !subagent_mailbox_best_effort_send_permitted(
874 best_effort_sent_at,
875 message,
876 Instant::now(),
877 ) {
878 return true;
879 }
880 return match tx.try_send(event) {
881 Ok(()) | Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => true,
882 Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => false,
883 };
884 }
885 tx.send(event).await.is_ok()
886 }
887
888 impl Engine {
889 /// Per-posture question discipline. Lives with the approval overlays in the
890 /// stable prefix / gate errors — not re-asserted every turn (#4780).
891 #[allow(dead_code)] // surface via approval-gate errors when those are tightened
892 fn permission_question_discipline(
893 approval_mode: crate::tui::approval::ApprovalMode,
894 ) -> &'static str {
895 use crate::tui::approval::ApprovalMode;
896
897 match approval_mode {
898 ApprovalMode::Suggest => {
899 "Tool approvals and user decisions are separate. Ask a concise question when an unresolved choice materially affects authority, cost, requested scope, or outcome; otherwise continue under the active approval policy."
900 }
901 ApprovalMode::Auto => {
902 "Auto-Review is fully autonomous. Do not ask the user questions or pause for a user decision. Resolve ambiguity from the available context, choose the safest reversible interpretation that still advances the request, and continue; if no safe in-scope action exists, report the constraint without opening a question prompt."
903 }
904 ApprovalMode::Bypass => {
905 "Tool calls do not need approval, but Full Access does not authorize invented intent. Ask one concise, deliberate question when a consequential choice cannot be recovered safely from context; otherwise proceed autonomously within the current sandbox, repository, and managed-policy boundaries."
906 }
907 ApprovalMode::Never => {
908 "Remain read-only. Ask when a missing user decision blocks a truthful plan or investigation; do not imply that this permission boundary can be bypassed."
909 }
910 }
911 }
912
913 pub(super) async fn emit_compaction_started(
914 &mut self,
915 id: String,
916 auto: bool,
917 message: String,
918 ) {
919 let _ = self
920 .tx_event
921 .send(Event::CompactionStarted { id, auto, message })
922 .await;
923 }
924
925 pub(super) async fn emit_compaction_completed(
926 &mut self,
927 id: String,
928 auto: bool,
929 message: String,
930 messages_before: Option<usize>,
931 messages_after: Option<usize>,
932 ) {
933 let summary_prompt = self.rendered_compaction_summary();
934 let _ = self
935 .tx_event
936 .send(Event::CompactionCompleted {
937 id,
938 auto,
939 message,
940 messages_before,
941 messages_after,
942 summary_prompt,
943 })
944 .await;
945 }
946
947 /// Render the accumulated compaction summary prompt to plain text so it
948 /// can travel in events and be persisted by host layers. All emit sites
949 /// run after `merge_compaction_summary`, so this reflects the summary
950 /// state the engine will use for subsequent requests.
951 fn rendered_compaction_summary(&self) -> Option<String> {
952 self.session
953 .compaction_summary_prompt
954 .as_ref()
955 .map(|prompt| match prompt {
956 SystemPrompt::Text(text) => text.clone(),
957 SystemPrompt::Blocks(blocks) => blocks
958 .iter()
959 .map(|block| block.text.as_str())
960 .collect::<Vec<_>>()
961 .join("\n\n"),
962 })
963 .filter(|text| !text.trim().is_empty())
964 }
965
966 pub(super) async fn emit_compaction_failed(&mut self, id: String, auto: bool, message: String) {
967 let _ = self
968 .tx_event
969 .send(Event::CompactionFailed { id, auto, message })
970 .await;
971 }
972
973 fn reset_cancel_token(&mut self) {
974 let token = CancellationToken::new();
975 self.cancel_token = token.clone();
976 match self.shared_cancel_token.lock() {
977 Ok(mut shared) => {
978 *shared = token;
979 }
980 Err(poisoned) => {
981 *poisoned.into_inner() = token;
982 }
983 }
984 // Fresh turn → clear any latched cancellation reason from the
985 // previous turn so a downstream "request cancelled" message
986 // doesn't inherit a stale cause.
987 match self.cancel_reason.lock() {
988 Ok(mut slot) => *slot = None,
989 Err(poisoned) => *poisoned.into_inner() = None,
990 }
991 match self.shared_paused.lock() {
992 Ok(mut paused) => *paused = false,
993 Err(poisoned) => *poisoned.into_inner() = false,
994 }
995 }
996
997 fn env_only_api_key_recovery_hint(api_config: &Config) -> Option<String> {
998 if !crate::config::active_provider_uses_env_only_api_key(api_config) {
999 return None;
1000 }
1001
1002 let provider = api_config.api_provider();
1003 let env_var = provider.env_vars_label();
1004
1005 Some(format!(
1006 "The rejected key came from {env_var}; no saved config key is present.\n\
1007 Run `codewhale auth status` to inspect credential sources, then \
1008 `codewhale auth set --provider {provider}` to save a valid key in ~/.codewhale/config.toml, \
1009 or remove the stale export and open a fresh shell.",
1010 provider = provider.as_str()
1011 ))
1012 }
1013
1014 pub(super) fn decorate_auth_error_message(&self, message: String) -> String {
1015 let Some(hint) = self.api_key_env_only_recovery.as_ref() else {
1016 return message;
1017 };
1018 if crate::error_taxonomy::classify_error_message(&message) != ErrorCategory::Authentication
1019 || message.contains("no saved config key is present")
1020 {
1021 return message;
1022 }
1023 format!("{message}\n\n{hint}")
1024 }
1025
1026 /// Install a route that the host already resolved and client-preflighted.
1027 /// No identity guessing or config re-resolution is allowed at this
1028 /// boundary: the descriptor is the single authority for the turn.
1029 fn install_validated_runtime_route(&mut self, route: ValidatedRuntimeRoute) {
1030 let provider = route.identity.provider;
1031 let identity = route.identity.key;
1032 let provider_id = route.identity.exact_id;
1033 let model = route.model;
1034 let limits = crate::route_budget::known_route_limits(route.candidate.limits());
1035 let capabilities = route.candidate.capabilities();
1036 let api_config = *route.config;
1037 let client = route.client;
1038
1039 self.api_provider = provider;
1040 self.api_provider_identity = identity;
1041 self.api_provider_id = provider_id;
1042 self.api_config = api_config;
1043 self.active_route_limits = limits;
1044 self.active_route_capabilities = capabilities;
1045 self.api_key_env_only_recovery = Self::env_only_api_key_recovery_hint(&self.api_config);
1046 self.deepseek_client = Some(client.clone());
1047 if !self.model_client_injected {
1048 self.model_client = Some(Arc::new(client.clone()));
1049 }
1050 self.deepseek_client_error = None;
1051 self.session.model = model;
1052 self.config.model.clone_from(&self.session.model);
1053 }
1054
1055 /// Activate a structurally resolved route at the engine boundary. Normal
1056 /// engines construct the concrete client before any turn state changes.
1057 /// Embedders/tests that explicitly injected a provider-neutral client keep
1058 /// that client as the I/O authority while still installing the exact route
1059 /// identity, model, config, and budget receipt.
1060 fn install_resolved_runtime_route(
1061 &mut self,
1062 mut route: ResolvedRuntimeRoute,
1063 ) -> Result<(), String> {
1064 if !self.model_client_injected {
1065 self.install_validated_runtime_route(route.validate()?);
1066 return Ok(());
1067 }
1068
1069 let preflighted_client = route.take_preflighted_client();
1070 let provider = route.identity.provider;
1071 let identity = route.identity.key;
1072 let provider_id = route.identity.exact_id;
1073 let model = route.model;
1074 let limits = crate::route_budget::known_route_limits(route.candidate.limits());
1075 let capabilities = route.candidate.capabilities();
1076 let api_config = *route.config;
1077 let concrete_client = preflighted_client
1078 .map(Ok)
1079 .unwrap_or_else(|| DeepSeekClient::from_candidate(&api_config, &route.candidate));
1080
1081 self.api_provider = provider;
1082 self.api_provider_identity = identity;
1083 self.api_provider_id = provider_id;
1084 self.api_config = api_config;
1085 self.active_route_limits = limits;
1086 self.active_route_capabilities = capabilities;
1087 self.api_key_env_only_recovery = Self::env_only_api_key_recovery_hint(&self.api_config);
1088 match concrete_client {
1089 Ok(client) => {
1090 self.deepseek_client = Some(client.clone());
1091 self.deepseek_client_error = None;
1092 }
1093 Err(err) => {
1094 self.deepseek_client = None;
1095 self.deepseek_client_error = Some(err.to_string());
1096 }
1097 }
1098 self.session.model = model;
1099 self.config.model.clone_from(&self.session.model);
1100 Ok(())
1101 }
1102
1103 fn current_runtime_route(&self) -> Result<ResolvedRuntimeRoute, String> {
1104 let config = self
1105 .authoritative_route_config
1106 .as_ref()
1107 .map(|config| config.read().clone())
1108 .unwrap_or_else(|| self.api_config.clone());
1109 let identity = config.resolve_persisted_provider_identity(
1110 Some(self.api_provider.as_str()),
1111 self.api_provider_id.as_deref(),
1112 )?;
1113 resolve_runtime_route_for_identity(&config, &identity, Some(&self.session.model))
1114 }
1115
1116 /// Create a new engine with the given configuration
1117 pub fn new(mut config: EngineConfig, api_config: &Config) -> (Self, EngineHandle) {
1118 crate::tls::ensure_rustls_crypto_provider();
1119
1120 // Unlike a Skill body, this instruction is visible on the first model
1121 // request. Keep selection semantic: the host supplies no keywords or
1122 // ranking and the model compares the full user context with the full
1123 // Registry catalog. Append it after configured instruction sources so
1124 // the Registry-first decision sits close to the current user turn.
1125 if config.features.enabled(Feature::Mcp) {
1126 config
1127 .instructions
1128 .push(crate::prompts::InstructionSource::Inline {
1129 name: MCP_REGISTRY_FIRST_INSTRUCTION_SOURCE.to_string(),
1130 content: MCP_REGISTRY_FIRST_INSTRUCTION.to_string(),
1131 });
1132 }
1133
1134 if let Some(objective) = normalized_goal_objective(config.goal_objective.as_deref()) {
1135 sync_goal_state_from_host(
1136 &config.goal_state,
1137 Some(&objective),
1138 config.goal_token_budget,
1139 config.goal_status,
1140 );
1141 }
1142
1143 let (tx_op, rx_op) = mpsc::channel(ENGINE_OP_CHANNEL_CAPACITY);
1144 let (tx_event, rx_event) = mpsc::channel(256);
1145 let (tx_approval, rx_approval) = mpsc::channel(64);
1146 let (tx_user_input, rx_user_input) = mpsc::channel(32);
1147 let (tx_steer, rx_steer) = mpsc::channel(64);
1148 let (tx_subagent_completion, rx_subagent_completion) = mpsc::unbounded_channel();
1149 let cancel_token = CancellationToken::new();
1150 let shared_cancel_token = Arc::new(StdMutex::new(cancel_token.clone()));
1151 let cancel_reason: Arc<StdMutex<Option<CancelReason>>> = Arc::new(StdMutex::new(None));
1152 let shared_paused = Arc::new(StdMutex::new(false));
1153 let live_runtime_authority = Arc::new(StdMutex::new(LiveRuntimeAuthorityState::new(
1154 LiveRuntimeAuthority::from_fields(
1155 AppMode::Agent,
1156 config.allow_shell,
1157 config.trust_mode,
1158 false,
1159 crate::tui::approval::ApprovalMode::Suggest,
1160 api_config.sandbox_mode.clone(),
1161 ),
1162 )));
1163 let tool_exec_lock = Arc::new(RwLock::new(()));
1164 let plugin_registry = config
1165 .plugin_registry
1166 .as_ref()
1167 .filter(|registry| registry.workspace() == config.workspace)
1168 .cloned()
1169 .unwrap_or_else(|| Arc::new(crate::plugins::PluginRegistry::empty(&config.workspace)));
1170
1171 // Create clients for both providers
1172 let (deepseek_client, deepseek_client_error) = match DeepSeekClient::new(api_config) {
1173 Ok(client) => (Some(client), None),
1174 Err(err) => (None, Some(err.to_string())),
1175 };
1176 let model_client = deepseek_client
1177 .as_ref()
1178 .map(|client| Arc::new(client.clone()) as SharedModelClient);
1179 let api_provider = api_config.api_provider();
1180 let (api_provider_identity, api_provider_id) = api_config
1181 .active_provider_identity(api_provider)
1182 .map(|identity| (identity.key, identity.exact_id))
1183 .unwrap_or_else(|_| {
1184 let key = api_config.provider_identity_for(api_provider);
1185 let exact_id = (!(api_provider == ApiProvider::Custom
1186 && api_config.uses_legacy_literal_custom_route()))
1187 .then(|| key.clone());
1188 (key, exact_id)
1189 });
1190 let api_key_env_only_recovery = Self::env_only_api_key_recovery_hint(api_config);
1191
1192 let mut session = Session::new(
1193 config.model.clone(),
1194 config.workspace.clone(),
1195 config.allow_shell,
1196 config.trust_mode,
1197 config.notes_path.clone(),
1198 config.mcp_config_path.clone(),
1199 );
1200 // Set up stable system prompt with project context (default to agent mode).
1201 // Per-turn working-set metadata is injected into the latest user
1202 // message at request time so file churn does not rewrite this prefix.
1203 let user_memory_block = crate::native_memory::native_prompt_block(
1204 config.memory_enabled,
1205 &config.memory_path,
1206 &config.workspace,
1207 );
1208 let prompt_goal_objective =
1209 goal_objective_for_prompt(config.goal_objective.as_deref(), &config.goal_state);
1210 let system_prompt =
1211 prompts::system_prompt_for_mode_with_context_skills_session_and_approval(
1212 &config.workspace,
1213 None,
1214 Some(&config.skills_dir),
1215 Some(&config.instructions),
1216 prompts::PromptSessionContext {
1217 user_memory_block: user_memory_block.as_deref(),
1218 goal_objective: prompt_goal_objective.as_deref(),
1219 project_context_pack_enabled: config.project_context_pack_enabled,
1220 locale_tag: &config.locale_tag,
1221 translation_enabled: config.translation_enabled,
1222 model_id: &config.model,
1223 context_window_override: Some(
1224 crate::route_budget::route_context_window_tokens(
1225 api_provider,
1226 &config.model,
1227 config.active_route_limits,
1228 ),
1229 ),
1230 verbosity: config.verbosity.as_deref(),
1231 skills_scan_codewhale_only: config.skills_scan_codewhale_only,
1232 plugin_registry: Some(plugin_registry.as_ref()),
1233 // Matches `current_mode`'s initial value below; a later
1234 // `/mode` switch re-runs `refresh_system_prompt`.
1235 mode: AppMode::Agent,
1236 },
1237 );
1238 let stable_prompt = Some(system_prompt);
1239 session.last_system_prompt_hash = Some(system_prompt_hash(stable_prompt.as_ref()));
1240 session.system_prompt = stable_prompt;
1241
1242 // Initialize prefix-cache stability monitor (lazy-pin).
1243 // The system prompt is available now but the tool catalog isn't
1244 // fully built until the first turn, so we start unpinned. The
1245 // first `check_and_update` call in the turn loop will pin the
1246 // fingerprint automatically.
1247 let _ = session.prefix_stability.get_or_insert_with(|| {
1248 // Use the tool registry's spec names for fingerprinting.
1249 // At this point tool spec builders may not be registered yet,
1250 // so we start with None — fingerprint will pin on first request.
1251 crate::prefix_cache::PrefixStabilityManager::new_unpinned()
1252 });
1253
1254 let subagent_manager = new_shared_subagent_manager_with_timeout(
1255 config.workspace.clone(),
1256 config.max_subagents,
1257 config.max_admitted_subagents,
1258 config.subagent_heartbeat_timeout,
1259 config.launch_concurrency,
1260 config.subagent_token_budget,
1261 );
1262 let shell_manager = config
1263 .runtime_services
1264 .shell_manager
1265 .clone()
1266 .unwrap_or_else(|| new_shared_shell_manager(config.workspace.clone()));
1267 match shell_manager.lock() {
1268 Ok(mut manager) => manager.set_prefer_bwrap(config.prefer_bwrap),
1269 Err(poisoned) => poisoned.into_inner().set_prefer_bwrap(config.prefer_bwrap),
1270 }
1271 let file_read_tracker = new_shared_file_read_tracker();
1272 let lsp_manager = Arc::new(match config.lsp_config.clone() {
1273 Some(cfg) => crate::lsp::LspManager::new(cfg, config.workspace.clone()),
1274 None => crate::lsp::LspManager::disabled(),
1275 });
1276
1277 // Workshop variable store (#548). Created unconditionally so the Arc
1278 // can be handed to every ToolContext; routing is gated on the router
1279 // field being Some rather than on the vars Arc being present.
1280 let workshop_vars: Option<
1281 std::sync::Arc<
1282 tokio::sync::Mutex<crate::tools::large_output_router::WorkshopVariables>,
1283 >,
1284 > = Some(std::sync::Arc::new(tokio::sync::Mutex::new(
1285 crate::tools::large_output_router::WorkshopVariables::default(),
1286 )));
1287
1288 // External sandbox backend (#516). Logged but non-fatal: if the
1289 // backend fails to construct, the engine continues with local
1290 // execution as the fallback.
1291 let sandbox_backend = crate::sandbox::backend::create_backend(api_config)
1292 .unwrap_or_else(|e| {
1293 tracing::warn!("Failed to create sandbox backend: {e}");
1294 None
1295 })
1296 .map(std::sync::Arc::from);
1297
1298 let active_route_limits = config.active_route_limits;
1299 let engine = Engine {
1300 config,
1301 api_config: api_config.clone(),
1302 authoritative_route_config: None,
1303 deepseek_client,
1304 model_client,
1305 model_client_injected: false,
1306 deepseek_client_error,
1307 api_key_env_only_recovery,
1308 session,
1309 repl_kernel: None,
1310 subagent_manager,
1311 shell_manager,
1312 file_read_tracker,
1313 mcp_pool: None,
1314 plugin_registry,
1315 api_provider,
1316 api_provider_identity,
1317 api_provider_id,
1318 active_route_limits,
1319 active_route_capabilities: codewhale_config::route::RouteCapabilities::default(),
1320 rx_op,
1321 live_runtime_authority: Arc::clone(&live_runtime_authority),
1322 tx_op: tx_op.clone(),
1323 scheduled_goal_continuation: None,
1324 goal_continuation_schedule_seq: 0,
1325 rx_approval,
1326 rx_user_input,
1327 rx_steer,
1328 tx_event,
1329 tx_subagent_completion,
1330 rx_subagent_completion,
1331 delivered_subagent_completion_ids: HashSet::new(),
1332 cancel_token: cancel_token.clone(),
1333 shared_cancel_token: shared_cancel_token.clone(),
1334 cancel_reason: cancel_reason.clone(),
1335 tool_exec_lock,
1336 turn_counter: 0,
1337 lsp_manager,
1338 pending_lsp_blocks: Vec::new(),
1339 workshop_vars,
1340 sandbox_backend,
1341 current_mode: AppMode::Agent,
1342 last_policy_narrowing: None,
1343 last_turn_meta_git_snapshot: StdMutex::new(None),
1344 token_estimate_cache: TokenEstimateCache::new(),
1345 shared_paused: shared_paused.clone(),
1346 advisor_emission_guard: None,
1347 };
1348 let handle = EngineHandle {
1349 tx_op,
1350 rx_event: Arc::new(RwLock::new(rx_event)),
1351 cancel_token: shared_cancel_token,
1352 cancel_reason,
1353 tx_approval,
1354 tx_user_input,
1355 tx_steer,
1356 shared_paused,
1357 client_preflight_required: true,
1358 live_runtime_authority,
1359 };
1360
1361 (engine, handle)
1362 }
1363
1364 /// Construct the real Engine with an injected provider-neutral model
1365 /// client. The event loop, prompt assembly, tool registry/execution,
1366 /// cancellation, and session projection are unchanged; only the model I/O
1367 /// boundary is replaced.
1368 #[allow(dead_code)] // Production injection seam; currently exercised by deterministic Engine tests.
1369 pub fn new_with_model_client(
1370 config: EngineConfig,
1371 api_config: &Config,
1372 client: SharedModelClient,
1373 ) -> (Self, EngineHandle) {
1374 let (mut engine, mut handle) = Self::new(config, api_config);
1375 engine.model_client = Some(client);
1376 engine.model_client_injected = true;
1377 engine.deepseek_client_error = None;
1378 handle.client_preflight_required = false;
1379 (engine, handle)
1380 }
1381
1382 async fn handle_run_shell_command(
1383 &mut self,
1384 command: String,
1385 mode: AppMode,
1386 allow_shell: bool,
1387 trust_mode: bool,
1388 auto_approve: bool,
1389 approval_mode: crate::tui::approval::ApprovalMode,
1390 ) {
1391 self.reset_cancel_token();
1392 self.turn_counter = self.turn_counter.saturating_add(1);
1393
1394 let turn_id = format!(
1395 "{}{seq}",
1396 USER_SHELL_TOOL_ID_PREFIX,
1397 seq = self.turn_counter
1398 );
1399 let tool_id = turn_id.clone();
1400 let tool_name = "Bash".to_string();
1401 let tool_input = json!({ "action": "run", "command": command, "source": "user" });
1402 let snapshot_prompt = tool_input["command"]
1403 .as_str()
1404 .unwrap_or_default()
1405 .to_string();
1406
1407 let authority = TurnAuthority::from_effective_fields(
1408 mode,
1409 allow_shell,
1410 trust_mode,
1411 auto_approve,
1412 approval_mode,
1413 );
1414 self.apply_runtime_mode_policy(&authority);
1415
1416 let _ = self
1417 .tx_event
1418 .send(Event::TurnStarted {
1419 turn_id: turn_id.clone(),
1420 created_at: chrono::Utc::now(),
1421 route: None,
1422 })
1423 .await;
1424
1425 if self.config.snapshots_enabled {
1426 let pre_workspace = self.session.workspace.clone();
1427 let pre_seq = self.turn_counter;
1428 let pre_cap = self.config.snapshots_max_workspace_bytes;
1429 let pre_prompt = snapshot_prompt.clone();
1430 let pre_sid = self.session.id.clone();
1431 let _ = tokio::task::spawn_blocking(move || {
1432 pre_turn_snapshot(
1433 &pre_workspace,
1434 pre_seq,
1435 pre_cap,
1436 Some(&pre_prompt),
1437 Some(&pre_sid),
1438 )
1439 })
1440 .await;
1441 }
1442
1443 let _ = self
1444 .tx_event
1445 .send(Event::ToolCallStarted {
1446 id: tool_id.clone(),
1447 name: tool_name.clone(),
1448 input: tool_input.clone(),
1449 })
1450 .await;
1451
1452 let tool_context = self.build_tool_context(mode, auto_approve);
1453 let registry = ToolRegistryBuilder::new()
1454 .with_shell_tools()
1455 .build(tool_context);
1456
1457 let result = if mode == AppMode::Plan {
1458 Err(ToolError::permission_denied(
1459 "Tool 'Bash' is unavailable in Plan mode".to_string(),
1460 ))
1461 } else if !self.config.features.enabled(Feature::ShellTool) {
1462 Err(ToolError::not_available(
1463 "Tool 'Bash' is disabled by feature flag".to_string(),
1464 ))
1465 } else if let Some(spec) = registry.get(&tool_name) {
1466 // #5191: the human typed this command — typing it IS the approval.
1467 // The tool-approval modal gates model-provenance calls; applying it
1468 // to a user-typed `!` command asks the user to re-approve what they
1469 // just typed. Typed exec ask-rules still apply as hard Block
1470 // denies, and the sandbox/execpolicy layer stays the real safety
1471 // boundary. Model-issued shell calls keep the standard approval
1472 // path; this branch is strictly composer provenance.
1473 let ask_rule_decision = exec_shell_ask_rule_decision(
1474 &self.config,
1475 &tool_name,
1476 &tool_input,
1477 &self.session.workspace,
1478 self.session.approval_mode,
1479 );
1480 if let Some(ToolAskRuleDecision::Block(reason)) = ask_rule_decision {
1481 Err(ToolError::permission_denied(reason))
1482 } else {
1483 emit_tool_audit(json!({
1484 "event": "tool.user_provenance_preapproved",
1485 "tool_id": tool_id.clone(),
1486 "tool_name": tool_name.clone(),
1487 "source": "composer_bang",
1488 }));
1489 Self::execute_tool_with_lock(
1490 self.tool_exec_lock.clone(),
1491 spec.supports_parallel(),
1492 false,
1493 self.tx_event.clone(),
1494 Some(self.cancel_token.clone()),
1495 tool_name.clone(),
1496 tool_input.clone(),
1497 self.session.workspace.clone(),
1498 Some(&registry),
1499 None,
1500 None,
1501 )
1502 .await
1503 }
1504 } else {
1505 Err(ToolError::not_available(
1506 "tool 'Bash' is not registered".to_string(),
1507 ))
1508 };
1509
1510 let mut result = result;
1511 if let Ok(tool_result) = result.as_mut()
1512 && let Some(path) = crate::tools::truncate::apply_spillover_with_artifact(
1513 tool_result,
1514 &tool_id,
1515 &tool_name,
1516 &self.session.id,
1517 )
1518 {
1519 emit_tool_audit(json!({
1520 "event": "tool.spillover",
1521 "tool_id": tool_id.clone(),
1522 "tool_name": tool_name.clone(),
1523 "path": path.display().to_string(),
1524 "source": "composer_bang",
1525 }));
1526 }
1527
1528 let status = user_shell_turn_outcome(&result, self.cancel_token.is_cancelled());
1529 let error = result.as_ref().err().map(ToString::to_string);
1530
1531 let _ = self
1532 .tx_event
1533 .send(Event::ToolCallComplete {
1534 id: tool_id,
1535 name: tool_name,
1536 result,
1537 })
1538 .await;
1539
1540 if status == TurnOutcomeStatus::Interrupted {
1541 self.emit_interrupted_survivor_status().await;
1542 }
1543 let _ = self
1544 .tx_event
1545 .send(Event::TurnComplete {
1546 usage: Usage::default(),
1547 status,
1548 error,
1549 tool_catalog: None,
1550 base_url: None,
1551 })
1552 .await;
1553
1554 if self.config.snapshots_enabled {
1555 let post_workspace = self.session.workspace.clone();
1556 let post_seq = self.turn_counter;
1557 let post_cap = self.config.snapshots_max_workspace_bytes;
1558 let post_sid = self.session.id.clone();
1559 crate::utils::spawn_blocking_supervised("post-shell-turn-snapshot", move || {
1560 post_turn_snapshot(
1561 &post_workspace,
1562 post_seq,
1563 post_cap,
1564 Some(&snapshot_prompt),
1565 Some(&post_sid),
1566 );
1567 });
1568 }
1569 }
1570
1571 /// Apply a user/host mode-or-posture change to the live session.
1572 ///
1573 /// Single authority source for mode/permission state: both the run loop
1574 /// and the active turn's typed live-authority drain land here.
1575 async fn apply_change_mode(
1576 &mut self,
1577 mode: AppMode,
1578 allow_shell: bool,
1579 trust_mode: bool,
1580 auto_approve: bool,
1581 approval_mode: crate::tui::approval::ApprovalMode,
1582 configured_sandbox_mode: Option<String>,
1583 ) {
1584 let authority = TurnAuthority::from_effective_fields(
1585 mode,
1586 allow_shell,
1587 trust_mode,
1588 auto_approve,
1589 approval_mode,
1590 );
1591 let effective_approval = authority.approval_mode_for_session();
1592 let changed = self.current_mode != authority.mode
1593 || self.session.allow_shell != authority.allow_shell
1594 || self.session.trust_mode != authority.trust_mode
1595 || self.session.auto_approve
1596 != (authority.auto_approve
1597 || effective_approval == crate::tui::approval::ApprovalMode::Bypass)
1598 || self.session.approval_mode != effective_approval
1599 || self.api_config.sandbox_mode != configured_sandbox_mode;
1600 self.api_config.sandbox_mode = configured_sandbox_mode;
1601 self.apply_runtime_mode_policy(&authority);
1602 if !changed {
1603 return;
1604 }
1605 self.emit_session_updated().await;
1606 let _ = self
1607 .tx_event
1608 .send(Event::status(format!(
1609 "Runtime policy changed to: {} / {}",
1610 mode.description(),
1611 effective_approval.permission_chip_label(),
1612 )))
1613 .await;
1614 }
1615
1616 fn take_pending_runtime_authority(&self) -> Option<LiveRuntimeAuthority> {
1617 let mut state = self
1618 .live_runtime_authority
1619 .lock()
1620 .unwrap_or_else(std::sync::PoisonError::into_inner);
1621 if state.applied_revision == state.revision {
1622 return None;
1623 }
1624 state.applied_revision = state.revision;
1625 Some(state.authority.clone())
1626 }
1627
1628 fn runtime_authority_snapshot(&self) -> LiveRuntimeAuthority {
1629 self.live_runtime_authority
1630 .lock()
1631 .unwrap_or_else(std::sync::PoisonError::into_inner)
1632 .authority
1633 .clone()
1634 }
1635
1636 async fn apply_runtime_authority(&mut self, authority: LiveRuntimeAuthority) {
1637 self.apply_change_mode(
1638 authority.mode,
1639 authority.allow_shell,
1640 authority.trust_mode,
1641 authority.auto_approve,
1642 authority.approval_mode,
1643 authority.configured_sandbox_mode,
1644 )
1645 .await;
1646 }
1647
1648 async fn apply_pending_runtime_authority(&mut self) -> bool {
1649 let Some(authority) = self.take_pending_runtime_authority() else {
1650 return false;
1651 };
1652 self.apply_runtime_authority(authority).await;
1653 true
1654 }
1655
1656 fn record_applied_runtime_authority(&self, authority: &TurnAuthority) {
1657 let applied = LiveRuntimeAuthority::from_turn_authority(
1658 authority,
1659 self.api_config.sandbox_mode.clone(),
1660 );
1661 let mut state = self
1662 .live_runtime_authority
1663 .lock()
1664 .unwrap_or_else(std::sync::PoisonError::into_inner);
1665 // Never overwrite a newer, not-yet-applied user change with the turn
1666 // posture that preceded it.
1667 if state.revision == state.applied_revision || state.authority == applied {
1668 state.authority = applied;
1669 state.applied_revision = state.revision;
1670 }
1671 }
1672
1673 fn apply_runtime_mode_policy(&mut self, authority: &TurnAuthority) {
1674 // Mode doctrine lives in the stable prefix (#4780), so a mode change
1675 // has to rebuild it. `refresh_system_prompt` is hash-guarded and only
1676 // swaps the prompt when the composed text actually differs, so modes
1677 // that share an overlay (Agent/Auto/Yolo) keep the prefix cache warm.
1678 let mode_changed = self.current_mode != authority.mode;
1679 self.current_mode = authority.mode;
1680 if mode_changed {
1681 self.refresh_system_prompt();
1682 }
1683 self.session.allow_shell = authority.allow_shell;
1684 self.config.allow_shell = authority.allow_shell;
1685 self.session.trust_mode = authority.trust_mode;
1686 self.config.trust_mode = authority.trust_mode;
1687 self.session.approval_mode = authority.approval_mode_for_session();
1688 self.session.auto_approve = authority.auto_approve
1689 || self.session.approval_mode == crate::tui::approval::ApprovalMode::Bypass;
1690 self.record_applied_runtime_authority(authority);
1691 }
1692
1693 fn schedule_goal_continuation(&mut self, dynamic_tools: Vec<DynamicToolSpec>) {
1694 if let Some(scheduled) = self.scheduled_goal_continuation.as_mut() {
1695 // A normal user turn or idle child handoff can finish while the
1696 // prior synthetic token is already queued. Refresh that one token
1697 // instead of multiplying autonomous turns and provider spend.
1698 scheduled.dynamic_tools = dynamic_tools;
1699 self.try_flush_pending_goal_continuation();
1700 return;
1701 }
1702
1703 self.goal_continuation_schedule_seq =
1704 self.goal_continuation_schedule_seq.wrapping_add(1).max(1);
1705 self.scheduled_goal_continuation = Some(ScheduledGoalContinuation {
1706 id: self.goal_continuation_schedule_seq,
1707 dynamic_tools,
1708 enqueued: false,
1709 });
1710 self.try_flush_pending_goal_continuation();
1711 }
1712
1713 fn cancel_scheduled_goal_continuation(&mut self) {
1714 if self.scheduled_goal_continuation.take().is_some() {
1715 tracing::debug!(
1716 "cancelled an outstanding goal continuation after a non-completed turn"
1717 );
1718 }
1719 }
1720
1721 fn take_scheduled_goal_continuation(
1722 &mut self,
1723 engine_schedule_id: Option<u64>,
1724 direct_dynamic_tools: Vec<DynamicToolSpec>,
1725 ) -> Option<Vec<DynamicToolSpec>> {
1726 let Some(schedule_id) = engine_schedule_id else {
1727 return Some(direct_dynamic_tools);
1728 };
1729 let Some(scheduled) = self.scheduled_goal_continuation.take() else {
1730 tracing::warn!(
1731 schedule_id,
1732 "discarding stale engine-owned goal continuation token"
1733 );
1734 return None;
1735 };
1736 if scheduled.id != schedule_id {
1737 tracing::warn!(
1738 schedule_id,
1739 current_schedule_id = scheduled.id,
1740 "discarding superseded engine-owned goal continuation token"
1741 );
1742 self.scheduled_goal_continuation = Some(scheduled);
1743 return None;
1744 }
1745
1746 // Clear before executing the synthetic turn. A successful execution
1747 // may now schedule exactly one replacement; inactive/failed turns do
1748 // not leave a phantom outstanding marker behind.
1749 Some(scheduled.dynamic_tools)
1750 }
1751
1752 fn has_scheduled_goal_continuation(&self) -> bool {
1753 self.scheduled_goal_continuation.is_some()
1754 }
1755
1756 fn bounded_redacted_goal_failure_detail(&self, detail: &str) -> Option<String> {
1757 let detail = detail.trim();
1758 if detail.is_empty() {
1759 return None;
1760 }
1761 // This message becomes durable goal state. Reuse the model boundary's
1762 // exact configured-secret redactor when available; that helper also
1763 // applies the config persistence redactor as a universal backstop.
1764 let detail = self.deepseek_client.as_ref().map_or_else(
1765 || codewhale_config::persistence::redact_secrets(detail),
1766 |client| client.redact_model_bound_text(detail),
1767 );
1768 Some(crate::utils::truncate_with_ellipsis(
1769 &detail,
1770 GOAL_CONTINUATION_FAILURE_DETAIL_MAX_BYTES,
1771 "…",
1772 ))
1773 }
1774
1775 fn goal_continuation_failure_message(&self, error: Option<&str>) -> String {
1776 self.bounded_redacted_goal_failure_detail(error.unwrap_or_default()).map_or_else(
1777 || {
1778 "Goal continuation blocked because the model turn failed without a provider reason. Fix the provider route or credentials, then resume the goal."
1779 .to_string()
1780 },
1781 |detail| {
1782 format!(
1783 "Goal continuation blocked because the model turn failed: {detail}. Fix the failure, then resume the goal."
1784 )
1785 },
1786 )
1787 }
1788
1789 fn goal_turn_not_started_message(&self, error: Option<&str>) -> String {
1790 self.bounded_redacted_goal_failure_detail(error.unwrap_or_default()).map_or_else(
1791 || {
1792 "Goal continuation blocked because the next model turn could not be started. Fix the provider route or credentials, then resume the goal."
1793 .to_string()
1794 },
1795 |detail| {
1796 format!(
1797 "Goal continuation blocked because the next model turn could not be started: {detail}. Fix the provider route or credentials, then resume the goal."
1798 )
1799 },
1800 )
1801 }
1802
1803 fn try_flush_pending_goal_continuation(&mut self) {
1804 let Some(scheduled) = self.scheduled_goal_continuation.as_ref() else {
1805 return;
1806 };
1807 if scheduled.enqueued {
1808 return;
1809 }
1810 let schedule_id = scheduled.id;
1811
1812 match self.tx_op.try_send(Op::ContinueGoal {
1813 // The authoritative set stays in `scheduled_goal_continuation` so
1814 // later completed turns can refresh it without moving this token.
1815 dynamic_tools: Vec::new(),
1816 engine_schedule_id: Some(schedule_id),
1817 }) {
1818 Ok(()) => {
1819 if let Some(scheduled) = self.scheduled_goal_continuation.as_mut()
1820 && scheduled.id == schedule_id
1821 {
1822 scheduled.enqueued = true;
1823 }
1824 }
1825 Err(mpsc::error::TrySendError::Closed(_)) => {
1826 tracing::warn!("goal continuation dropped because the engine mailbox is closed");
1827 if self
1828 .scheduled_goal_continuation
1829 .as_ref()
1830 .is_some_and(|scheduled| scheduled.id == schedule_id)
1831 {
1832 self.scheduled_goal_continuation = None;
1833 }
1834 }
1835 Err(mpsc::error::TrySendError::Full(_)) => {}
1836 }
1837 }
1838
1839 async fn next_run_input(&mut self, host_managed_turns: bool) -> Option<EngineRunInput> {
1840 // A full mailbox means queued controls must run first. Retrying at the
1841 // top of each receive appends the continuation behind the remaining
1842 // controls as soon as one slot becomes available.
1843 self.try_flush_pending_goal_continuation();
1844 if self.has_scheduled_goal_continuation() {
1845 // The synthetic token sits behind every operation that was already
1846 // queued when it was scheduled. Drain FIFO operations through that
1847 // token before accepting an idle child completion, whether or not
1848 // the mailbox happened to be full. Consuming or cancelling the
1849 // schedule marker restores normal select fairness immediately.
1850 self.rx_op
1851 .recv()
1852 .await
1853 .map(|op| EngineRunInput::Operation(Box::new(op)))
1854 } else {
1855 loop {
1856 let shell_wake_armed = !host_managed_turns && self.idle_shell_wake_armed();
1857 tokio::select! {
1858 op = self.rx_op.recv() => {
1859 return op.map(|op| EngineRunInput::Operation(Box::new(op)));
1860 }
1861 completion = self.rx_subagent_completion.recv(), if !host_managed_turns => {
1862 return completion.map(EngineRunInput::SubAgentCompletion);
1863 }
1864 // Background shells have no completion channel, so an
1865 // idle engine polls only while a goal is active and a
1866 // background job is outstanding; the arm disarms itself
1867 // the moment either condition clears.
1868 () = tokio::time::sleep(Duration::from_millis(SHELL_WAKE_POLL_MS)), if shell_wake_armed => {
1869 if self.finished_background_shell_pending() {
1870 return Some(EngineRunInput::ShellCompletionWake);
1871 }
1872 }
1873 }
1874 }
1875 }
1876 }
1877
1878 /// Whether the idle loop should poll for background shell completion:
1879 /// only while a goal is active and a background job is running or has
1880 /// finished without being claimed yet.
1881 fn idle_shell_wake_armed(&self) -> bool {
1882 let goal_active = self
1883 .config
1884 .goal_state
1885 .lock()
1886 .map(|state| state.snapshot().is_active())
1887 .unwrap_or(false);
1888 if !goal_active {
1889 return false;
1890 }
1891 self.shell_manager
1892 .lock()
1893 .map(|manager| manager.may_have_undelivered_completion())
1894 .unwrap_or(false)
1895 }
1896
1897 /// Whether a finished background job is waiting to be claimed.
1898 fn finished_background_shell_pending(&self) -> bool {
1899 self.shell_manager
1900 .lock()
1901 .map(|mut manager| manager.has_finished_unreported_jobs())
1902 .unwrap_or(false)
1903 }
1904
1905 /// An idle-engine wake for finished background shell work: queue a goal
1906 /// continuation. The evidence itself is claimed by the boundary drain in
1907 /// `handle_send_message`, so the continuation turn reads the completion
1908 /// payload the same way a user-initiated turn would.
1909 async fn handle_idle_shell_completion_wake(&mut self) {
1910 let _ = self
1911 .tx_event
1912 .send(Event::status(
1913 "Background shell work finished; continuing the active goal".to_string(),
1914 ))
1915 .await;
1916 self.schedule_goal_continuation(Vec::new());
1917 }
1918
1919 /// Run the engine event loop
1920 #[allow(clippy::too_many_lines)]
1921 pub async fn run(mut self) {
1922 // RuntimeThreadManager owns durable turn claims and installs a thread
1923 // id in runtime services. Only the interactive TUI may autonomously
1924 // create a new turn while the engine is otherwise idle; a hosted
1925 // engine must wait for its host to claim and explicitly dispatch the
1926 // next turn so events cannot be attached to the wrong durable record.
1927 let host_managed_turns = self.host_managed_turns();
1928
1929 loop {
1930 let Some(input) = self.next_run_input(host_managed_turns).await else {
1931 break;
1932 };
1933
1934 // Runtime posture updates publish through shared typed state
1935 // before attempting their best-effort wake-up. If the mailbox was
1936 // already full, its next queued operation is the wake-up: apply
1937 // the latest authority before doing any work under an obsolete
1938 // policy.
1939 if matches!(&input, EngineRunInput::Operation(_)) {
1940 self.apply_pending_runtime_authority().await;
1941 }
1942
1943 match input {
1944 EngineRunInput::SubAgentCompletion(completion) => {
1945 self.handle_idle_subagent_completion(completion).await;
1946 }
1947 EngineRunInput::ShellCompletionWake => {
1948 self.handle_idle_shell_completion_wake().await;
1949 }
1950 EngineRunInput::Operation(op) => match *op {
1951 Op::SendMessage {
1952 content,
1953 mode,
1954 route,
1955 compaction,
1956 goal_objective,
1957 goal_token_budget,
1958 goal_status,
1959 reasoning_effort,
1960 reasoning_effort_auto,
1961 auto_model,
1962 allow_shell,
1963 trust_mode,
1964 auto_approve,
1965 approval_mode,
1966 translation_enabled,
1967 allowed_tools,
1968 dynamic_tools,
1969 hook_executor,
1970 verbosity,
1971 provenance,
1972 } => {
1973 self.handle_send_message(
1974 content,
1975 mode,
1976 *route,
1977 *compaction,
1978 goal_objective,
1979 goal_token_budget,
1980 goal_status,
1981 reasoning_effort,
1982 reasoning_effort_auto,
1983 auto_model,
1984 allow_shell,
1985 trust_mode,
1986 auto_approve,
1987 approval_mode,
1988 translation_enabled,
1989 allowed_tools,
1990 dynamic_tools,
1991 hook_executor,
1992 verbosity,
1993 provenance,
1994 )
1995 .await;
1996 }
1997 Op::ContinueGoal {
1998 dynamic_tools,
1999 engine_schedule_id,
2000 } => {
2001 let Some(dynamic_tools) = self
2002 .take_scheduled_goal_continuation(engine_schedule_id, dynamic_tools)
2003 else {
2004 continue;
2005 };
2006 // Status controls queued while the previous turn was
2007 // running are processed before this operation. Re-read
2008 // the live goal now so pause/clear/complete/blocked can
2009 // cancel a stale continuation without starting a turn.
2010 let (content, goal_snapshot) = match self.goal_continuation_if_active() {
2011 GoalContinuationAction::Inactive => continue,
2012 GoalContinuationAction::Dispatch { content, snapshot } => {
2013 (content, *snapshot)
2014 }
2015 GoalContinuationAction::Stopped { message, reason } => {
2016 self.pause_goal_continuation(reason, message).await;
2017 continue;
2018 }
2019 };
2020 // Budget and inactive-state decisions are route
2021 // independent. Resolve the live route only for a real
2022 // dispatch so an exhausted goal still reaches its
2023 // truthful terminal state when provider config drifted.
2024 let route = match self.current_runtime_route() {
2025 Ok(route) => route,
2026 Err(err) => {
2027 let message = format!(
2028 "Goal continuation blocked because its provider route is no longer valid: {err}. Fix the route, then resume the goal."
2029 );
2030 let _ = self
2031 .tx_event
2032 .send(Event::error(ErrorEnvelope::fatal_auth(format!(
2033 "Goal continuation stopped because its provider route is no longer valid: {err}"
2034 ))))
2035 .await;
2036 self.block_goal_continuation(message).await;
2037 continue;
2038 }
2039 };
2040
2041 let _ = self
2042 .handle_send_message(
2043 content,
2044 self.current_mode,
2045 route,
2046 self.config.compaction.clone(),
2047 goal_snapshot.objective,
2048 goal_snapshot.token_budget,
2049 GoalStatus::Active,
2050 self.session.reasoning_effort.clone(),
2051 self.session.reasoning_effort_auto,
2052 self.session.auto_model,
2053 self.session.allow_shell,
2054 self.session.trust_mode,
2055 self.session.auto_approve,
2056 self.session.approval_mode,
2057 self.config.translation_enabled,
2058 self.config.allowed_tools.clone(),
2059 dynamic_tools,
2060 self.config.hook_executor.clone(),
2061 self.config.verbosity.clone(),
2062 UserInputProvenance::Runtime,
2063 )
2064 .await;
2065 }
2066 Op::RunShellCommand {
2067 command,
2068 mode,
2069 allow_shell,
2070 trust_mode,
2071 auto_approve,
2072 approval_mode,
2073 } => {
2074 self.handle_run_shell_command(
2075 command,
2076 mode,
2077 allow_shell,
2078 trust_mode,
2079 auto_approve,
2080 approval_mode,
2081 )
2082 .await;
2083 }
2084 Op::SetGoalStatus { status, clear } => {
2085 self.handle_set_goal_status(status, clear).await;
2086 }
2087 Op::CancelRequest => {
2088 self.cancel_token.cancel();
2089 self.reset_cancel_token();
2090 }
2091 Op::ApproveToolCall { id } => {
2092 // Tool approval handling will be implemented in tools module
2093 let _ = self
2094 .tx_event
2095 .send(Event::status(format!("Approved tool call: {id}")))
2096 .await;
2097 }
2098 Op::DenyToolCall { id } => {
2099 let _ = self
2100 .tx_event
2101 .send(Event::status(format!("Denied tool call: {id}")))
2102 .await;
2103 }
2104 Op::SpawnSubAgent { prompt } => {
2105 let Some(client) = self.deepseek_client.clone() else {
2106 let message = self
2107 .deepseek_client_error
2108 .as_deref()
2109 .map(|err| format!("Failed to spawn sub-agent: {err}"))
2110 .unwrap_or_else(|| {
2111 "Failed to spawn sub-agent: API client not configured"
2112 .to_string()
2113 });
2114 let _ = self
2115 .tx_event
2116 .send(Event::error(ErrorEnvelope::fatal(message)))
2117 .await;
2118 continue;
2119 };
2120
2121 let mcp_pool = if self.config.features.enabled(Feature::Mcp) {
2122 self.ensure_mcp_pool().await.ok()
2123 } else {
2124 None
2125 };
2126
2127 let mut runtime = SubAgentRuntime::new(
2128 client,
2129 self.session.model.clone(),
2130 // Sub-agents don't inherit YOLO mode - use Agent mode defaults
2131 self.build_tool_context(AppMode::Agent, self.session.auto_approve),
2132 self.session.allow_shell,
2133 Some(self.tx_event.clone()),
2134 Arc::clone(&self.subagent_manager),
2135 )
2136 .with_locale_tag(self.config.locale_tag.clone())
2137 .with_role_models(self.subagent_role_models())
2138 .with_api_config(self.api_config.clone())
2139 .with_fleet_roster(self.config.fleet_roster.clone())
2140 .with_auto_model(self.session.auto_model)
2141 .with_reasoning_effort(
2142 self.session.reasoning_effort.clone(),
2143 self.session.reasoning_effort_auto,
2144 )
2145 .with_agent_tool_surface_options(self.agent_tool_surface_options(
2146 shell_policy_for_mode(AppMode::Agent, self.session.allow_shell),
2147 ))
2148 .with_max_spawn_depth(self.config.max_spawn_depth)
2149 .with_step_api_timeout(self.config.subagent_api_timeout)
2150 .with_speech_output_dir(self.config.speech_output_dir.clone())
2151 .with_mcp_pool(mcp_pool)
2152 .with_parent_mode(self.current_mode)
2153 // #4810: no `with_todos` here — this runtime *is* the
2154 // spawned background agent, and `background_runtime()`
2155 // gives it its own list. Binding the session list would
2156 // be discarded anyway, and reading as if the background
2157 // agent writes the user's Work checklist.
2158 .background_runtime();
2159 // #4042: thread the session's --disallowed-tools into
2160 // the child so tool restrictions flow down to sub-agents.
2161 runtime.worker_profile.denied_tools =
2162 self.config.disallowed_tools.clone().unwrap_or_default();
2163 let route = resolve_subagent_assignment_route(
2164 &runtime,
2165 None,
2166 &prompt,
2167 &FleetRole::Worker,
2168 ModelRoute::Inherit,
2169 SubAgentThinking::Inherit,
2170 )
2171 .await;
2172 let effective_model = match ensure_subagent_model_for_provider(
2173 &runtime,
2174 &route.model_route,
2175 route.model,
2176 ) {
2177 Ok(model) => model,
2178 Err(err) => {
2179 let _ = self
2180 .tx_event
2181 .send(Event::error(ErrorEnvelope::fatal(format!(
2182 "Failed to spawn sub-agent: {err}"
2183 ))))
2184 .await;
2185 continue;
2186 }
2187 };
2188 runtime.model = effective_model;
2189 runtime.reasoning_effort = route.reasoning_effort;
2190 runtime.reasoning_effort_auto = false;
2191
2192 let result = {
2193 let mut manager = self.subagent_manager.write().await;
2194 manager.spawn_background(
2195 Arc::clone(&self.subagent_manager),
2196 runtime,
2197 FleetRole::Worker,
2198 prompt.clone(),
2199 None,
2200 )
2201 };
2202
2203 match result {
2204 Ok(snapshot) => {
2205 let _ = self
2206 .tx_event
2207 .send(Event::status(format!(
2208 "Spawned sub-agent {}",
2209 snapshot.agent_id
2210 )))
2211 .await;
2212 }
2213 Err(err) => {
2214 let _ = self
2215 .tx_event
2216 .send(Event::error(ErrorEnvelope::fatal(format!(
2217 "Failed to spawn sub-agent: {err}"
2218 ))))
2219 .await;
2220 }
2221 }
2222 }
2223 Op::PreviewOutboundRequest {
2224 inputs,
2225 json,
2226 base_prompt_only,
2227 } => {
2228 // Pure inspection: no turn is started, no message is
2229 // added, no engine state is written, and no provider
2230 // request is sent. Facts that are not exactly knowable
2231 // come back as typed unavailable sections rather than
2232 // as an error or a guess.
2233 let rendered = if base_prompt_only {
2234 crate::request_manifest::exact_base_prompt_only()
2235 } else {
2236 let manifest = self.build_request_manifest(*inputs).await;
2237 if json {
2238 manifest.to_json()
2239 } else {
2240 manifest.render()
2241 }
2242 };
2243 let _ = self
2244 .tx_event
2245 .send(Event::RequestManifestReady { rendered })
2246 .await;
2247 }
2248 Op::ListSubAgents => {
2249 // #3803: the sidebar refresh is a read-only snapshot.
2250 // Render from a read lock; only take the write lock to
2251 // run cleanup on a bounded cadence, so a UI refresh storm
2252 // during a sub-agent fanout no longer contends for the
2253 // write lock (against completions/persistence) on every
2254 // request. Cleanup still auto-cancels stale agents.
2255 self.touch_workers_with_running_shells().await;
2256 let due = {
2257 let manager = self.subagent_manager.read().await;
2258 manager.cleanup_due(
2259 crate::tools::subagent::SUBAGENT_LIST_CLEANUP_MIN_INTERVAL,
2260 )
2261 };
2262 let event = if due {
2263 let mut manager = self.subagent_manager.write().await;
2264 manager.cleanup(Duration::from_secs(60 * 60));
2265 agent_list_event(&manager)
2266 } else {
2267 let manager = self.subagent_manager.read().await;
2268 agent_list_event(&manager)
2269 };
2270 // #3802: use non-blocking send — this is a refresh event
2271 // that can safely be dropped when the channel is full.
2272 // The next drain cycle will re-request the list.
2273 if let Err(_e) = self.tx_event.try_send(event) {
2274 tracing::debug!(
2275 "Event channel full; dropping ListSubAgents refresh (will retry next drain)"
2276 );
2277 }
2278 }
2279 Op::CancelSubAgent { agent_id } => {
2280 let result = {
2281 let mut manager = self.subagent_manager.write().await;
2282 match manager.cancel_agent(&agent_id) {
2283 Ok(_) => Ok(agent_list_event(&manager)),
2284 Err(err) => Err(err),
2285 }
2286 };
2287 match result {
2288 Ok(event) => {
2289 if let Err(_e) = self.tx_event.try_send(event) {
2290 tracing::debug!(
2291 "Event channel full; dropping CancelSubAgent refresh"
2292 );
2293 }
2294 }
2295 Err(err) => {
2296 let _ =
2297 self.tx_event
2298 .try_send(Event::error(ErrorEnvelope::transient(format!(
2299 "Failed to cancel sub-agent {agent_id}: {err}"
2300 ))));
2301 }
2302 }
2303 }
2304 Op::ChangeMode { .. } => {
2305 // The mailbox payload may predate a newer posture that
2306 // was published while the channel was full. Apply the
2307 // single live snapshot so a stale queued ChangeMode
2308 // can never roll authority backward.
2309 let authority = self.runtime_authority_snapshot();
2310 self.apply_runtime_authority(authority).await;
2311 }
2312 Op::SetModel {
2313 model,
2314 mode: _,
2315 route_limits,
2316 } => {
2317 self.session.auto_model = model.trim().eq_ignore_ascii_case("auto");
2318 self.session.model = model;
2319 self.config.model.clone_from(&self.session.model);
2320 self.active_route_limits = route_limits;
2321 // This lightweight operation carries no executable
2322 // route candidate, so old provider/model capability
2323 // facts must not bleed into the new model.
2324 self.active_route_capabilities =
2325 codewhale_config::route::RouteCapabilities::default();
2326 self.refresh_system_prompt();
2327 self.emit_session_updated().await;
2328 let _ = self
2329 .tx_event
2330 .send(Event::status(format!(
2331 "Model set to: {}",
2332 self.session.model
2333 )))
2334 .await;
2335 }
2336 Op::SetCompaction { config } => {
2337 let enabled = config.enabled;
2338 self.config.compaction = config;
2339 let _ = self
2340 .tx_event
2341 .send(Event::status(format!(
2342 "Auto-compaction {}",
2343 if enabled { "enabled" } else { "disabled" }
2344 )))
2345 .await;
2346 }
2347 Op::SetPermissionRuleset { ruleset } => {
2348 self.config.exec_policy_engine.set_ruleset(ruleset);
2349 }
2350 Op::SetStreamChunkTimeout { timeout_secs } => {
2351 self.config.stream_chunk_timeout = Duration::from_secs(timeout_secs);
2352 let _ = self
2353 .tx_event
2354 .send(Event::status(format!(
2355 "Stream chunk timeout set to {timeout_secs}s"
2356 )))
2357 .await;
2358 }
2359 Op::SetSubagentRuntimeConfig {
2360 enabled,
2361 max_subagents,
2362 launch_concurrency,
2363 max_spawn_depth,
2364 api_timeout_secs,
2365 heartbeat_timeout_secs,
2366 } => {
2367 self.config.subagents_enabled = enabled;
2368 self.config.max_subagents =
2369 max_subagents.clamp(1, crate::config::MAX_SUBAGENTS);
2370 self.config.launch_concurrency =
2371 launch_concurrency.clamp(1, self.config.max_subagents);
2372 self.config.max_spawn_depth =
2373 max_spawn_depth.min(codewhale_config::MAX_SPAWN_DEPTH_CEILING);
2374 self.config.subagent_api_timeout = Duration::from_secs(api_timeout_secs);
2375 self.config.subagent_heartbeat_timeout =
2376 Duration::from_secs(heartbeat_timeout_secs);
2377 let launch_gate_applied = {
2378 let mut manager = self.subagent_manager.write().await;
2379 manager.update_runtime_limits(
2380 self.config.max_subagents,
2381 self.config.max_admitted_subagents,
2382 self.config.subagent_heartbeat_timeout,
2383 self.config.launch_concurrency,
2384 self.config.subagent_token_budget,
2385 )
2386 };
2387 let launch_note = if launch_gate_applied {
2388 ""
2389 } else {
2390 "; launch_concurrency takes full effect after active sub-agents finish or the session restarts"
2391 };
2392 let _ = self
2393 .tx_event
2394 .send(Event::status(format!(
2395 "Sub-agent runtime updated: enabled={enabled}, max_subagents={}, launch_concurrency={}, max_depth={}{}",
2396 self.config.max_subagents,
2397 self.config.launch_concurrency,
2398 self.config.max_spawn_depth,
2399 launch_note
2400 )))
2401 .await;
2402 }
2403 Op::SetFleetRoster { roster } => {
2404 self.config.fleet_roster = roster;
2405 let _ = self
2406 .tx_event
2407 .send(Event::status(
2408 "Fleet roster refreshed for subsequent turns".to_string(),
2409 ))
2410 .await;
2411 }
2412 Op::SyncSession {
2413 session_id,
2414 messages,
2415 system_prompt,
2416 system_prompt_override,
2417 model,
2418 workspace,
2419 mode,
2420 } => {
2421 let plugin_workspace_changed =
2422 self.plugin_registry.workspace() != workspace.as_path();
2423 if let Some(session_id) = session_id {
2424 self.session.id = session_id;
2425 } else if messages.is_empty() && system_prompt.is_none() {
2426 self.session.id = uuid::Uuid::new_v4().to_string();
2427 }
2428 self.session.messages =
2429 crate::runtime_handoff::project_messages_for_restore(&messages).into();
2430 self.session.compaction_summary_prompt =
2431 extract_compaction_summary_prompt(system_prompt.clone());
2432 self.session.system_prompt = system_prompt;
2433 self.session.last_system_prompt_hash =
2434 Some(system_prompt_hash(self.session.system_prompt.as_ref()));
2435 // Host-supplied prompts are persisted prefixes. Keep them
2436 // byte-stable; mode/runtime state is projected per request.
2437 self.session.system_prompt_override =
2438 system_prompt_override && self.session.system_prompt.is_some();
2439 self.session.auto_model = model.trim().eq_ignore_ascii_case("auto");
2440 self.session.model = model;
2441 self.session.workspace = workspace.clone();
2442 self.current_mode = mode;
2443 self.config.model.clone_from(&self.session.model);
2444 self.config.workspace = workspace.clone();
2445 if plugin_workspace_changed {
2446 self.plugin_registry =
2447 self.plugin_registry.rediscover_for_workspace(&workspace);
2448 self.config.plugin_registry = Some(Arc::clone(&self.plugin_registry));
2449 // A pool may contain plugin servers and authority
2450 // receipts from the previous workspace snapshot.
2451 self.mcp_pool = None;
2452 }
2453 let ctx =
2454 crate::project_context::load_project_context_with_parents(&workspace);
2455 self.session.project_context = if ctx.has_instructions() {
2456 Some(ctx)
2457 } else {
2458 None
2459 };
2460 self.session.rebuild_working_set();
2461 self.reconcile_restored_work_bindings().await;
2462 self.emit_session_updated().await;
2463 let _ = self
2464 .tx_event
2465 .send(Event::status("Session context synced".to_string()))
2466 .await;
2467 }
2468 Op::CompactContext { route, compaction } => {
2469 if let Err(err) = self.install_resolved_runtime_route(*route) {
2470 let _ = self
2471 .tx_event
2472 .send(Event::error(ErrorEnvelope::fatal_auth(format!(
2473 "Cannot compact context because its provider route is not ready: {err}"
2474 ))))
2475 .await;
2476 continue;
2477 }
2478 self.config.compaction = *compaction;
2479 self.handle_manual_compaction().await;
2480 }
2481 Op::GetSessionSnapshot { tx } => {
2482 let total_tokens = self.session.total_usage.input_tokens
2483 + self.session.total_usage.output_tokens;
2484 let snapshot = SessionSnapshot {
2485 messages: self.session.messages.to_vec(),
2486 total_tokens,
2487 model: self.session.model.clone(),
2488 model_provider: self.api_provider.as_str().to_string(),
2489 model_provider_id: self.api_provider_id.clone(),
2490 workspace: self.session.workspace.clone(),
2491 system_prompt: self.session.system_prompt.clone(),
2492 mode: self.current_mode.as_setting().to_string(),
2493 };
2494 if let Some(tx) = tx.lock().ok().and_then(|mut g| g.take()) {
2495 let _ = tx.send(snapshot);
2496 }
2497 }
2498 Op::GetProviderRuntimeStatus { tx } => {
2499 let status = if let Some(client) = self.deepseek_client.as_ref() {
2500 ProviderRuntimeStatus {
2501 provider: client.api_provider(),
2502 request_concurrency_limit: client
2503 .provider_request_concurrency_limit(),
2504 active_provider_requests: client.active_provider_requests(),
2505 }
2506 } else {
2507 let provider = self.api_config.api_provider();
2508 ProviderRuntimeStatus {
2509 provider,
2510 request_concurrency_limit: self
2511 .api_config
2512 .provider_max_concurrency(provider),
2513 active_provider_requests: 0,
2514 }
2515 };
2516 if let Some(tx) = tx.lock().ok().and_then(|mut g| g.take()) {
2517 let _ = tx.send(status);
2518 }
2519 }
2520 Op::ReloadMcp { config_path, tx } => {
2521 let result = self.reload_mcp_pool(config_path).await.map_err(|error| {
2522 codewhale_config::persistence::redact_secrets(&format!("{error:#}"))
2523 });
2524 if let Some(tx) = tx.lock().ok().and_then(|mut guard| guard.take()) {
2525 let _ = tx.send(result);
2526 }
2527 }
2528 Op::PurgeContext => {
2529 self.handle_purge().await;
2530 }
2531 Op::EditLastTurn { new_message } => {
2532 let route = match self.current_runtime_route() {
2533 Ok(route) => route,
2534 Err(err) => {
2535 let _ = self
2536 .tx_event
2537 .send(Event::error(ErrorEnvelope::fatal_auth(format!(
2538 "Cannot edit the last turn because its provider route is no longer valid: {err}"
2539 ))))
2540 .await;
2541 let outcome = SendMessageOutcome::NotStarted {
2542 error: Some(format!(
2543 "provider route is no longer valid: {err}"
2544 )),
2545 };
2546 self.reconcile_non_completed_goal_turn(&outcome).await;
2547 continue;
2548 }
2549 };
2550 // #383: /edit — remove the last user+assistant exchange
2551 // from the session, then re-send with the new content.
2552 // Pop messages from the tail until we've removed the
2553 // most recent user message and everything after it.
2554 // First, find the last user message index.
2555 let mut cut = None;
2556 for (idx, msg) in self.session.messages.iter().enumerate().rev() {
2557 if msg.role == "user" {
2558 cut = Some(idx);
2559 break;
2560 }
2561 }
2562 if let Some(idx) = cut {
2563 self.session.messages.truncate_to(idx);
2564 self.session.bump_messages_revision();
2565 }
2566 // Now dispatch the new message as a normal send,
2567 // reusing the engine's stored mode/model config.
2568 let mode = self.current_mode;
2569 self.handle_send_message(
2570 new_message,
2571 mode,
2572 route,
2573 self.config.compaction.clone(),
2574 self.config.goal_objective.clone(),
2575 self.config.goal_token_budget,
2576 self.config.goal_status,
2577 self.session.reasoning_effort.clone(),
2578 self.session.reasoning_effort_auto,
2579 self.session.auto_model,
2580 self.session.allow_shell,
2581 self.session.trust_mode,
2582 self.session.auto_approve,
2583 self.session.approval_mode,
2584 self.config.translation_enabled,
2585 self.config.allowed_tools.clone(),
2586 Vec::new(),
2587 self.config.hook_executor.clone(),
2588 self.config.verbosity.clone(),
2589 UserInputProvenance::ExternalUser,
2590 )
2591 .await;
2592 }
2593 Op::SetAdvisorEnabled { enabled } => {
2594 self.config.advisor_config.enabled = enabled;
2595 let state = if enabled { "enabled" } else { "disabled" };
2596 let _ = self
2597 .tx_event
2598 .send(Event::status(format!(
2599 "Advisor watcher {state}. Notes will appear after turns with tool calls."
2600 )))
2601 .await;
2602 tracing::info!(target: "advisor", "advisor watcher {state}");
2603 }
2604 Op::Shutdown => {
2605 break;
2606 }
2607 },
2608 }
2609 }
2610
2611 // #freeze: flush any sub-agent checkpoint that the hot-path debounce
2612 // coalesced away, so a graceful shutdown keeps the latest progress.
2613 {
2614 let mut manager = self.subagent_manager.write().await;
2615 manager.flush_pending_persist();
2616 }
2617
2618 // #420: graceful MCP shutdown — send SIGTERM and give stdio servers
2619 // a brief window to exit before drop fires SIGKILL via kill_on_drop.
2620 // Best-effort: pool may not exist (no MCP configured) and the lock
2621 // can fail under contention; either way the kill_on_drop fallback
2622 // still reaps the children.
2623 if let Some(pool) = self.mcp_pool.as_ref() {
2624 let mut guard = pool.lock().await;
2625 guard.shutdown_all().await;
2626 }
2627 }
2628
2629 fn host_managed_turns(&self) -> bool {
2630 self.config.runtime_services.active_thread_id.is_some()
2631 }
2632
2633 async fn emit_session_updated(&self) {
2634 let _ = self
2635 .tx_event
2636 .send(Event::SessionUpdated {
2637 session_id: self.session.id.clone(),
2638 messages: self.session.messages.clone().into(),
2639 system_prompt: self.session.system_prompt.clone(),
2640 model: self.session.model.clone(),
2641 workspace: self.session.workspace.clone(),
2642 })
2643 .await;
2644 }
2645
2646 fn goal_snapshot_for_event(&self) -> Option<GoalSnapshot> {
2647 match self.config.goal_state.lock() {
2648 Ok(state) => {
2649 let snapshot = state.snapshot();
2650 snapshot.objective.is_some().then_some(snapshot)
2651 }
2652 Err(err) => {
2653 tracing::warn!("goal state lock poisoned while emitting goal update: {err}");
2654 None
2655 }
2656 }
2657 }
2658
2659 async fn emit_goal_updated(&self) {
2660 if let Some(snapshot) = self.goal_snapshot_for_event() {
2661 let _ = self.tx_event.send(Event::GoalUpdated { snapshot }).await;
2662 }
2663 }
2664
2665 fn record_goal_usage_for_turn(&self, usage: &Usage, elapsed: std::time::Duration) {
2666 let token_delta =
2667 u64::from(usage.input_tokens).saturating_add(u64::from(usage.output_tokens));
2668 let time_delta_seconds = elapsed.as_secs();
2669 if token_delta == 0 && time_delta_seconds == 0 {
2670 return;
2671 }
2672 match self.config.goal_state.lock() {
2673 Ok(mut state) => state.record_usage(token_delta, time_delta_seconds),
2674 Err(err) => tracing::warn!("goal state lock poisoned while recording usage: {err}"),
2675 }
2676 }
2677
2678 fn active_input_tokens_with_current_text(
2679 &self,
2680 current_text: &str,
2681 system_prompt: Option<&SystemPrompt>,
2682 ) -> usize {
2683 let mut messages: Vec<Message> = self.session.messages.clone().into();
2684 if !current_text.trim().is_empty() {
2685 messages.push(Message {
2686 role: "user".to_string(),
2687 content: vec![ContentBlock::Text {
2688 text: current_text.to_string(),
2689 cache_control: None,
2690 }],
2691 });
2692 }
2693 estimate_input_tokens_conservative(&messages, system_prompt)
2694 }
2695
2696 fn append_resource_metadata_lines(
2697 &self,
2698 lines: &mut Vec<String>,
2699 current_text: &str,
2700 prompt_context: &NextTurnPromptContext,
2701 system_prompt: Option<&SystemPrompt>,
2702 ) {
2703 if let Some(line) = self.context_pressure_line(current_text, prompt_context, system_prompt)
2704 {
2705 lines.push(line);
2706 }
2707 if let Some(line) = self.active_goal_token_budget_line(prompt_context) {
2708 lines.push(line);
2709 }
2710 }
2711
2712 /// One-line context-pressure signal, emitted **only** while the input
2713 /// estimate sits at or above the warning/critical thresholds. No token
2714 /// counts, percentages, or headroom figures: the model only learns that
2715 /// the pressure band it is in has crossed a threshold. Between crossings
2716 /// the line is byte-stable, so ordinary turns do not bust the prefix
2717 /// cache.
2718 fn context_pressure_line(
2719 &self,
2720 current_text: &str,
2721 prompt_context: &NextTurnPromptContext,
2722 system_prompt: Option<&SystemPrompt>,
2723 ) -> Option<String> {
2724 let input_tokens = self.active_input_tokens_with_current_text(current_text, system_prompt);
2725 let budget = route_context_budget_for_route(
2726 prompt_context.provider,
2727 &prompt_context.model,
2728 prompt_context.route_limits,
2729 input_tokens,
2730 )?;
2731 context_pressure_message(budget.usage_percent()).map(str::to_string)
2732 }
2733
2734 /// Goal pacing for the model: the budget figure only, and only while a
2735 /// goal is actually active. Usage/time deltas, rates, and continuation
2736 /// counts are UI telemetry — they changed every turn and invalidated the
2737 /// prefix cache without adding model-steering signal.
2738 fn active_goal_token_budget_line(
2739 &self,
2740 prompt_context: &NextTurnPromptContext,
2741 ) -> Option<String> {
2742 let objective = prompt_context.goal_objective.as_deref()?;
2743 let snapshot = self.config.goal_state.lock().ok()?.snapshot();
2744 let same_goal =
2745 normalized_goal_objective(snapshot.objective.as_deref()).as_deref() == Some(objective);
2746 let token_budget = if same_goal {
2747 snapshot.token_budget
2748 } else {
2749 prompt_context.goal_token_budget
2750 }?;
2751 Some(format!("Active goal token budget: {token_budget}"))
2752 }
2753
2754 async fn add_session_message(&mut self, message: Message) {
2755 self.session.add_message(message);
2756 self.emit_session_updated().await;
2757 }
2758
2759 async fn add_interrupted_assistant_text(&mut self, text: &str) {
2760 if text.is_empty() {
2761 return;
2762 }
2763 let message = Message {
2764 role: crate::models::INTERRUPTED_ASSISTANT_ROLE.to_string(),
2765 content: vec![ContentBlock::Text {
2766 text: text.to_string(),
2767 cache_control: None,
2768 }],
2769 };
2770 let already_committed = self.session.messages.last().is_some_and(|last| {
2771 matches!(
2772 last.role.as_str(),
2773 "assistant" | crate::models::INTERRUPTED_ASSISTANT_ROLE
2774 ) && last.content == message.content
2775 });
2776 if already_committed {
2777 return;
2778 }
2779 self.add_session_message(message).await;
2780 }
2781
2782 #[allow(clippy::too_many_arguments)]
2783 fn turn_metadata_block(
2784 &self,
2785 routed_model: &str,
2786 auto_model: bool,
2787 reasoning_effort: Option<&str>,
2788 reasoning_effort_auto: bool,
2789 provenance: UserInputProvenance,
2790 current_text: &str,
2791 policy_narrowing: Option<&PolicyNarrowingEvent>,
2792 ) -> ContentBlock {
2793 let prompt_context = self.installed_next_turn_prompt_context();
2794 self.turn_metadata_block_from_snapshot(
2795 routed_model,
2796 auto_model,
2797 reasoning_effort,
2798 reasoning_effort_auto,
2799 provenance,
2800 current_text,
2801 TurnMetadataSnapshot {
2802 prompt_context: &prompt_context,
2803 system_prompt: self.session.system_prompt.as_ref(),
2804 approval_mode: self.session.approval_mode,
2805 working_set: &self.session.working_set,
2806 policy_narrowing,
2807 },
2808 )
2809 }
2810
2811 /// Build `<turn_meta>` from an explicit snapshot of the session state a
2812 /// turn installs *before* it writes the block.
2813 ///
2814 /// Production installs approval posture, policy narrowing, and the
2815 /// observed working set on `self`, then reads them back here.
2816 /// `/preview-request` cannot install any of that — it describes a turn
2817 /// that has not started — so it passes the values it would have installed,
2818 /// including a *clone* of the working set with the hypothetical message
2819 /// already observed. That is what makes the previewed block byte-identical
2820 /// to the real one without a single write.
2821 #[allow(clippy::too_many_arguments)]
2822 fn turn_metadata_block_from_snapshot(
2823 &self,
2824 _routed_model: &str,
2825 _auto_model: bool,
2826 _reasoning_effort: Option<&str>,
2827 _reasoning_effort_auto: bool,
2828 provenance: UserInputProvenance,
2829 current_text: &str,
2830 snapshot: TurnMetadataSnapshot<'_>,
2831 ) -> ContentBlock {
2832 let TurnMetadataSnapshot {
2833 prompt_context,
2834 system_prompt,
2835 approval_mode,
2836 working_set,
2837 policy_narrowing,
2838 } = snapshot;
2839 let today = chrono::Local::now().format("%Y-%m-%d").to_string();
2840 let working_set_summary = working_set
2841 .summary_block(&self.config.workspace)
2842 .map(|s| s.trim().to_string())
2843 .filter(|s| !s.is_empty());
2844
2845 // Facts only (#4780 + turn-meta diet). Mode doctrine ships once in the
2846 // stable system prefix. The permission posture does not: preserve its
2847 // compact label so the model can distinguish Ask, Auto-Review, Full
2848 // Access, and Never without repeating question-discipline prose.
2849 // Route/effort/model lines are telemetry the model cannot act on.
2850 // DGF-02 (dogfood 2026-08-02): the model was never told its own
2851 // sandbox posture, so an approved-then-sandbox-blocked write read as
2852 // a mystery failure it burned turns "debugging". Derive the posture
2853 // from the same resolver tool execution uses so the line and the
2854 // enforcement can never disagree. Stable per session (mode, config,
2855 // workspace), so ordinary turns stay byte-identical.
2856 let sandbox_posture = crate::core::authority::sandbox_policy_for_turn(
2857 prompt_context.mode,
2858 approval_mode,
2859 self.api_config.sandbox_mode.as_deref(),
2860 &self.config.workspace,
2861 );
2862 let mut lines = vec![
2863 format!("Current local date: {today}"),
2864 // Workspace path moved here from the static `## Environment` block so
2865 // the static system prefix stays byte-stable across sessions (see
2866 // `render_environment_block` for the prefix-cache rationale).
2867 format!("Current workspace: {}", self.config.workspace.display()),
2868 format!(
2869 "Current permission posture: {}",
2870 approval_mode.permission_chip_label()
2871 ),
2872 format!(
2873 "Current sandbox posture: {}",
2874 sandbox_posture.posture_label()
2875 ),
2876 ];
2877 // On ordinary external turns the user's own message is authoritative by
2878 // construction, so provenance is redundant. On non-external turns
2879 // (sub-agent handoff, runtime events) the *reduced* authority is the
2880 // sole signal, so surface it as one condensed line.
2881 if !provenance.can_authorize_work() {
2882 lines.push(format!(
2883 "Input provenance: {} (non-authoritative)",
2884 provenance.as_str()
2885 ));
2886 }
2887 // #3947: when runtime policy narrowed this turn's authority, the model
2888 // learns that it happened, why, and the exact sentence the user saw.
2889 // Emitted only on a narrowed turn, so the ordinary turn's metadata
2890 // stays byte-stable.
2891 if let Some(event) = policy_narrowing {
2892 lines.push(format!("Authority narrowing: {}", event.reason().as_str()));
2893 lines.push(format!("Authority transition: {}", event.transition()));
2894 lines.push(format!("Authority narrowing status: {}", event.message()));
2895 }
2896 self.append_resource_metadata_lines(
2897 &mut lines,
2898 current_text,
2899 prompt_context,
2900 system_prompt,
2901 );
2902 if let Some(working_set_summary) = working_set_summary {
2903 lines.push(working_set_summary);
2904 }
2905 // #5187 (k3-gap F3): the git snapshot re-collects branch/dirty state
2906 // every turn, so the line's bytes changed after every edit the model
2907 // itself made — churning the block and priming caution each turn.
2908 // Emit it only when the snapshot actually changed since the last
2909 // emitted block; the model can always run `git status` for a fresh
2910 // read.
2911 if let Some(git_snapshot) = crate::tui::workspace_context::collect(&self.config.workspace) {
2912 let mut last = self
2913 .last_turn_meta_git_snapshot
2914 .lock()
2915 .unwrap_or_else(std::sync::PoisonError::into_inner);
2916 if last.as_deref() != Some(git_snapshot.as_str()) {
2917 *last = Some(git_snapshot.clone());
2918 lines.push(format!("Git workspace: {git_snapshot}"));
2919 }
2920 }
2921 let summary = lines.join("\n");
2922
2923 ContentBlock::Text {
2924 text: format!("<turn_meta>\n{summary}\n</turn_meta>"),
2925 cache_control: None,
2926 }
2927 }
2928
2929 /// Assemble the content blocks of a user turn.
2930 ///
2931 /// The text comes first and the turn metadata last — both positions are
2932 /// load-bearing for prompt caching (see
2933 /// [`Self::turn_metadata_block`]), so resolved images are inserted between
2934 /// them rather than at either end.
2935 ///
2936 /// The composer stores an attachment as a `[Attached image: …]` text line
2937 /// and the bytes are read here, once, as the message is built. That keeps
2938 /// multi-megabyte payloads out of the composer and undo history, and it
2939 /// means deleting the line deletes the attachment for free. Anything that
2940 /// cannot be attached becomes a visible notice instead of vanishing.
2941 ///
2942 /// Whether the model can *see* the result is decided per request, not
2943 /// here — see `image_attach::strip_images_when_unsupported`.
2944 fn user_content_blocks(&self, text: String) -> Vec<ContentBlock> {
2945 let expanded = crate::image_attach::expand_attachment_blocks(&text);
2946 let mut content = Vec::with_capacity(2 + expanded.blocks.len());
2947 content.push(ContentBlock::Text {
2948 text,
2949 cache_control: None,
2950 });
2951 content.extend(expanded.blocks);
2952 if let Some(notice) = crate::image_attach::notice_block(&expanded.notices) {
2953 content.push(notice);
2954 }
2955 content
2956 }
2957
2958 /// The user message a turn would build, from an explicit state snapshot.
2959 ///
2960 /// Same block order and same constructor as
2961 /// [`Self::user_text_message_with_turn_metadata_for_route_and_provenance`];
2962 /// only the source of the turn-metadata inputs differs. See
2963 /// [`Self::turn_metadata_block_from_snapshot`].
2964 #[allow(clippy::too_many_arguments)]
2965 pub(super) fn user_text_message_from_snapshot(
2966 &self,
2967 text: String,
2968 routed_model: &str,
2969 auto_model: bool,
2970 reasoning_effort: Option<&str>,
2971 reasoning_effort_auto: bool,
2972 provenance: UserInputProvenance,
2973 snapshot: TurnMetadataSnapshot<'_>,
2974 ) -> Message {
2975 let turn_metadata = self.turn_metadata_block_from_snapshot(
2976 routed_model,
2977 auto_model,
2978 reasoning_effort,
2979 reasoning_effort_auto,
2980 provenance,
2981 &text,
2982 snapshot,
2983 );
2984 let mut content = self.user_content_blocks(text);
2985 content.push(turn_metadata);
2986 Message {
2987 role: "user".to_string(),
2988 content,
2989 }
2990 }
2991
2992 fn user_text_message_with_turn_metadata(&self, text: String) -> Message {
2993 self.user_text_message_with_turn_metadata_for_route(
2994 text,
2995 &self.session.model,
2996 self.session.auto_model,
2997 self.session.reasoning_effort.as_deref(),
2998 self.session.reasoning_effort_auto,
2999 )
3000 }
3001
3002 fn user_text_message_with_turn_metadata_for_route(
3003 &self,
3004 text: String,
3005 routed_model: &str,
3006 auto_model: bool,
3007 reasoning_effort: Option<&str>,
3008 reasoning_effort_auto: bool,
3009 ) -> Message {
3010 self.user_text_message_with_turn_metadata_for_route_and_provenance(
3011 text,
3012 routed_model,
3013 auto_model,
3014 reasoning_effort,
3015 reasoning_effort_auto,
3016 UserInputProvenance::ExternalUser,
3017 )
3018 }
3019
3020 fn runtime_text_message_with_turn_metadata(
3021 &self,
3022 text: String,
3023 provenance: UserInputProvenance,
3024 ) -> Message {
3025 self.user_text_message_with_turn_metadata_for_route_and_provenance(
3026 text,
3027 &self.session.model,
3028 self.session.auto_model,
3029 self.session.reasoning_effort.as_deref(),
3030 self.session.reasoning_effort_auto,
3031 provenance,
3032 )
3033 }
3034
3035 fn user_text_message_with_turn_metadata_for_route_and_provenance(
3036 &self,
3037 text: String,
3038 routed_model: &str,
3039 auto_model: bool,
3040 reasoning_effort: Option<&str>,
3041 reasoning_effort_auto: bool,
3042 provenance: UserInputProvenance,
3043 ) -> Message {
3044 // Place the user text first and turn_meta last so that the leading
3045 // bytes of each user message stay stable across date / model-route /
3046 // working-set changes. DeepSeek's KV prefix cache matches byte
3047 // sequences from the start of each message; when turn_meta (which
3048 // contains the current date) sits at position 0 the entire user
3049 // message prefix is invalidated at every date boundary. Moving it
3050 // to the tail preserves the user-input prefix and limits cache
3051 // invalidation to the trailing metadata block.
3052 let turn_metadata = self.turn_metadata_block(
3053 routed_model,
3054 auto_model,
3055 reasoning_effort,
3056 reasoning_effort_auto,
3057 provenance,
3058 &text,
3059 self.last_policy_narrowing.as_ref(),
3060 );
3061 let mut content = self.user_content_blocks(text);
3062 content.push(turn_metadata);
3063 Message {
3064 role: "user".to_string(),
3065 content,
3066 }
3067 }
3068
3069 async fn handle_idle_subagent_completion(&mut self, first: SubAgentCompletion) {
3070 let mut completions = Vec::new();
3071 if let Some(completion) =
3072 claim_subagent_completion(&mut self.delivered_subagent_completion_ids, first)
3073 {
3074 completions.push(completion);
3075 }
3076 while let Ok(completion) = self.rx_subagent_completion.try_recv() {
3077 if let Some(completion) =
3078 claim_subagent_completion(&mut self.delivered_subagent_completion_ids, completion)
3079 {
3080 completions.push(completion);
3081 }
3082 }
3083
3084 if completions.is_empty() {
3085 return;
3086 }
3087
3088 let claimed_ids = completions
3089 .iter()
3090 .map(|completion| completion.agent_id.clone())
3091 .collect::<Vec<_>>();
3092 let route = match self.current_runtime_route() {
3093 Ok(route) => route,
3094 Err(err) => {
3095 for agent_id in claimed_ids {
3096 self.delivered_subagent_completion_ids.remove(&agent_id);
3097 }
3098 let _ = self
3099 .tx_event
3100 .send(Event::error(ErrorEnvelope::fatal_auth(format!(
3101 "Cannot resume the turn because its provider route is no longer valid: {err}"
3102 ))))
3103 .await;
3104 let outcome = SendMessageOutcome::NotStarted {
3105 error: Some(format!("provider route is no longer valid: {err}")),
3106 };
3107 self.reconcile_non_completed_goal_turn(&outcome).await;
3108 return;
3109 }
3110 };
3111
3112 let count = completions.len();
3113 let content = completions
3114 .iter()
3115 .map(|completion| {
3116 if completion.is_high_priority_failure() {
3117 crate::runtime_handoff::subagent_failure_runtime_text(&completion.payload)
3118 } else {
3119 crate::runtime_handoff::subagent_completion_runtime_text(&completion.payload)
3120 }
3121 })
3122 .collect::<Vec<_>>()
3123 .join("\n\n");
3124
3125 let failed = completions
3126 .iter()
3127 .filter(|completion| completion.is_high_priority_failure())
3128 .count();
3129 let failure_suffix = if failed == 0 {
3130 String::new()
3131 } else {
3132 format!(" ({failed} failed)")
3133 };
3134
3135 let _ = self
3136 .tx_event
3137 .send(Event::status(format!(
3138 "Resuming turn with {count} idle sub-agent completion(s){failure_suffix}"
3139 )))
3140 .await;
3141
3142 let outcome = self
3143 .handle_send_message(
3144 content,
3145 self.current_mode,
3146 route,
3147 self.config.compaction.clone(),
3148 self.config.goal_objective.clone(),
3149 self.config.goal_token_budget,
3150 self.config.goal_status,
3151 self.session.reasoning_effort.clone(),
3152 self.session.reasoning_effort_auto,
3153 self.session.auto_model,
3154 self.session.allow_shell,
3155 self.session.trust_mode,
3156 self.session.auto_approve,
3157 self.session.approval_mode,
3158 self.config.translation_enabled,
3159 self.config.allowed_tools.clone(),
3160 Vec::new(),
3161 self.config.hook_executor.clone(),
3162 self.config.verbosity.clone(),
3163 UserInputProvenance::SubAgentHandoff,
3164 )
3165 .await;
3166 if !outcome.started() {
3167 for agent_id in claimed_ids {
3168 self.delivered_subagent_completion_ids.remove(&agent_id);
3169 }
3170 }
3171 }
3172
3173 /// Handle a send message operation
3174 #[allow(clippy::too_many_arguments)]
3175 /// After a turn completes, decide whether an active goal should keep going.
3176 /// Returns a continuation to dispatch, an explicit terminal backstop stop,
3177 /// or Inactive when no follow-up turn belongs in the queue.
3178 ///
3179 /// A goal runs until the model self-reports done/blocked or the user pauses
3180 /// or clears. Token/time accounting remains telemetry. The loop is "until
3181 /// done," not "until N turns" (#5052); a configurable safety
3182 /// backstop (`[goal] max_continuations`, `0` = unlimited) still halts a
3183 /// pathological loop that never emits a terminal signal.
3184 fn goal_continuation_if_active(&self) -> GoalContinuationAction {
3185 let mut state = match self.config.goal_state.lock() {
3186 Ok(state) => state,
3187 Err(err) => {
3188 tracing::warn!("goal state lock poisoned during continuation check: {err}");
3189 return GoalContinuationAction::Inactive;
3190 }
3191 };
3192 let snapshot = state.snapshot();
3193 if !snapshot.is_active() {
3194 return GoalContinuationAction::Inactive;
3195 }
3196
3197 // The snapshot status is a string ("active", "paused", "complete",
3198 // "blocked"). Map it to the goal-loop decision core's status enum.
3199 let status = match snapshot.status.as_str() {
3200 "active" => crate::goal_loop::GoalRunStatus::Active,
3201 "complete" => crate::goal_loop::GoalRunStatus::Completed,
3202 // Paused / Blocked / unknown → no continuation.
3203 _ => return GoalContinuationAction::Inactive,
3204 };
3205
3206 let decision = crate::goal_loop::decide_continuation(
3207 status,
3208 crate::goal_loop::GoalProgress {
3209 tokens_used: snapshot.tokens_used,
3210 time_used_seconds: snapshot.time_used_seconds,
3211 continuations: snapshot.continuation_count,
3212 },
3213 // Unbounded like grokbuild (agent-call cap) and kimicode swarm
3214 // (turnBudget per-task, resumable): token/time are telemetry only,
3215 // only Completed/Blocked/ContinuationLimit pause the loop.
3216 crate::goal_loop::GoalBudget::unbounded()
3217 .with_max_continuations(self.config.goal_max_continuations),
3218 );
3219
3220 match decision {
3221 crate::goal_loop::ContinuationDecision::Continue => {
3222 // A cross-turn dispatch is a real continuation pass just like
3223 // the bounded intra-turn retry in `turn_loop`. Record it before
3224 // rendering and carrying the snapshot so the durable prompt,
3225 // telemetry, and next host sync all agree on the pass number.
3226 state.record_continuation();
3227 let snapshot = state.snapshot();
3228 GoalContinuationAction::Dispatch {
3229 content: crate::tools::goal::render_continuation_prompt(
3230 &snapshot,
3231 snapshot.continuation_count,
3232 ),
3233 snapshot: Box::new(snapshot),
3234 }
3235 }
3236 crate::goal_loop::ContinuationDecision::Stop(reason) => {
3237 tracing::info!(?reason, "goal continuation stopped");
3238 let (message, pause_reason) = match reason {
3239 crate::goal_loop::StopReason::ContinuationLimit => (
3240 format!(
3241 "Goal paused after {} automatic continuations without a terminal result (safety backstop; raise or disable via [goal] max_continuations); inspect progress, then resume if useful.",
3242 self.config.goal_max_continuations,
3243 ),
3244 GoalPauseReason::Backoff,
3245 ),
3246 crate::goal_loop::StopReason::Completed
3247 | crate::goal_loop::StopReason::Blocked => {
3248 return GoalContinuationAction::Inactive;
3249 }
3250 };
3251 GoalContinuationAction::Stopped {
3252 message,
3253 reason: pause_reason,
3254 }
3255 }
3256 }
3257 }
3258
3259 /// Reconcile a turn that did not complete with the autonomous goal loop.
3260 /// Hosted engines leave lifecycle decisions to their durable host. The
3261 /// interactive engine must cancel any older queued synthetic token first,
3262 /// then project an active goal into a truthful non-running state.
3263 async fn reconcile_non_completed_goal_turn(&mut self, outcome: &SendMessageOutcome) {
3264 if self.host_managed_turns() {
3265 return;
3266 }
3267
3268 self.cancel_scheduled_goal_continuation();
3269 match outcome {
3270 SendMessageOutcome::NotStarted { error } => {
3271 let message = self.goal_turn_not_started_message(error.as_deref());
3272 self.block_goal_continuation(message).await;
3273 }
3274 SendMessageOutcome::Finished {
3275 status: TurnOutcomeStatus::Failed,
3276 error,
3277 } => {
3278 let message = self.goal_continuation_failure_message(error.as_deref());
3279 self.block_goal_continuation(message).await;
3280 }
3281 SendMessageOutcome::Finished {
3282 status: TurnOutcomeStatus::Interrupted,
3283 ..
3284 } => {
3285 // Goals are durable session objectives. An interrupted model
3286 // turn (Esc, steer, compaction, cancel) must cancel only the
3287 // auto-continuation timer — already done above — and leave the
3288 // goal Active. pause_reason=User is reserved for explicit
3289 // `/goal pause`. Requiring `/goal resume` after every interrupt
3290 // was a dogfood lie (2026-07-24).
3291 let _ = self
3292 .tx_event
3293 .send(Event::status(
3294 "Turn interrupted; session goal stays active.".to_string(),
3295 ))
3296 .await;
3297 }
3298 SendMessageOutcome::Finished {
3299 status: TurnOutcomeStatus::Completed,
3300 ..
3301 } => {}
3302 }
3303 }
3304
3305 /// A route/client rejection can happen before normal turn setup copies the
3306 /// host's just-declared goal into SharedGoalState. Seed only that goal
3307 /// descriptor so the rejection can publish a truthful Blocked snapshot;
3308 /// no user message or provider turn state is mutated here.
3309 fn sync_unstarted_goal_for_terminal_projection(
3310 &mut self,
3311 objective: Option<&str>,
3312 token_budget: Option<u32>,
3313 status: GoalStatus,
3314 ) {
3315 let objective = normalized_goal_objective(objective);
3316 if objective.is_none() || status != GoalStatus::Active {
3317 return;
3318 }
3319 sync_goal_state_from_host(
3320 &self.config.goal_state,
3321 objective.as_deref(),
3322 token_budget,
3323 status,
3324 );
3325 self.config.goal_objective = objective;
3326 self.config.goal_token_budget = token_budget;
3327 self.config.goal_status = status;
3328 }
3329
3330 /// Transition a still-active interactive goal to Blocked and publish every
3331 /// host projection in one ordered path. Continuation failures happen
3332 /// outside a model tool call, so without this bridge the loop can stop while
3333 /// the prompt and sidebar continue to claim the goal is actively running.
3334 async fn block_goal_continuation(&mut self, message: String) {
3335 let snapshot = match self.config.goal_state.lock() {
3336 Ok(mut state) => {
3337 if state.is_active()
3338 && let Err(err) = state.mark_blocked(message.clone())
3339 {
3340 tracing::warn!("failed to mark goal continuation blocked: {err}");
3341 return;
3342 }
3343 let snapshot = state.snapshot();
3344 if snapshot.status != GoalStatus::Blocked.as_str() {
3345 tracing::warn!(
3346 status = %snapshot.status,
3347 "goal changed before continuation blocker could be published"
3348 );
3349 return;
3350 }
3351 snapshot
3352 }
3353 Err(err) => {
3354 tracing::warn!("goal state lock poisoned while blocking continuation: {err}");
3355 return;
3356 }
3357 };
3358
3359 self.config.goal_objective.clone_from(&snapshot.objective);
3360 self.config.goal_token_budget = snapshot.token_budget;
3361 self.config.goal_status = GoalStatus::Blocked;
3362 self.refresh_system_prompt();
3363 self.emit_session_updated().await;
3364 let _ = self.tx_event.send(Event::GoalUpdated { snapshot }).await;
3365 let _ = self.tx_event.send(Event::status(message)).await;
3366 }
3367
3368 /// Pause a still-active goal with an inspectable reason and publish every
3369 /// host projection in one ordered path.
3370 async fn pause_goal_continuation(&mut self, reason: GoalPauseReason, message: String) {
3371 let snapshot = match self.config.goal_state.lock() {
3372 Ok(mut state) => {
3373 if !state.is_active() {
3374 return;
3375 }
3376 if let Err(err) = state.mark_paused(reason) {
3377 tracing::warn!("failed to pause goal continuation: {err}");
3378 return;
3379 }
3380 state.snapshot()
3381 }
3382 Err(err) => {
3383 tracing::warn!("goal state lock poisoned while pausing interruption: {err}");
3384 return;
3385 }
3386 };
3387
3388 self.config.goal_objective.clone_from(&snapshot.objective);
3389 self.config.goal_token_budget = snapshot.token_budget;
3390 self.config.goal_status = GoalStatus::Paused;
3391 self.refresh_system_prompt();
3392 self.emit_session_updated().await;
3393 let _ = self.tx_event.send(Event::GoalUpdated { snapshot }).await;
3394 let _ = self.tx_event.send(Event::status(message)).await;
3395 }
3396
3397 /// Handle `/goal pause|resume|clear|complete|blocked` by writing the new
3398 /// status to `SharedGoalState` so the cross-turn continuation loop respects
3399 /// it. This does NOT dispatch a model turn — it's a control-plane update.
3400 async fn handle_set_goal_status(&mut self, status: GoalStatus, clear: bool) {
3401 let snapshot = match self.config.goal_state.lock() {
3402 Ok(mut state) => {
3403 if clear {
3404 // `/goal clear` — wipe the objective entirely.
3405 state.sync_from_host_status(None, None, GoalStatus::Active);
3406 } else {
3407 // Update only the status; keep the objective and budget.
3408 // `sync_from_host_status` resets usage when the objective
3409 // changes, but here we pass the existing objective so usage
3410 // is preserved (pause/resume shouldn't reset the counter).
3411 let objective = state.objective().map(str::to_string);
3412 let budget = state.token_budget();
3413 state.sync_from_host_status(objective.as_deref(), budget, status);
3414 }
3415 state.snapshot()
3416 }
3417 Err(err) => {
3418 tracing::warn!("goal state lock poisoned during SetGoalStatus: {err}");
3419 return;
3420 }
3421 };
3422
3423 // Keep every host-side projection aligned with the authoritative
3424 // SharedGoalState. In particular, a cleared state must also clear the
3425 // configured fallback used by `goal_objective_for_prompt`; otherwise a
3426 // prompt refresh would silently restore the old <session_goal> block.
3427 self.config.goal_objective.clone_from(&snapshot.objective);
3428 self.config.goal_token_budget = snapshot.token_budget;
3429 self.config.goal_status = if snapshot.objective.is_some() {
3430 status
3431 } else {
3432 GoalStatus::Active
3433 };
3434 self.refresh_system_prompt();
3435 self.emit_session_updated().await;
3436 // Unlike routine end-of-turn updates, an explicit clear must publish
3437 // the canonical empty snapshot. Keeping this scoped to the control op
3438 // avoids an unrelated no-goal turn racing with a newly declared goal in
3439 // the UI while still letting the clear win over a preceding active
3440 // TurnComplete snapshot.
3441 let _ = self.tx_event.send(Event::GoalUpdated { snapshot }).await;
3442
3443 let label = if clear {
3444 "cleared"
3445 } else {
3446 match status {
3447 GoalStatus::Active => "resumed",
3448 GoalStatus::Paused => "paused",
3449 GoalStatus::Complete => "complete",
3450 GoalStatus::Blocked => "blocked",
3451 }
3452 };
3453 let _ = self
3454 .tx_event
3455 .send(Event::status(format!("Goal {label}.")))
3456 .await;
3457 }
3458
3459 /// Build the turn's tool registry and the model-facing tool catalog.
3460 ///
3461 /// This is the single authority for "what tools would the next request
3462 /// carry". `handle_send_message` calls it with [`SubAgentWiring::Live`]
3463 /// and [`McpAccess::Connect`]; `/preview-request` calls it with
3464 /// [`SubAgentWiring::Inert`] and [`McpAccess::PassiveSnapshot`], which
3465 /// together remove every side effect of the build — no fork snapshot, no
3466 /// spawned mailbox drainer, no pool creation, no `connect_all`, no status
3467 /// events — while producing a byte-identical catalog for the state that
3468 /// is already live.
3469 ///
3470 /// The session's `last_tool_catalog` is never an acceptable substitute:
3471 /// it is one turn stale and stores the pre-activation catalog rather than
3472 /// the active subset the provider would actually receive.
3473 ///
3474 /// `allowed_tools` is the command-scoped allow-list gate the catalog is
3475 /// filtered under. It is an explicit **parameter**, not a read of
3476 /// `self.config.allowed_tools`, because the preview's gate belongs to a
3477 /// turn that has not been installed: writing it onto the engine and
3478 /// restoring it afterwards would leave the wrong gate installed across
3479 /// every `.await` in this function, and would leave it installed
3480 /// permanently if the task were cancelled or panicked between the two
3481 /// writes.
3482 #[allow(clippy::too_many_arguments)]
3483 async fn build_turn_tool_registry_and_catalog(
3484 &mut self,
3485 input_policy: &TurnAuthority,
3486 dynamic_tools: &[DynamicToolSpec],
3487 allowed_tools: Option<Vec<String>>,
3488 wiring: SubAgentWiring,
3489 mcp_access: McpAccess,
3490 route: TurnRouteContext,
3491 turn_id: &str,
3492 ) -> TurnToolBuild {
3493 // Build tool registry and tool list for the current mode
3494 let todo_list = self.config.todos.clone();
3495 let plan_state = self.config.plan_state.clone();
3496
3497 let tool_context = self.build_tool_context_for_turn(input_policy, &route);
3498 // Ensure MCP pool is initialized before building the tool registry,
3499 // so start_mcp_server can be registered when Feature::Mcp is enabled.
3500 // A passive snapshot must not create the pool: allocating it is engine
3501 // state a preview has no business writing.
3502 if self.config.features.enabled(Feature::Mcp) && mcp_access.may_connect() {
3503 let _ = self.ensure_mcp_pool().await;
3504 }
3505 let builder = self
3506 .build_turn_tool_registry_builder_for_route(
3507 input_policy.mode,
3508 input_policy.allow_shell,
3509 route.client.clone(),
3510 &route.model,
3511 todo_list,
3512 plan_state,
3513 )
3514 .with_dynamic_tools(dynamic_tools);
3515
3516 let subagents_available =
3517 self.config.subagents_enabled && self.config.features.enabled(Feature::Subagents);
3518
3519 let fork_context_for_runtime = if subagents_available && wiring.is_live() {
3520 let state = StructuredState::capture(
3521 input_policy.mode.label(),
3522 self.config.workspace.clone(),
3523 std::env::current_dir().ok(),
3524 &self.session.working_set,
3525 Some(&self.subagent_manager),
3526 )
3527 .await;
3528 Some(SubAgentForkContext {
3529 messages: self.messages_with_turn_metadata(),
3530 structured_state_block: state.to_system_block(),
3531 // Resolve at spawn time so a work update earlier in this turn
3532 // reaches the child rather than freezing turn-start state.
3533 work_source: Some(self.work_state_source()),
3534 })
3535 } else {
3536 None
3537 };
3538
3539 // Mailbox for structured sub-agent envelopes (#128/#130). One per
3540 // turn: the receiver is drained by a short-lived task that converts
3541 // envelopes into `Event::SubAgentMailbox` so the UI can route them
3542 // to the matching in-transcript card. The drainer exits naturally
3543 // when every cloned sender is dropped at turn-end.
3544 let mailbox_for_runtime = if subagents_available && wiring.is_live() {
3545 let cancel_token = self.cancel_token.child_token();
3546 let (mailbox, mut receiver) = Mailbox::new(cancel_token.clone());
3547 let tx_event_clone = self.tx_event.clone();
3548 let mailbox_turn_id = turn_id.to_string();
3549 let (flush_tx, mut flush_rx) = tokio::sync::oneshot::channel();
3550 let drain_handle = spawn_supervised(
3551 "subagent-mailbox-drainer",
3552 std::panic::Location::caller(),
3553 async move {
3554 let mut best_effort_sent_at: HashMap<String, Instant> = HashMap::new();
3555 'drain: loop {
3556 tokio::select! {
3557 biased;
3558 _ = &mut flush_rx => {
3559 for envelope in receiver.drain_available() {
3560 if !forward_subagent_mailbox_message(
3561 &tx_event_clone,
3562 &mailbox_turn_id,
3563 envelope.seq,
3564 envelope.message,
3565 &mut best_effort_sent_at,
3566 ).await {
3567 break 'drain;
3568 }
3569 }
3570 break;
3571 }
3572 envelope = receiver.recv() => {
3573 let Some(envelope) = envelope else { break };
3574 if !forward_subagent_mailbox_message(
3575 &tx_event_clone,
3576 &mailbox_turn_id,
3577 envelope.seq,
3578 envelope.message,
3579 &mut best_effort_sent_at,
3580 ).await {
3581 break;
3582 }
3583 }
3584 }
3585 }
3586 },
3587 );
3588 Some(TurnMailboxBarrier {
3589 mailbox,
3590 cancel_token,
3591 flush_tx,
3592 drain_handle,
3593 })
3594 } else {
3595 None
3596 };
3597
3598 let mcp_pool = if self.config.features.enabled(Feature::Mcp) {
3599 if mcp_access.may_connect() {
3600 self.ensure_mcp_pool().await.ok()
3601 } else {
3602 self.mcp_pool.clone()
3603 }
3604 } else {
3605 None
3606 };
3607
3608 let mut subagent_runtime_model = None;
3609 let mut tool_registry = if subagents_available {
3610 let runtime = if let Some(client) = route.client.clone() {
3611 let runtime_allow_shell =
3612 input_policy.allow_shell && !matches!(input_policy.mode, AppMode::Plan);
3613 let runtime_shell_policy =
3614 shell_policy_for_mode(input_policy.mode, runtime_allow_shell);
3615 subagent_runtime_model = Some(route.model.clone());
3616 let mut rt = SubAgentRuntime::new(
3617 client,
3618 route.model.clone(),
3619 tool_context.clone(),
3620 runtime_allow_shell,
3621 Some(self.tx_event.clone()),
3622 Arc::clone(&self.subagent_manager),
3623 )
3624 .with_locale_tag(route.locale_tag.clone())
3625 .with_role_models(route.role_models.clone())
3626 .with_api_config((*route.api_config).clone())
3627 .with_fleet_roster(route.fleet_roster.clone())
3628 .with_auto_model(route.auto_model)
3629 .with_reasoning_effort(route.reasoning_effort.clone(), route.reasoning_effort_auto)
3630 .with_agent_tool_surface_options(
3631 self.agent_tool_surface_options(runtime_shell_policy),
3632 )
3633 .with_max_spawn_depth(self.config.max_spawn_depth)
3634 .with_step_api_timeout(self.config.subagent_api_timeout)
3635 .with_speech_output_dir(self.config.speech_output_dir.clone())
3636 .with_mcp_pool(mcp_pool.clone())
3637 .with_todos(self.config.todos.clone())
3638 .with_parent_completion_tx(self.tx_subagent_completion.clone())
3639 .with_runtime_cost_owner(self.config.compaction.runtime_cost_owner.as_deref())
3640 .with_parent_mode(input_policy.mode);
3641 if matches!(input_policy.mode, AppMode::Plan) {
3642 rt.worker_profile = WorkerRuntimeProfile::for_role(FleetRole::Planner);
3643 }
3644 // #4042: stamp the session's --disallowed-tools onto the parent
3645 // runtime so every model-spawned sub-agent inherits the deny-list
3646 // (plan-mode role override above is intentionally before this).
3647 rt.worker_profile.denied_tools =
3648 self.config.disallowed_tools.clone().unwrap_or_default();
3649 if let Some(context) = fork_context_for_runtime.clone() {
3650 rt = rt.with_fork_context(context);
3651 }
3652 if let Some(barrier) = mailbox_for_runtime.as_ref() {
3653 rt = rt
3654 .with_mailbox(barrier.mailbox.clone())
3655 .with_cancel_token(barrier.cancel_token.clone());
3656 }
3657 Some(rt)
3658 } else {
3659 None
3660 };
3661 if let Some(subagent_runtime) = runtime {
3662 builder
3663 .with_subagent_tools(self.subagent_manager.clone(), subagent_runtime)
3664 .build(tool_context)
3665 } else {
3666 tracing::warn!(
3667 "Sub-agents enabled but no API client available, falling back to basic tool set"
3668 );
3669 builder.build(tool_context)
3670 }
3671 } else {
3672 builder.build(tool_context)
3673 };
3674
3675 // Load plugin tools from the user's tools directory and apply any
3676 // config.toml overrides. Explicit overrides win over auto-discovered
3677 // scripts with the same tool name.
3678 let plugin_tool_names =
3679 configure_plugin_tools(&mut tool_registry, self.config.tools.as_ref());
3680
3681 let mcp_state = if self.config.features.enabled(Feature::Mcp) {
3682 if mcp_access.may_connect() {
3683 let tools = self.mcp_tools().await;
3684 let server_count = match self.mcp_pool.as_ref() {
3685 Some(pool) => pool.lock().await.connected_servers().len(),
3686 None => 0,
3687 };
3688 McpToolState::Live {
3689 tools,
3690 server_count,
3691 }
3692 } else {
3693 self.passive_mcp_snapshot().await
3694 }
3695 } else {
3696 McpToolState::Disabled
3697 };
3698 // Captured before the catalog closure consumes the tool list, so a
3699 // caller can attribute MCP contributions without a second connect.
3700 let mcp_tools = mcp_state.tools().to_vec();
3701 let mcp_tool_names: Vec<String> = mcp_tools.iter().map(|tool| tool.name.clone()).collect();
3702 // The surface budget belongs to the route the request would go to,
3703 // which is not necessarily the installed one under auto routing.
3704 let capability = route.capability_profile();
3705 let mut always_load = self.config.tools_always_load.clone();
3706 if self.config.features.enabled(Feature::Mcp) {
3707 always_load.insert("start_mcp_server".to_string());
3708 always_load.insert("registry_sync".to_string());
3709 always_load.insert("start_registry_mcp_server".to_string());
3710 }
3711 let bypass = input_policy.auto_approve
3712 || input_policy.approval_mode == crate::tui::approval::ApprovalMode::Bypass;
3713 let catalog_mode = if bypass {
3714 AppMode::Yolo
3715 } else {
3716 input_policy.mode
3717 };
3718 let mut catalog = build_model_tool_catalog_with_surface(
3719 tool_registry.to_api_tools_with_cache(true),
3720 mcp_tools,
3721 catalog_mode,
3722 &always_load,
3723 capability.tool_surface_budget,
3724 );
3725 if self.config.features.enabled(Feature::Mcp) {
3726 apply_registry_first_shell_guidance(&mut catalog);
3727 }
3728 for tool in &mut catalog {
3729 if plugin_tool_names.contains(&tool.name) {
3730 tool.defer_loading = Some(false);
3731 }
3732 }
3733 let surface = ToolSurfacePolicy::new(
3734 tool_registry,
3735 Some(catalog),
3736 input_policy.mode,
3737 &always_load,
3738 &input_policy.dynamic_active_tools,
3739 self.config.strict_tool_mode,
3740 allowed_tools,
3741 self.config.disallowed_tools.clone(),
3742 self.config.max_tool_calls,
3743 input_policy.approval_mode_for_session(),
3744 );
3745 TurnToolBuild {
3746 surface,
3747 mcp_tool_names,
3748 mcp: mcp_state,
3749 subagent_runtime_model,
3750 mailbox: mailbox_for_runtime,
3751 plugin_tool_names,
3752 }
3753 }
3754
3755 /// Read-only MCP snapshot for `/preview-request` (#1004).
3756 ///
3757 /// Never creates the pool, never calls `connect_all`, never reloads a
3758 /// config source, never starts a server, and never emits a status event.
3759 /// It answers exactly one question: *is the tool set the next turn would
3760 /// send already known?* It is known only when the pool exists, every
3761 /// enabled server is connected, and no config source has changed since
3762 /// the pool last read them. Otherwise the honest answer is "unavailable",
3763 /// because a real turn would connect and discover more tools.
3764 async fn passive_mcp_snapshot(&self) -> McpToolState {
3765 let Some(pool) = self.mcp_pool.as_ref() else {
3766 return McpToolState::Unavailable {
3767 reason: McpUnavailable::PoolNotStarted,
3768 };
3769 };
3770 let pool = pool.lock().await;
3771 if !pool.config_sources_unchanged() {
3772 return McpToolState::Unavailable {
3773 reason: McpUnavailable::ConfigChangedSinceConnect,
3774 };
3775 }
3776 let connected: Vec<&str> = pool.connected_servers();
3777 let pending = pool
3778 .enabled_server_names()
3779 .into_iter()
3780 .filter(|name| !connected.iter().any(|connected| *connected == name))
3781 .count();
3782 if pending > 0 {
3783 return McpToolState::Unavailable {
3784 reason: McpUnavailable::ServersNotConnected { pending },
3785 };
3786 }
3787 McpToolState::Live {
3788 tools: pool.to_api_tools(),
3789 server_count: connected.len(),
3790 }
3791 }
3792
3793 #[allow(clippy::too_many_arguments)]
3794 async fn handle_send_message(
3795 &mut self,
3796 content: String,
3797 mode: AppMode,
3798 route: ResolvedRuntimeRoute,
3799 compaction: CompactionConfig,
3800 goal_objective: Option<String>,
3801 goal_token_budget: Option<u32>,
3802 goal_status: GoalStatus,
3803 reasoning_effort: Option<String>,
3804 reasoning_effort_auto: bool,
3805 auto_model: bool,
3806 allow_shell: bool,
3807 trust_mode: bool,
3808 auto_approve: bool,
3809 approval_mode: crate::tui::approval::ApprovalMode,
3810 translation_enabled: bool,
3811 allowed_tools: Option<Vec<String>>,
3812 dynamic_tools: Vec<DynamicToolSpec>,
3813 hook_executor: Option<std::sync::Arc<crate::hooks::HookExecutor>>,
3814 verbosity: Option<String>,
3815 provenance: UserInputProvenance,
3816 ) -> SendMessageOutcome {
3817 let effective_provider = route.identity.provider;
3818 let provider_identity = route.identity.key.clone();
3819 let model = route.model.clone();
3820 let route_limits = crate::route_budget::known_route_limits(route.candidate.limits());
3821 let route_capabilities = route.candidate.capabilities();
3822 let route_api_config = route.config.clone();
3823 // Freeze the billing receipt here, while `route` is still the single
3824 // authority for this turn: `route.config` is the identity-scoped
3825 // Config the client is being built from, and `route.candidate` names
3826 // the endpoint it will call. After `install_resolved_runtime_route`
3827 // consumes `route`, the only sound source for these facts is this
3828 // receipt — an ambient `Config` read at TurnStarted or TurnComplete
3829 // would follow a later provider switch, auto-router hop, or custom
3830 // table change onto the wrong vendor.
3831 let dispatched_base_url = route.candidate.endpoint().base_url.clone();
3832 let dispatched_product =
3833 crate::route_billing::capture_product(&route.config, effective_provider);
3834 if let Err(err) = self.install_resolved_runtime_route(route) {
3835 let _ = self
3836 .tx_event
3837 .send(Event::error(ErrorEnvelope::fatal_auth(format!(
3838 "Cannot start the turn because its provider route is not ready: {err}"
3839 ))))
3840 .await;
3841 self.sync_unstarted_goal_for_terminal_projection(
3842 goal_objective.as_deref(),
3843 goal_token_budget,
3844 goal_status,
3845 );
3846 let outcome = SendMessageOutcome::NotStarted { error: Some(err) };
3847 self.reconcile_non_completed_goal_turn(&outcome).await;
3848 return outcome;
3849 }
3850
3851 // Deliver completions that arrived after the previous turn before the
3852 // next user request is sent. This keeps background shell work
3853 // model-visible without requiring an explicit wait/poll tool call.
3854 let shell_completions = self.drain_shell_completion_events();
3855 if !shell_completions.is_empty() {
3856 self.add_session_message(crate::runtime_handoff::shell_completion_runtime_message(
3857 &shell_completions,
3858 ))
3859 .await;
3860 if let Some(status) =
3861 crate::core::engine::turn_loop::shell_completion_status_text(&shell_completions, "")
3862 {
3863 let _ = self.tx_event.send(Event::status(status)).await;
3864 }
3865 }
3866
3867 let input_policy = effective_input_policy(
3868 provenance,
3869 mode,
3870 &content,
3871 allow_shell,
3872 trust_mode,
3873 mode == AppMode::Yolo || auto_approve,
3874 approval_mode,
3875 );
3876 let prompt_context = NextTurnPromptContext::for_planned_turn(
3877 effective_provider,
3878 model.clone(),
3879 route_limits,
3880 input_policy.mode,
3881 goal_objective.clone(),
3882 goal_status,
3883 goal_token_budget,
3884 translation_enabled,
3885 verbosity.clone(),
3886 );
3887 // #3947: an effective-mode change is never silent. The structured
3888 // event is recorded first (so doctor and this turn's metadata can read
3889 // it), then rendered to the UI from that same value.
3890 self.last_policy_narrowing = input_policy.narrowing.clone();
3891 if let Some(status) = input_policy.status() {
3892 let _ = self.tx_event.send(Event::status(status)).await;
3893 }
3894 // Reset cancel token for fresh turn (in case previous was cancelled)
3895 self.reset_cancel_token();
3896
3897 // Track the complete effective mode policy so mid-turn metadata, `/edit`,
3898 // idle worker resumptions, and approval gates cannot read a stale policy
3899 // after the UI changed modes (#3568).
3900 self.apply_runtime_mode_policy(&input_policy);
3901
3902 // Drain stale steer messages from previous turns.
3903 while self.rx_steer.try_recv().is_ok() {}
3904
3905 // Create turn context first so start event includes a stable turn id.
3906 let mut turn = TurnContext::new(self.config.max_steps);
3907 self.turn_counter = self.turn_counter.saturating_add(1);
3908 let turn_started_at = chrono::Utc::now();
3909 // Mint the route receipt from the client that `install_resolved_runtime_route`
3910 // actually installed above — the same client `Event::TurnComplete`
3911 // reports `base_url` from. Hosts must not re-derive this from config
3912 // when they process `TurnStarted`: by then config may already describe
3913 // a different endpoint or credential.
3914 let route_receipt = if self.model_client_injected {
3915 // Provider-neutral injected clients are the I/O authority, while
3916 // `deepseek_client` is only an auxiliary route-shaping client.
3917 // It cannot truthfully receipt a transport it did not perform.
3918 None
3919 } else {
3920 self.deepseek_client
3921 .as_ref()
3922 .map(|client| client.turn_route_receipt(&provider_identity))
3923 };
3924 let route_base_url = self
3925 .deepseek_client
3926 .as_ref()
3927 .map(|client| client.base_url());
3928 let turn_route = TurnRoute {
3929 provider: effective_provider,
3930 provider_identity,
3931 model: model.clone(),
3932 auto_model,
3933 receipt: route_receipt,
3934 // A start is not a dispatch. The billing envelope is attached
3935 // below, on the route held for the wire boundary only.
3936 billing: None,
3937 // The classification receipt, by contrast, is frozen here at the
3938 // client-freeze boundary and is readable from `TurnStarted` on.
3939 base_url: dispatched_base_url,
3940 billing_product: dispatched_product,
3941 };
3942 // Billing provenance follows the *route* that was installed for this
3943 // turn, which is authoritative even when a test or embedder injected the
3944 // transport: `deepseek_client`'s base URL is the resolved route's
3945 // endpoint either way. This is a weaker claim than `receipt`, which
3946 // digests the credential an injected client did not use and is therefore
3947 // withheld above.
3948 let dispatch_billing = crate::core::events::RouteBillingEnvelope {
3949 billing_surface: crate::route_billing::billing_surface_for_dispatch(
3950 Some(&self.api_config),
3951 effective_provider,
3952 route_base_url,
3953 )
3954 .map(str::to_string),
3955 endpoint_fingerprint: route_base_url.and_then(crate::cost_status::endpoint_fingerprint),
3956 // Classified from this turn's own frozen receipt, not from a
3957 // second ambient `for_route` read. Both halves of the route then
3958 // answer from the same captured endpoint + credential product, so
3959 // the envelope stamped on the wire and the receipt carried on
3960 // `TurnRoute` cannot disagree about how this turn bills.
3961 billing_mode: crate::route_billing::for_dispatched_receipt(
3962 crate::route_billing::DispatchedReceipt {
3963 provider: effective_provider,
3964 identity: Some(turn_route.provider_identity.as_str()),
3965 base_url: turn_route.base_url.as_str(),
3966 product: turn_route.billing_product,
3967 },
3968 )
3969 .into(),
3970 // Provisional. Replaced with the true wire-boundary instant
3971 // when `handle_deepseek_turn` emits `Event::RouteDispatched`.
3972 dispatched_at: turn_started_at,
3973 };
3974 turn.pending_route = Some(TurnRoute {
3975 billing: Some(dispatch_billing),
3976 ..turn_route.clone()
3977 });
3978
3979 // Emit turn started event IMMEDIATELY so the UI knows the turn is
3980 // active. The snapshot below can take 30+ seconds on slow filesystems
3981 // (e.g. WSL2 /mnt/c) and must not delay the TurnStarted event.
3982 let _ = self
3983 .tx_event
3984 .send(Event::TurnStarted {
3985 turn_id: turn.id.clone(),
3986 created_at: turn_started_at,
3987 route: Some(turn_route),
3988 })
3989 .await;
3990
3991 // Apply the host-resolved route budget before building the request.
3992 // The model, limits, and compaction policy arrive in one operation so
3993 // no provider request can observe a partially updated route.
3994 self.active_route_limits = route_limits;
3995 self.config.compaction = compaction;
3996
3997 // Snapshot the workspace BEFORE we touch a single tool. Run the git
3998 // work on the blocking pool so the async runtime stays responsive;
3999 // failure is non-fatal (the helper logs at WARN).
4000 if self.config.snapshots_enabled {
4001 // Clone the user prompt now — `content` is moved into
4002 // `user_text_message_with_turn_metadata_for_route` below, so we need
4003 // a copy for both pre- and post-turn snapshot labels. The
4004 // label carries a truncated first line so `/restore`
4005 // listings are human-readable.
4006 let snapshot_prompt = content.clone();
4007 let pre_workspace = self.session.workspace.clone();
4008 let pre_seq = self.turn_counter;
4009 let pre_cap = self.config.snapshots_max_workspace_bytes;
4010 let pre_sid = self.session.id.clone();
4011 let _ = tokio::task::spawn_blocking(move || {
4012 pre_turn_snapshot(
4013 &pre_workspace,
4014 pre_seq,
4015 pre_cap,
4016 Some(&snapshot_prompt),
4017 Some(&pre_sid),
4018 )
4019 })
4020 .await;
4021 }
4022
4023 // A new turn means any leftover retry banner (success cleared
4024 // it, failure pinned it) is no longer relevant — reset to idle
4025 // so the footer doesn't display a stale failure row across
4026 // turns (#499).
4027 crate::retry_status::clear();
4028
4029 // Clone user prompt for post-turn snapshot label before `content`
4030 // is moved into `user_text_message_with_turn_metadata_for_route` below.
4031 let snapshot_prompt_post = content.clone();
4032
4033 if self.model_client.is_none() {
4034 let message = self
4035 .deepseek_client_error
4036 .as_deref()
4037 .map(|err| format!("Failed to send message: {err}"))
4038 .unwrap_or_else(|| "Failed to send message: API client not configured".to_string());
4039 let _ = self
4040 .tx_event
4041 .send(Event::error(ErrorEnvelope::fatal_auth(message.clone())))
4042 .await;
4043 let _ = self
4044 .tx_event
4045 .send(Event::TurnComplete {
4046 usage: turn.usage.clone(),
4047 status: TurnOutcomeStatus::Failed,
4048 error: Some(message.clone()),
4049 tool_catalog: None,
4050 base_url: None,
4051 })
4052 .await;
4053 self.sync_unstarted_goal_for_terminal_projection(
4054 goal_objective.as_deref(),
4055 goal_token_budget,
4056 goal_status,
4057 );
4058 let outcome = SendMessageOutcome::NotStarted {
4059 error: Some(message),
4060 };
4061 self.reconcile_non_completed_goal_turn(&outcome).await;
4062 return outcome;
4063 }
4064
4065 let previous_goal_objective = self.config.goal_objective.clone();
4066 let previous_goal_token_budget = self.config.goal_token_budget;
4067 let previous_goal_status = self.config.goal_status;
4068
4069 self.session.model = model.clone();
4070 self.config.model.clone_from(&self.session.model);
4071 self.config.goal_objective = goal_objective.clone();
4072 self.config.goal_token_budget = goal_token_budget;
4073 self.config.goal_status = goal_status;
4074 if normalized_goal_objective(previous_goal_objective.as_deref())
4075 != normalized_goal_objective(goal_objective.as_deref())
4076 || previous_goal_token_budget != goal_token_budget
4077 || previous_goal_status != goal_status
4078 {
4079 sync_goal_state_from_host(
4080 &self.config.goal_state,
4081 normalized_goal_objective(goal_objective.as_deref()).as_deref(),
4082 goal_token_budget,
4083 goal_status,
4084 );
4085 }
4086 self.config.allowed_tools = allowed_tools;
4087 self.config.hook_executor = hook_executor;
4088 self.session.reasoning_effort = reasoning_effort;
4089 self.session.reasoning_effort_auto = reasoning_effort_auto;
4090 self.session.auto_model = auto_model;
4091 self.config.translation_enabled = translation_enabled;
4092 self.config.verbosity = verbosity;
4093
4094 // Compose from the immutable values accepted for this turn. Preview
4095 // receives the same context before anything is installed, so prompt
4096 // bytes cannot depend on stale session state or mutation order.
4097 self.refresh_system_prompt_from_context(&prompt_context);
4098
4099 self.session
4100 .working_set
4101 .observe_user_message(&content, &self.session.workspace);
4102
4103 // Add the user message through the same explicit snapshot constructor
4104 // preview uses. Route limits and mode in resource metadata therefore
4105 // belong to this turn even when the previous route was different.
4106 let user_msg = self.user_text_message_from_snapshot(
4107 content,
4108 &model,
4109 auto_model,
4110 self.session.reasoning_effort.as_deref(),
4111 self.session.reasoning_effort_auto,
4112 provenance,
4113 TurnMetadataSnapshot {
4114 prompt_context: &prompt_context,
4115 system_prompt: self.session.system_prompt.as_ref(),
4116 approval_mode: self.session.approval_mode,
4117 working_set: &self.session.working_set,
4118 policy_narrowing: self.last_policy_narrowing.as_ref(),
4119 },
4120 );
4121 self.session.add_message(user_msg);
4122
4123 self.emit_session_updated().await;
4124
4125 // Build tool registry and tool list for the current mode
4126 let turn_id_for_mailbox = turn.id.clone();
4127 let TurnToolBuild {
4128 surface,
4129 mailbox: mut mailbox_for_runtime,
4130 plugin_tool_names,
4131 ..
4132 } = self
4133 .build_turn_tool_registry_and_catalog(
4134 &input_policy,
4135 &dynamic_tools,
4136 self.config.allowed_tools.clone(),
4137 SubAgentWiring::Live,
4138 McpAccess::Connect,
4139 TurnRouteContext {
4140 provider: self.api_config.api_provider(),
4141 model: self.config.model.clone(),
4142 capabilities: route_capabilities,
4143 limits: self.active_route_limits,
4144 client: self.deepseek_client.clone(),
4145 api_config: route_api_config,
4146 locale_tag: self.config.locale_tag.clone(),
4147 role_models: self.subagent_role_models(),
4148 fleet_roster: self.config.fleet_roster.clone(),
4149 auto_model,
4150 reasoning_effort: self.session.reasoning_effort.clone(),
4151 reasoning_effort_auto: self.session.reasoning_effort_auto,
4152 },
4153 &turn_id_for_mailbox,
4154 )
4155 .await;
4156 let tool_catalog_for_event = Some(surface.catalog.clone());
4157
4158 // Resolve, once per turn, the out-of-request facts the read-only
4159 // request projection is allowed to report: flattened registry facts,
4160 // the MCP pool's own server attribution, and the engine-injected
4161 // catalog names. This is where `plugin_tool_names` and the pool lock
4162 // live; the snapshot itself is built later, at the request seam, from
4163 // the tools actually prepared for that step.
4164 let mut tool_surface = crate::tool_inspection::ToolSurfaceContext {
4165 registry: surface.registry.registry_facts(&plugin_tool_names),
4166 mcp_servers: match self.mcp_pool.as_ref() {
4167 Some(pool) => pool.lock().await.resolved_tool_servers(),
4168 None => std::collections::BTreeMap::new(),
4169 },
4170 synthetic_names: default_synthetic_catalog_tool_names(),
4171 provider: crate::tool_inspection::ProviderAvailability::Unknown,
4172 };
4173 tool_surface.provider = self.tool_surface_provider_receipt();
4174
4175 let base_url_for_event = if self.model_client_injected {
4176 None
4177 } else {
4178 self.deepseek_client
4179 .as_ref()
4180 .map(|client| client.base_url().to_string())
4181 };
4182
4183 // Main turn loop. Catch panics here so an internal error surfaces as a
4184 // failed TurnComplete instead of unwinding through `engine.run()` and
4185 // killing the whole engine-event-loop task — which left the UI stuck
4186 // on "working" forever with the engine silently dead (#2583, #1269).
4187 use futures_util::FutureExt as _;
4188 let turn_result = std::panic::AssertUnwindSafe(self.handle_deepseek_turn(
4189 &mut turn,
4190 surface,
4191 Some(tool_surface),
4192 ))
4193 .catch_unwind()
4194 .await;
4195 let (status, error) = match turn_result {
4196 Ok(outcome) => outcome,
4197 Err(panic) => {
4198 let detail = crate::utils::panic_message(&*panic);
4199 crate::utils::record_caught_panic("engine-event-loop", &detail);
4200 (
4201 TurnOutcomeStatus::Failed,
4202 Some(format!(
4203 "The engine hit an internal error and stopped this turn: {detail}. \
4204 Your session is intact — send your message again to retry. \
4205 A crash report was saved to ~/.codewhale/crashes/."
4206 )),
4207 )
4208 }
4209 };
4210
4211 // Update session usage
4212 self.session.total_usage.add(&turn.usage);
4213 self.record_goal_usage_for_turn(&turn.usage, turn.elapsed());
4214
4215 // Seal and fully forward every accepted mailbox envelope before the
4216 // terminal event. This is the durability barrier for child usage: an
4217 // event can no longer arrive after `TurnComplete` and be mistaken for
4218 // the following turn (or lost by a runtime monitor that already
4219 // settled the record).
4220 if let Some(barrier) = mailbox_for_runtime.take() {
4221 barrier.mailbox.seal();
4222 let _ = barrier.flush_tx.send(());
4223 let _ = barrier.drain_handle.await;
4224 }
4225
4226 // Emit turn complete event — after all post-turn bookkeeping so
4227 // the terminal is immediately responsive when the UI receives it.
4228 self.emit_goal_updated().await;
4229 if status == TurnOutcomeStatus::Interrupted {
4230 self.emit_interrupted_survivor_status().await;
4231 }
4232 let turn_complete_delivered = self
4233 .tx_event
4234 .send(Event::TurnComplete {
4235 usage: turn.usage,
4236 status,
4237 error: error.clone(),
4238 tool_catalog: tool_catalog_for_event,
4239 base_url: base_url_for_event,
4240 })
4241 .await
4242 .is_ok();
4243 tracing::info!(
4244 target: "engine.turn",
4245 status = ?status,
4246 delivered = turn_complete_delivered,
4247 "engine turn completion settled"
4248 );
4249
4250 // Post-turn snapshot. Fire-and-forget: TurnComplete is already
4251 // emitted, so the UI is unblocked and the user can type / select /
4252 // paste immediately (#234). The git work proceeds on the blocking
4253 // pool without forcing the engine loop to await it.
4254 if self.config.snapshots_enabled {
4255 // `snapshot_prompt_post` was cloned from `content` above,
4256 // before `content` was moved into the session messages.
4257 let post_workspace = self.session.workspace.clone();
4258 let post_seq = self.turn_counter;
4259 let post_cap = self.config.snapshots_max_workspace_bytes;
4260 let post_sid = self.session.id.clone();
4261 crate::utils::spawn_blocking_supervised("post-turn-snapshot", move || {
4262 post_turn_snapshot(
4263 &post_workspace,
4264 post_seq,
4265 post_cap,
4266 Some(&snapshot_prompt_post),
4267 Some(&post_sid),
4268 );
4269 });
4270 }
4271
4272 // ── Background advisor watcher (#3982) ────────────────────────────
4273 // Fire-and-forget: TurnComplete is already emitted. The advisor
4274 // reads a bounded snapshot of session messages (immutable clone),
4275 // makes a short LLM advisory call, and emits `Event::AdvisoryNote`.
4276 // Any failure is logged and swallowed — it must never affect the
4277 // parent turn's outcome.
4278 if self.config.advisor_config.enabled
4279 && matches!(status, TurnOutcomeStatus::Completed)
4280 && let Some(client) = self.deepseek_client.clone()
4281 {
4282 // Lazily create the shared emission guard on first use.
4283 let guard = self
4284 .advisor_emission_guard
4285 .get_or_insert_with(|| {
4286 Arc::new(tokio::sync::Mutex::new(
4287 crate::tools::subagent::EmissionGuard::new(),
4288 ))
4289 })
4290 .clone();
4291
4292 let advisor_messages: Vec<crate::models::Message> = self.session.messages.to_vec();
4293 let advisor_config = self.config.advisor_config.clone();
4294 let advisor_model = self.session.model.clone();
4295 let advisor_tx = self.tx_event.clone();
4296 let advisor_turn_id = turn.id.clone();
4297
4298 crate::utils::spawn_supervised(
4299 "advisor-watcher",
4300 std::panic::Location::caller(),
4301 async move {
4302 crate::tools::subagent::run_advisor_for_turn(
4303 advisor_turn_id,
4304 advisor_messages,
4305 advisor_config,
4306 client,
4307 advisor_model,
4308 guard,
4309 advisor_tx,
4310 )
4311 .await;
4312 },
4313 );
4314 }
4315
4316 // ── Cross-turn goal continuation ───────────────────────────────────
4317 // When the interactive engine owns turn lifecycle, a successful turn
4318 // with an active goal re-dispatches a synthetic continuation through
4319 // its own op channel. RuntimeThreadManager engines instead yield here:
4320 // their host must create the next durable claim before dispatching any
4321 // further turn. A Failed or Interrupted turn never continues.
4322 let outcome = SendMessageOutcome::Finished { status, error };
4323 if !self.host_managed_turns()
4324 && matches!(
4325 &outcome,
4326 SendMessageOutcome::Finished {
4327 status: TurnOutcomeStatus::Completed,
4328 ..
4329 }
4330 )
4331 {
4332 // Queue a typed continuation instead of freezing an Active goal
4333 // snapshot into a generic message. The operation re-reads the live
4334 // state when consumed, after any already-queued goal controls.
4335 self.schedule_goal_continuation(dynamic_tools);
4336 } else {
4337 self.reconcile_non_completed_goal_turn(&outcome).await;
4338 }
4339 outcome
4340 }
4341
4342 /// Capture typed live state for post-compact rehydrate (workers, shells,
4343 /// mode, permission). To-do state deliberately stays out of this stable
4344 /// prefix snapshot: the authoritative graph projection is appended fresh
4345 /// to each parent turn-loop and sub-agent step request by
4346 /// `work_state_tail_message`.
4347 async fn capture_compaction_live_state(&self) -> CompactionLiveState {
4348 self.touch_workers_with_running_shells().await;
4349 let running_workers = {
4350 let mut guard = self.subagent_manager.write().await;
4351 guard.cleanup(Duration::from_secs(60 * 60));
4352 guard
4353 .list()
4354 .into_iter()
4355 .filter(|s| matches!(s.status, SubAgentStatus::Running))
4356 .map(|s| {
4357 let role = s.assignment.role.as_deref().unwrap_or("-");
4358 let goal = if s.assignment.objective.is_empty() {
4359 "(no objective)"
4360 } else {
4361 s.assignment.objective.as_str()
4362 };
4363 format!("`{}` (role: {role}) — {goal}", s.agent_id)
4364 })
4365 .collect()
4366 };
4367
4368 let background_shells = match self.shell_manager.lock() {
4369 Ok(mut manager) => manager
4370 .list_jobs()
4371 .into_iter()
4372 .filter(|job| matches!(job.status, crate::tools::shell::ShellStatus::Running))
4373 .map(|job| format!("`{}`: `{}`", job.id, job.command))
4374 .collect(),
4375 Err(_) => Vec::new(),
4376 };
4377
4378 CompactionLiveState {
4379 mode: Some(self.current_mode.as_setting().to_string()),
4380 permission_posture: Some(
4381 self.session
4382 .approval_mode
4383 .permission_chip_label()
4384 .to_string(),
4385 ),
4386 background_shells,
4387 running_workers,
4388 open_approvals: Vec::new(),
4389 }
4390 }
4391
4392 async fn handle_manual_compaction(&mut self) {
4393 let id = format!("compact_{}", &uuid::Uuid::new_v4().to_string()[..8]);
4394 let zero_usage = Usage {
4395 input_tokens: 0,
4396 output_tokens: 0,
4397 ..Usage::default()
4398 };
4399 let Some(client) = self.deepseek_client.clone() else {
4400 let message = "Manual compaction unavailable: API client not configured".to_string();
4401 self.emit_compaction_failed(id, false, message.clone())
4402 .await;
4403 let _ = self
4404 .tx_event
4405 .send(Event::error(ErrorEnvelope::fatal_auth(message.clone())))
4406 .await;
4407 let _ = self
4408 .tx_event
4409 .send(Event::TurnComplete {
4410 usage: zero_usage,
4411 status: TurnOutcomeStatus::Failed,
4412 error: Some(message),
4413 tool_catalog: None,
4414 base_url: None,
4415 })
4416 .await;
4417 return;
4418 };
4419
4420 let start_message = "Manual context compaction started".to_string();
4421 self.emit_compaction_started(id.clone(), false, start_message)
4422 .await;
4423
4424 let compaction_pins = self
4425 .session
4426 .working_set
4427 .pinned_message_indices(&self.session.messages, &self.session.workspace);
4428 let compaction_paths = self.session.working_set.top_paths(24);
4429 let messages_before = self.session.messages.len();
4430 let mut turn_status = TurnOutcomeStatus::Completed;
4431 let mut turn_error = None;
4432
4433 let mut compaction_config = self.config.compaction.clone();
4434 let live = self.capture_compaction_live_state().await;
4435 if !live.is_empty() {
4436 compaction_config.live_state = Some(live);
4437 }
4438
4439 match compact_messages_safe(
4440 &client,
4441 &self.session.messages,
4442 &compaction_config,
4443 Some(&self.session.workspace),
4444 Some(&compaction_pins),
4445 Some(&compaction_paths),
4446 )
4447 .await
4448 {
4449 Ok(result) => {
4450 if !result.messages.is_empty() || self.session.messages.is_empty() {
4451 let messages_after = result.messages.len();
4452 self.session.replace_messages(result.messages);
4453 self.merge_compaction_summary(result.summary_prompt);
4454 self.emit_session_updated().await;
4455 let removed = messages_before.saturating_sub(messages_after);
4456 let message = if result.retries_used > 0 {
4457 format!(
4458 "Compaction complete: {messages_before} → {messages_after} messages ({removed} removed, {} retries)",
4459 result.retries_used
4460 )
4461 } else {
4462 format!(
4463 "Compaction complete: {messages_before} → {messages_after} messages ({removed} removed)"
4464 )
4465 };
4466 self.emit_compaction_completed(
4467 id,
4468 false,
4469 message,
4470 Some(messages_before),
4471 Some(messages_after),
4472 )
4473 .await;
4474 } else {
4475 let message = "Compaction skipped: produced empty result".to_string();
4476 self.emit_compaction_failed(id, false, message.clone())
4477 .await;
4478 turn_status = TurnOutcomeStatus::Failed;
4479 turn_error = Some(message);
4480 }
4481 }
4482 Err(err) => {
4483 let message = crate::compaction::report_compaction_failure(
4484 "Manual context compaction failed",
4485 &id,
4486 false,
4487 &err,
4488 );
4489 self.emit_compaction_failed(id, false, message.clone())
4490 .await;
4491 let _ = self.tx_event.send(Event::status(message.clone())).await;
4492 turn_status = TurnOutcomeStatus::Failed;
4493 turn_error = Some(message);
4494 }
4495 }
4496
4497 let _ = self
4498 .tx_event
4499 .send(Event::TurnComplete {
4500 usage: zero_usage,
4501 status: turn_status,
4502 error: turn_error,
4503 tool_catalog: None,
4504 base_url: None,
4505 })
4506 .await;
4507 }
4508
4509 async fn handle_purge(&mut self) {
4510 let zero_usage = Usage {
4511 input_tokens: 0,
4512 output_tokens: 0,
4513 ..Usage::default()
4514 };
4515 let Some(client) = self.deepseek_client.clone() else {
4516 let message = "Purge unavailable: API client not configured".to_string();
4517 emit_purge_failed(&self.tx_event, message.clone()).await;
4518 let _ = self
4519 .tx_event
4520 .send(Event::error(ErrorEnvelope::fatal_auth(message.clone())))
4521 .await;
4522 let _ = self
4523 .tx_event
4524 .send(Event::TurnComplete {
4525 usage: zero_usage,
4526 status: TurnOutcomeStatus::Failed,
4527 error: Some(message),
4528 tool_catalog: None,
4529 base_url: None,
4530 })
4531 .await;
4532 return;
4533 };
4534
4535 emit_purge_started(
4536 &self.tx_event,
4537 "Agent context purge in progress\u{2026}".to_string(),
4538 )
4539 .await;
4540 let messages_before = self.session.messages.len();
4541
4542 let (status, error) = match run_purge(
4543 &client,
4544 self.api_provider,
4545 &self.session.messages,
4546 &self.session.model,
4547 self.session.reasoning_effort.clone(),
4548 effective_max_output_tokens_for_route(
4549 self.api_provider,
4550 &self.session.model,
4551 self.active_route_limits,
4552 ),
4553 )
4554 .await
4555 {
4556 Ok(result) => {
4557 let messages_after = result.messages.len();
4558 self.session.replace_messages(result.messages);
4559 self.emit_session_updated().await;
4560
4561 let summary = format!(
4562 "Purge complete: {messages_before} → {messages_after} messages \
4563 ({} removed, {} condensed)",
4564 result.removed_count, result.replaced_count,
4565 );
4566 emit_purge_completed(
4567 &self.tx_event,
4568 messages_before,
4569 messages_after,
4570 result.removed_count,
4571 result.replaced_count,
4572 summary,
4573 )
4574 .await;
4575 (TurnOutcomeStatus::Completed, None)
4576 }
4577 Err(e) => {
4578 emit_purge_failed(&self.tx_event, e.clone()).await;
4579 (TurnOutcomeStatus::Failed, Some(e))
4580 }
4581 };
4582
4583 let _ = self
4584 .tx_event
4585 .send(Event::TurnComplete {
4586 usage: zero_usage,
4587 status,
4588 error,
4589 tool_catalog: None,
4590 base_url: None,
4591 })
4592 .await;
4593 }
4594
4595 /// Turn-visible background shell jobs still running right now, formatted
4596 /// for the interrupt-honesty status line (DGF-03, dogfood 2026-08-02):
4597 /// Esc stops the model turn, not detached shell work. Without this,
4598 /// files landing on disk after "Turn interrupted" read as a lie.
4599 fn running_background_shell_survivors(&self) -> Vec<String> {
4600 let Ok(mut manager) = self.shell_manager.lock() else {
4601 return Vec::new();
4602 };
4603 manager
4604 .list_jobs()
4605 .into_iter()
4606 .filter(|job| matches!(job.status, crate::tools::shell::ShellStatus::Running))
4607 .map(|job| {
4608 const MAX_COMMAND_CHARS: usize = 48;
4609 let mut command: String = job.command.chars().take(MAX_COMMAND_CHARS).collect();
4610 if job.command.chars().count() > MAX_COMMAND_CHARS {
4611 command.push('…');
4612 }
4613 format!("{} `{command}`", job.id)
4614 })
4615 .collect()
4616 }
4617
4618 /// Emit the interrupt-honesty status naming still-running background
4619 /// shell jobs. Called on the paths that can classify a turn as
4620 /// Interrupted, immediately before their `TurnComplete` event.
4621 async fn emit_interrupted_survivor_status(&self) {
4622 let survivors = self.running_background_shell_survivors();
4623 if survivors.is_empty() {
4624 return;
4625 }
4626 let _ = self
4627 .tx_event
4628 .send(Event::status(format!(
4629 "Turn interrupted, but {} background shell job(s) continue and may still write files: {}. Use /jobs to inspect or kill.",
4630 survivors.len(),
4631 survivors.join(", ")
4632 )))
4633 .await;
4634 }
4635
4636 fn estimated_input_tokens(&mut self) -> usize {
4637 // Memoized on (session.messages_revision, system-prompt fingerprint).
4638 // The cache invalidates as soon as either input changes; until then
4639 // repeated calls (capacity checkpoints, /status, context inspector,
4640 // TUI footer) all hit the cached value.
4641 self.token_estimate_cache.lookup_or_compute(
4642 self.session.messages_revision,
4643 self.session.system_prompt.as_ref(),
4644 &self.session.messages,
4645 )
4646 }
4647
4648 fn trim_oldest_messages_to_budget(&mut self, target_input_budget: usize) -> usize {
4649 let mut removed = 0usize;
4650 while self.session.messages.len() > MIN_RECENT_MESSAGES_TO_KEEP
4651 && self.estimated_input_tokens() > target_input_budget
4652 {
4653 self.session.messages.trim_front(1);
4654 self.session.bump_messages_revision();
4655 removed = removed.saturating_add(1);
4656 }
4657 removed
4658 }
4659
4660 fn compaction_pins_for_messages(
4661 &self,
4662 messages: &[Message],
4663 working_set: &crate::working_set::WorkingSet,
4664 ) -> Vec<usize> {
4665 let mut pins = working_set.pinned_message_indices(messages, &self.session.workspace);
4666
4667 pins.sort_unstable();
4668 pins.dedup();
4669 pins
4670 }
4671
4672 async fn recover_context_overflow(
4673 &mut self,
4674 client: &dyn crate::core::model_client::ModelClient,
4675 reason: &str,
4676 ) -> bool {
4677 let Some(target_budget) = context_input_budget_for_route(
4678 self.api_provider,
4679 &self.session.model,
4680 self.active_route_limits,
4681 0,
4682 ) else {
4683 return false;
4684 };
4685
4686 let id = format!("compact_{}", &uuid::Uuid::new_v4().to_string()[..8]);
4687 let start_message = format!("Emergency context compaction started ({reason})");
4688 self.emit_compaction_started(id.clone(), true, start_message)
4689 .await;
4690
4691 let before_tokens = self.estimated_input_tokens();
4692 let before_count = self.session.messages.len();
4693
4694 let mut retries_used = 0u32;
4695 let mut summary_prompt = None;
4696 let mut compacted_messages: Vec<Message> = self.session.messages.clone().into();
4697
4698 let mut forced_config = self.config.compaction.clone();
4699 forced_config.enabled = true;
4700 forced_config.token_threshold = forced_config
4701 .token_threshold
4702 .min(target_budget.saturating_sub(1))
4703 .max(1);
4704 let live = self.capture_compaction_live_state().await;
4705 if !live.is_empty() {
4706 forced_config.live_state = Some(live);
4707 }
4708
4709 // Preserve the working-set pins on the emergency/preflight path too.
4710 // Previously this passed None/None, so a compaction routed here (which,
4711 // on large windows, is the path that actually fires) could summarize
4712 // away pinned errors, patches, and the files the user is editing.
4713 let compaction_pins =
4714 self.compaction_pins_for_messages(&self.session.messages, &self.session.working_set);
4715 let compaction_paths = self.session.working_set.top_paths(24);
4716
4717 match compact_messages_safe(
4718 client,
4719 &self.session.messages,
4720 &forced_config,
4721 Some(&self.session.workspace),
4722 Some(&compaction_pins),
4723 Some(&compaction_paths),
4724 )
4725 .await
4726 {
4727 Ok(result) => {
4728 retries_used = result.retries_used;
4729 compacted_messages = result.messages;
4730 summary_prompt = result.summary_prompt;
4731 }
4732 Err(err) => {
4733 let _ = self
4734 .tx_event
4735 .send(Event::status(format!(
4736 "Emergency compaction API pass failed: {err}. Falling back to local trim."
4737 )))
4738 .await;
4739 }
4740 }
4741
4742 if !compacted_messages.is_empty() || self.session.messages.is_empty() {
4743 self.session.replace_messages(compacted_messages);
4744 }
4745 self.merge_compaction_summary(summary_prompt);
4746
4747 let trimmed = self.trim_oldest_messages_to_budget(target_budget);
4748 self.emit_session_updated().await;
4749 let after_tokens = self.estimated_input_tokens();
4750 let after_count = self.session.messages.len();
4751 let recovered = after_tokens <= target_budget
4752 && (after_tokens < before_tokens || after_count < before_count || trimmed > 0);
4753
4754 if recovered {
4755 let removed = before_count.saturating_sub(after_count);
4756 let mut details = format!(
4757 "Emergency compaction complete: {before_count} → {after_count} messages ({removed} removed), ~{before_tokens} → ~{after_tokens} tokens"
4758 );
4759 if retries_used > 0 {
4760 details.push_str(&format!(" ({retries_used} retries)"));
4761 }
4762 if trimmed > 0 {
4763 details.push_str(&format!(", trimmed {trimmed} oldest"));
4764 }
4765 self.emit_compaction_completed(
4766 id,
4767 true,
4768 details.clone(),
4769 Some(before_count),
4770 Some(after_count),
4771 )
4772 .await;
4773 let _ = self.tx_event.send(Event::status(details)).await;
4774 return true;
4775 }
4776
4777 let message = format!(
4778 "Emergency context compaction failed to reduce request below model limit \
4779 (estimate ~{after_tokens} tokens, budget ~{target_budget})."
4780 );
4781 self.emit_compaction_failed(id, true, message.clone()).await;
4782 let _ = self.tx_event.send(Event::status(message)).await;
4783 false
4784 }
4785
4786 /// Role/type model map for sub-agent runtimes: roster member pins first,
4787 /// then explicit `[subagents]` overrides on top so explicit config wins
4788 /// (#fleet-roster cutover (v0.8.67)).
4789 fn subagent_role_models(&self) -> HashMap<String, String> {
4790 let mut models = self.config.fleet_roster.model_overrides();
4791 models.extend(
4792 self.config
4793 .subagent_model_overrides
4794 .iter()
4795 .map(|(key, value)| (key.clone(), value.clone())),
4796 );
4797 models
4798 }
4799
4800 fn build_tool_context(&self, mode: AppMode, auto_approve: bool) -> ToolContext {
4801 let authority = TurnAuthority::from_effective_fields(
4802 mode,
4803 self.session.allow_shell,
4804 self.session.trust_mode,
4805 mode == AppMode::Yolo || auto_approve,
4806 self.session.approval_mode,
4807 );
4808 let route = TurnRouteContext {
4809 provider: self.api_provider,
4810 model: self.session.model.clone(),
4811 capabilities: self.active_route_capabilities,
4812 limits: self.active_route_limits,
4813 client: self.deepseek_client.clone(),
4814 api_config: Box::new(self.api_config.clone()),
4815 locale_tag: self.config.locale_tag.clone(),
4816 role_models: self.subagent_role_models(),
4817 fleet_roster: self.config.fleet_roster.clone(),
4818 auto_model: self.session.auto_model,
4819 reasoning_effort: self.session.reasoning_effort.clone(),
4820 reasoning_effort_auto: self.session.reasoning_effort_auto,
4821 };
4822 self.build_tool_context_for_turn(&authority, &route)
4823 }
4824
4825 /// Project the current engine authority onto an already-built registry.
4826 /// Registries own long-lived services and tool definitions; permission,
4827 /// shell, and sandbox policy are live turn state and must not be read from
4828 /// the registry's start-of-turn snapshot after a Runtime posture switch.
4829 fn live_tool_context(
4830 &self,
4831 registry: Option<&crate::tools::ToolRegistry>,
4832 ) -> Option<ToolContext> {
4833 let mut context = registry?.context().clone();
4834 let authority = TurnAuthority::from_effective_fields(
4835 self.current_mode,
4836 self.session.allow_shell,
4837 self.session.trust_mode,
4838 self.session.auto_approve,
4839 self.session.approval_mode,
4840 );
4841 context.trust_mode = authority.trust_mode;
4842 context.auto_approve = authority.auto_approve;
4843 context.shell_policy = authority.shell_policy();
4844 context.elevated_sandbox_policy = Some(authority.sandbox_policy(
4845 &self.session.workspace,
4846 self.api_config.sandbox_mode.as_deref(),
4847 ));
4848 context.shell_network_denied_hint = matches!(authority.mode, AppMode::Plan)
4849 .then(|| PLAN_SHELL_NETWORK_DENIED_HINT.to_string());
4850 Some(context)
4851 }
4852
4853 /// Build one tool context from the already-resolved turn authority and
4854 /// route. A preview owns values that are deliberately not installed on the
4855 /// session; rebuilding either from `self.session` would give it the prior
4856 /// turn's shell posture, context window, model, route capabilities, and
4857 /// provider-native search client.
4858 fn build_tool_context_for_turn(
4859 &self,
4860 authority: &TurnAuthority,
4861 route: &TurnRouteContext,
4862 ) -> ToolContext {
4863 // Load the per-workspace trusted-paths list (#29) on every tool-context
4864 // build. Cheap (a small JSON file) and always reflects the latest
4865 // `/trust add` / `/trust remove` mutations without an explicit cache
4866 // refresh hook.
4867 let trusted = crate::workspace_trust::WorkspaceTrust::load_for(&self.session.workspace);
4868 let mut trusted_external_paths = trusted.paths().to_vec();
4869 let clipboard_images_dir =
4870 crate::tui::clipboard::clipboard_images_dir(&self.session.workspace);
4871 if !trusted_external_paths
4872 .iter()
4873 .any(|path| path == &clipboard_images_dir)
4874 {
4875 trusted_external_paths.push(clipboard_images_dir);
4876 }
4877 let mut ctx = ToolContext::with_auto_approve(
4878 self.session.workspace.clone(),
4879 authority.trust_mode,
4880 self.session.notes_path.clone(),
4881 self.session.mcp_config_path.clone(),
4882 authority.auto_approve,
4883 )
4884 .with_state_namespace(self.session.id.clone())
4885 .with_route_context_window(crate::route_budget::route_context_window_tokens(
4886 route.provider,
4887 &route.model,
4888 route.limits,
4889 ))
4890 .with_features(self.config.features.clone())
4891 .with_shell_manager(self.shell_manager.clone())
4892 .with_file_read_tracker(self.file_read_tracker.clone())
4893 .with_runtime_services(self.config.runtime_services.clone())
4894 .with_skills_config(
4895 self.config.skills_dir.clone(),
4896 self.config.skills_scan_codewhale_only,
4897 )
4898 .with_plugin_registry(Arc::clone(&self.plugin_registry))
4899 .with_session_objects(crate::rlm::session::SessionObjectSnapshot::new(
4900 self.session.id.clone(),
4901 route.model.clone(),
4902 self.session.workspace.clone(),
4903 self.session.system_prompt.clone(),
4904 self.session.messages.clone().into(),
4905 ))
4906 .with_cancel_token(self.cancel_token.clone())
4907 .with_shell_policy(authority.shell_policy())
4908 .with_trusted_external_paths(trusted_external_paths)
4909 .with_follow_symlinks(self.config.workspace_follow_symlinks);
4910
4911 // Hand the user-memory path to tools so the model-callable
4912 // `remember` tool can append entries (#489). `None` when the
4913 // feature is disabled — tools short-circuit on that.
4914 if self.config.memory_enabled {
4915 ctx.memory_path = Some(self.config.memory_path.clone());
4916 }
4917
4918 if let Some(decider) = self.config.network_policy.as_ref() {
4919 ctx = ctx.with_network_policy(decider.clone());
4920 }
4921
4922 // Adaptive evidence routing is engine-native and always present.
4923 // `[workshop]` only customizes thresholds; it no longer gates storage.
4924 if let Some(vars_arc) = self.workshop_vars.as_ref() {
4925 let router = crate::tools::large_output_router::LargeOutputRouter::new(
4926 self.config.workshop.clone().unwrap_or_default(),
4927 );
4928 ctx = ctx.with_large_output_router(router, vars_arc.clone());
4929 }
4930
4931 // Wire the external sandbox backend (#516). exec_shell checks this
4932 // field and routes commands through the backend instead of spawning
4933 // a local process when it's set.
4934 if let Some(backend) = self.sandbox_backend.as_ref() {
4935 ctx = ctx.with_sandbox_backend(std::sync::Arc::clone(backend));
4936 }
4937
4938 // Wire search provider config.
4939 ctx.search_provider = self.config.search_provider;
4940 ctx.search_api_key = self.config.search_api_key.clone();
4941 ctx.search_base_url = self.config.search_base_url.clone();
4942 ctx.route_capabilities = route.capabilities;
4943 if route.capabilities.server_side_web_search.is_supported() {
4944 ctx.provider_native_search = route
4945 .client
4946 .as_ref()
4947 .cloned()
4948 .and_then(crate::client::ProviderNativeSearchClient::new);
4949 }
4950
4951 let policy = authority.sandbox_policy(
4952 &self.session.workspace,
4953 self.api_config.sandbox_mode.as_deref(),
4954 );
4955 let mut ctx = ctx.with_elevated_sandbox_policy(policy);
4956 if matches!(authority.mode, AppMode::Plan) {
4957 ctx = ctx.with_shell_network_denied_hint(PLAN_SHELL_NETWORK_DENIED_HINT);
4958 }
4959 ctx
4960 }
4961
4962 /// Revalidate durable owners after a saved session is installed. Owner
4963 /// stores apply restart recovery first; the graph consumes only their
4964 /// monotonic snapshots and never infers liveness from prior UI state.
4965 async fn reconcile_restored_work_bindings(&self) {
4966 let Some(work) = self.config.runtime_services.work.as_ref() else {
4967 return;
4968 };
4969 let session_id = self.session.id.as_str();
4970 let candidates = work
4971 .reconcilable_durable_bindings(Some(session_id))
4972 .into_iter()
4973 .collect::<HashSet<_>>();
4974 let checked_at = chrono::Utc::now().timestamp_millis();
4975
4976 let mut seen_tasks = HashSet::new();
4977 if let Some(task_manager) = self.config.runtime_services.task_manager.as_ref() {
4978 for task in task_manager.list_tasks(None).await {
4979 let external = format!("task:{}", task.id);
4980 if !candidates.contains(&external) {
4981 continue;
4982 }
4983 seen_tasks.insert(external.clone());
4984 if let Err(err) = work.reconcile_operation(
4985 session_id,
4986 crate::work_graph::task_owner_snapshot(
4987 &task.id,
4988 task.status,
4989 task.lifecycle_seq,
4990 task.created_at,
4991 task.started_at,
4992 task.ended_at,
4993 ),
4994 ) {
4995 tracing::warn!(task_id = %task.id, error = %err, "failed to reconcile restored task owner");
4996 }
4997 }
4998 }
4999 for external in candidates
5000 .iter()
5001 .filter(|external| external.starts_with("task:"))
5002 .filter(|external| !seen_tasks.contains(*external))
5003 {
5004 if let Err(err) = work.reconcile_observation(
5005 session_id,
5006 external,
5007 crate::work_graph::OperationObservation::OwnerMissing { checked_at },
5008 ) {
5009 tracing::warn!(%external, error = %err, "failed to mark missing task owner");
5010 }
5011 }
5012
5013 let worker_records = self.subagent_manager.read().await.list_worker_records();
5014 let mut seen_workers = HashSet::new();
5015 for record in worker_records {
5016 let Some(snapshot) = agent_worker_owner_snapshot(&record) else {
5017 continue;
5018 };
5019 if !candidates.contains(&snapshot.external) {
5020 continue;
5021 }
5022 seen_workers.insert(snapshot.external.clone());
5023 if let Err(err) = work.reconcile_operation(session_id, snapshot) {
5024 tracing::warn!(worker_id = %record.spec.worker_id, error = %err, "failed to reconcile restored worker owner");
5025 }
5026 }
5027 for external in candidates
5028 .iter()
5029 .filter(|external| external.starts_with("worker:"))
5030 .filter(|external| !seen_workers.contains(*external))
5031 {
5032 if let Err(err) = work.reconcile_observation(
5033 session_id,
5034 external,
5035 crate::work_graph::OperationObservation::OwnerMissing { checked_at },
5036 ) {
5037 tracing::warn!(%external, error = %err, "failed to mark missing worker owner");
5038 }
5039 }
5040
5041 if let Err(err) = crate::tools::workflow::reconcile_persisted_workflow_bindings(
5042 work,
5043 session_id,
5044 &self.session.workspace,
5045 ) {
5046 tracing::warn!(error = %err, "failed to reconcile restored workflow owners");
5047 }
5048 }
5049
5050 async fn ensure_mcp_pool(&mut self) -> Result<Arc<AsyncMutex<McpPool>>, ToolError> {
5051 if let Some(pool) = self.mcp_pool.as_ref() {
5052 return Ok(Arc::clone(pool));
5053 }
5054 let mut pool = McpPool::from_config_path_with_workspace_and_plugins(
5055 &self.session.mcp_config_path,
5056 &self.session.workspace,
5057 Arc::clone(&self.plugin_registry),
5058 )
5059 .unwrap_or_else(|e| {
5060 tracing::debug!(
5061 "MCP config unavailable: {}",
5062 crate::mcp::format_mcp_error_for_display(&e)
5063 );
5064 McpPool::empty_with_workspace_config_sources(
5065 &self.session.mcp_config_path,
5066 &self.session.workspace,
5067 Arc::clone(&self.plugin_registry),
5068 )
5069 .unwrap_or_else(|fallback_error| {
5070 tracing::debug!(
5071 "MCP reload source setup failed: {}",
5072 crate::mcp::format_mcp_error_for_display(&fallback_error)
5073 );
5074 McpPool::new(McpConfig::default())
5075 })
5076 });
5077 if let Some(decider) = self.config.network_policy.as_ref() {
5078 pool = pool.with_network_policy(decider.clone());
5079 }
5080 let pool = Arc::new(AsyncMutex::new(pool));
5081 self.mcp_pool = Some(Arc::clone(&pool));
5082 Ok(pool)
5083 }
5084
5085 async fn reload_mcp_pool(
5086 &mut self,
5087 config_path: PathBuf,
5088 ) -> anyhow::Result<crate::mcp::McpManagerSnapshot> {
5089 let pool = self
5090 .ensure_mcp_pool()
5091 .await
5092 .map_err(|error| anyhow::anyhow!(error.to_string()))?;
5093 let mut pool = pool.lock().await;
5094 let connection_errors = if config_path == self.session.mcp_config_path {
5095 pool.reload_and_connect_all().await?
5096 } else {
5097 pool.switch_workspace_config_source_and_connect_all(
5098 &config_path,
5099 &self.session.workspace,
5100 Arc::clone(&self.plugin_registry),
5101 )
5102 .await?
5103 };
5104 let errors = connection_errors
5105 .into_iter()
5106 .map(|(name, error)| (name, crate::mcp::format_mcp_error_for_display(&error)))
5107 .collect::<HashMap<_, _>>();
5108 self.session.mcp_config_path = config_path;
5109 Ok(pool.manager_snapshot(&self.session.mcp_config_path, false, &errors))
5110 }
5111
5112 async fn mcp_tools(&mut self) -> Vec<Tool> {
5113 let pool = match self.ensure_mcp_pool().await {
5114 Ok(pool) => pool,
5115 Err(err) => {
5116 let _ = self.tx_event.send(Event::status(format!("{err:#}"))).await;
5117 return Vec::new();
5118 }
5119 };
5120
5121 let mut pool = pool.lock().await;
5122 let errors = pool.connect_all().await;
5123 for (server, err) in errors {
5124 let _ = self
5125 .tx_event
5126 .send(Event::status(format!(
5127 "Failed to connect MCP server '{server}': {err:#}"
5128 )))
5129 .await;
5130 }
5131
5132 pool.to_api_tools()
5133 }
5134
5135 /// Handle a turn using the DeepSeek API.
5136 #[allow(clippy::too_many_lines)]
5137 /// Refresh the stable system prompt based on current non-mode context.
5138 fn refresh_system_prompt(&mut self) {
5139 let context = self.installed_next_turn_prompt_context();
5140 self.refresh_system_prompt_from_context(&context);
5141 }
5142
5143 fn refresh_system_prompt_from_context(&mut self, context: &NextTurnPromptContext) {
5144 let stable_prompt = self.compose_stable_system_prompt(context);
5145
5146 let stable_hash = system_prompt_hash(stable_prompt.as_ref());
5147 if self.session.system_prompt_override {
5148 return;
5149 }
5150 if self.session.last_system_prompt_hash != Some(stable_hash) {
5151 self.session.system_prompt = stable_prompt;
5152 self.session.last_system_prompt_hash = Some(stable_hash);
5153 }
5154 }
5155
5156 /// Compose the stable system prompt for an explicit route, without
5157 /// touching session state.
5158 ///
5159 /// [`Self::refresh_system_prompt`] calls it for the installed route;
5160 /// `/preview-request` calls it for the route the *next* turn would use,
5161 /// which may be a different model with a different context window when
5162 /// auto routing is on. Extracting it is what lets a preview describe the
5163 /// next prompt exactly without mutating the session to find out.
5164 pub(super) fn compose_stable_system_prompt(
5165 &self,
5166 context: &NextTurnPromptContext,
5167 ) -> Option<SystemPrompt> {
5168 let user_memory_block = crate::native_memory::native_prompt_block(
5169 self.config.memory_enabled,
5170 &self.config.memory_path,
5171 &self.config.workspace,
5172 );
5173 let base = prompts::system_prompt_for_mode_with_context_skills_session_and_approval(
5174 &self.config.workspace,
5175 None,
5176 Some(&self.config.skills_dir),
5177 Some(&self.config.instructions),
5178 prompts::PromptSessionContext {
5179 user_memory_block: user_memory_block.as_deref(),
5180 goal_objective: context.goal_objective.as_deref(),
5181 project_context_pack_enabled: self.config.project_context_pack_enabled,
5182 locale_tag: &self.config.locale_tag,
5183 translation_enabled: context.translation_enabled,
5184 model_id: &context.model,
5185 context_window_override: Some(crate::route_budget::route_context_window_tokens(
5186 context.provider,
5187 &context.model,
5188 context.route_limits,
5189 )),
5190 verbosity: context.verbosity.as_deref(),
5191 skills_scan_codewhale_only: self.config.skills_scan_codewhale_only,
5192 plugin_registry: Some(self.plugin_registry.as_ref()),
5193 mode: context.mode,
5194 },
5195 );
5196 merge_system_prompts(Some(&base), self.session.compaction_summary_prompt.clone())
5197 }
5198
5199 fn installed_next_turn_prompt_context(&self) -> NextTurnPromptContext {
5200 NextTurnPromptContext::for_planned_turn(
5201 self.api_provider,
5202 self.config.model.clone(),
5203 self.active_route_limits,
5204 self.current_mode,
5205 goal_objective_for_prompt(
5206 self.config.goal_objective.as_deref(),
5207 &self.config.goal_state,
5208 ),
5209 self.config.goal_status,
5210 self.config.goal_token_budget,
5211 self.config.translation_enabled,
5212 self.config.verbosity.clone(),
5213 )
5214 }
5215
5216 /// Merge a compaction summary into the system prompt.
5217 ///
5218 /// **Zone affiliation (#2264)**: this mutates the system prompt, which is
5219 /// part of the `PinnedPrefix` zone in the three-zone contract. Compaction
5220 /// is the one intentional mid-session prefix mutation — the engine
5221 /// intentionally accepts the cache-invalidation cost because the
5222 /// context-reduction benefit outweighs it.
5223 fn merge_compaction_summary(&mut self, summary_prompt: Option<SystemPrompt>) {
5224 let Some(summary_prompt) = summary_prompt else {
5225 return;
5226 };
5227 let reanchor = self
5228 .config
5229 .runtime_services
5230 .work
5231 .as_ref()
5232 .and_then(|work| work.active_operation_summary(Some(&self.session.id)))
5233 .map(SystemPrompt::Text);
5234 let summary_prompt =
5235 merge_system_prompts(Some(&summary_prompt), reanchor).or(Some(summary_prompt));
5236 let prior_compaction =
5237 strip_active_operation_reanchor(self.session.compaction_summary_prompt.as_ref());
5238 self.session.compaction_summary_prompt =
5239 merge_system_prompts(prior_compaction.as_ref(), summary_prompt.clone());
5240 let prior_system = strip_active_operation_reanchor(self.session.system_prompt.as_ref());
5241 let merged = merge_system_prompts(prior_system.as_ref(), summary_prompt);
5242 self.session.last_system_prompt_hash = Some(system_prompt_hash(merged.as_ref()));
5243 self.session.system_prompt = merged;
5244 }
5245 }
5246
5247 fn strip_active_operation_reanchor(prompt: Option<&SystemPrompt>) -> Option<SystemPrompt> {
5248 fn strip_text(mut text: String) -> Option<String> {
5249 while let Some(start) = text.find(crate::work_graph::ACTIVE_OPERATION_SUMMARY_START) {
5250 let tail = start + crate::work_graph::ACTIVE_OPERATION_SUMMARY_START.len();
5251 let end = text[tail..]
5252 .find(crate::work_graph::ACTIVE_OPERATION_SUMMARY_END)
5253 .map_or(text.len(), |offset| {
5254 tail + offset + crate::work_graph::ACTIVE_OPERATION_SUMMARY_END.len()
5255 });
5256 text.replace_range(start..end, "");
5257 }
5258 let text = text.trim().to_string();
5259 (!text.is_empty()).then_some(text)
5260 }
5261
5262 match prompt.cloned()? {
5263 SystemPrompt::Text(text) => strip_text(text).map(SystemPrompt::Text),
5264 SystemPrompt::Blocks(blocks) => {
5265 let blocks = blocks
5266 .into_iter()
5267 .filter_map(|mut block| {
5268 block.text = strip_text(block.text)?;
5269 Some(block)
5270 })
5271 .collect::<Vec<_>>();
5272 (!blocks.is_empty()).then_some(SystemPrompt::Blocks(blocks))
5273 }
5274 }
5275 }
5276
5277 fn default_plugin_tools_dir() -> PathBuf {
5278 codewhale_config::codewhale_home()
5279 .unwrap_or_else(|_| {
5280 crate::config::effective_home_dir()
5281 .map_or_else(|| PathBuf::from(".codewhale"), |h| h.join(".codewhale"))
5282 })
5283 .join("tools")
5284 }
5285
5286 fn plugin_tools_dir(tools_config: Option<&crate::config::ToolsConfig>) -> PathBuf {
5287 if let Some(tools_config) = tools_config
5288 && let Some(custom_dir) = tools_config.plugin_dir.as_deref()
5289 {
5290 return PathBuf::from(shellexpand::tilde(custom_dir).as_ref());
5291 }
5292 default_plugin_tools_dir()
5293 }
5294
5295 fn configure_plugin_tools(
5296 tool_registry: &mut crate::tools::ToolRegistry,
5297 tools_config: Option<&crate::config::ToolsConfig>,
5298 ) -> std::collections::HashSet<String> {
5299 let names_before: std::collections::HashSet<String> = tool_registry
5300 .names()
5301 .into_iter()
5302 .map(|s| s.to_string())
5303 .collect();
5304
5305 let plugin_dir = plugin_tools_dir(tools_config);
5306 tool_registry.load_plugins(&plugin_dir);
5307
5308 if let Some(tools_config) = tools_config
5309 && let Some(ref overrides) = tools_config.overrides
5310 {
5311 tool_registry.apply_overrides(overrides, &plugin_dir);
5312 }
5313
5314 let names_after: std::collections::HashSet<String> = tool_registry
5315 .names()
5316 .into_iter()
5317 .map(|s| s.to_string())
5318 .collect();
5319 &names_after - &names_before
5320 }
5321
5322 fn system_prompt_hash(prompt: Option<&SystemPrompt>) -> u64 {
5323 let mut hasher = DefaultHasher::new();
5324 match prompt {
5325 Some(SystemPrompt::Text(text)) => {
5326 0u8.hash(&mut hasher);
5327 text.hash(&mut hasher);
5328 }
5329 Some(SystemPrompt::Blocks(blocks)) => {
5330 1u8.hash(&mut hasher);
5331 for block in blocks {
5332 block.block_type.hash(&mut hasher);
5333 block.text.hash(&mut hasher);
5334 if let Some(cache_control) = &block.cache_control {
5335 cache_control.cache_type.hash(&mut hasher);
5336 }
5337 }
5338 }
5339 None => {
5340 2u8.hash(&mut hasher);
5341 }
5342 }
5343 hasher.finish()
5344 }
5345
5346 fn normalized_goal_objective(value: Option<&str>) -> Option<String> {
5347 value
5348 .map(str::trim)
5349 .filter(|value| !value.is_empty())
5350 .map(str::to_string)
5351 }
5352
5353 fn sync_goal_state_from_host(
5354 goal_state: &SharedGoalState,
5355 objective: Option<&str>,
5356 token_budget: Option<u32>,
5357 status: GoalStatus,
5358 ) {
5359 match goal_state.lock() {
5360 Ok(mut state) => state.sync_from_host_status(objective, token_budget, status),
5361 Err(err) => tracing::warn!("goal state lock poisoned while syncing host goal: {err}"),
5362 }
5363 }
5364
5365 fn goal_objective_for_prompt(
5366 configured_goal: Option<&str>,
5367 goal_state: &SharedGoalState,
5368 ) -> Option<String> {
5369 match goal_state.lock() {
5370 Ok(state) => {
5371 if let Some(objective) = state.objective() {
5372 // Preserve original behavior: return None (not fallback) when
5373 // objective exists but goal is inactive.
5374 return state.is_active().then(|| objective.to_string());
5375 }
5376 }
5377 Err(err) => tracing::warn!("goal state lock poisoned while building prompt: {err}"),
5378 }
5379 normalized_goal_objective(configured_goal)
5380 }
5381
5382 // ── Mode & approval prompts as request-time runtime metadata ─────────
5383 //
5384 // Mode contracts and approval policies are not persisted in the session
5385 // history and are not sent as extra system messages. Instead, each API
5386 // request projects a transient user-role runtime metadata message at the
5387 // tail. The stable system prompt remains byte-stable, stored history remains
5388 // byte-stable, and strict chat-template providers never see a system message
5389 // outside messages[0].
5390
5391 #[derive(Debug, Clone, PartialEq, Eq)]
5392 pub(super) enum ToolAskRuleDecision {
5393 Allow,
5394 Prompt(String),
5395 Block(String),
5396 }
5397
5398 #[derive(Debug, Clone, PartialEq, Eq)]
5399 pub(super) enum AutoReviewPlanDecision {
5400 NoChange,
5401 Allow,
5402 ForcePrompt(String),
5403 Block(String),
5404 }
5405
5406 pub(super) fn auto_review_run_origin_for_plan(
5407 detached_start: bool,
5408 ) -> crate::tui::auto_review::RunOrigin {
5409 if detached_start {
5410 crate::tui::auto_review::RunOrigin::Background
5411 } else {
5412 crate::tui::auto_review::RunOrigin::Interactive
5413 }
5414 }
5415
5416 // The parameter list intentionally mirrors `AutoReviewContext::from_tool_call`,
5417 // which this thin wrapper builds; the 8 call sites (1 prod + tests) read clearer
5418 // passing the fields than constructing a context first.
5419 #[allow(clippy::too_many_arguments)]
5420 pub(super) fn auto_review_plan_decision(
5421 policy: &crate::tui::auto_review::AutoReviewPolicy,
5422 tool_name: &str,
5423 tool_input: &Value,
5424 run_origin: crate::tui::auto_review::RunOrigin,
5425 approval_mode: crate::tui::approval::ApprovalMode,
5426 user_intent: Option<&str>,
5427 workspace_trusted: bool,
5428 dirty_worktree: bool,
5429 ) -> (AutoReviewPlanDecision, Value) {
5430 let context = crate::tui::auto_review::AutoReviewContext::from_tool_call(
5431 tool_name,
5432 tool_input,
5433 run_origin,
5434 approval_mode,
5435 user_intent,
5436 workspace_trusted,
5437 dirty_worktree,
5438 );
5439 let decision = policy.evaluate(&context);
5440 let audit_event = policy.audit_event(&context, &decision);
5441 let plan_decision = if approval_mode == crate::tui::approval::ApprovalMode::Auto
5442 && tool_name == REQUEST_USER_INPUT_NAME
5443 {
5444 // This synthetic tool does not execute user work. Let the turn loop
5445 // return its ordinary autonomous guidance result instead of treating
5446 // a hallucinated question as an unknown external action.
5447 AutoReviewPlanDecision::Allow
5448 } else {
5449 match decision.action {
5450 crate::tui::auto_review::AutoReviewAction::Allow
5451 if approval_mode == crate::tui::approval::ApprovalMode::Auto =>
5452 {
5453 AutoReviewPlanDecision::Allow
5454 }
5455 crate::tui::auto_review::AutoReviewAction::Allow => AutoReviewPlanDecision::NoChange,
5456 crate::tui::auto_review::AutoReviewAction::AskUser
5457 if approval_mode == crate::tui::approval::ApprovalMode::Auto =>
5458 {
5459 AutoReviewPlanDecision::Block(format!(
5460 "Auto-Review held tool '{tool_name}': {}",
5461 decision.reason
5462 ))
5463 }
5464 crate::tui::auto_review::AutoReviewAction::AskUser => AutoReviewPlanDecision::NoChange,
5465 crate::tui::auto_review::AutoReviewAction::HoldForReview => {
5466 // HoldForReview only originates from the built-in safety floor
5467 // (configured rules produce Allow/Block), so name the gate
5468 // honestly instead of blaming an "auto-review policy" the user
5469 // may never have configured (#3883).
5470 let reason = format!(
5471 "Built-in safety gate requires approval: {}",
5472 decision.reason
5473 );
5474 if matches!(
5475 approval_mode,
5476 crate::tui::approval::ApprovalMode::Auto
5477 | crate::tui::approval::ApprovalMode::Never
5478 | crate::tui::approval::ApprovalMode::Bypass
5479 ) {
5480 // Auto-Review, Never, and Full Access are non-interactive for
5481 // approval holds. Full Access auto-runs ordinary calls, but a
5482 // non-bypassable safety floor always fails closed.
5483 AutoReviewPlanDecision::Block(reason)
5484 } else {
5485 AutoReviewPlanDecision::ForcePrompt(reason)
5486 }
5487 }
5488 crate::tui::auto_review::AutoReviewAction::Block => {
5489 AutoReviewPlanDecision::Block(format!(
5490 "Auto-review policy blocked tool '{tool_name}': {}",
5491 decision.reason
5492 ))
5493 }
5494 }
5495 };
5496 (plan_decision, audit_event)
5497 }
5498
5499 pub(super) fn exec_shell_ask_rule_decision(
5500 config: &EngineConfig,
5501 tool_name: &str,
5502 tool_input: &Value,
5503 workspace: &Path,
5504 approval_mode: crate::tui::approval::ApprovalMode,
5505 ) -> Option<ToolAskRuleDecision> {
5506 let policy_tool_name =
5507 crate::tools::canonical_action::canonical_action_alias(tool_name, tool_input);
5508 if policy_tool_name != "exec_shell" {
5509 return None;
5510 }
5511 let command = tool_input.get("command").and_then(Value::as_str)?;
5512 tool_ask_rule_decision_for_context(
5513 config,
5514 policy_tool_name,
5515 command,
5516 None,
5517 workspace,
5518 approval_mode,
5519 )
5520 }
5521
5522 pub(super) fn file_tool_ask_rule_decision(
5523 config: &EngineConfig,
5524 tool_name: &str,
5525 tool_input: &Value,
5526 workspace: &Path,
5527 approval_mode: crate::tui::approval::ApprovalMode,
5528 ) -> Option<ToolAskRuleDecision> {
5529 let policy_tool_name =
5530 crate::tools::canonical_action::canonical_action_alias(tool_name, tool_input);
5531 let paths = file_tool_permission_paths(policy_tool_name, tool_input)?;
5532 if paths.is_empty() {
5533 return tool_ask_rule_decision_for_context(
5534 config,
5535 policy_tool_name,
5536 "",
5537 None,
5538 workspace,
5539 approval_mode,
5540 );
5541 }
5542
5543 let mut prompt: Option<String> = None;
5544 let mut all_allowed = true;
5545 for path in paths {
5546 match tool_ask_rule_decision_for_context(
5547 config,
5548 policy_tool_name,
5549 "",
5550 Some(&path),
5551 workspace,
5552 approval_mode,
5553 ) {
5554 Some(ToolAskRuleDecision::Block(reason)) => {
5555 return Some(ToolAskRuleDecision::Block(reason));
5556 }
5557 Some(ToolAskRuleDecision::Prompt(reason)) => {
5558 prompt.get_or_insert(reason);
5559 all_allowed = false;
5560 }
5561 Some(ToolAskRuleDecision::Allow) => {}
5562 None => all_allowed = false,
5563 }
5564 }
5565 if let Some(prompt) = prompt {
5566 Some(ToolAskRuleDecision::Prompt(prompt))
5567 } else if all_allowed {
5568 Some(ToolAskRuleDecision::Allow)
5569 } else {
5570 None
5571 }
5572 }
5573
5574 fn tool_ask_rule_decision_for_context(
5575 config: &EngineConfig,
5576 tool_name: &str,
5577 command: &str,
5578 path: Option<&str>,
5579 workspace: &Path,
5580 approval_mode: crate::tui::approval::ApprovalMode,
5581 ) -> Option<ToolAskRuleDecision> {
5582 let cwd = workspace.to_string_lossy();
5583 let ask_for_approval = match approval_mode {
5584 crate::tui::approval::ApprovalMode::Never => AskForApproval::Never,
5585 crate::tui::approval::ApprovalMode::Auto
5586 | crate::tui::approval::ApprovalMode::Bypass
5587 | crate::tui::approval::ApprovalMode::Suggest => AskForApproval::OnFailure,
5588 };
5589 let decision = config
5590 .exec_policy_engine
5591 .check(ExecPolicyContext {
5592 command,
5593 cwd: cwd.as_ref(),
5594 tool: Some(tool_name),
5595 path,
5596 ask_for_approval,
5597 sandbox_mode: None,
5598 })
5599 .ok()?;
5600 if !decision.allow {
5601 Some(ToolAskRuleDecision::Block(decision.reason().to_string()))
5602 } else if decision.requires_approval {
5603 Some(ToolAskRuleDecision::Prompt(decision.reason().to_string()))
5604 } else if decision.matched_action == Some(codewhale_execpolicy::PermissionAction::Allow) {
5605 // Count only. Never `matched_rule`, never `reason()`, never the
5606 // command or its argv: `auto_allow` patterns are user-authored command
5607 // strings.
5608 codewhale_telemetry::session_counters()
5609 .bump(codewhale_telemetry::Counter::ApprovalAutoAllowed);
5610 Some(ToolAskRuleDecision::Allow)
5611 } else {
5612 None
5613 }
5614 }
5615
5616 fn file_tool_permission_paths(tool_name: &str, input: &Value) -> Option<Vec<String>> {
5617 match tool_name {
5618 "read_file" | "write_file" | "edit_file" | "file_search" | "grep_files" => {
5619 Some(string_field(input, "path").into_iter().collect())
5620 }
5621 "list_dir" => Some(vec![
5622 string_field(input, "path").unwrap_or_else(|| ".".to_string()),
5623 ]),
5624 "apply_patch" => Some(apply_patch_permission_paths(input)),
5625 _ => None,
5626 }
5627 }
5628
5629 /// Target paths when a call is one of the canonical workspace file-write
5630 /// tools (`write_file` / `edit_file` / `apply_patch`), `None` for any other
5631 /// tool. Feeds the in-workspace write carve-out (#5185).
5632 fn file_write_tool_target_paths(tool_name: &str, input: &Value) -> Option<Vec<String>> {
5633 let canonical = crate::tools::canonical_action::canonical_action_alias(tool_name, input);
5634 if !matches!(canonical, "write_file" | "edit_file" | "apply_patch") {
5635 return None;
5636 }
5637 file_tool_permission_paths(canonical, input)
5638 }
5639
5640 fn string_field(input: &Value, key: &str) -> Option<String> {
5641 input
5642 .get(key)
5643 .and_then(Value::as_str)
5644 .map(str::trim)
5645 .filter(|value| !value.is_empty())
5646 .map(str::to_string)
5647 }
5648
5649 fn apply_patch_permission_paths(input: &Value) -> Vec<String> {
5650 crate::tools::apply_patch::preflight_apply_patch(input)
5651 .map(|preflight| preflight.touched_files)
5652 .unwrap_or_default()
5653 }
5654
5655 /// Spawn the engine in a background task
5656 pub fn spawn_engine(config: EngineConfig, api_config: &Config) -> EngineHandle {
5657 let (engine, handle) = Engine::new(config, api_config);
5658
5659 spawn_supervised(
5660 "engine-event-loop",
5661 std::panic::Location::caller(),
5662 async move {
5663 engine.run().await;
5664 },
5665 );
5666
5667 handle
5668 }
5669
5670 /// Spawn a runtime-owned engine whose autonomous later turns resolve against
5671 /// the manager's atomic config snapshot. This does not mutate an active turn.
5672 pub(crate) fn spawn_engine_with_authoritative_route_config(
5673 config: EngineConfig,
5674 api_config: &Config,
5675 authoritative_route_config: Arc<parking_lot::RwLock<Config>>,
5676 ) -> EngineHandle {
5677 let (mut engine, handle) = Engine::new(config, api_config);
5678 engine.authoritative_route_config = Some(authoritative_route_config);
5679
5680 spawn_supervised(
5681 "engine-event-loop",
5682 std::panic::Location::caller(),
5683 async move {
5684 engine.run().await;
5685 },
5686 );
5687
5688 handle
5689 }
5690
5691 #[cfg(test)]
5692 pub(crate) struct MockEngineHandle {
5693 pub handle: EngineHandle,
5694 pub rx_op: mpsc::Receiver<Op>,
5695 rx_approval: mpsc::Receiver<ApprovalDecision>,
5696 rx_user_input: mpsc::Receiver<UserInputDecision>,
5697 pub rx_steer: mpsc::Receiver<String>,
5698 pub tx_event: mpsc::Sender<Event>,
5699 pub cancel_token: CancellationToken,
5700 }
5701
5702 #[cfg(test)]
5703 #[derive(Debug, Clone, PartialEq, Eq)]
5704 pub(crate) enum MockApprovalEvent {
5705 Approved {
5706 id: String,
5707 },
5708 Denied {
5709 id: String,
5710 },
5711 RetryWithPolicy {
5712 id: String,
5713 policy: crate::sandbox::SandboxPolicy,
5714 },
5715 }
5716
5717 #[cfg(test)]
5718 impl MockEngineHandle {
5719 pub(crate) async fn recv_approval_event(&mut self) -> Option<MockApprovalEvent> {
5720 match self.rx_approval.recv().await? {
5721 ApprovalDecision::Approved { id } => Some(MockApprovalEvent::Approved { id }),
5722 ApprovalDecision::Denied { id } => Some(MockApprovalEvent::Denied { id }),
5723 ApprovalDecision::RetryWithPolicy { id, policy } => {
5724 Some(MockApprovalEvent::RetryWithPolicy { id, policy })
5725 }
5726 }
5727 }
5728
5729 pub(crate) async fn recv_user_input_submission(
5730 &mut self,
5731 ) -> Option<(String, UserInputResponse)> {
5732 match self.rx_user_input.recv().await? {
5733 UserInputDecision::Submitted { id, response } => Some((id, response)),
5734 UserInputDecision::Cancelled { .. } => None,
5735 }
5736 }
5737
5738 pub(crate) async fn recv_user_input_cancellation(&mut self) -> Option<String> {
5739 match self.rx_user_input.recv().await? {
5740 UserInputDecision::Cancelled { id } => Some(id),
5741 UserInputDecision::Submitted { .. } => None,
5742 }
5743 }
5744
5745 /// Close the engine event stream without moving fields out of the handle,
5746 /// so failure-path tests can keep using the receiver helpers afterwards.
5747 pub(crate) fn close_event_stream(&mut self) {
5748 let (tx_event, _rx_event) = mpsc::channel(1);
5749 self.tx_event = tx_event;
5750 }
5751 }
5752
5753 #[cfg(test)]
5754 pub(crate) fn mock_engine_handle() -> MockEngineHandle {
5755 let (tx_op, rx_op) = mpsc::channel(32);
5756 let (tx_event, rx_event) = mpsc::channel(256);
5757 let (tx_approval, rx_approval) = mpsc::channel(64);
5758 let (tx_user_input, rx_user_input) = mpsc::channel(32);
5759 let (tx_steer, rx_steer) = mpsc::channel(64);
5760 let cancel_token = CancellationToken::new();
5761 let shared_cancel_token = Arc::new(StdMutex::new(cancel_token.clone()));
5762 let cancel_reason: Arc<StdMutex<Option<CancelReason>>> = Arc::new(StdMutex::new(None));
5763 let shared_paused = Arc::new(StdMutex::new(false));
5764 let live_runtime_authority = Arc::new(StdMutex::new(LiveRuntimeAuthorityState::new(
5765 LiveRuntimeAuthority::from_fields(
5766 AppMode::Agent,
5767 false,
5768 false,
5769 false,
5770 crate::tui::approval::ApprovalMode::Suggest,
5771 None,
5772 ),
5773 )));
5774 let handle = EngineHandle {
5775 tx_op,
5776 rx_event: Arc::new(RwLock::new(rx_event)),
5777 cancel_token: shared_cancel_token,
5778 cancel_reason,
5779 tx_approval,
5780 tx_user_input,
5781 tx_steer,
5782 shared_paused,
5783 client_preflight_required: false,
5784 live_runtime_authority,
5785 };
5786
5787 MockEngineHandle {
5788 handle,
5789 rx_op,
5790 rx_approval,
5791 rx_user_input,
5792 rx_steer,
5793 tx_event,
5794 cancel_token,
5795 }
5796 }
5797
5798 /// The session state a turn installs before it writes `<turn_meta>`.
5799 ///
5800 /// Production reads it back off `self` after installing it; `/preview-request`
5801 /// supplies the values it *would* install, so an inspection can reproduce the
5802 /// block exactly without writing any of them.
5803 pub(crate) struct TurnMetadataSnapshot<'a> {
5804 pub(crate) prompt_context: &'a NextTurnPromptContext,
5805 pub(crate) system_prompt: Option<&'a SystemPrompt>,
5806 pub(crate) approval_mode: crate::tui::approval::ApprovalMode,
5807 pub(crate) working_set: &'a crate::working_set::WorkingSet,
5808 pub(crate) policy_narrowing: Option<&'a PolicyNarrowingEvent>,
5809 }
5810
5811 /// Immutable prompt facts for the next accepted turn.
5812 ///
5813 /// Both production and `/preview-request` compose through this value. It owns
5814 /// every per-turn field resolved by submit or route planning that can change
5815 /// the stable system prompt, so a hypothetical route cannot accidentally
5816 /// inherit the installed turn's goal, translation, verbosity, mode,
5817 /// model, or context window. Workspace-scoped prompt inputs remain engine
5818 /// configuration and are documented separately as snapshot dependencies.
5819 #[derive(Debug, Clone, PartialEq, Eq)]
5820 pub(crate) struct NextTurnPromptContext {
5821 pub(crate) provider: ApiProvider,
5822 pub(crate) model: String,
5823 pub(crate) route_limits: Option<codewhale_config::route::RouteLimits>,
5824 pub(crate) mode: AppMode,
5825 pub(crate) goal_objective: Option<String>,
5826 pub(crate) goal_token_budget: Option<u32>,
5827 pub(crate) translation_enabled: bool,
5828 pub(crate) verbosity: Option<String>,
5829 }
5830
5831 impl NextTurnPromptContext {
5832 #[allow(clippy::too_many_arguments)]
5833 pub(crate) fn for_planned_turn(
5834 provider: ApiProvider,
5835 model: String,
5836 route_limits: Option<codewhale_config::route::RouteLimits>,
5837 mode: AppMode,
5838 goal_objective: Option<String>,
5839 goal_status: GoalStatus,
5840 goal_token_budget: Option<u32>,
5841 translation_enabled: bool,
5842 verbosity: Option<String>,
5843 ) -> Self {
5844 Self {
5845 provider,
5846 model,
5847 route_limits,
5848 mode,
5849 goal_objective: (goal_status == GoalStatus::Active)
5850 .then(|| normalized_goal_objective(goal_objective.as_deref()))
5851 .flatten(),
5852 goal_token_budget,
5853 translation_enabled,
5854 verbosity,
5855 }
5856 }
5857 }
5858
5859 /// Result of one turn tool-catalog build.
5860 /// Turn-scoped mailbox handle plus the machinery needed to close it exactly
5861 /// once. Held by the engine (never by the child runtime) so the flush barrier
5862 /// is owned by the same code that emits the terminal turn event.
5863 pub(crate) struct TurnMailboxBarrier {
5864 pub(crate) mailbox: Mailbox,
5865 pub(crate) cancel_token: tokio_util::sync::CancellationToken,
5866 pub(crate) flush_tx: tokio::sync::oneshot::Sender<()>,
5867 pub(crate) drain_handle: tokio::task::JoinHandle<()>,
5868 }
5869
5870 struct TurnToolBuild {
5871 /// One authority for executable, searchable, and initially active tools.
5872 surface: ToolSurfacePolicy,
5873 /// Names of the MCP-contributed tools in this build.
5874 mcp_tool_names: Vec<String>,
5875 /// What is known about the MCP contribution to this catalog.
5876 mcp: McpToolState,
5877 /// Route model installed into the child runtime, when sub-agent tools were
5878 /// available. This is an internal receipt, not a manifest field.
5879 #[cfg_attr(not(test), allow(dead_code))]
5880 subagent_runtime_model: Option<String>,
5881 /// Turn-scoped sub-agent mailbox and its flush barrier, when sub-agent
5882 /// wiring was live. The engine must seal, flush, and await this before it
5883 /// emits `TurnComplete`: that is what makes detached-child usage accounting
5884 /// exactly-once rather than "whatever arrived in time".
5885 mailbox: Option<TurnMailboxBarrier>,
5886 /// Tools this build loaded from the plugin surface rather than the built-in
5887 /// registry builder. Carried out so the read-only request projection can
5888 /// tell `plugin` provenance from `builtin` instead of collapsing both.
5889 plugin_tool_names: std::collections::HashSet<String>,
5890 }
5891
5892 /// The route a tool catalog is being shaped for.
5893 ///
5894 /// A real turn installs its route before building the catalog, so this is
5895 /// simply the installed route. `/preview-request` has a *planned* route that
5896 /// is deliberately not installed, so it passes that one instead — otherwise
5897 /// an auto-routed preview would report the previous route's tool budget.
5898 #[derive(Clone)]
5899 pub(crate) struct TurnRouteContext {
5900 pub(crate) provider: ApiProvider,
5901 pub(crate) model: String,
5902 pub(crate) capabilities: codewhale_config::route::RouteCapabilities,
5903 pub(crate) limits: Option<codewhale_config::route::RouteLimits>,
5904 /// Client for this exact route. Tool contexts use it only for
5905 /// provider-native helper capabilities; previews pass their throw-away
5906 /// planned client instead of inheriting the installed session client.
5907 pub(crate) client: Option<DeepSeekClient>,
5908 /// Route-scoped runtime config, captured by the planner. A preview must
5909 /// never construct child agents from the previously installed config.
5910 pub(crate) api_config: Box<crate::config::Config>,
5911 pub(crate) locale_tag: String,
5912 pub(crate) role_models: HashMap<String, String>,
5913 pub(crate) fleet_roster: Arc<crate::fleet::roster::FleetRoster>,
5914 pub(crate) auto_model: bool,
5915 pub(crate) reasoning_effort: Option<String>,
5916 pub(crate) reasoning_effort_auto: bool,
5917 }
5918
5919 impl TurnRouteContext {
5920 pub(crate) fn capability_profile(&self) -> crate::model_profile::CapabilityProfile {
5921 crate::model_profile::resolved_capability_profile_for_route(
5922 self.provider,
5923 &self.model,
5924 self.capabilities,
5925 self.limits.unwrap_or_default(),
5926 )
5927 }
5928 }
5929
5930 /// Whether a tool-catalog build may start or connect MCP servers.
5931 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
5932 pub(crate) enum McpAccess {
5933 /// A real turn: create the pool if needed and connect every enabled
5934 /// server, exactly as before.
5935 Connect,
5936 /// An inspection: use only what is already connected, and report the
5937 /// tool surface as unavailable when that is not the whole picture.
5938 PassiveSnapshot,
5939 }
5940
5941 impl McpAccess {
5942 fn may_connect(self) -> bool {
5943 matches!(self, Self::Connect)
5944 }
5945 }
5946
5947 /// The MCP contribution to one tool-catalog build.
5948 #[derive(Debug, Clone)]
5949 pub(crate) enum McpToolState {
5950 /// MCP is off for this session; a turn would send no MCP tools.
5951 Disabled,
5952 /// The exact MCP tool set the next request would carry.
5953 Live {
5954 tools: Vec<Tool>,
5955 server_count: usize,
5956 },
5957 /// The exact set is not knowable without connecting, which an inspection
5958 /// must not do.
5959 Unavailable { reason: McpUnavailable },
5960 }
5961
5962 impl McpToolState {
5963 pub(crate) fn tools(&self) -> &[Tool] {
5964 match self {
5965 Self::Live { tools, .. } => tools,
5966 Self::Disabled | Self::Unavailable { .. } => &[],
5967 }
5968 }
5969
5970 /// Connected server count, or `None` when the state is unavailable.
5971 pub(crate) fn server_count(&self) -> Option<usize> {
5972 match self {
5973 Self::Disabled => Some(0),
5974 Self::Live { server_count, .. } => Some(*server_count),
5975 Self::Unavailable { .. } => None,
5976 }
5977 }
5978 }
5979
5980 /// Why a passive MCP snapshot could not describe the next turn exactly.
5981 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
5982 pub(crate) enum McpUnavailable {
5983 /// No pool exists yet: the first turn of the session would create and
5984 /// connect one.
5985 PoolNotStarted,
5986 /// An MCP config source changed since the pool last read it, so the next
5987 /// turn would reload before connecting.
5988 ConfigChangedSinceConnect,
5989 /// Some enabled servers are configured but not connected.
5990 ServersNotConnected { pending: usize },
5991 }
5992
5993 impl McpUnavailable {
5994 /// Short, path-free explanation for the manifest.
5995 pub(crate) fn label(self) -> String {
5996 match self {
5997 Self::PoolNotStarted => {
5998 "MCP is enabled but no server has been connected in this session yet".to_string()
5999 }
6000 Self::ConfigChangedSinceConnect => {
6001 "an MCP configuration source changed since the last connect".to_string()
6002 }
6003 Self::ServersNotConnected { pending } => {
6004 format!("{pending} enabled MCP server(s) are not connected yet")
6005 }
6006 }
6007 }
6008 }
6009
6010 /// Whether a tool-catalog build may establish sub-agent runtime side effects.
6011 ///
6012 /// Both variants register exactly the same tools; only the runtime plumbing
6013 /// differs (the structured fork snapshot and the spawned mailbox drainer),
6014 /// which is what makes an offline inspection safe to run at any time.
6015 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
6016 pub(crate) enum SubAgentWiring {
6017 /// A real turn: wire the fork snapshot and the mailbox drainer.
6018 Live,
6019 /// An inspection: build the catalog, spawn nothing.
6020 Inert,
6021 }
6022
6023 impl SubAgentWiring {
6024 fn is_live(self) -> bool {
6025 matches!(self, Self::Live)
6026 }
6027 }
6028
6029 mod approval;
6030 mod context;
6031 mod handle;
6032 pub mod preview;
6033 #[cfg(test)]
6034 pub(crate) use context::compact_tool_result_for_context;
6035 pub(crate) use context::compact_tool_result_for_route;
6036 /// Public so external hosts/wrappers can reuse the engine's input-budget math
6037 /// (see `context_input_budget_for_route`'s doc) instead of re-deriving it.
6038 pub use context::context_input_budget_for_route;
6039 #[cfg(test)]
6040 use context::route_context_budget_for_provider;
6041 use context::{
6042 MAX_CONTEXT_RECOVERY_ATTEMPTS, MIN_RECENT_MESSAGES_TO_KEEP,
6043 effective_max_output_tokens_for_route, estimate_input_tokens_conservative,
6044 extract_compaction_summary_prompt, is_context_length_error_message,
6045 route_context_budget_for_route, summarize_text,
6046 };
6047 #[cfg(test)]
6048 use context::{context_input_budget_for_provider, effective_max_output_tokens};
6049 mod dispatch;
6050 mod lsp_hooks;
6051 mod read_repeat_guard;
6052 mod streaming;
6053 mod stuck_guard;
6054 mod token_estimate_cache;
6055 pub(crate) mod tool_catalog;
6056 mod tool_execution;
6057 mod tool_preparation;
6058 mod tool_setup;
6059 mod turn_loop;
6060 pub(crate) use token_estimate_cache::TokenEstimateCache;
6061
6062 pub(super) const MAX_PARALLEL_SHELL_EXEC: usize = 4;
6063
6064 #[cfg(test)]
6065 pub(crate) fn default_active_native_tool_names() -> &'static [&'static str] {
6066 tool_catalog::DEFAULT_ACTIVE_NATIVE_TOOLS
6067 }
6068
6069 use self::approval::{ApprovalDecision, ApprovalResult, UserInputDecision};
6070 use self::dispatch::{
6071 ParallelToolResult, ParallelToolResultEntry, ToolApprovalStamp, ToolExecGuard, ToolExecOutcome,
6072 ToolExecutionBatch, ToolExecutionPlan, caller_allowed_for_tool, caller_type_for_tool_use,
6073 final_tool_input, format_tool_error_with_schema, malformed_tool_arguments_error,
6074 malformed_tool_arguments_input, mcp_tool_is_parallel_safe, parse_parallel_tool_calls,
6075 parse_tool_input, plan_tool_execution_batches, stamp_tool_result_approval,
6076 };
6077 #[cfg(test)]
6078 use self::dispatch::{format_tool_error, should_parallelize_tool_batch};
6079 #[cfg(test)]
6080 use self::lsp_hooks::edited_paths_for_tool;
6081 #[cfg(test)]
6082 use self::streaming::TOOL_CALL_START_MARKERS;
6083 #[cfg(test)]
6084 use self::streaming::filter_tool_call_delta;
6085 use self::streaming::{
6086 ContentBlockKind, FAKE_WRAPPER_NOTICE, MAX_STREAM_ERRORS_BEFORE_FAIL, MAX_STREAM_RETRIES,
6087 MAX_TRANSPARENT_STREAM_RETRIES, STREAM_MAX_CONTENT_BYTES, STREAM_MAX_DURATION_SECS,
6088 ToolCallDeltaFilterState, ToolUseState, contains_fake_tool_wrapper,
6089 filter_tool_call_delta_with_state, flush_tool_call_delta_state,
6090 should_resume_after_network_drop, should_resume_after_sleep, should_transparently_retry_stream,
6091 sleep_gap_detected, stream_read_error_user_message,
6092 };
6093 use self::tool_catalog::{
6094 CODE_EXECUTION_TOOL_NAME, JS_EXECUTION_TOOL_NAME, MULTI_TOOL_PARALLEL_NAME,
6095 REQUEST_USER_INPUT_NAME, ToolSurfacePolicy, active_tools_for_request,
6096 apply_registry_first_shell_guidance, build_model_tool_catalog_with_surface,
6097 default_synthetic_catalog_tool_names, execute_code_execution_tool, execute_tool_search,
6098 is_tool_search_tool, maybe_hydrate_requested_deferred_tool, missing_tool_error_message,
6099 };
6100 #[cfg(test)]
6101 use self::tool_catalog::{
6102 TOOL_SEARCH_NAME, active_tools_for_step, build_model_tool_catalog, ensure_advanced_tooling,
6103 initial_active_tools, maybe_activate_requested_deferred_tool,
6104 preflight_requested_deferred_tool, should_default_defer_tool, tool_allowed,
6105 tool_catalog_consistency_issues, tool_denied,
6106 };
6107 use self::tool_execution::emit_tool_audit;
6108 use self::tool_preparation::{prepare_tool_call, reprepare_tool_call_after_hook};
6109 use crate::tools::js_execution::execute_js_execution_tool;
6110
6111 #[cfg(test)]
6112 mod tests;
6113
6113 lines RUST