mermaid_cli/effect/

mod.rs

//! The effect runner: dispatches `Cmd` values into tokio tasks.
//!
//! There are exactly two places in the codebase that spawn a tokio
//! task: this module and tests. Everywhere else asks the
//! reducer to return a `Cmd`, and the runner handles it. That
//! centralization is what makes structured concurrency per turn
//! actually work — nothing can accidentally spawn a detached task
//! that outlives the turn it was started for.
//!
//! Architecture:
//!
//! ```text
//!   main loop ── reducer ── Cmd ── dispatch ── EffectRunner
//!                                                 ├── TurnScope(turn A) ── JoinSet
//!                                                 ├── TurnScope(turn B) ── JoinSet
//!                                                 └── detached effects (Save, Exit, …)
//!                                                       ↓
//!                                              Msg via mpsc::Sender<Msg>
//!                                                       ↓
//!                                                 main loop (next iteration)
//! ```
//!
//! The runner dispatches every `Cmd` variant to a real handler —
//! model streaming (`CallModel` → `ModelProvider::chat`), tool
//! execution (`ExecuteTool` → `ToolExecutor::execute`), persistence
//! (`SaveConversation`, `LoadConversation`, `PersistLastModel`,
//! `PersistReasoningFor`, `RefreshInstructions`), MCP lifecycle
//! (`InitMcpServers`, `StopMcpServer`), local side-effects
//! (`WriteImageToTemp`, `OpenInSystem`, `PullOllamaModel`,
//! `SetTerminalTitle`, `DismissStatusAfter`). Cancellation flows
//! through `Cmd::CancelScope(TurnId)` → the scope's
//! `CancellationToken`.

mod config_watch;
mod middleware;
mod turn_scope;

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;

use tokio::sync::mpsc;

use crate::app::{Config, MemoryConfig};
use crate::domain::{
    Cmd, CompactionPolicy, CompactionRequest, CompactionResult, CompactionTrigger, Msg, TurnId,
};
use crate::models::{ModelError, TokenUsage};
use crate::providers::ctx::{ExecContext, StreamContext};
use crate::providers::model::ModelProvider;
use crate::providers::{ProviderFactory, StreamEvent, ToolRegistry};

pub use middleware::{DEFAULT_MAX_ATTEMPTS, retry_transient_http};
pub use turn_scope::TurnScope;

#[cfg(not(test))]
const CANCEL_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
#[cfg(test)]
const CANCEL_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(50);

/// Single channel back to the reducer. `EffectRunner` holds the
/// sender; every spawned task clones this so it can emit `Msg` as
/// work progresses. Bounded capacity applies natural backpressure —
/// if the main loop can't keep up, the provider's streaming send
/// `.await`s and the whole pipeline throttles.
pub type MsgSender = mpsc::Sender<Msg>;

/// Bounded channel capacity for the effect → reducer stream. 512 is
/// generous — a single streaming chunk fits comfortably, and the
/// main loop drains at ~60 Hz so backlog rarely grows. Bigger wastes
/// RAM; smaller introduces spurious backpressure on bursty tool
/// output.
pub const MSG_CHANNEL_CAPACITY: usize = 512;

/// The runner. One instance per process, constructed by
/// `app::run` and consumed when the main loop exits.
pub struct EffectRunner {
    msg_tx: MsgSender,
    /// Per-turn scopes. Populated lazily: the first `Cmd` bearing a
    /// TurnId creates a scope; `Cmd::CancelScope` tears it down.
    /// Empty (drained) scopes are reaped by `reap_empty_scopes`, which
    /// runs at the top of every `dispatch` call so the map stays
    /// bounded across long sessions (F12).
    scopes: HashMap<TurnId, TurnScope>,
    /// Detached work (saves, persists, MCP lifecycle) lives here.
    /// This one set never gets cancelled piecemeal — shutdown drains
    /// it during `EffectRunner::shutdown`.
    detached: tokio::task::JoinSet<()>,
    /// MCP manager handle is held elsewhere (`crate::mcp` has a
    /// `OnceLock` for its global manager); we just note workdir so
    /// handlers can construct absolute paths.
    workdir: PathBuf,
    /// Lazy provider registry. `CallModel` resolves through this.
    /// Tests that don't care about real providers leave this `None`
    /// and observe the fallback `UpstreamError` Msg; production
    /// construction via `with_bindings` sets it.
    providers: Option<Arc<ProviderFactory>>,
    /// Shared tool registry. See `providers` — same optionality
    /// rationale for unit tests.
    tools: Option<Arc<ToolRegistry>>,
    /// Durable runtime task that owns work launched by this runner.
    task_id: Option<String>,
    /// Interactive TUI runners write OSC 2 terminal-title updates.
    /// Headless `mermaid run` must suppress them so stdout stays
    /// machine-readable for JSON/markdown/text output modes.
    terminal_title_enabled: bool,
    /// Inline-approval broker. `Some` only for interactive TUI runs (set via
    /// `with_interactive_approvals`); headless + child runners leave it `None`,
    /// so the gate falls back to the out-of-band DB-approval flow.
    approval: Option<crate::providers::ApprovalBroker>,
    /// Abort handle for the background config watcher (#45). It's a perpetual
    /// loop living in `detached`, so `shutdown` aborts it explicitly before
    /// draining — otherwise the drain would block on it until the timeout.
    config_watch: Option<tokio::task::AbortHandle>,
}

impl EffectRunner {
    /// Create an unused runner. Pair with `msg_rx` from `channel()`.
    pub fn new(msg_tx: MsgSender, workdir: PathBuf) -> Self {
        Self {
            msg_tx,
            scopes: HashMap::new(),
            detached: tokio::task::JoinSet::new(),
            workdir,
            providers: None,
            tools: None,
            task_id: None,
            terminal_title_enabled: true,
            approval: None,
            config_watch: None,
        }
    }

    /// Enable inline approval prompts (interactive TUI only). The gate then
    /// pauses gated tools and routes the user's decision through the
    /// `ApprovalBroker` instead of writing an out-of-band DB approval row.
    pub fn with_interactive_approvals(mut self) -> Self {
        self.approval = Some(crate::providers::ApprovalBroker::new(self.msg_tx.clone()));
        self
    }

    /// Start the background config watcher (#45): it polls `MERMAID.md` + memory
    /// and emits `Msg::InstructionsChanged`/`MemoryChanged` on change, so the
    /// reducer reads them as injected data instead of refreshing inline. Call
    /// once at startup. Live-loop only — a replay driver feeds the recorded
    /// Changed Msgs rather than polling.
    pub fn spawn_config_watcher(&mut self, cwd: PathBuf, memory: MemoryConfig) {
        let handle = self.detached.spawn(config_watch::config_watcher(
            self.msg_tx.clone(),
            cwd,
            memory,
        ));
        self.config_watch = Some(handle);
    }

    /// Attach a durable runtime task id so tool runs, approvals,
    /// checkpoints, compactions, and background processes can be linked.
    pub fn with_task_id(mut self, task_id: Option<String>) -> Self {
        self.task_id = task_id;
        self
    }

    /// Disable terminal-title writes for non-interactive callers.
    pub fn without_terminal_title(mut self) -> Self {
        self.terminal_title_enabled = false;
        self
    }

    /// Attach provider + tool registries. Production wiring uses
    /// this; unit tests that don't need real dispatch can skip.
    /// Without bindings, `CallModel` / `ExecuteTool` emit well-
    /// formed error Msgs so the reducer still transitions cleanly.
    pub fn with_bindings(
        mut self,
        providers: Arc<ProviderFactory>,
        tools: Arc<ToolRegistry>,
    ) -> Self {
        self.providers = Some(providers);
        self.tools = Some(tools);
        self
    }

    /// Pair-constructor: returns both the runner and the receiving
    /// end of the Msg channel. Preferred for production wiring
    /// because it keeps the channel capacity constant in one place.
    pub fn pair(workdir: PathBuf) -> (Self, mpsc::Receiver<Msg>) {
        let (tx, rx) = mpsc::channel(MSG_CHANNEL_CAPACITY);
        (Self::new(tx, workdir), rx)
    }

    /// Pair constructor that also wires the real provider factory +
    /// tool registry. Used by `app::run_interactive`.
    pub fn pair_with_bindings(
        workdir: PathBuf,
        config: Config,
        tools: Arc<ToolRegistry>,
    ) -> (Self, mpsc::Receiver<Msg>) {
        let providers = Arc::new(ProviderFactory::new(config));
        Self::pair_from(workdir, providers, tools)
    }

    /// Pair constructor that takes a pre-built `ProviderFactory`.
    /// Used when the caller needs to share a `ProviderFactory` with
    /// the `SubagentSpawner` so subagents can issue model calls
    /// through the same cache.
    pub fn pair_from(
        workdir: PathBuf,
        providers: Arc<ProviderFactory>,
        tools: Arc<ToolRegistry>,
    ) -> (Self, mpsc::Receiver<Msg>) {
        let (tx, rx) = mpsc::channel(MSG_CHANNEL_CAPACITY);
        (Self::new(tx, workdir).with_bindings(providers, tools), rx)
    }

    pub fn pair_from_with_task(
        workdir: PathBuf,
        providers: Arc<ProviderFactory>,
        tools: Arc<ToolRegistry>,
        task_id: Option<String>,
    ) -> (Self, mpsc::Receiver<Msg>) {
        let (runner, rx) = Self::pair_from(workdir, providers, tools);
        (runner.with_task_id(task_id), rx)
    }

    /// Construct a runner that shares a pre-derived cancellation
    /// token for its turn scopes. Used by `SubagentSpawner` so the
    /// child runner's work aborts as soon as the parent's `ctx.token`
    /// fires.
    pub fn new_child(
        msg_tx: MsgSender,
        workdir: PathBuf,
        providers: Arc<ProviderFactory>,
        tools: Arc<ToolRegistry>,
    ) -> Self {
        // A subagent's runner is never the interactive top-level, so it must
        // NOT emit OSC 2 terminal-title escapes: in a headless `mermaid run`
        // the parent suppresses them, but an un-suppressed child leaks
        // `\x1b]2;…\x07` into stdout and corrupts `--format json`/`text` output.
        Self::new(msg_tx, workdir)
            .with_bindings(providers, tools)
            .without_terminal_title()
    }

    /// Get or create the scope for a turn. Idempotent. The scope is
    /// retained until `CancelScope` tears it down or it naturally
    /// drains.
    fn scope_mut(&mut self, turn: TurnId) -> &mut TurnScope {
        self.scopes
            .entry(turn)
            .or_insert_with(|| TurnScope::new(turn))
    }

    /// Drop the scope for a turn, signalling cancellation to every
    /// child first. Safe to call for non-existent turns.
    ///
    /// After the scope is cancelled, a detached task moves it off the
    /// runner, drains its `JoinSet` (so child tasks unwind), then emits
    /// `Msg::TurnCancelled(turn)` so the reducer can transition
    /// `Cancelling → Idle`. Without this terminal event the TUI would
    /// stick in `Cancelling` — the reducer has no other way to learn
    /// that the abort fully landed.
    fn drop_scope(&mut self, turn: TurnId) {
        if let Some(mut scope) = self.scopes.remove(&turn) {
            scope.cancel();
            let tx = self.msg_tx.clone();
            self.detached.spawn(async move {
                if tokio::time::timeout(CANCEL_DRAIN_TIMEOUT, scope.drain())
                    .await
                    .is_err()
                {
                    tracing::warn!(
                        turn = %turn,
                        timeout_ms = CANCEL_DRAIN_TIMEOUT.as_millis(),
                        "cancel drain timed out; aborting remaining scoped tasks"
                    );
                }
                let _ = tx.send(Msg::TurnCancelled(turn)).await;
            });
        } else {
            // The scope was already reaped — its `JoinSet` drained to empty
            // and `reap_empty_scopes` (top of `dispatch`) removed it before
            // this cancel landed. The reducer is still in `Cancelling` with
            // no other way to learn the turn ended, so emit the terminal
            // event anyway. Idempotent: `handle_turn_cancelled` no-ops on
            // any turn that isn't currently `Cancelling`.
            let tx = self.msg_tx.clone();
            self.detached.spawn(async move {
                let _ = tx.send(Msg::TurnCancelled(turn)).await;
            });
        }
    }

