capo_agent/

app.rs

#![cfg_attr(test, allow(clippy::expect_used, clippy::unwrap_used))]

use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, MutexGuard};

use futures::{Stream, StreamExt};
use motosan_agent_loop::{
    AgentEvent, AgentOp, AgentSession, AgentStreamItem, AutocompactConfig, AutocompactExtension,
    CoreEvent, Engine, LlmClient, SessionStore,
};
use motosan_agent_tool::ToolContext;
use tokio::sync::mpsc;

use crate::agent::build_system_prompt;
use crate::config::Config;
use crate::error::{AppError, Result};
use crate::events::{ProgressChunk, UiEvent, UiToolResult};
use crate::llm::build_llm_client;
use crate::permissions::{NoOpPermissionGate, PermissionGate, PromptStrategy};
use crate::tools::{builtin_tools, SharedCancelToken, ToolCtx, ToolProgressChunk};

/// Internal accumulator for per-session cumulative token counts. Distinct
/// from the wire-form `UiEvent::TurnStats` (which also carries per-turn
/// deltas + model). See spec §3.1 and §3.4.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TurnStatsAccum {
    pub cumulative_input: u64,
    pub cumulative_output: u64,
    pub turn_count: u64,
}

impl TurnStatsAccum {
    /// Add a `TokenUsage` from a completed turn. Saturating arithmetic
    /// — `u64::MAX` is the cap (overflow is practically unreachable
    /// at realistic token counts).
    pub fn add(&mut self, usage: motosan_agent_loop::TokenUsage) {
        self.cumulative_input = self.cumulative_input.saturating_add(usage.input_tokens);
        self.cumulative_output = self.cumulative_output.saturating_add(usage.output_tokens);
        self.turn_count = self.turn_count.saturating_add(1);
    }
}

/// Extract `Reasoning` (thinking/chain-of-thought) ContentParts from any
/// `Assistant` Messages in `messages`. Returns one `UiEvent::ThinkingComplete`
/// per Reasoning part, in source order. Pure function — unit-testable.
///
/// Used by `run_turn` to surface thinking blocks to the TUI after a turn
/// completes (motosan-agent-loop 0.21.1 doesn't stream thinking deltas;
/// see v0.9 design spec §3.5).
pub(crate) fn extract_thinking_events(messages: &[motosan_agent_loop::Message]) -> Vec<UiEvent> {
    let mut events = Vec::new();
    for msg in messages {
        if let motosan_agent_loop::Message::Assistant { content, .. } = msg {
            for part in content {
                if let motosan_agent_loop::AssistantContent::Reasoning { text, .. } = part {
                    events.push(UiEvent::ThinkingComplete { text: text.clone() });
                }
            }
        }
    }
    events
}

/// Extract thinking events only for messages appended during the just-finished
/// turn. `AgentResult.messages` / terminal messages contain full session
/// history; slicing at `previous_len` prevents historical reasoning blocks
/// from being re-emitted on every later turn.
pub(crate) fn extract_new_thinking_events(
    messages: &[motosan_agent_loop::Message],
    previous_len: usize,
) -> Vec<UiEvent> {
    extract_thinking_events(messages.get(previous_len..).unwrap_or(&[]))
}

/// How a fresh `AgentSession` should be constructed by `SessionFactory`.
#[derive(Debug, Clone)]
pub(crate) enum SessionMode {
    /// A brand-new session (fresh ulid; persisted if a store is configured).
    New,
    /// Resume an existing session by id (requires a configured store).
    Resume(String),
}

/// Everything needed to (re)build an `AgentSession` + its `LlmClient`.
/// Stored on `App` so `new_session` / `load_session` / `switch_model` can
/// rebuild without re-running `AppBuilder`. All fields are cheap to clone
/// (owned values or `Arc`s).
struct SharedLlm {
    client: Arc<dyn LlmClient>,
}

impl SharedLlm {
    fn new(client: Arc<dyn LlmClient>) -> Self {
        Self { client }
    }

    fn client(&self) -> Arc<dyn LlmClient> {
        Arc::clone(&self.client)
    }
}

pub(crate) struct SessionFactory {
    cwd: PathBuf,
    settings: Arc<Mutex<crate::settings::Settings>>,
    auth: crate::auth::Auth,
    policy: Arc<crate::permissions::Policy>,
    session_cache: Arc<crate::permissions::SessionCache>,
    ui_tx: Option<mpsc::Sender<UiEvent>>,
    headless_permissions: bool,
    permission_gate: Arc<dyn PermissionGate>,
    permission_strategy_handle: Option<Arc<tokio::sync::RwLock<PromptStrategy>>>,
    /// The SHARED progress sender. Rebuilt tools must use this exact
    /// sender so `App.progress_rx` keeps receiving — see Task 5.
    progress_tx: mpsc::Sender<ToolProgressChunk>,
    skills: Arc<Vec<crate::skills::Skill>>,
    install_builtin_tools: bool,
    extra_tools: Vec<Arc<dyn motosan_agent_tool::Tool>>,
    max_iterations: usize,
    context_discovery_disabled: bool,
    autocompact_enabled: bool,
    session_store: Option<Arc<dyn SessionStore>>,
    llm_override: Option<Arc<dyn LlmClient>>,
    current_model: Arc<Mutex<Option<crate::model::ModelId>>>,
    cancel_token: SharedCancelToken,
}