    /// Number of active per-turn scopes. Tests use this to observe
    /// lifecycle without racing on internal state.
    pub fn scope_count(&self) -> usize {
        self.scopes.len()
    }

    /// F12: remove scope entries whose `JoinSet` is empty — every
    /// child task has completed, so the scope is just an orphan key
    /// in the map. Called at the top of `dispatch` so the map stays
    /// bounded over long sessions. Cheap: one linear walk, no async.
    ///
    /// `JoinSet::is_empty` only returns true after completed tasks are
    /// harvested via `join_next`/`try_join_next`, so we first drain
    /// any ready completions per scope.
    fn reap_empty_scopes(&mut self) {
        self.reap_detached();
        self.scopes.retain(|_, scope| {
            scope.drain_completed();
            !scope.is_empty()
        });
    }

    /// Harvest finished detached tasks. Without this the `detached` JoinSet
    /// grows for the whole session (every fire-and-forget effect lingers as a
    /// completed-but-unjoined handle), and a panicking detached task vanishes
    /// without a trace. Non-blocking — only already-finished tasks are taken (#38).
    fn reap_detached(&mut self) {
        while let Some(result) = self.detached.try_join_next() {
            if let Err(e) = result
                && !e.is_cancelled()
            {
                tracing::warn!(error = %e, "effect: detached task panicked");
            }
        }
    }

    /// Route a single `Cmd` into the appropriate spawn + handler.
    /// Returns immediately; handlers work asynchronously and emit
    /// `Msg` back through the sender channel.
    pub fn dispatch(&mut self, cmd: Cmd) {
        // F12: reap any drained scopes before touching the map. Keeps
        // `scope_count()` bounded as the session grows.
        self.reap_empty_scopes();
        tracing::trace!(cmd = %cmd.summary(), "effect: dispatch");

        match cmd {
            Cmd::CallModel { turn, mut request } => {
                let tx = self.msg_tx.clone();
                let providers = self.providers.clone();
                // Enrich `request.tools` with every user-facing
                // tool in the bound registry. The reducer has
                // already populated MCP tools from `state.mcp`;
                // built-ins come from the runner (which holds the
                // registry). This keeps `ChatRequest.tools` the
                // single source of truth for what the model sees.
                if let Some(tools) = &self.tools {
                    let mut enriched = tools.describe_all();
                    // Report the built-in tool-schema token cost so the
                    // reducer's /context preview can fold it into its MCP-only
                    // estimate and agree with what the model actually sees.
                    let builtin_tokens = crate::domain::estimate_tool_schema_tokens(&enriched);
                    let _ = tx.try_send(Msg::BuiltinToolSchemaTokens(builtin_tokens));
                    enriched.append(&mut request.tools);
                    request.tools = enriched;
                }
                // Detached + off the blocking pool: never run a plugin hook on
                // the synchronous dispatch path (it would freeze input/render).
                self.detached.spawn(fire_plugin_hooks(
                    "prompt_submit",
                    serde_json::json!({
                        "turn_id": turn.0,
                        "model_id": request.model_id.clone(),
                        "message_count": request.messages.len(),
                        "tool_count": request.tools.len(),
                    }),
                ));
                let scope = self.scope_mut(turn);
                let token = scope.token();
                scope.spawn(async move {
                    use futures::FutureExt;
                    let fallback_tx = tx.clone();
                    if std::panic::AssertUnwindSafe(dispatch_call_model(
                        tx, providers, turn, request, token,
                    ))
                    .catch_unwind()
                    .await
                    .is_err()
                    {
                        // The dispatch task panicked. A turn whose model call
                        // never emits a terminal Msg stays in `Generating`
                        // forever; emit one so the reducer can leave that state
                        // instead of wedging (#43).
                        tracing::error!(turn = %turn, "dispatch_call_model panicked");
                        let _ = fallback_tx
                            .send(Msg::UpstreamError {
                                turn,
                                error: crate::models::UserFacingError {
                                    summary: "Internal error".to_string(),
                                    message: "The model dispatch task panicked unexpectedly."
                                        .to_string(),
                                    suggestion: "This is a bug. Please retry; if it persists, \
                                                 check the logs."
                                        .to_string(),
                                    category: crate::models::ErrorCategory::Internal,
                                    recoverable: true,
                                },
                            })
                            .await;
                    }
                });
            },
            Cmd::CompactConversation { turn, mut request } => {
                let tx = self.msg_tx.clone();
                let providers = self.providers.clone();
                if let Some(tools) = &self.tools {
                    let mut enriched = tools.describe_all();
                    enriched.append(&mut request.chat.tools);
                    request.chat.tools = enriched;
                }
                let scope = self.scope_mut(turn);
                let token = scope.token();
                scope.spawn(async move {
                    dispatch_compact_conversation(tx, providers, turn, request, token).await;
                });
            },
            Cmd::ExecuteTool {
                turn,
                call_id,
                source,
                model_id,
                safety_mode,
                intent,
            } => {
                let tx = self.msg_tx.clone();
                let tools = self.tools.clone();
                let workdir = self.workdir.clone();
                // Pass the shared Config from ProviderFactory so
                // subagents inherit it (F7). Falls back to
                // Config::default() when providers aren't bound (unit
                // tests without real wiring).
                let config = self
                    .providers
                    .as_ref()
                    .map(|p| Arc::new(p.config().clone()))
                    .unwrap_or_else(|| Arc::new(crate::app::Config::default()));
                // Auto mode: build an LLM classifier to vet borderline
                // actions. Only when a provider is bound (real wiring); the
                // gate fails safe to "escalate" when it's `None`. The vet
                // uses the configured classifier model, else the session model.
                let classifier: Option<Arc<dyn crate::providers::AutoClassifier>> =
                    if safety_mode == crate::runtime::SafetyMode::Auto {
                        self.providers.as_ref().map(|p| {
                            let model = config
                                .safety
                                .auto_classifier_model
                                .clone()
                                .unwrap_or_else(|| model_id.clone());
                            Arc::new(crate::providers::ModelAutoClassifier::new(p.clone(), model))
                                as Arc<dyn crate::providers::AutoClassifier>
                        })
                    } else {
                        None
                    };
                let task_id = self.task_id.clone();
                let approval = self.approval.clone();
                let scope = self.scope_mut(turn);
                let token = scope.token();
                let background = scope.background_token();
                scope.spawn(async move {
                    use futures::FutureExt;
                    let fallback_tx = tx.clone();
                    if std::panic::AssertUnwindSafe(dispatch_execute_tool(
                        tx,
                        tools,
                        workdir,
                        turn,
                        call_id,
                        source,
                        token,
                        background,
                        config,
                        model_id,
                        task_id,
                        safety_mode,
                        intent,
                        classifier,
                        approval,
                    ))
                    .catch_unwind()
                    .await
                    .is_err()
                    {
                        // The tool task panicked. Its turn waits on a
                        // `ToolFinished` for this `call_id` that will now never
                        // arrive; emit a terminal error outcome so the turn
                        // doesn't wedge (#43).
                        tracing::error!(
                            turn = %turn,
                            call_id = call_id.0,
                            "dispatch_execute_tool panicked"
                        );
                        let _ = fallback_tx
                            .send(Msg::ToolFinished {
                                turn,
                                call_id,
                                outcome: crate::domain::ToolOutcome::error(
                                    "internal error: the tool execution task panicked".to_string(),
                                    0.0,
                                ),
                            })
                            .await;
                    }
                });
            },
            Cmd::ResolveApproval { call_id, decision } => {
                // Deliver the user's inline decision to the parked tool task.
                // Not turn-scoped — fire-and-forget to the broker.
                if let Some(broker) = &self.approval {
                    broker.resolve(call_id, decision.into());
                }
            },
            Cmd::CancelScope(turn) => {
                self.drop_scope(turn);
            },
            Cmd::BackgroundScope(turn) => {
                // Fire the scope's background token (don't drop the scope):
                // detachable tools move their child to a background process and
                // return a normal outcome, so the turn finishes naturally.
                self.scope_mut(turn).background();
            },
            Cmd::SaveConversation(history) => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    if let Ok(manager) = crate::session::ConversationManager::new(&workdir)
                        && manager.save_conversation(&history).is_ok()
                    {
                        let _ = tx.send(Msg::SessionSaved).await;
                    } else {
                        tracing::warn!("SaveConversation: failed to write to disk");
                    }
                });
            },
            Cmd::SaveCompactionArchive {
                archive,
                record,
                conversation,
            } => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                let task_id = self.task_id.clone();
                self.detached.spawn(async move {
                    let Ok(manager) = crate::session::ConversationManager::new(&workdir) else {
                        return;
                    };
                    // Archive FIRST — it is the only durable copy of the
                    // dropped messages. Only overwrite the (stripped)
                    // conversation once the archive has persisted, so a
                    // failed archive can never lose messages.
                    match manager.save_compaction_archive(&archive) {
                        Ok(path) => {
                            if let Ok(store) = crate::runtime::RuntimeStore::open_default() {
                                let compaction_id = record.id.clone();
                                let conversation_id = archive.conversation_id.clone();
                                let _ =
                                    store.compactions().create(crate::runtime::NewCompaction {
                                        id: Some(record.id),
                                        task_id: task_id.clone(),
                                        session_id: Some(archive.conversation_id),
                                        source_token_estimate: Some(record.before_tokens as i64),
                                        summary_token_count: Some(record.summary_tokens as i64),
                                        preserved_turns: Some(record.preserved_message_count as i64),
                                        archive_path: Some(path.display().to_string()),
                                        verification_status: Some("verified".to_string()),
                                    });
                                fire_plugin_hooks(
                                    "compaction",
                                    serde_json::json!({
                                        "id": compaction_id,
                                        "task_id": task_id.clone(),
                                        "session_id": conversation_id,
                                        "archive_path": path.display().to_string(),
                                    }),
                                )
                                .await;
                            }
                            if manager.save_conversation(&conversation).is_ok() {
                                let _ = tx.send(Msg::SessionSaved).await;
                            } else {
                                tracing::warn!(
                                    "SaveCompactionArchive: archive persisted but conversation save failed"
                                );
                            }
                        },
                        Err(err) => {
                            tracing::warn!(
                                error = %err,
                                "SaveCompactionArchive: archive write failed; NOT overwriting conversation",
                            );
                        },
                    }
                });
            },
            Cmd::SaveProcess(process) => {
                let task_id = self.task_id.clone();
                self.detached.spawn(async move {
                    let status = match process.status {
                        crate::domain::ManagedProcessStatus::Running => {
                            crate::runtime::ProcessStatus::Running
                        },
                        crate::domain::ManagedProcessStatus::Exited => {
                            crate::runtime::ProcessStatus::Exited
                        },
                        crate::domain::ManagedProcessStatus::Unknown => {
                            crate::runtime::ProcessStatus::Unknown
                        },
                    };
                    if let Ok(store) = crate::runtime::RuntimeStore::open_default() {
                        let _ = store.processes().upsert(crate::runtime::NewProcess {
                            id: Some(process.id),
                            task_id,
                            pid: process.pid,
                            command: process.command,
                            cwd: process.cwd,
                            log_path: Some(process.log_path),
                            detected_url: process.detected_url,
                            status,
                            health: None,
                        });
                    }
                });
            },
            Cmd::PersistLastModel(model) => {
                self.detached.spawn(async move {
                    let _ = crate::app::persist_last_model(&model);
                });
            },
            Cmd::PersistReasoningFor { model_id, level } => {
                self.detached.spawn(async move {
                    let _ = crate::app::persist_reasoning_for_model(&model_id, level);
                });
            },
            Cmd::PersistOllamaNumCtxFor { model_id, num_ctx } => {
                self.detached.spawn(async move {
                    let _ = crate::app::persist_ollama_num_ctx_for_model(&model_id, num_ctx);
                });
            },
            Cmd::PersistOllamaOffload(enabled) => {
                self.detached.spawn(async move {
                    let _ = crate::app::persist_ollama_allow_ram_offload(enabled);
                });
            },
            Cmd::RefreshInstructions => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    let (loaded, _outcome) = crate::app::instructions::refresh(None, &workdir);
                    let _ = tx.send(Msg::InstructionsChanged(loaded)).await;
                });
            },
            Cmd::RefreshMemory => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    let cfg = crate::app::load_config().unwrap_or_default().memory;
                    let (loaded, _) = crate::app::memory::refresh(None, &workdir, &cfg);
                    let _ = tx.send(Msg::MemoryChanged(loaded)).await;
                });
            },
            Cmd::ListMemory => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    let cfg = crate::app::load_config().unwrap_or_default().memory;
                    let text = match crate::app::memory::load(&workdir, &cfg) {
                        Some(mem) => mem.index,
                        None => "No memories saved yet. Durable facts (yours or mine) show up here — use `/remember <fact>` or just ask me to remember something.".to_string(),
                    };
                    let _ = tx.send(Msg::RuntimeText(text)).await;
                });
            },
            Cmd::RememberMemory { text } => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    let cfg = crate::app::load_config().unwrap_or_default().memory;
                    let name = memory_title_from_text(&text);
                    let (status, kind) = match crate::app::memory::write_memory(
                        &workdir,
                        crate::app::memory::MemoryScope::ProjectPrivate,
                        &name,
                        &text,
                        &[],
                        &text,
                    ) {
                        Ok(_) => (
                            format!("Remembered: {name}"),
                            crate::domain::StatusKind::Info,
                        ),
                        Err(e) => (
                            format!("Couldn't save memory: {e}"),
                            crate::domain::StatusKind::Error,
                        ),
                    };
                    let (loaded, _) = crate::app::memory::refresh(None, &workdir, &cfg);
                    let _ = tx.send(Msg::MemoryChanged(loaded)).await;
                    let _ = tx
                        .send(Msg::TransientStatus {
                            text: status,
                            kind,
                            dismiss_ms: 3_000,
                        })
                        .await;
                });
            },
            Cmd::ForgetMemory { id } => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    let cfg = crate::app::load_config().unwrap_or_default().memory;
                    let (status, kind) = match crate::app::memory::delete_memory(&workdir, &id) {
                        Ok(Some(_)) => (format!("Forgot: {id}"), crate::domain::StatusKind::Info),
                        Ok(None) => (
                            format!("No memory named '{id}'"),
                            crate::domain::StatusKind::Info,
                        ),
                        Err(e) => (
                            format!("Couldn't forget memory: {e}"),
                            crate::domain::StatusKind::Error,
                        ),
                    };
                    let (loaded, _) = crate::app::memory::refresh(None, &workdir, &cfg);
                    let _ = tx.send(Msg::MemoryChanged(loaded)).await;
                    let _ = tx
                        .send(Msg::TransientStatus {
                            text: status,
                            kind,
                            dismiss_ms: 3_000,
                        })
                        .await;
                });
            },
            Cmd::ConsolidateMemory { model_id } => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                let providers = self.providers.clone();
                self.detached.spawn(async move {
                    consolidate_memory(tx, providers, workdir, model_id).await;
                });
            },
            Cmd::LoadConversation(id) => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    match crate::session::ConversationManager::new(&workdir) {
                        Ok(mgr) => match mgr.load_conversation(&id) {
                            Ok(history) => {
                                let _ = tx.send(Msg::ConversationLoaded(history)).await;
                            },
                            Err(e) => {
                                tracing::warn!(id = %id, error = %e, "LoadConversation failed");
                            },
                        },
                        Err(e) => {
                            tracing::warn!(error = %e, "ConversationManager init failed");
                        },
                    }
                });
            },
            Cmd::ListConversations => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn(async move {
                    let summaries = match crate::session::ConversationManager::new(&workdir) {
                        Ok(mgr) => mgr
                            .list_conversations()
                            .unwrap_or_default()
                            .into_iter()
                            .map(|h| crate::domain::ConversationSummary {
                                id: h.id.clone(),
                                title: h.title.clone(),
                                message_count: h.messages.len(),
                                updated_at: h.updated_at.to_rfc3339(),
                            })
                            .collect(),
                        Err(_) => Vec::new(),
                    };
                    let _ = tx.send(Msg::ConversationsListed(summaries)).await;
                });
            },
            Cmd::ListRuntimeTasks { limit } => {
                let tx = self.msg_tx.clone();
                // Synchronous rusqlite read — run on the blocking pool so it
                // never stalls an async worker thread (#40).
                self.detached.spawn_blocking(move || {
                    let tasks = crate::runtime::RuntimeClient::auto()
                        .list_tasks(limit)
                        .map(|read| read.value)
                        .unwrap_or_default();
                    let _ = tx.blocking_send(Msg::RuntimeTasksListed(tasks));
                });
            },
            Cmd::LoadRuntimeTask { id } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let (task, events) = crate::runtime::RuntimeClient::auto()
                        .task_detail(&id)
                        .map(|read| (Some(read.value.task), read.value.events))
                        .unwrap_or((None, Vec::new()));
                    let _ = tx.blocking_send(Msg::RuntimeTaskLoaded { task, events });
                });
            },
            Cmd::ListRuntimeProcesses { limit } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let processes = crate::runtime::RuntimeClient::auto()
                        .list_processes(limit)
                        .map(|read| read.value)
                        .unwrap_or_default();
                    let _ = tx.blocking_send(Msg::RuntimeProcessesListed(processes));
                });
            },
            Cmd::ShowRuntimeProcessLogs { id } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let text = crate::runtime::RuntimeClient::auto()
                        .process_log(&id, None)
                        .map(|log| format!("Process log {}\n\n{}", id, log.content))
                        .unwrap_or_else(|err| format!("Process log error: {}", err));
                    let _ = tx.blocking_send(Msg::RuntimeText(text));
                });
            },
            Cmd::StopRuntimeProcess { id } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let msg = match crate::runtime::RuntimeClient::auto().stop_process(&id) {
                        Ok(response) => Msg::TransientStatus {
                            text: format!("Stopped process {} (pid {})", id, response.item.pid),
                            kind: crate::domain::StatusKind::Info,
                            dismiss_ms: 3_000,
                        },
                        Err(err) => Msg::TransientStatus {
                            text: format!("Process stop failed: {}", err),
                            kind: crate::domain::StatusKind::Warn,
                            dismiss_ms: 5_000,
                        },
                    };
                    let _ = tx.blocking_send(msg);
                });
            },
            Cmd::RestartRuntimeProcess { id } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let msg = match crate::runtime::RuntimeClient::auto().restart_process(&id) {
                        Ok(response) => Msg::TransientStatus {
                            text: format!("Restarted process {} (pid {})", id, response.item.pid),
                            kind: crate::domain::StatusKind::Info,
                            dismiss_ms: 3_000,
                        },
                        Err(err) => Msg::TransientStatus {
                            text: format!("Process restart failed: {}", err),
                            kind: crate::domain::StatusKind::Warn,
                            dismiss_ms: 5_000,
                        },
                    };
                    let _ = tx.blocking_send(msg);
                });
            },
            Cmd::OpenRuntimeTarget { target } => {
                self.detached.spawn_blocking(move || {
                    let resolved = crate::runtime::RuntimeService::open_default()
                        .and_then(|service| service.resolve_open_target(&target))
                        .unwrap_or(target);
                    // #63: the resolved value can be a `detected_url`/`log_path`
                    // from a `processes` row — validate before the OS opener,
                    // exactly like `open_process`.
                    if let Err(err) = crate::runtime::validate_open_target(&resolved) {
                        tracing::warn!(error = %err, "refusing to open runtime target");
                        return;
                    }
                    crate::utils::open_file(resolved);
                });
            },
            Cmd::ShowRuntimePorts => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let text = crate::runtime::RuntimeClient::auto()
                        .ports()
                        .map(|ports| format!("Listening TCP ports\n\n{}", ports.ports))
                        .unwrap_or_else(|err| format!("Port inspection failed: {}", err));
                    let _ = tx.blocking_send(Msg::RuntimeText(text));
                });
            },
            Cmd::ListRuntimeApprovals => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let approvals = crate::runtime::RuntimeClient::auto()
                        .list_approvals()
                        .map(|read| read.value)
                        .unwrap_or_default();
                    let _ = tx.blocking_send(Msg::RuntimeApprovalsListed(approvals));
                });
            },
            Cmd::DecideRuntimeApproval { id, decision } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let result = if decision == "approved" {
                        crate::runtime::RuntimeClient::auto().approve(&id)
                    } else {
                        crate::runtime::RuntimeClient::auto().deny(&id)
                    };
                    let msg = match result {
                        Ok(result) => Msg::TransientStatus {
                            text: if result.replayed {
                                format!("Approval {} {}: {}", id, decision, result.summary)
                            } else {
                                format!("Approval {} {}", id, decision)
                            },
                            kind: crate::domain::StatusKind::Info,
                            dismiss_ms: 3_000,
                        },
                        Err(err) => Msg::TransientStatus {
                            text: format!("Approval update failed: {}", err),
                            kind: crate::domain::StatusKind::Warn,
                            dismiss_ms: 5_000,
                        },
                    };
                    let _ = tx.blocking_send(msg);
                });
            },
            Cmd::ListRuntimeCheckpoints { limit } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let checkpoints = crate::runtime::RuntimeClient::auto()
                        .list_checkpoints(limit)
                        .map(|read| read.value)
                        .unwrap_or_default();
                    let _ = tx.blocking_send(Msg::RuntimeCheckpointsListed(checkpoints));
                });
            },
            Cmd::ListRuntimePlugins => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let plugins = crate::runtime::RuntimeClient::auto()
                        .list_plugins()
                        .map(|read| read.value)
                        .unwrap_or_default();
                    let _ = tx.blocking_send(Msg::RuntimePluginsListed(plugins));
                });
            },
            Cmd::UpdateRuntimeTaskStatus {
                id,
                status,
                final_report,
            } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let msg = match crate::runtime::RuntimeStore::open_default().and_then(|store| {
                        store
                            .tasks()
                            .update_status(&id, status, final_report.as_deref())
                    }) {
                        Ok(()) => Msg::TransientStatus {
                            text: format!("Task {} -> {}", id, status),
                            kind: crate::domain::StatusKind::Info,
                            dismiss_ms: 3_000,
                        },
                        Err(err) => Msg::TransientStatus {
                            text: format!("Task update failed: {}", err),
                            kind: crate::domain::StatusKind::Warn,
                            dismiss_ms: 4_000,
                        },
                    };
                    let _ = tx.blocking_send(msg);
                });
            },
            Cmd::CreateRuntimeCheckpoint { paths } => {
                let tx = self.msg_tx.clone();
                let workdir = self.workdir.clone();
                self.detached.spawn_blocking(move || {
                    let pending_action = Some(serde_json::json!({
                        "source": "tui",
                        "command": "checkpoint",
                    }));
                    let msg =
                        match crate::runtime::create_checkpoint(&workdir, &paths, pending_action) {
                            Ok(manifest) => Msg::TransientStatus {
                                text: format!(
                                    "Checkpoint {} created for {} path(s)",
                                    manifest.id,
                                    manifest.files.len()
                                ),
                                kind: crate::domain::StatusKind::Info,
                                dismiss_ms: 4_000,
                            },
                            Err(err) => Msg::TransientStatus {
                                text: format!("Checkpoint failed: {}", err),
                                kind: crate::domain::StatusKind::Warn,
                                dismiss_ms: 5_000,
                            },
                        };
                    let _ = tx.blocking_send(msg);
                });
            },
            Cmd::RestoreRuntimeCheckpoint { id } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let msg = match crate::runtime::RuntimeClient::auto().restore_checkpoint(&id) {
                        Ok(result) => Msg::TransientStatus {
                            text: format!(
                                "Restored checkpoint {} ({} file(s)){}",
                                result.checkpoint.id,
                                result.checkpoint.files.len(),
                                if result.checkpoint.pending_action.is_some() {
                                    "; pending action available in checkpoint manifest"
                                } else {
                                    ""
                                }
                            ),
                            kind: crate::domain::StatusKind::Info,
                            dismiss_ms: 4_000,
                        },
                        Err(err) => Msg::TransientStatus {
                            text: format!("Restore failed: {}", err),
                            kind: crate::domain::StatusKind::Warn,
                            dismiss_ms: 5_000,
                        },
                    };
                    let _ = tx.blocking_send(msg);
                });
            },
            Cmd::ShowRuntimeModelInfo { model } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn_blocking(move || {
                    let text = runtime_model_info_text(&model);
                    let _ = tx.blocking_send(Msg::RuntimeText(text));
                });
            },
            Cmd::InitMcpServers(configs) => {
                let tx = self.msg_tx.clone();
                self.detached.spawn(async move {
                    if configs.is_empty() {
                        return;
                    }
                    crate::mcp::manager_ref::mark_init_started();
                    let manager =
                        std::sync::Arc::new(crate::mcp::McpServerManager::start(&configs).await);
                    // Emit a Ready or Errored per server based on
                    // what came up. Zero-tool servers are still ready
                    // if the manager has an active client for them.
                    for (name, _cfg) in configs.iter() {
                        let server_tools: Vec<crate::domain::McpToolSpec> = manager
                            .get_all_tools()
                            .iter()
                            .filter(|(server, _)| server == name)
                            .map(|(_, def)| crate::domain::McpToolSpec {
                                name: def.name.clone(),
                                description: def.description.clone(),
                                input_schema: def.input_schema.clone(),
                            })
                            .collect();
                        let msg = mcp_startup_msg(name, manager.has_server(name), server_tools);
                        let _ = tx.send(msg).await;
                    }
                    crate::mcp::manager_ref::set_manager(manager);
                    crate::mcp::manager_ref::mark_init_complete();
                });
            },
            Cmd::StopMcpServer { name } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn(async move {
                    // Actually kill the child before claiming it's stopped —
                    // otherwise the UI says "stopped" while the server runs on.
                    if let Some(mgr) = crate::mcp::manager_ref::get() {
                        mgr.stop_server(&name).await;
                    }
                    let _ = tx.send(Msg::McpServerStopped { name }).await;
                });
            },
            Cmd::PullOllamaModel { model } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn(async move {
                    dispatch_pull_ollama_model(tx, model).await;
                });
            },
            Cmd::OpenInSystem(path) => {
                self.detached.spawn(async move {
                    let _ = tokio::task::spawn_blocking(move || {
                        crate::utils::open_file(&path);
                    })
                    .await;
                });
            },
            Cmd::DismissStatusAfter { ms } => {
                let tx = self.msg_tx.clone();
                self.detached.spawn(async move {
                    tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
                    let _ = tx.send(Msg::StatusDismiss).await;
                });
            },
            Cmd::WriteImageToTemp {
                path,
                bytes,
                format: _,
            } => {
                self.detached.spawn(async move {
                    if let Err(e) = tokio::fs::write(&path, &bytes).await {
                        tracing::warn!(path = %path.display(), error = %e, "WriteImageToTemp failed");
                    }
                });
            },
            Cmd::ReadClipboard => {
                let tx = self.msg_tx.clone();
                self.detached.spawn(async move {
                    dispatch_read_clipboard(tx).await;
                });
            },
            Cmd::CopyToClipboard(text) => {
                let tx = self.msg_tx.clone();
                self.detached.spawn(async move {
                    dispatch_copy_to_clipboard(text, tx).await;
                });
            },
            Cmd::Exit => {
                // The main loop observes `state.should_exit` after
                // the reducer returns; the runner doesn't need to
                // take any special action. Documented here for
                // exhaustiveness.
            },
            Cmd::SetTerminalTitle(title) => {
                if !self.terminal_title_enabled {
                    return;
                }
                // Offload the terminal write to the blocking pool: writing to
                // stdout can block when the terminal (or a downstream pipe) is
                // slow, and an async worker must not block on it (#44). The
                // OSC-2 title sequence is out-of-band relative to the renderer's
                // frame draws, so it doesn't corrupt them.
                self.detached.spawn_blocking(move || {
                    use std::io::Write;
                    let seq = format!("\x1b]2;{}\x07", title);
                    let mut stdout = std::io::stdout();
                    let _ = stdout.write_all(seq.as_bytes());
                    let _ = stdout.flush();
                });
            },
        }
    }

    /// Async shutdown: cancel every scope, then wait for all spawned
    /// work to drain. Bounded by 5 seconds — a hung task past that
    /// gets aborted outright by `JoinSet::drop`.
    pub async fn shutdown(mut self) {
        for (id, scope) in self.scopes.iter() {
            tracing::debug!(turn = %id, "shutdown: cancelling scope");
            scope.cancel();
        }

        // The config watcher (#45) is a perpetual loop in `detached`; abort it
        // so the drain below doesn't block on it until the bounded timeout.
        if let Some(handle) = self.config_watch.take() {
            handle.abort();
        }

        // Drain with a bounded timeout.
        let shutdown_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);

        let drain = async {
            // If an MCP init is still in flight, its child processes are already
            // spawned but `set_manager` hasn't run yet — `get()` below would
            // return `None` and we'd leak those children. Wait (bounded) for init
            // to settle so the manager is installed before we reap it (#59).
            let _ = tokio::time::timeout(
                std::time::Duration::from_secs(2),
                crate::mcp::manager_ref::wait_ready(),
            )
            .await;
            // Gracefully shut down MCP server children (the stdin-EOF →
            // terminate → kill ladder in `McpServerManager::shutdown`). The
            // manager lives in a `'static OnceLock` that never drops, so this
            // explicit call on the exit path is the only thing that reaps those
            // child processes. No-op when no servers were configured.
            if let Some(mgr) = crate::mcp::manager_ref::get() {
                mgr.shutdown().await;
            }
            for (_, mut scope) in self.scopes.drain() {
                scope.drain().await;
            }
            while let Some(result) = self.detached.join_next().await {
                if let Err(e) = result
                    && !e.is_cancelled()
                {
                    tracing::warn!(error = %e, "shutdown: detached task panic");
                }
            }
        };

        let _ = tokio::time::timeout_at(shutdown_deadline, drain).await;
    }

    /// Test helper: clone the Msg sender so a test can synthesize a
    /// message as if it came from an effect handler.
    #[doc(hidden)]
    pub fn msg_sender(&self) -> MsgSender {
        self.msg_tx.clone()
    }
}