impl SessionFactory {
    fn settings(&self) -> crate::settings::Settings {
        match self.settings.lock() {
            Ok(guard) => guard.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    fn store_settings(&self, settings: crate::settings::Settings) {
        match self.settings.lock() {
            Ok(mut guard) => *guard = settings,
            Err(poisoned) => *poisoned.into_inner() = settings,
        }
    }

    fn current_model(&self) -> Option<crate::model::ModelId> {
        match self.current_model.lock() {
            Ok(guard) => guard.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    fn set_current_model(&self, model: crate::model::ModelId) {
        match self.current_model.lock() {
            Ok(mut guard) => *guard = Some(model),
            Err(poisoned) => *poisoned.into_inner() = Some(model),
        }
    }

    fn clear_current_model(&self) {
        match self.current_model.lock() {
            Ok(mut guard) => *guard = None,
            Err(poisoned) => *poisoned.into_inner() = None,
        }
    }

    /// Build a fresh `AgentSession` + its `LlmClient`. `model_override`,
    /// when set, replaces `settings.model.name` for this build (used by
    /// `switch_model`). `settings_override`, when set, supplies the full
    /// settings struct for this build and bypasses any sticky `/model`
    /// override (used by `/settings`). The returned `LlmClient` is what
    /// `App.llm` should be swapped to.
    async fn build(
        &self,
        mode: SessionMode,
        model_override: Option<&crate::model::ModelId>,
        settings_override: Option<&crate::settings::Settings>,
    ) -> Result<(AgentSession, Arc<dyn LlmClient>)> {
        // Apply the model override (or last switched model) onto a settings clone.
        // A settings override is authoritative: `/settings` should not be
        // shadowed by an earlier `/model` sticky override.
        let effective_model = model_override.cloned().or_else(|| {
            if settings_override.is_some() {
                None
            } else {
                self.current_model()
            }
        });
        let mut settings = settings_override
            .cloned()
            .unwrap_or_else(|| self.settings());
        if let Some(m) = &effective_model {
            settings.model.name = m.as_str().to_string();
        }

        let llm = if effective_model.is_none() {
            self.llm_override.as_ref().map_or_else(
                || build_llm_client(&settings, &self.auth),
                |llm| Ok(Arc::clone(llm)),
            )?
        } else {
            build_llm_client(&settings, &self.auth)?
        };

        // Tools — rebuilt fresh each time, sharing the SAME progress_tx so
        // `App.progress_rx` keeps receiving from whatever session is live.
        let tool_ctx = ToolCtx::new_with_cancel_token(
            &self.cwd,
            Arc::clone(&self.permission_gate),
            self.progress_tx.clone(),
            self.cancel_token.clone(),
        );
        let mut tools = if self.install_builtin_tools {
            builtin_tools(tool_ctx.clone())
        } else {
            Vec::new()
        };
        tools.extend(self.extra_tools.iter().cloned());

        let tool_names: Vec<String> = tools.iter().map(|t| t.def().name).collect();
        let base_prompt = build_system_prompt(&tool_names, &self.skills);
        let system_prompt = if self.context_discovery_disabled {
            base_prompt
        } else {
            let agent_dir = crate::paths::agent_dir();
            let context = crate::context_files::load_project_context_files(&self.cwd, &agent_dir);
            crate::context_files::assemble_system_prompt(&base_prompt, &context, &self.cwd)
        };
        let motosan_tool_context = ToolContext::new("capo", "capo").with_cwd(&self.cwd);

        let mut engine_builder = Engine::builder()
            .max_iterations(self.max_iterations)
            .system_prompt(system_prompt)
            .tool_context(motosan_tool_context);
        for tool in tools {
            engine_builder = engine_builder.tool(tool);
        }
        if let Some(ui_tx) = &self.ui_tx {
            let ext = if let Some(handle) = &self.permission_strategy_handle {
                crate::permissions::PermissionExtension::with_strategy_handle(
                    Arc::clone(&self.policy),
                    Arc::clone(&self.session_cache),
                    self.cwd.clone(),
                    Some(ui_tx.clone()),
                    Arc::clone(handle),
                )
            } else {
                crate::permissions::PermissionExtension::new(
                    Arc::clone(&self.policy),
                    Arc::clone(&self.session_cache),
                    self.cwd.clone(),
                    ui_tx.clone(),
                )
            };
            engine_builder = engine_builder.extension(Box::new(ext));
        } else if self.headless_permissions {
            let ext = if let Some(handle) = &self.permission_strategy_handle {
                crate::permissions::PermissionExtension::with_strategy_handle(
                    Arc::clone(&self.policy),
                    Arc::clone(&self.session_cache),
                    self.cwd.clone(),
                    None,
                    Arc::clone(handle),
                )
            } else {
                crate::permissions::PermissionExtension::headless(
                    Arc::clone(&self.policy),
                    Arc::clone(&self.session_cache),
                    self.cwd.clone(),
                )
            };
            engine_builder = engine_builder.extension(Box::new(ext));
        }
        if self.autocompact_enabled
            && settings.session.compact_at_context_pct > 0.0
            && settings.session.compact_at_context_pct < 1.0
        {
            let cfg = AutocompactConfig {
                threshold: settings.session.compact_at_context_pct,
                max_context_tokens: settings.session.max_context_tokens,
                keep_turns: settings.session.keep_turns.max(1),
            };
            engine_builder = engine_builder
                .extension(Box::new(AutocompactExtension::new(cfg, Arc::clone(&llm))));
        }
        let engine = engine_builder.build();

        let session = match (&mode, &self.session_store) {
            (SessionMode::Resume(id), Some(store)) => {
                let s = AgentSession::resume(id, Arc::clone(store), engine, Arc::clone(&llm))
                    .await
                    .map_err(|e| AppError::Config(format!("resume failed: {e}")))?;
                let entries = s
                    .entries()
                    .await
                    .map_err(|e| AppError::Config(format!("entries failed: {e}")))?;
                crate::session::hydrate_read_files(&entries, &tool_ctx).await?;
                s
            }
            (SessionMode::Resume(_), None) => {
                return Err(AppError::Config("resume requires a session store".into()));
            }
            (SessionMode::New, Some(store)) => {
                let id = crate::session::SessionId::new();
                AgentSession::new_with_store(
                    id.into_string(),
                    Arc::clone(store),
                    engine,
                    Arc::clone(&llm),
                )
            }
            (SessionMode::New, None) => AgentSession::new(engine, Arc::clone(&llm)),
        };

        Ok((session, llm))
    }
}

pub struct App {
    session: arc_swap::ArcSwap<AgentSession>,
    llm: arc_swap::ArcSwap<SharedLlm>,
    factory: SessionFactory,
    config: Config,
    cancel_token: SharedCancelToken,
    progress_rx: Arc<tokio::sync::Mutex<mpsc::Receiver<ToolProgressChunk>>>,
    next_tool_id: Arc<Mutex<ToolCallTracker>>,
    skills: Arc<Vec<crate::skills::Skill>>,
    mcp_servers: Vec<(String, Arc<dyn motosan_agent_loop::mcp::McpServer>)>,
    /// Loaded extensions for hook dispatch. Cheap to clone (Arc).
    pub(crate) extension_registry: Arc<crate::extensions::ExtensionRegistry>,
    /// Cumulative token counts for this session. Cheap to clone (Arc).
    /// Updated after each turn's `AgentResult.usage`; not persisted across
    /// `/resume` (motosan-agent-loop 0.21.1 doesn't carry per-turn
    /// `TokenUsage` in `MessageMeta` — see spec §3.4 graceful degradation).
    pub(crate) token_tally: Arc<tokio::sync::Mutex<TurnStatsAccum>>,
    extension_diagnostics: Arc<Vec<crate::extensions::ExtensionDiagnostic>>,
    pub(crate) session_cache: Arc<crate::permissions::SessionCache>,
    /// v0.10 Phase A: current permission mode (RwLock for runtime mutation).
    /// Read by App::permission_mode(); written by App::set_permission_mode().
    pub(crate) permission_mode: Arc<tokio::sync::RwLock<crate::permissions::PermissionMode>>,
    /// v0.10 Phase A: handle to the PermissionExtension's strategy lock.
    /// Some when a PermissionExtension is registered.
    pub(crate) permission_strategy_handle: Option<Arc<tokio::sync::RwLock<PromptStrategy>>>,
    /// v0.10 Phase A: ui_tx kept on App for emitting PermissionModeChanged.
    /// Same Sender already used elsewhere (extension events, etc).
    pub(crate) ui_tx_owned: Option<mpsc::Sender<UiEvent>>,
}

impl App {
    pub fn config(&self) -> &Config {
        &self.config
    }

    /// Request graceful cancellation of the currently-running turn (if any).
    /// Safe to call from any task; the engine observes the token and halts
    /// at the next safe point.
    pub fn cancel(&self) {
        self.cancel_token.cancel();
    }

    /// Exposes the session cache so front-ends can write "Allow for session"
    /// decisions resolved by the user. Exposed as `Arc` so callers can drop
    /// their reference freely.
    pub fn permissions_cache(&self) -> Arc<crate::permissions::SessionCache> {
        Arc::clone(&self.session_cache)
    }

    /// Read-only handle on the loaded extensions registry. For the
    /// `Command::ListExtensions` dispatcher reply.
    pub fn extension_registry(&self) -> Arc<crate::extensions::ExtensionRegistry> {
        Arc::clone(&self.extension_registry)
    }

    /// Snapshot of the live settings currently used for session rebuilds.
    pub fn settings(&self) -> crate::settings::Settings {
        let mut settings = self.factory.settings();
        if let Some(model) = self.factory.current_model() {
            settings.model.name = model.to_string();
        }
        settings
    }

    /// Read-only handle on the cumulative token tally.
    pub fn token_tally(&self) -> Arc<tokio::sync::Mutex<TurnStatsAccum>> {
        Arc::clone(&self.token_tally)
    }

    /// v0.10 Phase A: change the permission mode for the current session.
    /// Mutates the PermissionExtension's strategy lock; emits
    /// UiEvent::PermissionModeChanged so RPC/TUI clients can sync.
    /// Session-only — does NOT persist to settings.toml.
    pub async fn set_permission_mode(&self, mode: crate::permissions::PermissionMode) {
        let new_strategy = match mode {
            crate::permissions::PermissionMode::Bypass => PromptStrategy::AllowAll,
            crate::permissions::PermissionMode::AcceptEdits => PromptStrategy::AcceptEdits,
            crate::permissions::PermissionMode::Prompt if self.ui_tx_owned.is_none() => {
                PromptStrategy::HeadlessDeny
            }
            crate::permissions::PermissionMode::Prompt => PromptStrategy::Prompt,
        };

        if let Some(handle) = &self.permission_strategy_handle {
            *handle.write().await = new_strategy;
        }

        *self.permission_mode.write().await = mode;

        if let Some(ui_tx) = &self.ui_tx_owned {
            let _ = ui_tx.send(UiEvent::PermissionModeChanged { mode }).await;
        }
    }

    /// v0.10 Phase B-1: emit `UiEvent::SettingsSnapshot` carrying the
    /// current live settings so clients learn startup settings without
    /// issuing `Command::ReloadSettings`.
    pub async fn emit_settings_snapshot(&self) {
        if let Some(ui_tx) = &self.ui_tx_owned {
            let _ = ui_tx
                .send(UiEvent::SettingsSnapshot {
                    settings: self.settings(),
                })
                .await;
        }
    }

    /// Read the current permission mode.
    pub async fn permission_mode(&self) -> crate::permissions::PermissionMode {
        *self.permission_mode.read().await
    }

    /// Load-time diagnostics collected while building the extension registry.
    pub fn extension_diagnostics(&self) -> Arc<Vec<crate::extensions::ExtensionDiagnostic>> {
        Arc::clone(&self.extension_diagnostics)
    }

    /// M3: the underlying motosan session id. Always populated; ephemeral
    /// sessions use a synthetic id internally.
    pub fn session_id(&self) -> String {
        self.session.load().session_id().to_string()
    }

    /// M3: snapshot of the session's persisted message history. Used by the
    /// binary on `--continue` / `--session` to seed the TUI transcript so
    /// the user can see what was said in prior runs. `AgentSession::resume`
    /// already populates this internally for the *agent* to use as context
    /// on the next turn; this method exposes it so the *front-end* can
    /// render it too. Returns motosan's `Vec<Message>` verbatim (callers
    /// decide how to map roles → UI blocks; system messages are typically
    /// dropped because they're the prompt, not transcript).
    pub async fn session_history(
        &self,
    ) -> motosan_agent_loop::Result<Vec<motosan_agent_loop::Message>> {
        self.session.load_full().history().await
    }

    /// Manually compact the current session's context. Forces a one-shot
    /// compaction (a zero-threshold `ThresholdStrategy` so `should_compact`
    /// is always true), keeping the most recent few user turns. The
    /// compaction marker is appended to the session jsonl like an
    /// autocompact event.
    pub async fn compact(&self) -> Result<()> {
        use motosan_agent_loop::ThresholdStrategy;
        let strategy = ThresholdStrategy {
            threshold: 0.0,
            ..ThresholdStrategy::default()
        };
        let llm = self.llm.load_full().client();
        self.session
            .load_full()
            .maybe_compact(&strategy, llm)
            .await
            .map_err(|e| AppError::Config(format!("compaction failed: {e}")))?;
        Ok(())
    }

    /// Replace the live session with a brand-new empty one. An in-flight
    /// turn (if any) keeps its own session snapshot and is unaffected; the
    /// next `send_user_message` uses the new session.
    pub async fn new_session(&self) -> Result<()> {
        self.fire_session_before_switch("new", None).await?;
        let (session, llm) = self.factory.build(SessionMode::New, None, None).await?;
        self.session.store(Arc::new(session));
        self.llm.store(Arc::new(SharedLlm::new(llm)));
        self.reset_token_tally().await;
        Ok(())
    }

    /// Replace the live session with a stored session loaded by id.
    /// Errors if no session store is configured or the id is unknown.
    ///
    /// **Not turn-safe** — see `switch_model`'s doc. Phase D2's command
    /// handler must gate this on no active turn.
    pub async fn load_session(&self, id: &str) -> Result<()> {
        self.fire_session_before_switch("load", Some(id)).await?;
        self.load_session_without_hook(id).await
    }

    async fn load_session_without_hook(&self, id: &str) -> Result<()> {
        let (session, llm) = self
            .factory
            .build(SessionMode::Resume(id.to_string()), None, None)
            .await?;
        self.session.store(Arc::new(session));
        self.llm.store(Arc::new(SharedLlm::new(llm)));
        self.reset_token_tally().await;
        Ok(())
    }

    async fn reset_token_tally(&self) {
        let mut tally = self.token_tally.lock().await;
        *tally = TurnStatsAccum::default();
    }

    /// Copy the live session to a brand-new, fully independent session file
    /// and switch to the copy. Unlike `/fork` (an in-file branch), the clone
    /// is a separate file and appears in the `/resume` picker. Returns the
    /// new session id. Requires a session store.
    ///
    /// **Not turn-safe** — like `load_session`, the caller (the
    /// `Command::CloneSession` arm in `forward_commands`) must gate this on
    /// no active turn.
    pub async fn clone_session(&self) -> Result<String> {
        self.fire_session_before_switch("clone", None).await?;
        let Some(store) = self.factory.session_store.as_ref() else {
            return Err(AppError::Config("clone requires a session store".into()));
        };
        let source_id = self.session.load().session_id().to_string();
        let new_id = crate::session::SessionId::new().into_string();
        let catalog = motosan_agent_loop::SessionCatalog::new(Arc::clone(store));
        catalog
            .fork(&source_id, &new_id)
            .await
            .map_err(|e| AppError::Config(format!("clone failed: {e}")))?;
        self.load_session_without_hook(&new_id).await?;
        Ok(new_id)
    }

    /// Rebuild the live session with a different model, preserving the
    /// conversation. Requires a session store (the current session is
    /// resumed by id under the new model). Errors otherwise.
    ///
    /// **Not turn-safe.** Like `load_session`, this resumes a session by
    /// id. If a turn is still in flight on the *old* session when this is
    /// called, that turn keeps writing entries under the same `session_id`
    /// while the resumed session reads from it — a write/read interleaving
    /// that the resumed session won't observe until its next reload. The
    /// `App` API does not expose a turn-in-progress flag, so the *caller*
    /// (Phase D2's `/model` / `/resume` command handlers in
    /// `forward_commands`) MUST gate these calls on no active turn — the
    /// `active_turn` bookkeeping already in `forward_commands` is exactly
    /// that gate. D2's plan must wire that guard; D1 documents the contract.
    pub async fn switch_model(&self, model: &crate::model::ModelId) -> Result<()> {
        self.fire_session_before_switch("model_switch", None)
            .await?;
        let current_id = self.session.load().session_id().to_string();
        let (session, llm) = self
            .factory
            .build(SessionMode::Resume(current_id), Some(model), None)
            .await?;
        self.factory.set_current_model(model.clone());
        self.session.store(Arc::new(session));
        self.llm.store(Arc::new(SharedLlm::new(llm)));
        Ok(())
    }

    /// Replace the live session's settings + rebuild the underlying
    /// session and LLM client. Used by the TUI `/settings` editor (via
    /// `Command::ApplySettings`) and external RPC clients (via
    /// `Command::ReloadSettings`).
    ///
    /// Does NOT fire `session_before_switch` — settings save is
    /// user-intentional + the editor is right there; blocking via an
    /// extension would be hostile UX. See v0.8 Phase B plan.
    ///
    /// Per the v0.8 Phase A mutable-accumulator audit lesson, resets
    /// `token_tally` (the cumulative count is per-session; new settings
    /// = effectively a new session even if the session id is preserved).
    ///
    /// **Not turn-safe** — like `switch_model`, the caller must gate on
    /// no active turn.
    pub async fn reload_settings(&self, new_settings: crate::settings::Settings) -> Result<()> {
        // Rebuild session + LLM with the override. SessionMode::Resume
        // preserves the session id and history; the LLM is rebuilt from
        // the new settings (could be a different provider entirely).
        let current_id = self.session.load().session_id().to_string();
        let (session, llm) = self
            .factory
            .build(SessionMode::Resume(current_id), None, Some(&new_settings))
            .await?;
        self.factory.store_settings(new_settings);
        self.factory.clear_current_model();
        self.session.store(Arc::new(session));
        self.llm.store(Arc::new(SharedLlm::new(llm)));

        // Per v0.8 Phase A audit rule (mutable-accumulator extension):
        // reset token tally on session replacement.
        self.reset_token_tally().await;

        Ok(())
    }

    /// M4 Phase B: disconnect every registered MCP server (2s per-server
    /// timeout, best-effort). Call from the binary's ctrl-C handler.
    pub async fn disconnect_mcp(&self) {
        for (name, server) in &self.mcp_servers {
            let _ =
                tokio::time::timeout(std::time::Duration::from_secs(2), server.disconnect()).await;
            tracing::debug!(target: "mcp", server = %name, "disconnected");
        }
    }

    fn run_turn(
        &self,
        msg: crate::user_message::UserMessage,
        fork_from: Option<motosan_agent_loop::EntryId>,
    ) -> impl Stream<Item = UiEvent> + Send + 'static {
        let session = self.session.load_full();
        let skills = Arc::clone(&self.skills);
        let cancel_token = self.cancel_token.clone();
        let tracker = Arc::clone(&self.next_tool_id);
        let progress = Arc::clone(&self.progress_rx);
        let token_tally = Arc::clone(&self.token_tally);
        let settings_model_name = self
            .factory
            .current_model()
            .map(|model| model.to_string())
            .unwrap_or_else(|| self.factory.settings().model.name);

        async_stream::stream! {
            // Validate + encode attachments BEFORE the single-turn guard so a
            // bad attachment error is never masked by "another turn is already
            // running". See spec §3.
            let new_user = {
                // Skill expansion only applies to the user-visible text, not
                // attachments. We thread the expanded text back into the
                // UserMessage before preparation.
                let expanded_text = crate::skills::expand::expand_skill_command(&msg.text, &skills);
                let expanded_msg = crate::user_message::UserMessage {
                    text: expanded_text,
                    attachments: msg.attachments.clone(),
                };
                match crate::user_message::prepare_user_message(&expanded_msg) {
                    Ok(m) => m,
                    Err(err) => {
                        yield UiEvent::AttachmentError {
                            kind: err.kind(),
                            message: err.to_string(),
                        };
                        return;
                    }
                }
            };

            // Single-turn guard (M2 contract preserved).
            let mut progress_guard = match progress.try_lock() {
                Ok(guard) => guard,
                Err(_) => {
                    yield UiEvent::Error(
                        "another turn is already running; capo is single-turn-per-App".into(),
                    );
                    return;
                }
            };

            // Reset cancel token for THIS turn.
            let cancel = cancel_token.reset();

            yield UiEvent::AgentTurnStarted;
            yield UiEvent::AgentThinking;

            // Start the turn (gives us a TurnHandle with stream + previous_len + branch_parent + ops_tx).
            let handle = match fork_from {
                None => {
                    // Linear turn: load history, append, start_turn.
                    let history = match session.history().await {
                        Ok(h) => h,
                        Err(err) => {
                            yield UiEvent::Error(format!("session.history failed: {err}"));
                            return;
                        }
                    };
                    let mut messages = history;
                    messages.push(new_user);
                    match session.start_turn(messages).await {
                        Ok(h) => h,
                        Err(err) => {
                            yield UiEvent::Error(format!("session.start_turn failed: {err}"));
                            return;
                        }
                    }
                }
                Some(from) => {
                    // Fork: fork_turn assembles the branch's prior history itself.
                    match session.fork_turn(from, vec![new_user]).await {
                        Ok(h) => h,
                        Err(err) => {
                            yield UiEvent::Error(format!("session.fork_turn failed: {err}"));
                            return;
                        }
                    }
                }
            };
            let previous_len = handle.previous_len;
            let epoch = handle.epoch;
            let branch_parent = handle.branch_parent;
            let ops_tx = handle.ops_tx.clone();
            let mut agent_stream = handle.stream;

            // Bridge our SharedCancelToken to motosan's AgentOp::Interrupt.
            //
            // `AgentSession::start_turn` does NOT take a CancellationToken —
            // the engine's cancel-token path used in M2's direct `.run().cancel(tok)`
            // is bypassed when going through AgentSession. The control plane is
            // `ops_tx`. We spawn a tiny task that waits on `cancel.cancelled()`
            // and forwards an `Interrupt` op when fired.
            let interrupt_bridge = tokio::spawn(async move {
                cancel.cancelled().await;
                let _ = ops_tx.send(AgentOp::Interrupt).await;
            });

            // Drain events.
            let mut terminal_messages: Option<Vec<motosan_agent_loop::Message>> = None;
            let mut terminal_result: Option<motosan_agent_loop::Result<motosan_agent_loop::AgentResult>> = None;

            loop {
                // Forward any progress chunks that arrived in this iteration.
                while let Ok(chunk) = progress_guard.try_recv() {
                    yield UiEvent::ToolCallProgress {
                        id: progress_event_id(&tracker),
                        chunk: ProgressChunk::from(chunk),
                    };
                }

                tokio::select! {
                    biased;
                    maybe_item = agent_stream.next() => {
                        match maybe_item {
                            Some(AgentStreamItem::Event(ev)) => {
                                if let Some(ui) = map_event(ev, &tracker) {
                                    yield ui;
                                }
                            }
                            Some(AgentStreamItem::Terminal(term)) => {
                                terminal_result = Some(term.result);
                                terminal_messages = Some(term.messages);
                                break;
                            }
                            None => break,
                        }
                    }
                    Some(chunk) = progress_guard.recv() => {
                        yield UiEvent::ToolCallProgress {
                            id: progress_event_id(&tracker),
                            chunk: ProgressChunk::from(chunk),
                        };
                    }
                }
            }

            // Tear down the interrupt bridge whether or not cancellation fired.
            interrupt_bridge.abort();

            // Persist new messages via record_turn_outcome (only when terminal reached).
            if let Some(msgs) = terminal_messages.as_ref() {
                if let Err(err) = session
                    .record_turn_outcome(epoch, previous_len, msgs, branch_parent)
                    .await
                {
                    yield UiEvent::Error(format!("session.record_turn_outcome: {err}"));
                }
            }

            // Translate the terminal Result into final UiEvents.
            match terminal_result {
                Some(Ok(result)) => {
                    // v0.9 Phase A: surface thinking ContentParts BEFORE the final
                    // assistant text. Spec §3.5: transcript order = thinking → text.
                    if let Some(msgs) = terminal_messages.as_ref() {
                        for ev in extract_new_thinking_events(msgs, previous_len) {
                            yield ev;
                        }
                    }

                    let final_text = terminal_messages
                        .as_ref()
                        .and_then(|msgs| {
                            msgs.iter()
                                .rev()
                                .find(|m| m.role() == motosan_agent_loop::Role::Assistant)
                                .map(|m| m.text())
                        })
                        .unwrap_or_default();
                    if !final_text.is_empty() {
                        yield UiEvent::AgentMessageComplete(final_text);
                    }
                    // Emit TurnStats after AgentMessageComplete, before
                    // AgentTurnComplete. See spec §3.2 emit ordering.
                    let usage = result.usage;
                    let (cumulative_input, cumulative_output) = {
                        let mut tally = token_tally.lock().await;
                        tally.add(usage);
                        (tally.cumulative_input, tally.cumulative_output)
                    };
                    yield UiEvent::TurnStats {
                        input_tokens: usage.input_tokens,
                        output_tokens: usage.output_tokens,
                        cumulative_input,
                        cumulative_output,
                        model: settings_model_name.clone(),
                    };
                    // Flush any remaining progress chunks.
                    while let Ok(chunk) = progress_guard.try_recv() {
                        yield UiEvent::ToolCallProgress {
                            id: progress_event_id(&tracker),
                            chunk: ProgressChunk::from(chunk),
                        };
                    }
                    yield UiEvent::AgentTurnComplete;
                }
                Some(Err(err)) => {
                    yield UiEvent::Error(format!("{err}"));
                }
                None => { /* stream closed without terminal — cancelled */ }
            }
        }
    }

    pub fn send_user_message(
        &self,
        msg: crate::user_message::UserMessage,
    ) -> impl Stream<Item = UiEvent> + Send + 'static {
        self.run_turn(msg, None)
    }

    /// Continue the conversation from an earlier entry, creating a branch.
    /// `from` is the `EntryId` to attach under (typically a past user
    /// message from `fork_candidates`). Streams `UiEvent`s exactly as
    /// `send_user_message` does.
    pub fn fork_from(
        &self,
        from: motosan_agent_loop::EntryId,
        message: crate::user_message::UserMessage,
    ) -> impl Stream<Item = UiEvent> + Send + 'static {
        let registry = Arc::clone(&self.extension_registry);
        let inner = self.run_turn(message, Some(from));
        async_stream::stream! {
            use crate::extensions::dispatcher::{dispatch_session_before_switch, HookOutcome};
            match dispatch_session_before_switch(&registry, "fork", None).await {
                HookOutcome::Continue => {}
                HookOutcome::Cancelled { extension_name, reason } => {
                    let msg = match reason {
                        Some(r) => format!("extension `{extension_name}` cancelled fork: {r}"),
                        None => format!("extension `{extension_name}` cancelled fork"),
                    };
                    yield UiEvent::Error(msg);
                    return;
                }
            }
            let mut inner = Box::pin(inner);
            while let Some(ev) = futures::StreamExt::next(&mut inner).await {
                yield ev;
            }
        }
    }

    /// User messages on the current (active) branch, newest first, each
    /// paired with its `EntryId` and a one-line preview — the rows the
    /// `/fork` picker offers.
    pub async fn fork_candidates(&self) -> Result<Vec<(motosan_agent_loop::EntryId, String)>> {
        let entries = self
            .session
            .load_full()
            .entries()
            .await
            .map_err(|e| AppError::Config(format!("entries failed: {e}")))?;
        let branch = motosan_agent_loop::active_branch(&entries);
        let mut out: Vec<(motosan_agent_loop::EntryId, String)> = branch
            .iter()
            .filter_map(|stored| {
                let msg = stored.entry.as_message()?;
                if !matches!(msg.role(), motosan_agent_loop::Role::User) {
                    return None;
                }
                let preview: String = msg
                    .text()
                    .lines()
                    .next()
                    .unwrap_or("")
                    .chars()
                    .take(80)
                    .collect();
                Some((stored.id.clone(), preview))
            })
            .collect();
        out.reverse();
        Ok(out)
    }

    /// The session log as a navigable branch tree — for the `/tree` UI.
    pub async fn branches(&self) -> Result<motosan_agent_loop::BranchTree> {
        self.session
            .load_full()
            .branches()
            .await
            .map_err(|e| AppError::Config(format!("branches failed: {e}")))
    }

    /// Fire `session_before_switch` and translate `Cancelled` into
    /// `AppError::HookCancelled`. Returns `Ok(())` on `Continue`.
    async fn fire_session_before_switch(
        &self,
        reason: &str,
        session_id: Option<&str>,
    ) -> Result<()> {
        use crate::extensions::dispatcher::{dispatch_session_before_switch, HookOutcome};
        match dispatch_session_before_switch(&self.extension_registry, reason, session_id).await {
            HookOutcome::Continue => Ok(()),
            HookOutcome::Cancelled {
                extension_name,
                reason,
            } => Err(AppError::HookCancelled {
                extension_name,
                reason,
            }),
        }
    }
}

#[derive(Debug, Default)]
struct ToolCallTracker {
    next_id: usize,
    pending: VecDeque<(String, String)>,
}

impl ToolCallTracker {
    fn start(&mut self, name: &str) -> String {
        self.next_id += 1;
        let id = format!("tool_{}", self.next_id);
        self.pending.push_back((name.to_string(), id.clone()));
        id
    }