/// Dispatch a `CallModel` command. Resolves the provider (lazy,
/// cached) and streams its events onto the Msg channel. Without a
/// bound `ProviderFactory` (unit tests), emits a single
/// `UpstreamError` so the reducer ends the turn cleanly.
/// Await a sibling relay task's handle, logging a panic (but not a normal
/// post-cancellation abort). These relays run alongside the provider/tool call
/// for streaming backpressure; awaiting their handle keeps a stray panic from
/// vanishing the way a bare `let _ = handle.await` would (#58, #60).
async fn join_logged(handle: tokio::task::JoinHandle<()>, what: &str) {
    if let Err(e) = handle.await
        && !e.is_cancelled()
    {
        tracing::warn!(error = %e, task = what, "effect: sibling relay task panicked");
    }
}

async fn dispatch_call_model(
    msg_tx: MsgSender,
    providers: Option<Arc<ProviderFactory>>,
    turn: TurnId,
    mut request: crate::domain::ChatRequest,
    token: tokio_util::sync::CancellationToken,
) {
    use crate::models::UserFacingError;

    let Some(factory) = providers else {
        let error = UserFacingError {
            summary: "not wired".to_string(),
            message: "EffectRunner has no ProviderFactory bound".to_string(),
            suggestion: "construct via EffectRunner::pair_with_bindings".to_string(),
            category: crate::models::ErrorCategory::Internal,
            recoverable: false,
        };
        let _ = msg_tx.send(Msg::UpstreamError { turn, error }).await;
        return;
    };

    // Lazily resolve the provider for this model.
    let provider = match factory.resolve(&request.model_id).await {
        Ok(p) => p,
        Err(e) => {
            let error = classify_error_for_ui(&e);
            let _ = msg_tx.send(Msg::UpstreamError { turn, error }).await;
            return;
        },
    };
    {
        // Telemetry write — offload the synchronous DB upserts to the blocking
        // pool so they never stall this model-call dispatch path, which runs on
        // every turn (#39).
        let model_id = request.model_id.clone();
        let caps = provider.capabilities().clone();
        tokio::task::spawn_blocking(move || record_provider_capabilities(&model_id, &caps));
    }
    if !request.tools.is_empty() && !provider.capabilities().supports_tools {
        let _ = msg_tx
            .send(Msg::TransientStatus {
                text: format!(
                    "{} does not advertise tool support; Mermaid will send the turn without tools",
                    request.model_id
                ),
                kind: crate::domain::StatusKind::Warn,
                dismiss_ms: 6_000,
            })
            .await;
        request.tools.clear();
    }

    // Resolve the *effective* context window. For Ollama this probes the model's
    // real window and auto-fits num_ctx to memory (cache-first, off the UI
    // thread); for other providers it's the static advertised window. Using the
    // effective value here is what un-skips auto-compaction for Ollama (which had
    // `NoKnownContextLimit`) and gives the status bar real numbers.
    let sizing = provider.resolve_context_window(&request).await;
    let max_context_tokens = sizing.effective.or_else(|| {
        crate::domain::runtime::infer_static_context_window_for_model_id(&request.model_id)
    });
    // Report the resolved window to the reducer for the `/context` display +
    // truncation quick-fix. Harmless for non-Ollama (source is None → no extra
    // detail shown).
    let _ = msg_tx
        .send(Msg::ProviderContextResolved {
            model_max: sizing.model_max,
            effective: sizing.effective,
            source: sizing.source,
        })
        .await;
    let context_snapshot =
        crate::domain::estimate_context_usage_for_request(&request, max_context_tokens);
    let _ = msg_tx
        .send(Msg::ContextUsageEstimated {
            turn,
            snapshot: context_snapshot.clone(),
        })
        .await;

    let policy = CompactionPolicy::default();
    let mut compacted_before_stream = false;
    if crate::domain::should_auto_compact(&context_snapshot, &request, policy).is_ok() {
        let compaction = CompactionRequest::auto(request.clone(), CompactionTrigger::AutoThreshold);
        match run_compaction(
            Arc::clone(&provider),
            turn,
            compaction,
            context_snapshot.clone(),
            max_context_tokens,
            token.clone(),
        )
        .await
        {
            Ok(result) => {
                request.messages = result.replacement_messages.clone();
                compacted_before_stream = true;
                let _ = msg_tx.send(Msg::CompactionFinished { turn, result }).await;
            },
            Err(err) => {
                // Auto-compaction is best-effort. If it can't reduce the
                // context — the estimate is roughest exactly at the limit, so
                // a large preserved tail can read `after >= before` — don't
                // kill the turn. Log it, surface a soft warning, and proceed
                // with the original request; the provider's own context limit
                // is the real gate. (Manual `/compact` keeps its hard error
                // via `run_compaction`'s reduction guard.)
                if token.is_cancelled() {
                    return;
                }
                tracing::warn!(
                    turn = %turn,
                    error = %err,
                    "auto-compaction failed; proceeding with the un-compacted request",
                );
                let _ = msg_tx
                    .send(Msg::CompactionFailed {
                        turn,
                        trigger: CompactionTrigger::AutoThreshold,
                        message: err.to_string(),
                        kind: crate::domain::StatusKind::Warn,
                    })
                    .await;
            },
        }
    }

    // Build a StreamContext — provider writes typed events into the
    // internal sink; we relay each to the reducer as a Msg.
    let (stream_tx, mut stream_rx) = mpsc::channel::<StreamEvent>(256);
    let ctx = StreamContext::new(token.clone(), stream_tx, turn);

    // Drain stream events into Msgs on a sibling task. Ends when the sink
    // closes (provider's final `Done` or completion) OR the turn token is
    // cancelled — `select!`ing on the token ties this relay to the turn's
    // structured cancellation so a cancel drops it within a tick instead of
    // waiting on the next event. (A separate task is required: the relay must
    // run concurrently with `provider.chat` for streaming backpressure.)
    let relay_tx = msg_tx.clone();
    let relay_token = token.clone();
    let relay = tokio::spawn(async move {
        loop {
            let event = tokio::select! {
                biased;
                _ = relay_token.cancelled() => break,
                ev = stream_rx.recv() => match ev {
                    Some(ev) => ev,
                    None => break,
                },
            };
            let msg = match event {
                StreamEvent::Text(chunk) => Msg::StreamText { turn, chunk },
                StreamEvent::Reasoning(chunk) => Msg::StreamReasoning { turn, chunk },
                StreamEvent::ToolCall(call) => Msg::StreamToolCall { turn, call },
                StreamEvent::ThinkingSignature(_) => continue, // folded into Done below
                StreamEvent::Done {
                    usage,
                    thinking_signature,
                    stop_reason,
                } => Msg::StreamDone {
                    turn,
                    usage,
                    thinking_signature,
                    stop_reason,
                },
            };
            if relay_tx.send(msg).await.is_err() {
                break;
            }
        }
    });

    // Run the actual provider. On error, the relay will have
    // already emitted partial events; we follow with a single
    // UpstreamError to terminate the turn cleanly.
    //
    // `ModelError::Cancelled` is swallowed — the terminal
    // `Msg::TurnCancelled` is emitted from `drop_scope` after the
    // turn's `TurnScope` drains. Emitting `UpstreamError` here would
    // commit a "cancelled" message the user didn't ask to see.
    let mut completed_ok = false;
    match provider.chat(request.clone(), ctx).await {
        Ok(_final_response) => {
            // Success — the final `Done` flowed through the sink.
            completed_ok = true;
        },
        Err(crate::models::ModelError::Cancelled) => {
            // Silent: `drop_scope` will emit `Msg::TurnCancelled`.
        },
        Err(e) => {
            let retry_context_limit = !compacted_before_stream && is_context_limit_error(&e);
            if retry_context_limit {
                let latest_snapshot =
                    crate::domain::estimate_context_usage_for_request(&request, max_context_tokens);
                let compaction =
                    CompactionRequest::auto(request.clone(), CompactionTrigger::ContextLimitRetry);
                match run_compaction(
                    Arc::clone(&provider),
                    turn,
                    compaction,
                    latest_snapshot,
                    max_context_tokens,
                    token.clone(),
                )
                .await
                {
                    Ok(result) => {
                        let mut retry_request = request;
                        retry_request.messages = result.replacement_messages.clone();
                        let _ = msg_tx.send(Msg::CompactionFinished { turn, result }).await;
                        join_logged(relay, "stream_relay").await;
                        dispatch_provider_stream(msg_tx, provider, turn, retry_request, token)
                            .await;
                        return;
                    },
                    Err(compact_err) => {
                        let _ = msg_tx
                            .send(Msg::CompactionFailed {
                                turn,
                                trigger: CompactionTrigger::ContextLimitRetry,
                                message: compact_err.to_string(),
                                kind: crate::domain::StatusKind::Error,
                            })
                            .await;
                    },
                }
            }
            let error = classify_error_for_ui(&e);
            run_provider_error_hook(&request.model_id, &error).await;
            let _ = msg_tx.send(Msg::UpstreamError { turn, error }).await;
        },
    }

    join_logged(relay, "stream_relay").await;

    // Post-turn (success only): verify the model actually fit VRAM. Skipped when
    // the user allowed RAM offload (no warning possible) and a no-op for
    // non-Ollama providers (verify_placement returns None). Off the critical path
    // — StreamDone is already enqueued, so any warning renders after the answer.
    if completed_ok
        && request.ollama_allow_ram_offload != Some(true)
        && let Some(p) = provider.verify_placement(sizing.effective).await
    {
        tracing::debug!(
            size_vram_bytes = p.size_vram_bytes,
            total_bytes = p.total_bytes,
            offloaded = p.size_vram_bytes < p.total_bytes,
            suggested_num_ctx = ?p.suggested_num_ctx,
            "Ollama placement"
        );
        let _ = msg_tx
            .send(Msg::OllamaPlacementResolved {
                model_id: request.model_id.clone(),
                size_vram_bytes: p.size_vram_bytes,
                total_bytes: p.total_bytes,
                suggested_num_ctx: p.suggested_num_ctx,
            })
            .await;
    }
}

async fn dispatch_provider_stream(
    msg_tx: MsgSender,
    provider: Arc<dyn ModelProvider>,
    turn: TurnId,
    request: crate::domain::ChatRequest,
    token: tokio_util::sync::CancellationToken,
) {
    let (stream_tx, mut stream_rx) = mpsc::channel::<StreamEvent>(256);
    let ctx = StreamContext::new(token.clone(), stream_tx, turn);
    let relay_tx = msg_tx.clone();
    let relay_token = token.clone();
    let relay = tokio::spawn(async move {
        loop {
            let event = tokio::select! {
                biased;
                _ = relay_token.cancelled() => break,
                ev = stream_rx.recv() => match ev {
                    Some(ev) => ev,
                    None => break,
                },
            };
            let msg = match event {
                StreamEvent::Text(chunk) => Msg::StreamText { turn, chunk },
                StreamEvent::Reasoning(chunk) => Msg::StreamReasoning { turn, chunk },
                StreamEvent::ToolCall(call) => Msg::StreamToolCall { turn, call },
                StreamEvent::ThinkingSignature(_) => continue,
                StreamEvent::Done {
                    usage,
                    thinking_signature,
                    stop_reason,
                } => Msg::StreamDone {
                    turn,
                    usage,
                    thinking_signature,
                    stop_reason,
                },
            };
            if relay_tx.send(msg).await.is_err() {
                break;
            }
        }
    });

    let model_id = request.model_id.clone();
    match provider.chat(request, ctx).await {
        Ok(_) | Err(ModelError::Cancelled) => {},
        Err(e) => {
            let error = classify_error_for_ui(&e);
            run_provider_error_hook(&model_id, &error).await;
            let _ = msg_tx.send(Msg::UpstreamError { turn, error }).await;
        },
    }

    join_logged(relay, "stream_relay").await;
}

/// Run plugin hooks OFF the async executor. `run_plugin_hooks` is synchronous —
/// it spawns hook children and bounded-waits on them — so calling it inline
/// would block a tokio worker, or (on the `dispatch` path) the whole event loop.
/// `spawn_blocking` moves it to the blocking pool. Hooks are fire-and-forget
/// observers, so the result is dropped.
async fn fire_plugin_hooks(event: &'static str, payload: serde_json::Value) {
    let _ = tokio::task::spawn_blocking(move || crate::runtime::run_plugin_hooks(event, &payload))
        .await;
}

async fn run_provider_error_hook(model_id: &str, error: &crate::models::UserFacingError) {
    fire_plugin_hooks(
        "provider_error",
        serde_json::json!({
            "model_id": model_id,
            "summary": &error.summary,
            "message": &error.message,
            "category": format!("{:?}", error.category),
            "recoverable": error.recoverable,
        }),
    )
    .await;
}

/// Derive a short title for a `/remember` memory from free-text input: the
/// first non-empty line, capped to ~8 words / 60 chars. `write_memory`
/// slugifies it into the filename.
fn memory_title_from_text(text: &str) -> String {
    let first = text
        .lines()
        .find(|l| !l.trim().is_empty())
        .unwrap_or("memory")
        .trim();
    let title: String = first
        .split_whitespace()
        .take(8)
        .collect::<Vec<_>>()
        .join(" ")
        .chars()
        .take(60)
        .collect();
    if title.trim().is_empty() {
        "memory".to_string()
    } else {
        title
    }
}

const CONSOLIDATE_SYSTEM_PROMPT: &str = "You maintain a coding agent's durable memory: a set of atomic facts. Your only job is to find facts that are EXACT DUPLICATES or CLEARLY OBSOLETE/SUPERSEDED by another fact, and list their ids for pruning. Never prune facts that are merely related or similar but carry distinct information. Never rewrite or merge facts. When in doubt, keep. Reply with ONLY a JSON object: {\"prune\": [\"id1\", \"id2\"], \"reason\": \"one short sentence\"}. If nothing should be pruned, return an empty prune list.";

#[derive(Debug)]
struct PrunePlan {
    prune: Vec<String>,
    reason: String,
}

/// Extract a `{prune:[...], reason:""}` plan from a model response, tolerating
/// prose or code fences around the JSON object.
fn parse_prune_plan(text: &str) -> Option<PrunePlan> {
    let start = text.find('{')?;
    let end = text.rfind('}')?;
    if end < start {
        return None;
    }
    let json: serde_json::Value = serde_json::from_str(&text[start..=end]).ok()?;
    let prune = json
        .get("prune")?
        .as_array()?
        .iter()
        .filter_map(|v| v.as_str().map(str::to_string))
        .collect();
    let reason = json
        .get("reason")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    Some(PrunePlan { prune, reason })
}