    fn complete(&mut self, name: &str) -> String {
        if let Some(pos) = self
            .pending
            .iter()
            .position(|(pending_name, _)| pending_name == name)
        {
            if let Some((_, id)) = self.pending.remove(pos) {
                return id;
            }
        }

        self.next_id += 1;
        format!("tool_{}", self.next_id)
    }

    // ToolProgressChunk does not carry a tool-call id. When exactly one tool is
    // pending we can attribute progress safely; otherwise we must not guess from
    // queue order (for example `pending.back()`), because concurrent tool calls
    // would mislabel output.
    fn progress_id(&self) -> Option<String> {
        match self.pending.len() {
            1 => self.pending.front().map(|(_, id)| id.clone()),
            _ => None,
        }
    }
}

fn lock_tool_tracker(tracker: &Arc<Mutex<ToolCallTracker>>) -> MutexGuard<'_, ToolCallTracker> {
    match tracker.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}

fn progress_event_id(tracker: &Arc<Mutex<ToolCallTracker>>) -> String {
    lock_tool_tracker(tracker)
        .progress_id()
        .unwrap_or_else(|| "tool_unknown".to_string())
}

fn anthropic_api_key_from<F>(auth: &crate::auth::Auth, env_lookup: F) -> Option<String>
where
    F: Fn(&str) -> Option<String>,
{
    env_lookup("ANTHROPIC_API_KEY")
        .map(|key| key.trim().to_string())
        .filter(|key| !key.is_empty())
        .or_else(|| auth.api_key("anthropic").map(str::to_string))
}