/// `/consolidate-memory`: a one-shot model pass that names duplicate/obsolete
/// facts to prune (never rewrites — that's the anti-drift rule). The pruned
/// files are snapshotted into a checkpoint first, so the prune is reversible.
async fn consolidate_memory(
    tx: MsgSender,
    providers: Option<Arc<ProviderFactory>>,
    workdir: std::path::PathBuf,
    model_id: String,
) {
    let items = crate::app::memory::entries_with_bodies(&workdir);
    if items.len() < 2 {
        let _ = tx
            .send(Msg::RuntimeText(format!(
                "Nothing to consolidate — {} memor{} saved.",
                items.len(),
                if items.len() == 1 { "y" } else { "ies" }
            )))
            .await;
        return;
    }
    let Some(factory) = providers else {
        let _ = tx
            .send(Msg::RuntimeText(
                "Memory consolidation needs a model provider, which isn't bound in this session."
                    .to_string(),
            ))
            .await;
        return;
    };

    let mut listing = String::new();
    for (entry, body) in &items {
        let id = entry
            .path
            .file_stem()
            .and_then(|s| s.to_str())
            .unwrap_or(entry.name.as_str());
        listing.push_str(&format!(
            "- id: {id}\n  scope: {}\n  description: {}\n  body: {}\n",
            entry.scope.as_str(),
            entry.description,
            body.replace('\n', " ").trim(),
        ));
    }
    let user = format!(
        "Here are {} durable memory facts. Identify exact duplicates and clearly obsolete or superseded facts to prune.\n\n{}",
        items.len(),
        listing
    );
    let request = crate::domain::ChatRequest {
        model_id: model_id.clone(),
        messages: vec![crate::models::ChatMessage::user(user)],
        system_prompt: CONSOLIDATE_SYSTEM_PROMPT.to_string(),
        instructions: None,
        reasoning: crate::models::ReasoningLevel::None,
        temperature: 0.0,
        max_tokens: 1024,
        tools: Vec::new(),
        ollama_num_ctx: None,
        ollama_allow_ram_offload: None,
    };

    let provider = match factory.resolve(&model_id).await {
        Ok(p) => p,
        Err(e) => {
            let _ = tx
                .send(Msg::RuntimeText(format!(
                    "Memory consolidation failed: {e}"
                )))
                .await;
            return;
        },
    };
    let token = tokio_util::sync::CancellationToken::new();
    let text =
        match crate::providers::model::collect_text(provider, TurnId(0), request, token).await {
            Ok((t, _)) => t,
            Err(e) => {
                let _ = tx
                    .send(Msg::RuntimeText(format!(
                        "Memory consolidation failed: {e}"
                    )))
                    .await;
                return;
            },
        };

    let Some(plan) = parse_prune_plan(&text) else {
        let _ = tx
            .send(Msg::RuntimeText(
                "Memory consolidation: couldn't parse the model's plan; nothing changed."
                    .to_string(),
            ))
            .await;
        return;
    };
    if plan.prune.is_empty() {
        let reason = if plan.reason.is_empty() {
            String::new()
        } else {
            format!(" {}", plan.reason)
        };
        let _ = tx
            .send(Msg::RuntimeText(format!(
                "Memory consolidation: nothing to prune.{reason}"
            )))
            .await;
        return;
    }

    // Snapshot the to-be-pruned files first so the prune is reversible.
    let paths: Vec<std::path::PathBuf> = plan
        .prune
        .iter()
        .filter_map(|id| crate::app::memory::find(&workdir, id).map(|e| e.path))
        .collect();
    if !paths.is_empty() {
        let _ = crate::runtime::create_checkpoint(
            &workdir,
            &paths,
            Some(serde_json::json!({ "tool": "consolidate_memory", "reason": plan.reason })),
        );
    }

    let mut pruned = Vec::new();
    for id in &plan.prune {
        if let Ok(Some(_)) = crate::app::memory::delete_memory(&workdir, id) {
            pruned.push(id.clone());
        }
    }

    let cfg = crate::app::load_config().unwrap_or_default().memory;
    let (loaded, _) = crate::app::memory::refresh(None, &workdir, &cfg);
    let _ = tx.send(Msg::MemoryChanged(loaded)).await;

    let report = if pruned.is_empty() {
        "Memory consolidation: the model named facts to prune, but none matched existing memories."
            .to_string()
    } else {
        format!(
            "Consolidated memory — pruned {} fact{}: {}.{} Recoverable from the latest checkpoint (/checkpoints, /restore).",
            pruned.len(),
            if pruned.len() == 1 { "" } else { "s" },
            pruned.join(", "),
            if plan.reason.is_empty() {
                String::new()
            } else {
                format!(" Reason: {}.", plan.reason)
            },
        )
    };
    let _ = tx.send(Msg::RuntimeText(report)).await;
}

async fn dispatch_compact_conversation(
    msg_tx: MsgSender,
    providers: Option<Arc<ProviderFactory>>,
    turn: TurnId,
    request: CompactionRequest,
    token: tokio_util::sync::CancellationToken,
) {
    let Some(factory) = providers else {
        let _ = msg_tx
            .send(Msg::CompactionFailed {
                turn,
                trigger: request.trigger,
                message: "EffectRunner has no ProviderFactory bound".to_string(),
                kind: crate::domain::StatusKind::Error,
            })
            .await;
        return;
    };

    let provider = match factory.resolve(&request.chat.model_id).await {
        Ok(provider) => provider,
        Err(err) => {
            let _ = msg_tx
                .send(Msg::CompactionFailed {
                    turn,
                    trigger: request.trigger,
                    message: err.to_string(),
                    kind: crate::domain::StatusKind::Error,
                })
                .await;
            return;
        },
    };

    let max_context_tokens = provider.capabilities().max_context_tokens.or_else(|| {
        crate::domain::runtime::infer_static_context_window_for_model_id(&request.chat.model_id)
    });
    let before_snapshot =
        crate::domain::estimate_context_usage_for_request(&request.chat, max_context_tokens);

    let trigger = request.trigger;
    match run_compaction(
        provider,
        turn,
        request,
        before_snapshot,
        max_context_tokens,
        token,
    )
    .await
    {
        Ok(result) => {
            let _ = msg_tx.send(Msg::CompactionFinished { turn, result }).await;
        },
        Err(err) => {
            let _ = msg_tx
                .send(Msg::CompactionFailed {
                    turn,
                    trigger,
                    message: err.to_string(),
                    kind: crate::domain::StatusKind::Error,
                })
                .await;
        },
    }
}

async fn run_compaction(
    provider: Arc<dyn ModelProvider>,
    turn: TurnId,
    request: CompactionRequest,
    before_snapshot: crate::domain::ContextUsageSnapshot,
    max_context_tokens: Option<usize>,
    token: tokio_util::sync::CancellationToken,
) -> Result<CompactionResult, ModelError> {
    let started = Instant::now();
    let prepared = crate::domain::prepare_compaction(&request, max_context_tokens)
        .map_err(|skip| ModelError::InvalidRequest(skip.to_string()))?;

    let summary_request = crate::domain::build_summary_request(
        &request.chat,
        &prepared,
        request.instructions.as_deref(),
        request.policy,
    );
    let (draft, draft_usage) =
        collect_compaction_text(Arc::clone(&provider), turn, summary_request, token.clone())
            .await?;
    let draft_summary = crate::domain::normalize_summary(&draft);
    if draft_summary.trim().is_empty() {
        return Err(ModelError::InvalidRequest(
            "compaction produced an empty summary".to_string(),
        ));
    }

    let verify_request = crate::domain::build_verification_request(
        &request.chat,
        &prepared,
        &draft_summary,
        request.instructions.as_deref(),
        request.policy,
    );
    let (final_summary, verify_usage, verified, verification_error) =
        match collect_compaction_text(Arc::clone(&provider), turn, verify_request, token).await {
            Ok((verified_text, verify_usage)) => {
                let verified_summary = crate::domain::normalize_summary(&verified_text);
                if verified_summary.trim().is_empty() {
                    (
                        draft_summary,
                        verify_usage,
                        false,
                        Some("verification returned an empty summary".to_string()),
                    )
                } else {
                    (verified_summary, verify_usage, true, None)
                }
            },
            Err(ModelError::Cancelled) => return Err(ModelError::Cancelled),
            Err(err) => (draft_summary, None, false, Some(err.to_string())),
        };

    let id = format!(
        "compact_{}",
        chrono::Local::now().format("%Y%m%d_%H%M%S_%3f")
    );
    let mut record = crate::domain::CompactionRecord {
        id,
        trigger: request.trigger,
        created_at: chrono::Local::now(),
        before_tokens: before_snapshot.used_tokens,
        after_tokens: 0,
        archived_message_count: prepared.archived_messages.len(),
        preserved_message_count: prepared.preserved_messages.len(),
        summary_tokens: final_summary.len().div_ceil(4),
        duration_secs: started.elapsed().as_secs_f64(),
        verified,
        verification_error,
        focus: request.instructions.clone(),
        archive_path: None,
    };

    let mut replacement =
        crate::domain::build_replacement_messages(&final_summary, &prepared, &record);
    let mut compacted_request = request.chat.clone();
    compacted_request.messages = replacement.clone();
    let mut after_snapshot =
        crate::domain::estimate_context_usage_for_request(&compacted_request, max_context_tokens);
    record.after_tokens = after_snapshot.used_tokens;
    record.duration_secs = started.elapsed().as_secs_f64();
    replacement = crate::domain::build_replacement_messages(&final_summary, &prepared, &record);
    compacted_request.messages = replacement.clone();
    after_snapshot =
        crate::domain::estimate_context_usage_for_request(&compacted_request, max_context_tokens);
    record.after_tokens = after_snapshot.used_tokens;

    if after_snapshot.used_tokens >= before_snapshot.used_tokens {
        return Err(ModelError::InvalidRequest(format!(
            "compaction did not reduce context ({} -> {} tokens)",
            before_snapshot.used_tokens, after_snapshot.used_tokens
        )));
    }

    if crate::domain::context_exceeds_hard_limit(
        &after_snapshot,
        &compacted_request,
        request.policy,
    ) {
        return Err(ModelError::InvalidRequest(format!(
            "compacted context still exceeds response reserve ({} tokens used)",
            after_snapshot.used_tokens
        )));
    }

    Ok(CompactionResult {
        record,
        replacement_messages: replacement,
        archived_messages: prepared.archived_messages,
        before_snapshot,
        after_snapshot,
        usage: crate::domain::combine_usage(draft_usage, verify_usage),
    })
}

async fn collect_compaction_text(
    provider: Arc<dyn ModelProvider>,
    turn: TurnId,
    request: crate::domain::ChatRequest,
    token: tokio_util::sync::CancellationToken,
) -> Result<(String, Option<TokenUsage>), ModelError> {
    // Shared with the Auto-mode safety classifier — see
    // `crate::providers::model::collect_text`.
    crate::providers::model::collect_text(provider, turn, request, token).await
}

fn record_provider_capabilities(
    model_id: &str,
    caps: &crate::providers::capabilities::Capabilities,
) {
    let (provider, model) = split_model_id(model_id);
    if let Ok(store) = crate::runtime::RuntimeStore::open_default() {
        for (key, value) in [
            ("tools_support", caps.supports_tools.to_string()),
            ("vision_support", caps.supports_vision.to_string()),
            (
                "context_limit",
                caps.max_context_tokens
                    .map(|v| v.to_string())
                    .unwrap_or_else(|| "unknown".to_string()),
            ),
            (
                "reasoning_parameter_shape",
                format!("{:?}", caps.supports_reasoning),
            ),
            (
                "streaming_usage_available",
                "provider_dependent".to_string(),
            ),
            ("token_usage_field_shape", "normalized".to_string()),
        ] {
            let _ = store
                .provider_probes()
                .upsert(crate::runtime::NewProviderProbe {
                    provider: provider.clone(),
                    model_id: model.clone(),
                    capability_key: key.to_string(),
                    capability_value: value,
                    confidence: "verified".to_string(),
                    error: None,
                });
        }
    }
}

fn split_model_id(model_id: &str) -> (String, String) {
    match model_id.split_once('/') {
        Some((provider, model)) if !provider.is_empty() && !model.is_empty() => {
            (provider.to_ascii_lowercase(), model.to_string())
        },
        _ => ("ollama".to_string(), model_id.to_string()),
    }
}

fn is_context_limit_error(error: &ModelError) -> bool {
    let text = error.to_string().to_lowercase();
    text.contains("context")
        && (text.contains("too large")
            || text.contains("exceed")
            || text.contains("maximum")
            || text.contains("token"))
}

/// Dispatch an `ExecuteTool` command.
#[allow(clippy::too_many_arguments)]
async fn dispatch_execute_tool(
    msg_tx: MsgSender,
    tools: Option<Arc<ToolRegistry>>,
    workdir: PathBuf,
    turn: TurnId,
    call_id: crate::domain::ToolCallId,
    source: crate::models::tool_call::ToolCall,
    token: tokio_util::sync::CancellationToken,
    background: tokio_util::sync::CancellationToken,
    config: Arc<crate::app::Config>,
    model_id: String,
    task_id: Option<String>,
    safety_mode: crate::runtime::SafetyMode,
    intent: Option<String>,
    classifier: Option<Arc<dyn crate::providers::AutoClassifier>>,
    approval: Option<crate::providers::ApprovalBroker>,
) {
    let _ = msg_tx.send(Msg::ToolStarted { turn, call_id }).await;

    let Some(registry) = tools else {
        let _ = msg_tx
            .send(Msg::ToolFinished {
                turn,
                call_id,
                outcome: crate::domain::ToolOutcome::error(
                    "EffectRunner has no ToolRegistry bound",
                    0.0,
                ),
            })
            .await;
        return;
    };

    // Route MCP-prefixed calls to the mcp proxy, which takes
    // {server_name, tool_name, arguments}. The raw model call has
    // those embedded in the function name and arguments respectively.
    let (tool_key, args) = if source.function.name.starts_with("mcp__") {
        let rest = &source.function.name[5..];
        if let Some((server, tool)) = rest.split_once("__") {
            (
                "mcp_proxy",
                serde_json::json!({
                    "server_name": server,
                    "tool_name": tool,
                    "arguments": source.function.arguments.clone(),
                }),
            )
        } else {
            let _ = msg_tx
                .send(Msg::ToolFinished {
                    turn,
                    call_id,
                    outcome: crate::domain::ToolOutcome::error(
                        format!("invalid MCP tool name: {}", source.function.name),
                        0.0,
                    ),
                })
                .await;
            return;
        }
    } else {
        (
            source.function.name.as_str(),
            source.function.arguments.clone(),
        )
    };
    let tool_run_id =
        start_runtime_tool_run(task_id.as_deref(), turn, call_id, tool_key, &args).await;

    let Some(tool) = registry.get(tool_key) else {
        let outcome = crate::domain::ToolOutcome::error(format!("unknown tool: {}", tool_key), 0.0);
        finish_runtime_tool_run(tool_run_id.as_deref(), &outcome);
        let _ = msg_tx
            .send(Msg::ToolFinished {
                turn,
                call_id,
                outcome,
            })
            .await;
        return;
    };

    // Bridge the tool's progress channel to `Msg::ToolProgress`.
    // A sibling task drains progress events while the tool runs.
    // The channel closes when `progress_tx` drops (when `ctx`
    // drops at the end of `tool.execute`), which terminates the
    // relay loop cleanly.
    let (progress_tx, mut progress_rx) = mpsc::channel(16);
    let relay_tx = msg_tx.clone();
    let relay_token = token.clone();
    let progress_relay = tokio::spawn(async move {
        loop {
            let event = tokio::select! {
                biased;
                _ = relay_token.cancelled() => break,
                ev = progress_rx.recv() => match ev {
                    Some(ev) => ev,
                    None => break,
                },
            };
            if relay_tx
                .send(Msg::ToolProgress {
                    turn,
                    call_id,
                    event,
                })
                .await
                .is_err()
            {
                break;
            }
        }
    });

    let mut ctx = ExecContext::new(
        token,
        progress_tx,
        call_id,
        turn,
        workdir,
        config,
        model_id,
        task_id,
        safety_mode,
        intent,
        classifier,
        approval,
    );
    ctx.background = background;
    let before_payload = serde_json::json!({
        "turn_id": turn.0,
        "call_id": call_id.0,
        "tool": tool_key,
    });
    fire_plugin_hooks("before_tool_use", before_payload).await;
    let outcome = tool.execute(args, ctx).await;
    let after_payload = serde_json::json!({
        "turn_id": turn.0,
        "call_id": call_id.0,
        "tool": tool_key,
        "status": tool_status_label(outcome.status),
        "summary": &outcome.summary,
    });
    fire_plugin_hooks("after_tool_use", after_payload).await;
    finish_runtime_tool_run(tool_run_id.as_deref(), &outcome);
    join_logged(progress_relay, "tool_progress_relay").await;
    let _ = msg_tx
        .send(Msg::ToolFinished {
            turn,
            call_id,
            outcome,
        })
        .await;
}

async fn start_runtime_tool_run(
    task_id: Option<&str>,
    turn: TurnId,
    call_id: crate::domain::ToolCallId,
    tool_name: &str,
    args: &serde_json::Value,
) -> Option<String> {
    // Synchronous rusqlite write on the hot tool-execution path — offload it to
    // the blocking pool. The id is needed by `finish`, so we await the result
    // (unlike `finish`, which is fire-and-forget) (#39).
    let task_id = task_id.map(str::to_string);
    let tool_name = tool_name.to_string();
    let args_json = serde_json::to_string(args).ok();
    tokio::task::spawn_blocking(move || {
        crate::runtime::RuntimeStore::open_default()
            .and_then(|store| {
                store.tool_runs().start(crate::runtime::NewToolRun {
                    id: None,
                    task_id,
                    turn_id: Some(turn.0.to_string()),
                    call_id: Some(call_id.0.to_string()),
                    tool_name,
                    args_json,
                })
            })
            .map(|record| record.id)
            .ok()
    })
    .await
    .ok()
    .flatten()
}

fn finish_runtime_tool_run(tool_run_id: Option<&str>, outcome: &crate::domain::ToolOutcome) {
    let Some(tool_run_id) = tool_run_id else {
        return;
    };
    let tool_run_id = tool_run_id.to_string();
    let status = tool_status_label(outcome.status).to_string();
    let output_json = serde_json::to_string(&serde_json::json!({
        "status": tool_status_label(outcome.status),
        "summary": &outcome.summary,
        "model_content": &outcome.model_content,
        "error": &outcome.error,
        "metadata": &outcome.metadata,
        "artifacts": &outcome.artifacts,
        "duration_secs": outcome.duration_secs,
    }))
    .ok();
    // Fire-and-forget telemetry write on the blocking pool — don't stall the
    // tool-finish path waiting on rusqlite (#39).
    tokio::task::spawn_blocking(move || {
        if let Ok(store) = crate::runtime::RuntimeStore::open_default() {
            let _ = store
                .tool_runs()
                .finish(&tool_run_id, &status, output_json.as_deref());
        }
    });
}

fn tool_status_label(status: crate::domain::ToolStatus) -> &'static str {
    match status {
        crate::domain::ToolStatus::Success => "success",
        crate::domain::ToolStatus::Error => "error",
        crate::domain::ToolStatus::Cancelled => "cancelled",
    }
}

fn runtime_model_info_text(model: &str) -> String {
    let snapshot = crate::domain::runtime::ProviderCapabilitySnapshot::from_model_id(model);
    let mut lines = vec![
        format!("Model info: {}", model),
        format!("- provider: {}", snapshot.provider),
        format!("- model: {}", snapshot.model),
        format!("- supports tools: {}", snapshot.supports_tools),
        format!("- supports vision: {}", snapshot.supports_vision),
        format!("- reasoning: {}", snapshot.reasoning),
        format!(
            "- context limit: {}",
            snapshot
                .max_context_tokens
                .map(|value: usize| value.to_string())
                .unwrap_or_else(|| "unknown".to_string())
        ),
    ];
    if let Ok(store) = crate::runtime::RuntimeStore::open_default()
        && let Ok(probes) = store
            .provider_probes()
            .list(Some(&snapshot.provider), Some(&snapshot.model))
        && !probes.is_empty()
    {
        lines.push(String::new());
        lines.push("Cached provider reality records:".to_string());
        for probe in probes {
            lines.push(format!(
                "- {} = {} ({})",
                probe.capability_key, probe.capability_value, probe.confidence
            ));
        }
    }
    lines.join("\n")
}

/// Spawn `ollama pull <model>` and stream its stdout lines as
/// `Msg::ModelPullProgress` status updates. Emits a final
/// `Msg::ModelPullFinished` on successful exit; on failure, emits a
/// single `ModelPullProgress` with the error text.
async fn dispatch_pull_ollama_model(tx: MsgSender, model: String) {
    use tokio::io::{AsyncBufReadExt, BufReader};
    use tokio::process::Command;

    let mut cmd = Command::new("ollama");
    cmd.arg("pull")
        .arg(&model)
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true);

    let mut child = match cmd.spawn() {
        Ok(c) => c,
        Err(e) => {
            let _ = tx
                .send(Msg::ModelPullProgress(format!(
                    "ollama pull failed to start: {}",
                    e
                )))
                .await;
            return;
        },
    };

    // Capture the reader's handle instead of orphaning it: the child's stdout
    // closes when it exits, so this task finishes right after `child.wait`
    // below — we join it there so a panic is logged, not silently lost (#60).
    let reader_handle = child.stdout.take().map(|stdout| {
        let tx_inner = tx.clone();
        tokio::spawn(async move {
            let mut reader = BufReader::new(stdout).lines();
            while let Ok(Some(line)) = reader.next_line().await {
                let _ = tx_inner.send(Msg::ModelPullProgress(line)).await;
            }
        })
    });

    match child.wait().await {
        Ok(status) if status.success() => {
            let _ = tx.send(Msg::ModelPullFinished { model }).await;
        },
        Ok(status) => {
            let _ = tx
                .send(Msg::ModelPullProgress(format!(
                    "ollama pull exited with status {}",
                    status.code().unwrap_or(-1)
                )))
                .await;
        },
        Err(e) => {
            let _ = tx
                .send(Msg::ModelPullProgress(format!(
                    "ollama pull wait error: {}",
                    e
                )))
                .await;
        },
    }

    // The child has exited; its stdout is closed, so the reader is finishing.
    // Join it (logging a panic) so it isn't left orphaned (#60).
    if let Some(handle) = reader_handle {
        join_logged(handle, "ollama_pull_reader").await;
    }
}

fn mcp_startup_msg(name: &str, started: bool, tools: Vec<crate::domain::McpToolSpec>) -> Msg {
    if started {
        Msg::McpServerReady {
            name: name.to_string(),
            tools,
        }
    } else {
        Msg::McpServerErrored {
            name: name.to_string(),
            reason: "server failed to start or initialize".to_string(),
        }
    }
}

/// Read the system clipboard on a blocking thread and emit a `Msg`
/// back into the main loop. Image content wins when present; falls
/// back to text; empty or error surface as `Msg::TransientStatus` so
/// the user gets visible feedback (a silent no-op on Ctrl+V would be
/// confusing, especially on macOS where `osascript` can take ~300ms).
///
/// `tokio::task::spawn_blocking` is the right primitive: `clipboard::
/// has_image` / `read_image_bytes` / `read_text` shell out to xclip /
/// wl-paste / pngpaste / PowerShell, all of which block synchronously.
async fn dispatch_read_clipboard(tx: MsgSender) {
    use crate::domain::{Paste, StatusKind};

    enum Outcome {
        Image { bytes: Vec<u8>, format: String },
        Text(String),
        Empty,
        Error(String),
    }

    let outcome = tokio::task::spawn_blocking(|| {
        if crate::clipboard::has_image() {
            match crate::clipboard::read_image_bytes() {
                Ok((bytes, format)) => Outcome::Image { bytes, format },
                Err(e) => Outcome::Error(format!("Clipboard image read failed: {}", e)),
            }
        } else {
            match crate::clipboard::read_text() {
                Ok(t) if !t.is_empty() => Outcome::Text(t),
                Ok(_) => Outcome::Empty,
                Err(e) => Outcome::Error(format!("Clipboard empty / read failed: {}", e)),
            }
        }
    })
    .await
    .unwrap_or_else(|e| Outcome::Error(format!("clipboard spawn_blocking: {}", e)));

    let msg = match outcome {
        Outcome::Image { bytes, format } => Msg::Paste(Paste::Image { bytes, format }),
        Outcome::Text(text) => Msg::Paste(Paste::Text(text)),
        Outcome::Empty => Msg::TransientStatus {
            text: "Clipboard is empty".to_string(),
            kind: StatusKind::Info,
            dismiss_ms: 2_000,
        },
        Outcome::Error(text) => Msg::TransientStatus {
            text,
            kind: StatusKind::Warn,
            dismiss_ms: 4_000,
        },
    };
    let _ = tx.send(msg).await;
}

/// Write text to the system clipboard on a blocking thread (the platform
/// tools shell out and block), then report the result via a transient status.
async fn dispatch_copy_to_clipboard(text: String, tx: MsgSender) {
    use crate::domain::StatusKind;

    let char_count = text.chars().count();
    let result = tokio::task::spawn_blocking(move || crate::clipboard::write_text(&text))
        .await
        .unwrap_or_else(|e| Err(anyhow::anyhow!("clipboard spawn_blocking: {e}")));

    let msg = match result {
        Ok(()) => Msg::TransientStatus {
            text: format!("Copied {char_count} chars to clipboard"),
            kind: StatusKind::Info,
            dismiss_ms: 2_000,
        },
        Err(e) => Msg::TransientStatus {
            text: format!("Copy failed: {e}"),
            kind: StatusKind::Warn,
            dismiss_ms: 4_000,
        },
    };
    let _ = tx.send(msg).await;
}