fn map_event(ev: AgentEvent, tool_tracker: &Arc<Mutex<ToolCallTracker>>) -> Option<UiEvent> {
    match ev {
        AgentEvent::Core(CoreEvent::TextChunk(delta)) => Some(UiEvent::AgentTextDelta(delta)),
        AgentEvent::Core(CoreEvent::ToolStarted { name }) => {
            let id = lock_tool_tracker(tool_tracker).start(&name);
            Some(UiEvent::ToolCallStarted {
                id,
                name,
                args: serde_json::json!({}),
            })
        }
        AgentEvent::Core(CoreEvent::ToolCompleted { name, result }) => {
            let id = lock_tool_tracker(tool_tracker).complete(&name);
            Some(UiEvent::ToolCallCompleted {
                id,
                result: UiToolResult {
                    is_error: result.is_error,
                    text: format!("{name}: {result:?}"),
                },
            })
        }
        _ => None,
    }
}

type CustomToolsFactory = Box<dyn FnOnce(ToolCtx) -> Vec<Arc<dyn motosan_agent_tool::Tool>>>;

pub struct AppBuilder {
    config: Option<Config>,
    cwd: Option<PathBuf>,
    permission_gate: Option<Arc<dyn PermissionGate>>,
    install_builtin_tools: bool,
    max_iterations: usize,
    llm_override: Option<Arc<dyn LlmClient>>,
    custom_tools_factory: Option<CustomToolsFactory>,
    permissions_policy_path: Option<PathBuf>,
    ui_tx: Option<mpsc::Sender<crate::events::UiEvent>>,
    headless_permissions: bool,
    settings: Option<crate::settings::Settings>,
    auth: Option<crate::auth::Auth>,
    context_discovery_disabled: bool,
    // M3 Phase A:
    session_store: Option<Arc<dyn SessionStore>>,
    resume_session_id: Option<crate::session::SessionId>,
    autocompact_enabled: bool,
    // M4 Phase A:
    skills: Vec<crate::skills::Skill>,
    // M4 Phase B:
    extra_tools: Vec<Arc<dyn motosan_agent_tool::Tool>>,
    mcp_servers: Vec<(String, Arc<dyn motosan_agent_loop::mcp::McpServer>)>,
    extension_registry: Option<Arc<crate::extensions::ExtensionRegistry>>,
    extension_diagnostics: Option<Arc<Vec<crate::extensions::ExtensionDiagnostic>>>,
    token_tally: Option<Arc<tokio::sync::Mutex<TurnStatsAccum>>>,
}

impl Default for AppBuilder {
    fn default() -> Self {
        Self {
            config: None,
            cwd: None,
            permission_gate: None,
            install_builtin_tools: false,
            max_iterations: 20,
            llm_override: None,
            custom_tools_factory: None,
            permissions_policy_path: None,
            ui_tx: None,
            headless_permissions: false,
            settings: None,
            auth: None,
            context_discovery_disabled: false,
            session_store: None,
            resume_session_id: None,
            autocompact_enabled: false,
            skills: Vec::new(),
            extra_tools: Vec::new(),
            mcp_servers: Vec::new(),
            extension_registry: None,
            extension_diagnostics: None,
            token_tally: None,
        }
    }
}

impl AppBuilder {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn with_config(mut self, cfg: Config) -> Self {
        self.config = Some(cfg);
        self
    }

    pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
        self.cwd = Some(cwd.into());
        self
    }

    pub fn with_permission_gate(mut self, gate: Arc<dyn PermissionGate>) -> Self {
        self.permission_gate = Some(gate);
        self
    }

    /// Install Capo's builtin tools (`read`, `bash`).
    ///
    /// This is mutually exclusive with `with_custom_tools_factory` /
    /// `build_with_custom_tools`; `build()` returns a configuration error if
    /// both are set.
    pub fn with_builtin_tools(mut self) -> Self {
        self.install_builtin_tools = true;
        self
    }

    pub fn with_max_iterations(mut self, n: usize) -> Self {
        self.max_iterations = n;
        self
    }

    pub fn with_llm(mut self, llm: Arc<dyn LlmClient>) -> Self {
        self.llm_override = Some(llm);
        self
    }

    pub fn with_permissions_config(mut self, path: PathBuf) -> Self {
        self.permissions_policy_path = Some(path);
        self
    }

    pub fn with_ui_channel(mut self, tx: mpsc::Sender<UiEvent>) -> Self {
        self.ui_tx = Some(tx);
        self
    }

    /// Test/library hook to inject a pre-seeded token tally.
    pub fn with_token_tally(mut self, tally: Arc<tokio::sync::Mutex<TurnStatsAccum>>) -> Self {
        self.token_tally = Some(tally);
        self
    }

    /// Register `PermissionExtension` in headless-deny mode — `permissions.toml`
    /// and read-only auto-allow still apply, but a tool call that would
    /// otherwise prompt is denied (there is no UI to ask). Used by `--json`.
    /// Ignored if `with_ui_channel` is also set (an interactive channel wins).
    pub fn with_headless_permissions(mut self) -> Self {
        self.headless_permissions = true;
        self
    }

    /// M3: install user `Settings`. Replaces `with_config` for new code.
    pub fn with_settings(mut self, settings: crate::settings::Settings) -> Self {
        self.settings = Some(settings);
        self
    }

    /// M3: install `Auth` (credentials for LLM providers).
    pub fn with_auth(mut self, auth: crate::auth::Auth) -> Self {
        self.auth = Some(auth);
        self
    }

    /// M3: disable AGENTS.md / CLAUDE.md discovery for this App.
    /// Opt-out — discovery is on by default in `build()`.
    pub fn disable_context_discovery(mut self) -> Self {
        self.context_discovery_disabled = true;
        self
    }

    /// M3 Phase A: install a `SessionStore` (e.g. `motosan_agent_loop::FileSessionStore`)
    /// for persistence. When omitted, sessions are ephemeral (no jsonl on disk).
    pub fn with_session_store(mut self, store: Arc<dyn SessionStore>) -> Self {
        self.session_store = Some(store);
        self
    }

    /// M3 Phase A: enable autocompact at the `Settings::session.compact_at_context_pct`
    /// threshold. Requires `with_settings` to have been called; otherwise uses
    /// `Settings::default()`. Settings provide `max_context_tokens` and
    /// `keep_turns`. No-op when `settings.session.compact_at_context_pct == 0.0`.
    pub fn with_autocompact(mut self) -> Self {
        self.autocompact_enabled = true;
        self
    }

    /// M4 Phase A: register skills. Pass an empty Vec (or omit the call,
    /// or call `without_skills()`) to disable skill injection. Skills are
    /// rendered into the system prompt's `<available_skills>` block when
    /// the `read` tool is available, and matched against `/skill:<name>`
    /// expansion before user messages reach the LLM.
    pub fn with_skills(mut self, skills: Vec<crate::skills::Skill>) -> Self {
        self.skills = skills;
        self
    }

    pub fn without_skills(mut self) -> Self {
        self.skills.clear();
        self
    }

    /// M4 Phase B: register additional tools (typically MCP). Unlike
    /// `with_custom_tools_factory`, this APPENDS to the builtin tools
    /// and does NOT replace them.
    pub fn with_extra_tools(mut self, tools: Vec<Arc<dyn motosan_agent_tool::Tool>>) -> Self {
        self.extra_tools = tools;
        self
    }

    pub fn with_extension_registry(
        mut self,
        registry: Arc<crate::extensions::ExtensionRegistry>,
    ) -> Self {
        self.extension_registry = Some(registry);
        self
    }

    pub fn with_extension_diagnostics(
        mut self,
        diagnostics: Arc<Vec<crate::extensions::ExtensionDiagnostic>>,
    ) -> Self {
        self.extension_diagnostics = Some(diagnostics);
        self
    }

    /// M4 Phase B: register MCP server handles so `App::disconnect_mcp`
    /// can iterate them on shutdown. Storage only — does not connect.
    pub fn with_mcp_servers(
        mut self,
        servers: Vec<(String, Arc<dyn motosan_agent_loop::mcp::McpServer>)>,
    ) -> Self {
        self.mcp_servers = servers;
        self
    }

    /// Install a custom tool set for this app.
    ///
    /// This is mutually exclusive with `with_builtin_tools`; `build()` returns
    /// a configuration error if both are set.
    pub fn with_custom_tools_factory(
        mut self,
        factory: impl FnOnce(ToolCtx) -> Vec<Arc<dyn motosan_agent_tool::Tool>> + 'static,
    ) -> Self {
        self.custom_tools_factory = Some(Box::new(factory));
        self
    }

    /// Convenience wrapper for `with_custom_tools_factory(...).build()`.
    ///
    /// This is mutually exclusive with `with_builtin_tools`.
    pub async fn build_with_custom_tools(
        self,
        factory: impl FnOnce(ToolCtx) -> Vec<Arc<dyn motosan_agent_tool::Tool>> + 'static,
    ) -> Result<App> {
        self.with_custom_tools_factory(factory).build().await
    }

    /// M3 Phase A: build the App with an optional resume session id.
    ///
    /// - `Some(id)` + `session_store: Some(_)` → `AgentSession::resume(id, store, engine, llm)`,
    ///   then replay history into `ToolCtx.read_files`.
    /// - `Some(id)` + `session_store: None` → error (resume requires a store).
    /// - `None` + `session_store: Some(_)` → `AgentSession::new_with_store(fresh_id, store, engine, llm)`.
    /// - `None` + `session_store: None` → `AgentSession::new(engine, llm)` (ephemeral).
    pub async fn build_with_session(
        mut self,
        resume: Option<crate::session::SessionId>,
    ) -> Result<App> {
        if let Some(id) = resume {
            if self.session_store.is_none() {
                return Err(AppError::Config(
                    "build_with_session(Some(id)) requires with_session_store(...)".into(),
                ));
            }
            self.resume_session_id = Some(id);
        }
        self.build_internal().await
    }

    /// Legacy entry point; equivalent to `build_with_session(None)`.
    pub async fn build(self) -> Result<App> {
        self.build_with_session(None).await
    }

    async fn build_internal(mut self) -> Result<App> {
        let mcp_servers = std::mem::take(&mut self.mcp_servers);
        let extra_tools = std::mem::take(&mut self.extra_tools);
        let skills = self.skills.clone();
        if self.install_builtin_tools && self.custom_tools_factory.is_some() {
            return Err(AppError::Config(
                "with_builtin_tools and with_custom_tools_factory are mutually exclusive".into(),
            ));
        }

        // Synthesise the legacy `Config` so the rest of `build()` keeps
        // working without a wholesale rewrite. Settings + Auth override the
        // deprecated `Config` when both APIs are supplied.
        let has_config = self.config.is_some();
        let has_auth = self.auth.is_some();
        let mut config = self.config.unwrap_or_default();
        let settings = match self.settings {
            Some(settings) => settings,
            None => {
                let mut settings = crate::settings::Settings::default();
                settings.model.provider = config.model.provider.clone();
                settings.model.name = config.model.name.clone();
                settings.model.max_tokens = config.model.max_tokens;
                settings
            }
        };
        config.model.provider = settings.model.provider.clone();
        config.model.name = settings.model.name.clone();
        config.model.max_tokens = settings.model.max_tokens;
        let mut auth = self.auth.unwrap_or_default();
        if !has_auth {
            if let Some(key) = config.anthropic.api_key.as_deref() {
                auth.0.insert(
                    "anthropic".into(),
                    crate::auth::ProviderAuth::ApiKey {
                        key: key.to_string(),
                    },
                );
            }
        }
        let env_or_auth_key = anthropic_api_key_from(&auth, |name| std::env::var(name).ok());
        if env_or_auth_key.is_some() || has_auth || !has_config {
            config.anthropic.api_key = env_or_auth_key;
        }
        let cwd = self
            .cwd
            .or_else(|| std::env::current_dir().ok())
            .unwrap_or_else(|| PathBuf::from("."));
        let agent_dir = crate::paths::agent_dir();
        let permission_gate = self.permission_gate.unwrap_or_else(|| {
            // When no gate is provided *and* no ui channel is wired,
            // fall back to NoOp with a warning log; when ui channel IS
            // wired, the PermissionExtension handles the real decisions.
            if self.ui_tx.is_some() || self.headless_permissions {
                Arc::new(NoOpPermissionGate) as Arc<dyn PermissionGate>
            } else {
                tracing::warn!("no PermissionGate and no UI channel — tools run unchecked");
                Arc::new(NoOpPermissionGate) as Arc<dyn PermissionGate>
            }
        });

        // Permissions.
        let policy: Arc<crate::permissions::Policy> =
            Arc::new(match self.permissions_policy_path.as_ref() {
                Some(path) => crate::permissions::Policy::load_or_default(path)?,
                None => crate::permissions::Policy::default(),
            });
        let session_cache = Arc::new(crate::permissions::SessionCache::new());
        let permission_strategy_handle = if self.ui_tx.is_some() || self.headless_permissions {
            let initial_strategy = if self.ui_tx.is_some() {
                PromptStrategy::Prompt
            } else {
                PromptStrategy::HeadlessDeny
            };
            Some(Arc::new(tokio::sync::RwLock::new(initial_strategy)))
        } else {
            None
        };
        let permission_mode = Arc::new(tokio::sync::RwLock::new(
            crate::permissions::PermissionMode::default(),
        ));
        let token_tally = self
            .token_tally
            .unwrap_or_else(|| Arc::new(tokio::sync::Mutex::new(TurnStatsAccum::default())));

        // Shared progress channel consumed by `send_user_message`.
        let (progress_tx, progress_rx) = mpsc::channel::<ToolProgressChunk>(64);

        // Resolve the tool set ONCE. Builtin tools, or a custom factory's
        // output, plus MCP `extra_tools`. A FnOnce custom factory can't be
        // re-run per session, so its output is captured here as a Vec.
        // `probe_ctx` also yields the build-time cancel token for `App`.
        let probe_ctx = ToolCtx::new(&cwd, Arc::clone(&permission_gate), progress_tx.clone());
        let cancel_token = probe_ctx.cancel_token.clone();
        let (install_builtin, factory_extra_tools): (bool, Vec<Arc<dyn motosan_agent_tool::Tool>>) =
            if let Some(factory_fn) = self.custom_tools_factory.take() {
                let mut t = factory_fn(probe_ctx);
                t.extend(extra_tools.clone());
                (false, t)
            } else {
                (self.install_builtin_tools, extra_tools.clone())
            };

        let factory = SessionFactory {
            cwd: cwd.clone(),
            settings: Arc::new(Mutex::new(settings.clone())),
            auth: auth.clone(),
            policy: Arc::clone(&policy),
            session_cache: Arc::clone(&session_cache),
            ui_tx: self.ui_tx.clone(),
            headless_permissions: self.headless_permissions,
            permission_gate: Arc::clone(&permission_gate),
            permission_strategy_handle: permission_strategy_handle.clone(),
            progress_tx: progress_tx.clone(),
            skills: Arc::new(skills.clone()),
            install_builtin_tools: install_builtin,
            extra_tools: factory_extra_tools,
            max_iterations: self.max_iterations,
            context_discovery_disabled: self.context_discovery_disabled,
            autocompact_enabled: self.autocompact_enabled,
            session_store: self.session_store.clone(),
            llm_override: self.llm_override.clone(),
            current_model: Arc::new(Mutex::new(None)),
            cancel_token: cancel_token.clone(),
        };

        let mode = match self.resume_session_id.take() {
            Some(id) => SessionMode::Resume(id.into_string()),
            None => SessionMode::New,
        };
        let (session, llm) = factory.build(mode, None, None).await?;
        let (extension_registry, extension_diagnostics) =
            if let Some(reg) = self.extension_registry.take() {
                let diagnostics = self
                    .extension_diagnostics
                    .unwrap_or_else(|| Arc::new(Vec::new()));
                (reg, diagnostics)
            } else {
                let manifest_path = crate::paths::extensions_manifest_path(&agent_dir);
                let (registry, diagnostics) =
                    crate::extensions::load_extensions_manifest(&manifest_path).await;
                for d in &diagnostics {
                    let message = d.message.as_str();
                    match d.severity {
                        crate::extensions::DiagnosticSeverity::Warn => {
                            tracing::warn!("extensions: {message}");
                        }
                        crate::extensions::DiagnosticSeverity::Error => {
                            tracing::error!("extensions: {message}");
                        }
                    }
                }
                (Arc::new(registry), Arc::new(diagnostics))
            };

        Ok(App {
            session: arc_swap::ArcSwap::from_pointee(session),
            llm: arc_swap::ArcSwap::from_pointee(SharedLlm::new(llm)),
            factory,
            config,
            cancel_token,
            progress_rx: Arc::new(tokio::sync::Mutex::new(progress_rx)),
            next_tool_id: Arc::new(Mutex::new(ToolCallTracker::default())),
            skills: Arc::new(skills),
            mcp_servers,
            extension_registry,
            token_tally,
            extension_diagnostics,
            session_cache,
            permission_mode,
            permission_strategy_handle,
            ui_tx_owned: self.ui_tx.clone(),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{AnthropicConfig, ModelConfig};
    use crate::events::UiEvent;
    use crate::user_message::UserMessage;
    use async_trait::async_trait;
    use motosan_agent_loop::{ChatOutput, LlmClient, LlmResponse, Message, ToolCallItem};
    use motosan_agent_tool::ToolDef;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[test]
    fn turn_stats_accum_accumulates_across_calls() {
        let mut accum = TurnStatsAccum::default();
        accum.add(motosan_agent_loop::TokenUsage {
            input_tokens: 100,
            output_tokens: 25,
        });
        accum.add(motosan_agent_loop::TokenUsage {
            input_tokens: 1000,
            output_tokens: 250,
        });
        assert_eq!(accum.cumulative_input, 1100);
        assert_eq!(accum.cumulative_output, 275);
        assert_eq!(accum.turn_count, 2);
    }

    #[test]
    fn turn_stats_accum_saturates_on_overflow() {
        let mut accum = TurnStatsAccum {
            cumulative_input: u64::MAX - 5,
            cumulative_output: 0,
            turn_count: 1,
        };
        accum.add(motosan_agent_loop::TokenUsage {
            input_tokens: 100,
            output_tokens: 0,
        });
        assert_eq!(accum.cumulative_input, u64::MAX);
    }

    #[test]
    fn extract_thinking_events_extracts_reasoning_in_source_order() {
        use motosan_agent_loop::{new_message_id, AssistantContent, MessageMeta};
        let messages = vec![Message::Assistant {
            id: new_message_id(),
            meta: MessageMeta::default(),
            content: vec![
                AssistantContent::Reasoning {
                    text: "first thought".into(),
                    signature: None,
                },
                AssistantContent::Text {
                    text: "answer 1".into(),
                },
                AssistantContent::Reasoning {
                    text: "second thought".into(),
                    signature: None,
                },
            ],
        }];
        let events = extract_thinking_events(&messages);
        assert_eq!(events.len(), 2);
        match (&events[0], &events[1]) {
            (UiEvent::ThinkingComplete { text: t1 }, UiEvent::ThinkingComplete { text: t2 }) => {
                assert_eq!(t1, "first thought");
                assert_eq!(t2, "second thought");
            }
            _ => panic!("expected two ThinkingComplete events"),
        }
    }

    #[test]
    fn extract_new_thinking_events_skips_historical_reasoning_before_previous_len() {
        use motosan_agent_loop::{new_message_id, AssistantContent, ContentPart, MessageMeta};
        let messages = vec![
            Message::User {
                id: new_message_id(),
                meta: MessageMeta::default(),
                content: vec![ContentPart::text("old question")],
            },
            Message::Assistant {
                id: new_message_id(),
                meta: MessageMeta::default(),
                content: vec![AssistantContent::Reasoning {
                    text: "old thought".into(),
                    signature: None,
                }],
            },
            Message::User {
                id: new_message_id(),
                meta: MessageMeta::default(),
                content: vec![ContentPart::text("new question")],
            },
            Message::Assistant {
                id: new_message_id(),
                meta: MessageMeta::default(),
                content: vec![AssistantContent::Reasoning {
                    text: "new thought".into(),
                    signature: None,
                }],
            },
        ];

        let events = extract_new_thinking_events(&messages, 2);
        assert_eq!(events.len(), 1, "should only emit current-turn reasoning");
        match &events[0] {
            UiEvent::ThinkingComplete { text } => assert_eq!(text, "new thought"),
            other => panic!("expected ThinkingComplete; got {other:?}"),
        }
    }

    #[test]
    fn extract_thinking_events_skips_non_assistant_and_non_reasoning() {
        use motosan_agent_loop::{new_message_id, AssistantContent, ContentPart, MessageMeta};
        let messages = vec![
            Message::User {
                id: new_message_id(),
                meta: MessageMeta::default(),
                content: vec![ContentPart::text("question?")],
            },
            Message::Assistant {
                id: new_message_id(),
                meta: MessageMeta::default(),
                content: vec![AssistantContent::Text {
                    text: "answer".into(),
                }],
            },
        ];
        let events = extract_thinking_events(&messages);
        assert!(events.is_empty(), "expected no events; got {events:?}");
    }

    #[tokio::test]
    async fn builder_fails_without_api_key() {
        let cfg = Config {
            anthropic: AnthropicConfig {
                api_key: None,
                base_url: "https://api.anthropic.com".into(),
            },
            model: ModelConfig {
                provider: "anthropic".into(),
                name: "claude-sonnet-4-6".into(),
                max_tokens: 4096,
            },
        };
        let err = match AppBuilder::new()
            .with_config(cfg)
            .with_builtin_tools()
            .build()
            .await
        {
            Ok(_) => panic!("must fail without key"),
            Err(err) => err,
        };
        assert!(format!("{err}").contains("ANTHROPIC_API_KEY"));
    }

    struct ToolOnlyLlm {
        turn: AtomicUsize,
    }

    #[async_trait]
    impl LlmClient for ToolOnlyLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            let turn = self.turn.fetch_add(1, Ordering::SeqCst);
            if turn == 0 {
                Ok(ChatOutput::new(LlmResponse::ToolCalls(vec![
                    ToolCallItem {
                        id: "t1".into(),
                        name: "read".into(),
                        args: serde_json::json!({"path":"nope.txt"}),
                    },
                ])))
            } else {
                Ok(ChatOutput::new(LlmResponse::Message(String::new())))
            }
        }
    }

    #[tokio::test]
    async fn empty_final_message_is_not_emitted() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(std::sync::Arc::new(ToolOnlyLlm {
                turn: AtomicUsize::new(0),
            }))
            .build()
            .await
            .expect("build");
        let events: Vec<UiEvent> =
            futures::StreamExt::collect(app.send_user_message(UserMessage::text("x"))).await;
        let empties = events
            .iter()
            .filter(|e| matches!(e, UiEvent::AgentMessageComplete(t) if t.is_empty()))
            .count();
        assert_eq!(
            empties, 0,
            "should not emit empty final message, got: {events:?}"
        );
    }

    struct EchoLlm;

    struct UsageLlm {
        responses: std::sync::Mutex<VecDeque<ChatOutput>>,
    }

    #[async_trait]
    impl LlmClient for UsageLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            let next = match self.responses.lock() {
                Ok(mut responses) => responses.pop_front(),
                Err(poisoned) => poisoned.into_inner().pop_front(),
            };
            Ok(next.unwrap_or_else(|| panic!("UsageLlm script exhausted")))
        }
    }

    #[async_trait]
    impl LlmClient for EchoLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            Ok(ChatOutput::new(LlmResponse::Message("ok".into())))
        }
    }

    #[tokio::test]
    async fn turn_stats_emitted_with_cumulative_after_each_turn() {
        use motosan_agent_loop::{LlmResponse, TokenUsage};

        let dir = tempfile::tempdir().expect("tempdir");
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let llm = Arc::new(UsageLlm {
            responses: std::sync::Mutex::new(VecDeque::from([
                ChatOutput::with_usage(
                    LlmResponse::Message("first".into()),
                    TokenUsage {
                        input_tokens: 100,
                        output_tokens: 50,
                    },
                ),
                ChatOutput::with_usage(
                    LlmResponse::Message("second".into()),
                    TokenUsage {
                        input_tokens: 200,
                        output_tokens: 80,
                    },
                ),
            ])),
        });
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm)
            .build()
            .await
            .expect("build");

        let events_1: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("hi"))
            .collect()
            .await;
        let events_2: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("again"))
            .collect()
            .await;

        let ts1 = events_1
            .iter()
            .find_map(|e| match e {
                UiEvent::TurnStats {
                    input_tokens,
                    output_tokens,
                    cumulative_input,
                    cumulative_output,
                    ..
                } => Some((
                    *input_tokens,
                    *output_tokens,
                    *cumulative_input,
                    *cumulative_output,
                )),
                _ => None,
            })
            .expect("turn 1 had no TurnStats");
        assert_eq!(ts1, (100, 50, 100, 50), "turn 1 stats");

        let ts2 = events_2
            .iter()
            .find_map(|e| match e {
                UiEvent::TurnStats {
                    input_tokens,
                    output_tokens,
                    cumulative_input,
                    cumulative_output,
                    ..
                } => Some((
                    *input_tokens,
                    *output_tokens,
                    *cumulative_input,
                    *cumulative_output,
                )),
                _ => None,
            })
            .expect("turn 2 had no TurnStats");
        assert_eq!(ts2, (200, 80, 300, 130), "turn 2 stats");

        let positions: Vec<&str> = events_1
            .iter()
            .filter_map(|e| match e {
                UiEvent::AgentMessageComplete(_) => Some("msg_complete"),
                UiEvent::TurnStats { .. } => Some("stats"),
                UiEvent::AgentTurnComplete => Some("turn_complete"),
                _ => None,
            })
            .collect();
        assert_eq!(
            positions,
            vec!["msg_complete", "stats", "turn_complete"],
            "wrong ordering"
        );
    }

    #[tokio::test]
    async fn turn_stats_reset_after_new_session() {
        use motosan_agent_loop::{LlmResponse, TokenUsage};

        let dir = tempfile::tempdir().expect("tempdir");
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let llm = Arc::new(UsageLlm {
            responses: std::sync::Mutex::new(VecDeque::from([
                ChatOutput::with_usage(
                    LlmResponse::Message("first".into()),
                    TokenUsage {
                        input_tokens: 100,
                        output_tokens: 50,
                    },
                ),
                ChatOutput::with_usage(
                    LlmResponse::Message("after-new".into()),
                    TokenUsage {
                        input_tokens: 7,
                        output_tokens: 3,
                    },
                ),
            ])),
        });
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm)
            .build()
            .await
            .expect("build");

        let _: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("hi"))
            .collect()
            .await;
        app.new_session().await.expect("new session");
        let events: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("after new"))
            .collect()
            .await;
        let stats = events
            .iter()
            .find_map(|event| match event {
                UiEvent::TurnStats {
                    input_tokens,
                    output_tokens,
                    cumulative_input,
                    cumulative_output,
                    ..
                } => Some((
                    *input_tokens,
                    *output_tokens,
                    *cumulative_input,
                    *cumulative_output,
                )),
                _ => None,
            })
            .expect("turn had no TurnStats");
        assert_eq!(stats, (7, 3, 7, 3));
    }

    #[tokio::test]
    async fn with_headless_permissions_builds_an_app() {
        let dir = tempfile::tempdir().expect("tempdir");
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_headless_permissions()
            .build()
            .await
            .expect("build");
        // Build succeeds and yields a usable session id.
        assert!(!app.session_id().is_empty());
    }

    #[tokio::test]
    async fn new_session_swaps_in_a_fresh_empty_session() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .build()
            .await
            .expect("build");

        let _: Vec<_> = app
            .send_user_message(UserMessage::text("hello"))
            .collect()
            .await;
        let id_before = app.session_id();
        assert!(!app.session_history().await.expect("history").is_empty());

        app.new_session().await.expect("new_session");

        assert_ne!(app.session_id(), id_before, "a fresh session has a new id");
        assert!(
            app.session_history().await.expect("history").is_empty(),
            "fresh session has no history"
        );
    }

    #[tokio::test]
    async fn load_session_restores_a_stored_session_by_id() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let store_dir = dir.path().join("sessions");
        let store: Arc<dyn SessionStore> =
            Arc::new(motosan_agent_loop::FileSessionStore::new(store_dir));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(Arc::clone(&store))
            .build()
            .await
            .expect("build");

        let _: Vec<_> = app
            .send_user_message(UserMessage::text("remember this"))
            .collect()
            .await;
        let original_id = app.session_id();

        app.new_session().await.expect("new_session");
        assert_ne!(app.session_id(), original_id);

        app.load_session(&original_id).await.expect("load_session");
        assert_eq!(app.session_id(), original_id);
        let history = app.session_history().await.expect("history");
        assert!(
            history.iter().any(|m| m.text().contains("remember this")),
            "loaded session should carry the original turn"
        );
    }

    #[tokio::test]
    async fn clone_session_copies_to_a_new_id_and_switches_to_it() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        let _: Vec<_> = app
            .send_user_message(UserMessage::text("hello"))
            .collect()
            .await;
        let original_id = app.session_id();

        let new_id = app.clone_session().await.expect("clone_session");

        // The clone has a distinct id, and the live session switched to it.
        assert_ne!(new_id, original_id);
        assert_eq!(app.session_id(), new_id);
        // The copy carries the same conversation.
        let history = app.session_history().await.expect("history");
        assert!(history.iter().any(|m| m.text().contains("hello")));
    }

    #[tokio::test]
    async fn fork_from_creates_a_branch_off_an_earlier_entry() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        // Two linear turns.
        let _: Vec<_> = app
            .send_user_message(UserMessage::text("first"))
            .collect()
            .await;
        let _: Vec<_> = app
            .send_user_message(UserMessage::text("second"))
            .collect()
            .await;

        // The EntryId of the first user message.
        let entries = app.session.load_full().entries().await.expect("entries");
        let first_id = entries
            .iter()
            .find_map(|stored| {
                let msg = stored.entry.as_message()?;
                (msg.role() == motosan_agent_loop::Role::User && msg.text().contains("first"))
                    .then(|| stored.id.clone())
            })
            .expect("first user message present");

        // Fork from it.
        let _: Vec<_> = app
            .fork_from(first_id, UserMessage::text("branched"))
            .collect()
            .await;

        let history = app.session_history().await.expect("history");
        let texts: Vec<String> = history.iter().map(|m| m.text()).collect();
        assert!(
            texts.iter().any(|t| t.contains("first")),
            "fork keeps the fork-point ancestor"
        );
        assert!(
            texts.iter().any(|t| t.contains("branched")),
            "fork includes the new message"
        );
        assert!(
            !texts.iter().any(|t| t.contains("second")),
            "fork excludes the abandoned branch"
        );
    }

    #[tokio::test]
    async fn fork_candidates_lists_active_branch_user_messages_newest_first() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        let _: Vec<_> = app
            .send_user_message(UserMessage::text("alpha"))
            .collect()
            .await;
        let _: Vec<_> = app
            .send_user_message(UserMessage::text("bravo"))
            .collect()
            .await;

        let candidates = app.fork_candidates().await.expect("candidates");
        let previews: Vec<&str> = candidates.iter().map(|(_, p)| p.as_str()).collect();
        // Newest first.
        assert!(previews[0].contains("bravo"), "got {previews:?}");
        assert!(previews.iter().any(|p| p.contains("alpha")));
        // Every id is non-empty and distinct.
        assert!(candidates.iter().all(|(id, _)| !id.is_empty()));
    }

    #[tokio::test]
    async fn branches_returns_a_tree_for_a_linear_session() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        let _: Vec<_> = app
            .send_user_message(UserMessage::text("hello"))
            .collect()
            .await;
        let tree = app.branches().await.expect("branches");
        // A linear 1-turn session: non-empty node list, an active leaf.
        assert!(!tree.nodes.is_empty());
        assert!(tree.active_leaf.is_some());
    }

    #[tokio::test]
    async fn reload_settings_rebuilds_session_and_resets_token_tally() {
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        let llm = Arc::new(UsageLlm {
            responses: std::sync::Mutex::new(VecDeque::from([
                ChatOutput::with_usage(
                    LlmResponse::Message("first".into()),
                    motosan_agent_loop::TokenUsage {
                        input_tokens: 100,
                        output_tokens: 50,
                    },
                ),
                ChatOutput::with_usage(
                    LlmResponse::Message("after-reload".into()),
                    motosan_agent_loop::TokenUsage {
                        input_tokens: 5,
                        output_tokens: 2,
                    },
                ),
            ])),
        });
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        let _: Vec<UiEvent> = app
            .send_user_message(crate::user_message::UserMessage::text("hi"))
            .collect()
            .await;
        {
            let token_tally = app.token_tally();
            let tally = token_tally.lock().await;
            assert_eq!(tally.cumulative_input, 100);
            assert_eq!(tally.cumulative_output, 50);
            assert_eq!(tally.turn_count, 1);
        }

        let mut new_settings = crate::settings::Settings::default();
        new_settings.model.name = "claude-opus-4-7".into();
        new_settings.ui.footer_show_cost = false;
        app.reload_settings(new_settings).await.expect("reload");
        assert_eq!(app.factory.settings().model.name, "claude-opus-4-7");
        assert!(!app.factory.settings().ui.footer_show_cost);
        assert_eq!(app.factory.current_model(), None);

        {
            let token_tally = app.token_tally();
            let tally = token_tally.lock().await;
            assert_eq!(tally.cumulative_input, 0, "tally not reset on reload");
            assert_eq!(tally.cumulative_output, 0);
            assert_eq!(tally.turn_count, 0);
        }

        let _: Vec<UiEvent> = app
            .send_user_message(crate::user_message::UserMessage::text("after"))
            .collect()
            .await;
        {
            let token_tally = app.token_tally();
            let tally = token_tally.lock().await;
            assert_eq!(tally.cumulative_input, 5);
            assert_eq!(tally.cumulative_output, 2);
            assert_eq!(tally.turn_count, 1);
        }
    }

    #[tokio::test]
    async fn switch_model_preserves_history() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        let _: Vec<_> = app
            .send_user_message(UserMessage::text("keep me"))
            .collect()
            .await;
        let id_before = app.session_id();

        app.switch_model(&crate::model::ModelId::from("claude-opus-4-7"))
            .await
            .expect("switch_model");

        assert_eq!(
            app.session_id(),
            id_before,
            "switch_model keeps the same session"
        );
        let history = app.session_history().await.expect("history");
        assert!(history.iter().any(|m| m.text().contains("keep me")));
    }

    #[tokio::test]
    async fn switch_model_is_sticky_for_future_session_rebuilds() {
        let dir = tempfile::tempdir().expect("tempdir");
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("s"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(EchoLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        let selected = crate::model::ModelId::from("claude-opus-4-7");
        app.switch_model(&selected).await.expect("switch_model");
        app.new_session().await.expect("new_session");

        assert_eq!(app.factory.current_model(), Some(selected.clone()));
        assert_eq!(app.settings().model.name, selected.to_string());
    }

    struct SleepThenDoneLlm {
        turn: AtomicUsize,
    }

    #[async_trait]
    impl LlmClient for SleepThenDoneLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            let turn = self.turn.fetch_add(1, Ordering::SeqCst);
            if turn == 0 {
                Ok(ChatOutput::new(LlmResponse::ToolCalls(vec![
                    ToolCallItem {
                        id: "sleep".into(),
                        name: "bash".into(),
                        args: serde_json::json!({"command":"sleep 5", "timeout_ms": 10000}),
                    },
                ])))
            } else {
                Ok(ChatOutput::new(LlmResponse::Message("done".into())))
            }
        }
    }

    #[tokio::test]
    async fn cancel_reaches_builtin_tools_after_session_rebuild() {
        use futures::StreamExt;
        let dir = tempfile::tempdir().expect("tempdir");
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = Arc::new(
            AppBuilder::new()
                .with_config(config)
                .with_cwd(dir.path())
                .with_builtin_tools()
                .with_llm(Arc::new(SleepThenDoneLlm {
                    turn: AtomicUsize::new(0),
                }) as Arc<dyn LlmClient>)
                .build()
                .await
                .expect("build"),
        );

        app.new_session().await.expect("new_session");
        let running_app = Arc::clone(&app);
        let handle = tokio::spawn(async move {
            running_app
                .send_user_message(UserMessage::text("run a slow command"))
                .collect::<Vec<_>>()
                .await
        });
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        app.cancel();

        let events = tokio::time::timeout(std::time::Duration::from_secs(2), handle)
            .await
            .expect("turn should finish after cancellation")
            .expect("join");
        assert!(
            events.iter().any(|event| {
                matches!(
                    event,
                    UiEvent::ToolCallCompleted { result, .. }
                        if result.text.contains("command cancelled by user")
                )
            }),
            "cancel should reach the rebuilt bash tool: {events:?}"
        );
    }

    #[tokio::test]
    async fn compact_summarizes_a_session_with_enough_history() {
        struct DoneLlm;
        #[async_trait]
        impl LlmClient for DoneLlm {
            async fn chat(
                &self,
                _messages: &[Message],
                _tools: &[ToolDef],
            ) -> motosan_agent_loop::Result<ChatOutput> {
                Ok(ChatOutput::new(LlmResponse::Message("done".into())))
            }
        }

        let dir = tempfile::tempdir().expect("tempdir");
        let store = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("sessions"),
        ));
        let mut config = Config::default();
        config.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(config)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(DoneLlm) as Arc<dyn LlmClient>)
            .with_session_store(store)
            .build()
            .await
            .expect("build");

        // Run 4 user turns. `App::compact()` uses ThresholdStrategy with the
        // default keep_turns (3); `select_cutoff` only returns a cutoff once
        // there are >= keep_turns user messages — so a 1-turn session would
        // make compact() a no-op (Ok, but nothing summarized). 4 turns
        // guarantees the real compaction path (and the DoneLlm summarizer
        // call) is exercised.
        for i in 0..4 {
            let _: Vec<_> = app
                .send_user_message(UserMessage::text(format!("turn {i}")))
                .collect()
                .await;
        }

        app.compact().await.expect("compact should succeed");

        // The compaction marker is appended as a session entry — history
        // after compaction is shorter than the raw 4-turn transcript.
        let history = app.session_history().await.expect("history");
        assert!(
            !history.is_empty(),
            "session should still have content post-compaction"
        );
    }

    #[test]
    fn anthropic_env_api_key_overrides_auth_json_key() {
        let mut auth = crate::auth::Auth::default();
        auth.0.insert(
            "anthropic".into(),
            crate::auth::ProviderAuth::ApiKey {
                key: "sk-auth".into(),
            },
        );

        let key = anthropic_api_key_from(&auth, |name| {
            (name == "ANTHROPIC_API_KEY").then(|| " sk-env ".to_string())
        });
        assert_eq!(key.as_deref(), Some("sk-env"));
    }

    #[tokio::test]
    async fn with_settings_overrides_deprecated_config_model() {
        use crate::settings::Settings;

        let mut config = Config::default();
        config.model.name = "from-config".into();
        config.anthropic.api_key = Some("sk-config".into());

        let mut settings = Settings::default();
        settings.model.name = "from-settings".into();

        let tmp = tempfile::tempdir().unwrap();
        let app = AppBuilder::new()
            .with_config(config)
            .with_settings(settings)
            .with_cwd(tmp.path())
            .disable_context_discovery()
            .with_llm(Arc::new(EchoLlm))
            .build()
            .await
            .expect("build");
        assert_eq!(app.config().model.name, "from-settings");
        assert_eq!(app.config().anthropic.api_key.as_deref(), Some("sk-config"));
    }

    #[tokio::test]
    async fn with_settings_synthesises_legacy_config_for_build() {
        use crate::auth::{Auth, ProviderAuth};
        use crate::settings::Settings;

        let mut settings = Settings::default();
        settings.model.name = "claude-sonnet-4-6".into();

        let mut auth = Auth::default();
        auth.0.insert(
            "anthropic".into(),
            ProviderAuth::ApiKey {
                key: "sk-test".into(),
            },
        );

        let tmp = tempfile::tempdir().unwrap();
        let app = AppBuilder::new()
            .with_settings(settings)
            .with_auth(auth)
            .with_cwd(tmp.path())
            .with_builtin_tools()
            .disable_context_discovery()
            .with_llm(Arc::new(EchoLlm))
            .build()
            .await
            .expect("build");
        let _ = app;
    }

    #[tokio::test]
    async fn cancel_before_turn_does_not_poison_future_turns() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(std::sync::Arc::new(EchoLlm))
            .build()
            .await
            .expect("build");

        app.cancel();
        let events: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("x"))
            .collect()
            .await;

        assert!(
            events
                .iter()
                .any(|e| matches!(e, UiEvent::AgentMessageComplete(text) if text == "ok")),
            "turn should use a fresh cancellation token: {events:?}"
        );
    }

    #[test]
    fn map_event_matches_started_and_completed_ids_by_tool_name() {
        let tracker = Arc::new(Mutex::new(ToolCallTracker::default()));

        let started_bash = map_event(
            AgentEvent::Core(CoreEvent::ToolStarted {
                name: "bash".into(),
            }),
            &tracker,
        );
        let started_read = map_event(
            AgentEvent::Core(CoreEvent::ToolStarted {
                name: "read".into(),
            }),
            &tracker,
        );
        let completed_bash = map_event(
            AgentEvent::Core(CoreEvent::ToolCompleted {
                name: "bash".into(),
                result: motosan_agent_tool::ToolResult::text("ok"),
            }),
            &tracker,
        );
        let completed_read = map_event(
            AgentEvent::Core(CoreEvent::ToolCompleted {
                name: "read".into(),
                result: motosan_agent_tool::ToolResult::text("ok"),
            }),
            &tracker,
        );

        assert!(matches!(
            started_bash,
            Some(UiEvent::ToolCallStarted { ref id, ref name, .. }) if id == "tool_1" && name == "bash"
        ));
        assert!(matches!(
            started_read,
            Some(UiEvent::ToolCallStarted { ref id, ref name, .. }) if id == "tool_2" && name == "read"
        ));
        assert!(matches!(
            completed_bash,
            Some(UiEvent::ToolCallCompleted { ref id, .. }) if id == "tool_1"
        ));
        assert!(matches!(
            completed_read,
            Some(UiEvent::ToolCallCompleted { ref id, .. }) if id == "tool_2"
        ));
    }

    #[test]
    fn tool_tracker_handles_two_concurrent_invocations_of_same_tool() {
        let tracker = Arc::new(Mutex::new(ToolCallTracker::default()));
        let s1 = map_event(
            AgentEvent::Core(CoreEvent::ToolStarted {
                name: "bash".into(),
            }),
            &tracker,
        );
        let s2 = map_event(
            AgentEvent::Core(CoreEvent::ToolStarted {
                name: "bash".into(),
            }),
            &tracker,
        );
        let c1 = map_event(
            AgentEvent::Core(CoreEvent::ToolCompleted {
                name: "bash".into(),
                result: motosan_agent_tool::ToolResult::text("a"),
            }),
            &tracker,
        );
        let c2 = map_event(
            AgentEvent::Core(CoreEvent::ToolCompleted {
                name: "bash".into(),
                result: motosan_agent_tool::ToolResult::text("b"),
            }),
            &tracker,
        );

        let id_s1 = match s1 {
            Some(UiEvent::ToolCallStarted { id, .. }) => id,
            other => panic!("{other:?}"),
        };
        let id_s2 = match s2 {
            Some(UiEvent::ToolCallStarted { id, .. }) => id,
            other => panic!("{other:?}"),
        };
        let id_c1 = match c1 {
            Some(UiEvent::ToolCallCompleted { id, .. }) => id,
            other => panic!("{other:?}"),
        };
        let id_c2 = match c2 {
            Some(UiEvent::ToolCallCompleted { id, .. }) => id,
            other => panic!("{other:?}"),
        };

        assert_eq!(id_s1, id_c1);
        assert_eq!(id_s2, id_c2);
        assert_ne!(id_s1, id_s2);
    }

    #[tokio::test]
    async fn concurrent_send_user_message_returns_an_error_instead_of_hanging() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(std::sync::Arc::new(ToolOnlyLlm {
                turn: AtomicUsize::new(0),
            }))
            .build()
            .await
            .expect("build");

        let mut first = Box::pin(app.send_user_message(UserMessage::text("first")));
        let first_event = first.next().await;
        assert!(matches!(first_event, Some(UiEvent::AgentTurnStarted)));

        let second_events: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("second"))
            .collect()
            .await;
        assert_eq!(
            second_events.len(),
            1,
            "expected immediate single error event, got: {second_events:?}"
        );
        assert!(matches!(
            &second_events[0],
            UiEvent::Error(msg) if msg.contains("single-turn-per-App")
        ));
    }

    #[tokio::test]
    async fn attachment_error_is_emitted_before_the_single_turn_guard_fires() {
        // Mirror of the concurrent_send_user_message... test above, but the
        // second call carries a bad attachment. The single-turn guard would
        // emit UiEvent::Error("...single-turn-per-App"); we expect the EARLIER
        // prepare_user_message step to short-circuit with AttachmentError
        // instead. See spec §3 ("prepare runs before guard") and §7.1.
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(std::sync::Arc::new(ToolOnlyLlm {
                turn: AtomicUsize::new(0),
            }))
            .build()
            .await
            .expect("build");

        // Park the first turn (claims the single-turn lock).
        let mut first =
            Box::pin(app.send_user_message(crate::user_message::UserMessage::text("first")));
        let first_event = first.next().await;
        assert!(matches!(first_event, Some(UiEvent::AgentTurnStarted)));

        // Fire a second call with a bad attachment.
        let bad = crate::user_message::UserMessage {
            text: "second".into(),
            attachments: vec![crate::user_message::Attachment::Image {
                path: std::path::PathBuf::from("/tmp/this-file-must-not-exist-capo-v06-test.png"),
            }],
        };
        let second_events: Vec<UiEvent> = app.send_user_message(bad).collect().await;

        assert_eq!(
            second_events.len(),
            1,
            "expected exactly one event (the attachment error); got: {second_events:?}"
        );
        assert!(
            matches!(
                &second_events[0],
                UiEvent::AttachmentError {
                    kind: crate::user_message::AttachmentErrorKind::NotFound,
                    ..
                }
            ),
            "expected AttachmentError::NotFound as first event; got {second_events:?}"
        );
    }

    #[test]
    fn progress_events_only_claim_an_id_when_exactly_one_tool_is_pending() {
        let tracker = Arc::new(Mutex::new(ToolCallTracker::default()));
        assert_eq!(progress_event_id(&tracker), "tool_unknown");

        let only = map_event(
            AgentEvent::Core(CoreEvent::ToolStarted {
                name: "bash".into(),
            }),
            &tracker,
        );
        let only_id = match only {
            Some(UiEvent::ToolCallStarted { id, .. }) => id,
            other => panic!("{other:?}"),
        };
        assert_eq!(progress_event_id(&tracker), only_id);

        let _second = map_event(
            AgentEvent::Core(CoreEvent::ToolStarted {
                name: "read".into(),
            }),
            &tracker,
        );
        assert_eq!(progress_event_id(&tracker), "tool_unknown");
    }

    #[tokio::test]
    async fn builder_rejects_builtin_and_custom_tools_together() {
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let dir = tempfile::tempdir().unwrap();
        let err = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_custom_tools_factory(|_| Vec::new())
            .build()
            .await
        {
            Ok(_) => panic!("must reject conflicting tool configuration"),
            Err(err) => err,
        };

        assert!(format!("{err}").contains("mutually exclusive"));
    }

    /// M3 Phase A smoke: two turns share history when a SessionStore is wired.
    #[tokio::test]
    async fn two_turns_in_same_session_share_history() {
        #[derive(Default)]
        struct CounterLlm {
            turn: AtomicUsize,
        }
        #[async_trait]
        impl LlmClient for CounterLlm {
            async fn chat(
                &self,
                messages: &[Message],
                _tools: &[ToolDef],
            ) -> motosan_agent_loop::Result<ChatOutput> {
                let turn = self.turn.fetch_add(1, Ordering::SeqCst);
                let answer = format!("turn-{turn}-saw-{}-messages", messages.len());
                Ok(ChatOutput::new(LlmResponse::Message(answer)))
            }
        }

        let tmp = tempfile::tempdir().unwrap();
        let store = std::sync::Arc::new(motosan_agent_loop::FileSessionStore::new(
            tmp.path().to_path_buf(),
        ));

        let app = AppBuilder::new()
            .with_settings(crate::settings::Settings::default())
            .with_auth(crate::auth::Auth::default())
            .with_cwd(tmp.path())
            .with_builtin_tools()
            .disable_context_discovery()
            .with_llm(std::sync::Arc::new(CounterLlm::default()))
            .with_session_store(store)
            .build_with_session(None)
            .await
            .expect("build");

        let _events1: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("hi"))
            .collect()
            .await;
        let events2: Vec<UiEvent> = app
            .send_user_message(UserMessage::text("again"))
            .collect()
            .await;

        // Turn 2's LLM saw turn 1's user message + turn 1's assistant + turn 2's new user.
        let saw_more_than_one = events2.iter().any(|e| {
            matches!(
                e,
                UiEvent::AgentMessageComplete(t) if t.contains("messages") && !t.contains("saw-1-")
            )
        });
        assert!(
            saw_more_than_one,
            "second turn should have seen history; events: {events2:?}"
        );
    }
}