fn classify_error_for_ui(e: &crate::models::ModelError) -> crate::models::UserFacingError {
    use crate::models::{ErrorCategory, ModelError, UserFacingError};
    match e {
        ModelError::Backend(b) => UserFacingError {
            summary: "Backend error".to_string(),
            message: b.to_string(),
            suggestion: "Check the provider endpoint / API key.".to_string(),
            category: ErrorCategory::Connection,
            recoverable: true,
        },
        ModelError::Authentication(msg) => UserFacingError {
            summary: "Auth error".to_string(),
            message: msg.clone(),
            suggestion: "Set the env var the provider expects.".to_string(),
            category: ErrorCategory::Auth,
            recoverable: false,
        },
        ModelError::RateLimit { retry_after } => UserFacingError {
            summary: "Rate limit".to_string(),
            message: format!("retry after {:?}", retry_after),
            suggestion: "Wait and try again.".to_string(),
            category: ErrorCategory::Temporary,
            recoverable: true,
        },
        ModelError::StreamError(msg) => UserFacingError {
            summary: "Stream error".to_string(),
            message: msg.clone(),
            suggestion: "Retry the request.".to_string(),
            category: ErrorCategory::Connection,
            recoverable: true,
        },
        other => UserFacingError {
            summary: "Model error".to_string(),
            message: other.to_string(),
            suggestion: String::new(),
            category: ErrorCategory::Internal,
            recoverable: false,
        },
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::ToolCallId;
    use std::time::Duration;

    fn runner() -> (EffectRunner, mpsc::Receiver<Msg>) {
        EffectRunner::pair(PathBuf::from("/tmp"))
    }

    #[test]
    fn new_child_suppresses_terminal_title() {
        // A subagent's child runner must not emit OSC 2 terminal titles —
        // otherwise they leak into a headless parent's stdout and corrupt
        // `--format json`/`text` output (caught during live headless testing).
        let (tx, _rx) = mpsc::channel::<Msg>(MSG_CHANNEL_CAPACITY);
        let providers = Arc::new(ProviderFactory::new(crate::app::Config::default()));
        let tools = Arc::new(ToolRegistry::new());
        let child = EffectRunner::new_child(tx, PathBuf::from("/tmp"), providers, tools);
        assert!(
            !child.terminal_title_enabled,
            "subagent child runner must suppress terminal-title escapes"
        );
    }

    #[test]
    fn parse_prune_plan_extracts_json_amid_prose() {
        let plan = parse_prune_plan(
            "Sure, here's the plan:\n```json\n{\"prune\": [\"a\", \"b\"], \"reason\": \"dupes\"}\n```\nDone.",
        )
        .expect("should parse");
        assert_eq!(plan.prune, vec!["a".to_string(), "b".to_string()]);
        assert_eq!(plan.reason, "dupes");
    }

    #[test]
    fn parse_prune_plan_handles_empty_and_garbage() {
        let empty = parse_prune_plan("{\"prune\": [], \"reason\": \"all distinct\"}")
            .expect("empty plan parses");
        assert!(empty.prune.is_empty());
        assert!(parse_prune_plan("no json here").is_none());
    }

    #[test]
    fn memory_title_from_text_is_short_and_nonempty() {
        assert_eq!(
            memory_title_from_text("prefer ripgrep over grep"),
            "prefer ripgrep over grep"
        );
        assert_eq!(memory_title_from_text("   "), "memory");
        let long = memory_title_from_text("one two three four five six seven eight nine ten");
        assert!(long.split_whitespace().count() <= 8);
    }

    #[tokio::test]
    async fn dispatch_exit_is_noop_on_runner_state() {
        let (mut r, _rx) = runner();
        r.dispatch(Cmd::Exit);
        assert_eq!(r.scope_count(), 0);
    }

    #[tokio::test]
    async fn dispatch_save_emits_session_saved() {
        let (mut r, mut rx) = runner();
        r.dispatch(Cmd::SaveConversation(
            crate::session::ConversationHistory::new("/p".to_string(), "m".to_string()),
        ));
        let msg = tokio::time::timeout(Duration::from_millis(200), rx.recv())
            .await
            .expect("sender emits")
            .expect("channel alive");
        assert!(matches!(msg, Msg::SessionSaved));
    }

    #[tokio::test]
    async fn dispatch_dismiss_after_delay_emits_status_dismiss() {
        let (mut r, mut rx) = runner();
        let t0 = std::time::Instant::now();
        r.dispatch(Cmd::DismissStatusAfter { ms: 30 });
        let msg = tokio::time::timeout(Duration::from_millis(300), rx.recv())
            .await
            .expect("sender emits")
            .expect("channel alive");
        assert!(matches!(msg, Msg::StatusDismiss));
        assert!(t0.elapsed() >= Duration::from_millis(25));
    }

    #[test]
    fn mcp_startup_msg_treats_zero_tool_started_server_as_ready() {
        let msg = mcp_startup_msg("empty", true, Vec::new());
        assert!(matches!(
            msg,
            Msg::McpServerReady { name, tools } if name == "empty" && tools.is_empty()
        ));
    }

    #[test]
    fn mcp_startup_msg_reports_unstarted_server_as_error() {
        let msg = mcp_startup_msg("bad", false, Vec::new());
        assert!(matches!(
            msg,
            Msg::McpServerErrored { name, reason }
                if name == "bad" && reason.contains("failed to start")
        ));
    }

    #[tokio::test]
    async fn cancel_scope_emits_turn_cancelled_after_bounded_timeout() {
        let (mut r, mut rx) = runner();
        let turn = TurnId(77);
        {
            let scope = r.scope_mut(turn);
            scope.spawn(async {
                std::future::pending::<()>().await;
            });
        }
        assert_eq!(r.scope_count(), 1);

        let start = std::time::Instant::now();
        r.dispatch(Cmd::CancelScope(turn));
        assert_eq!(r.scope_count(), 0);
        let msg = tokio::time::timeout(Duration::from_millis(500), rx.recv())
            .await
            .expect("bounded cancel should emit terminal message")
            .expect("channel alive");
        assert!(matches!(msg, Msg::TurnCancelled(t) if t == turn));
        assert!(
            start.elapsed() < Duration::from_millis(500),
            "cancel terminal message took {:?}",
            start.elapsed()
        );
    }

    #[tokio::test]
    async fn cancel_scope_emits_turn_cancelled_even_after_reaping() {
        // Regression (Axis 1 #9): if a turn's tasks complete and
        // `reap_empty_scopes` removes the now-empty scope before the user's
        // cancel lands, `drop_scope` used to be a silent no-op and the reducer
        // stuck forever in `Cancelling`. The terminal `TurnCancelled` must fire
        // even when the scope is already gone.
        let (mut r, mut rx) = runner();
        let turn = TurnId(88);
        {
            let scope = r.scope_mut(turn);
            scope.spawn(async {}); // completes immediately
        }
        assert_eq!(r.scope_count(), 1);

        // Let the task finish, then any dispatch reaps the now-empty scope.
        tokio::time::sleep(Duration::from_millis(20)).await;
        r.dispatch(Cmd::Exit);
        assert_eq!(r.scope_count(), 0, "completed scope should be reaped");

        // The scope is gone, but the reducer is still `Cancelling`: cancel must
        // still produce a terminal message.
        r.dispatch(Cmd::CancelScope(turn));
        let msg = tokio::time::timeout(Duration::from_millis(500), rx.recv())
            .await
            .expect("cancel on a reaped scope must still emit a terminal message")
            .expect("channel alive");
        assert!(matches!(msg, Msg::TurnCancelled(t) if t == turn));
    }

    #[tokio::test]
    async fn dispatch_call_model_creates_scope() {
        let (mut r, _rx) = runner();
        let turn = TurnId(7);
        let request = crate::domain::ChatRequest {
            model_id: "test/m".to_string(),
            messages: vec![],
            system_prompt: String::new(),
            instructions: None,
            reasoning: crate::models::ReasoningLevel::Medium,
            temperature: 0.7,
            max_tokens: 4096,
            tools: vec![],

            ollama_num_ctx: None,
            ollama_allow_ram_offload: None,
        };
        r.dispatch(Cmd::CallModel { turn, request });
        assert_eq!(r.scope_count(), 1);
    }

    /// F12: after a spawned task completes (here via the
    /// no-ProviderFactory error path), the next `dispatch` call reaps
    /// the empty scope instead of leaving an orphan entry in the map.
    #[tokio::test]
    async fn empty_scopes_are_reaped_on_next_dispatch() {
        let (mut r, mut rx) = runner();
        let turn = TurnId(42);
        let request = crate::domain::ChatRequest {
            model_id: "test/m".to_string(),
            messages: vec![],
            system_prompt: String::new(),
            instructions: None,
            reasoning: crate::models::ReasoningLevel::Medium,
            temperature: 0.7,
            max_tokens: 4096,
            tools: vec![],

            ollama_num_ctx: None,
            ollama_allow_ram_offload: None,
        };
        r.dispatch(Cmd::CallModel { turn, request });
        assert_eq!(r.scope_count(), 1);

        // Runner has no provider bindings → dispatch_call_model hits
        // the "not wired" error path and emits UpstreamError, then the
        // spawned task returns. Drain that message so we know the task
        // ran to completion.
        let msg = tokio::time::timeout(Duration::from_millis(200), rx.recv())
            .await
            .expect("upstream error arrived")
            .expect("channel alive");
        assert!(matches!(msg, Msg::UpstreamError { .. }));

        // Give the JoinSet a tick to notice the task finished.
        tokio::task::yield_now().await;

        // Any subsequent dispatch reaps the now-empty scope.
        r.dispatch(Cmd::DismissStatusAfter { ms: 10 });
        assert_eq!(
            r.scope_count(),
            0,
            "completed scope must be reaped on next dispatch"
        );
    }

    #[tokio::test]
    async fn dispatch_execute_tool_under_turn_emits_tool_started() {
        let (mut r, mut rx) = runner();
        let turn = TurnId(7);
        let call_id = ToolCallId(1);
        let source = crate::models::tool_call::ToolCall {
            id: Some("c1".to_string()),
            function: crate::models::tool_call::FunctionCall {
                name: "read_file".to_string(),
                arguments: serde_json::json!({"path": "x"}),
            },
        };
        r.dispatch(Cmd::ExecuteTool {
            turn,
            call_id,
            source,
            model_id: "ollama/test".to_string(),
            safety_mode: crate::runtime::SafetyMode::Ask,
            intent: None,
        });
        let first = tokio::time::timeout(Duration::from_millis(200), rx.recv())
            .await
            .expect("some msg")
            .expect("channel alive");
        assert!(matches!(
            first,
            Msg::ToolStarted {
                turn: t,
                call_id: c,
            } if t == turn && c == call_id
        ));
    }

    #[tokio::test]
    async fn cancel_scope_before_execute_tool_drops_pending_work() {
        let (mut r, _rx) = runner();
        let turn = TurnId(9);
        r.dispatch(Cmd::CallModel {
            turn,
            request: crate::domain::ChatRequest {
                model_id: "m".to_string(),
                messages: vec![],
                system_prompt: String::new(),
                instructions: None,
                reasoning: crate::models::ReasoningLevel::Medium,
                temperature: 0.7,
                max_tokens: 4096,
                tools: vec![],

                ollama_num_ctx: None,
                ollama_allow_ram_offload: None,
            },
        });
        assert_eq!(r.scope_count(), 1);

        r.dispatch(Cmd::CancelScope(turn));
        assert_eq!(r.scope_count(), 0);
    }

    #[tokio::test]
    async fn shutdown_drains_pending_saves() {
        let (mut r, _rx) = runner();
        for _ in 0..5 {
            r.dispatch(Cmd::SaveConversation(
                crate::session::ConversationHistory::new("/p".to_string(), "m".to_string()),
            ));
        }
        // Shutdown waits for all five to complete (should be instant).
        let start = std::time::Instant::now();
        r.shutdown().await;
        assert!(start.elapsed() < Duration::from_secs(2));
    }
}