#[cfg(test)]
mod skills_builder_tests {
    use super::*;
    use crate::skills::types::{Skill, SkillSource};
    use std::path::PathBuf;

    fn fixture() -> Skill {
        Skill {
            name: "x".into(),
            description: "d".into(),
            file_path: PathBuf::from("/x.md"),
            base_dir: PathBuf::from("/"),
            disable_model_invocation: false,
            source: SkillSource::Global,
        }
    }

    #[test]
    fn with_skills_stores_skills() {
        let b = AppBuilder::new().with_skills(vec![fixture()]);
        assert_eq!(b.skills.len(), 1);
        assert_eq!(b.skills[0].name, "x");
    }

    #[test]
    fn without_skills_clears() {
        let b = AppBuilder::new()
            .with_skills(vec![fixture()])
            .without_skills();
        assert!(b.skills.is_empty());
    }
}

#[cfg(test)]
mod mcp_builder_tests {
    use super::*;
    use motosan_agent_tool::Tool;

    // Trivial fake Tool just to verify with_extra_tools stores Arcs.
    struct FakeTool;
    impl Tool for FakeTool {
        fn def(&self) -> motosan_agent_tool::ToolDef {
            motosan_agent_tool::ToolDef {
                name: "fake__echo".into(),
                description: "test".into(),
                input_schema: serde_json::json!({"type": "object"}),
            }
        }
        fn call(
            &self,
            _args: serde_json::Value,
            _ctx: &motosan_agent_tool::ToolContext,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = motosan_agent_tool::ToolResult> + Send + '_>,
        > {
            Box::pin(async { motosan_agent_tool::ToolResult::text("ok") })
        }
    }

    #[test]
    fn with_extra_tools_stores_tools() {
        let tools: Vec<Arc<dyn Tool>> = vec![Arc::new(FakeTool)];
        let b = AppBuilder::new().with_extra_tools(tools);
        assert_eq!(b.extra_tools.len(), 1);
    }
}