kiss_coding/

session_runner.rs

//! AgentSession: the high-level facade every mode drives. Owns the session
//! manager, tool set, queues, retry, auto-compaction, and cost accounting.

use crate::advisor::{AdvisorTool, DEFAULT_ADVISOR_MAX_USES, advisor_system_prompt};
use crate::compaction::{
    self, extract_file_ops, file_ops_details, plan_compaction, should_compact,
};
use crate::decision::DecisionModel;
use crate::iterative::IterativeRuntime;
use crate::session::manager::SessionManager;
use crate::settings::{CacheWarmingMode, DecisionMode as _, QueueMode, Settings};
use crate::subagents::{ForkTurns, SUBAGENT_SYSTEM_PROMPT, SubagentRuntime, fork_messages};
use crate::workflows::{WorkflowApprover, WorkflowRuntime};
use anyhow::Context as _;
use futures::FutureExt as _;
use futures::future::BoxFuture;
use kiss_agent::{
    AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, DynTool, EventSink, StreamFn,
    TurnUpdate,
};
use kiss_ai::{Model, Registry, StopReason, ThinkingLevel, Usage};
use std::collections::VecDeque;
use std::future::Future;
use std::sync::{Arc, Mutex, OnceLock};
use tokio_util::sync::CancellationToken;

/// Harness-level events layered over the loop's AgentEvents.
#[derive(Debug, Clone)]
pub enum SessionEvent {
    Agent(Box<AgentEvent>),
    /// The full prompt finished, after all retries and automatic compaction.
    PromptEnd {
        cancelled: bool,
    },
    QueueUpdate {
        steering: Vec<String>,
        follow_up: Vec<String>,
    },
    CompactionStart {
        auto: bool,
    },
    CompactionEnd {
        summary: String,
        tokens_before: u64,
        error: Option<String>,
    },
    Retry {
        attempt: u32,
        max: u32,
        delay_ms: u64,
        error: String,
    },
    ModelChanged {
        provider: String,
        model_id: String,
    },
    ReasoningEffortChanged {
        decision_model: String,
        level: ThinkingLevel,
        generations: u8,
    },
    ReasoningEffortFallback {
        level: ThinkingLevel,
        reason: String,
    },
    /// A dynamic workflow changed. The terminal redraws from the shared
    /// snapshot. The event carries only the cheap version marker.
    Workflow {
        run: crate::workflows::RunId,
        version: u64,
    },
    /// The verified result of a workflow tool call. Frontends show this after
    /// the assistant answer so model text cannot replace the real run state.
    WorkflowOutcome {
        run: Option<crate::workflows::RunId>,
        name: String,
        status: WorkflowTurnStatus,
    },
    /// A loop or autoresearch job changed. The terminal reads its snapshot.
    Iterative {
        job: crate::iterative::JobId,
        version: u64,
    },
    /// A pooled account hit its rate limit and the request moved to the
    /// next account. `index` is zero-based; `total` is the pool size.
    /// The session file could not be written. The entries stay in the live
    /// session and the next successful write saves them.
    SaveFailed {
        error: String,
    },
    AccountSwitched(kiss_ai::auth::accounts::AccountSwitch),
}

pub type SessionEventSink = Arc<dyn Fn(SessionEvent) + Send + Sync>;

/// Tools and instructions that belong to one submitted user turn.
///
/// Workflow mode is explicit and local to the prompt invocation. This keeps
/// one prompt from enabling or disabling workflow tools for another prompt.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum PromptMode {
    #[default]
    Ordinary,
    Workflow,
}

/// The result that the workflow runtime, rather than the model, observed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkflowTurnStatus {
    Cancelled,
    Completed,
    Failed,
    Stopped,
}

/// What a session is doing now. This is the only record of an active run.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunPhase {
    Idle,
    /// A run is requesting the model or executing tools.
    Streaming,
    /// A run waits before it retries a transient error.
    RetryWait,
    /// A run or a manual request is compacting the conversation.
    Compacting,
}

/// The queue that receives a message when the session is busy.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueueKind {
    /// Deliver between the turns of the active run.
    Steering,
    /// Deliver when the active run would otherwise stop.
    FollowUp,
}

/// Holds a session out of [`RunPhase::Idle`]. Dropping it returns the session
/// to idle, so a panic or a dropped future cannot leave the session busy.
struct ActivePhase {
    session: Arc<AgentSession>,
    cancel: CancellationToken,
    armed: bool,
}

impl ActivePhase {
    /// Take the next queued prompts, or return the session to idle.
    ///
    /// Both happen under the phase lock. [`AgentSession::submit`] queues under
    /// the same lock, so a message cannot arrive between the check and idle.
    /// A cancelled run leaves its queue for the caller to reclaim.
    fn take_queued(&mut self) -> Option<(Vec<AgentMessage>, PromptMode)> {
        let session = &self.session;
        let settings = session.settings();
        let mut phase = session.phase.lock().unwrap();
        if !self.cancel.is_cancelled() {
            for (queue, queue_mode) in [
                (&session.steering, settings.steering_mode),
                (&session.follow_up, settings.follow_up_mode),
            ] {
                if let Some(prompt_mode) = queued_mode(queue, queue_mode) {
                    let prompts = drain_queue(queue, queue_mode);
                    *phase = RunPhase::Streaming;
                    drop(phase);
                    session.emit_queues();
                    return Some((prompts, prompt_mode));
                }
            }
        }
        *phase = RunPhase::Idle;
        drop(phase);
        session.flush_user_shell();
        session.idle.notify_waiters();
        self.armed = false;
        None
    }
}

impl Drop for ActivePhase {
    fn drop(&mut self) {
        if self.armed {
            self.cancel.cancel();
            *self.session.phase.lock().unwrap() = RunPhase::Idle;
            self.session.flush_user_shell();
            self.session.idle.notify_waiters();
        }
    }
}

#[derive(Clone)]
struct QueuedPrompt {
    message: AgentMessage,
    mode: PromptMode,
}

pub struct TreeNavigationOutcome {
    pub editor_text: Option<String>,
    pub summarized: bool,
    pub cancelled: bool,
}

#[derive(Debug, Clone)]
pub struct EphemeralResponse {
    pub text: String,
    pub usage: Usage,
}

const SESSION_TITLE_MAX_CHARS: usize = 36;
const SESSION_TITLE_PROMPT_MAX_BYTES: usize = 960;

fn bounded_session_title_prompt(prompt: &str) -> &str {
    let prompt = prompt.trim();
    let end = prompt.floor_char_boundary(SESSION_TITLE_PROMPT_MAX_BYTES);
    &prompt[..end]
}

fn normalize_session_title(text: &str) -> Option<String> {
    let line = text.lines().find(|line| !line.trim().is_empty())?;
    let cleaned: String = line
        .chars()
        .filter(|character| !character.is_control())
        .collect();
    let normalized = cleaned
        .trim()
        .trim_matches(|character| matches!(character, '"' | '\'' | '`' | '“' | '”' | '‘' | '’'))
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ");
    let title: String = normalized.chars().take(SESSION_TITLE_MAX_CHARS).collect();
    let title = title
        .trim_end_matches(['.', '?', '!', ',', ':', ';'])
        .trim();
    (!title.is_empty()).then(|| title.to_string())
}

pub struct AgentSession {
    /// Lock order: take `manager` before `phase`, never the other way; take
    /// `context_file` before `manager`.
    pub manager: Mutex<SessionManager>,
    pub registry: Arc<Registry>,
    base_tools: Mutex<Vec<DynTool>>,
    session_tools: Mutex<Vec<DynTool>>,
    tools: Mutex<Vec<DynTool>>,
    settings: Mutex<Settings>,
    system_prompt: Mutex<String>,
    model: Mutex<Model>,
    thinking: Mutex<ThinkingLevel>,
    fast_mode: Mutex<bool>,
    steering: Arc<Mutex<VecDeque<QueuedPrompt>>>,
    follow_up: Arc<Mutex<VecDeque<QueuedPrompt>>>,
    /// User shell results that finished during a run, with the id of the
    /// session they belong to. They are saved when the run ends.
    pending_user_shell: Mutex<Vec<(String, AgentMessage)>>,
    cancel: Mutex<CancellationToken>,
    cache_warm_cancel: Mutex<CancellationToken>,
    phase: Mutex<RunPhase>,
    idle: tokio::sync::Notify,
    totals: Mutex<Usage>,
    context_usage_cache: Mutex<Option<(u64, u64)>>,
    context_file: Mutex<Option<crate::context_file::ContextFile>>,
    api_key_override: Option<(String, String)>,
    sink: SessionEventSink,
    subagents_allowed: bool,
    subagents: OnceLock<Arc<SubagentRuntime>>,
    workflows: OnceLock<Arc<WorkflowRuntime>>,
    iterative: OnceLock<Arc<IterativeRuntime>>,
    workflow_approver: Mutex<Option<WorkflowApprover>>,
    /// The model and thinking level the `advisor` tool asks, if any. Only a
    /// root session has one.
    advisor: Mutex<Option<(Model, ThinkingLevel)>>,
    advisor_tool: OnceLock<DynTool>,
    /// Advisor calls made during the current user prompt.
    pub(crate) advisor_calls: std::sync::atomic::AtomicU32,
    /// Optional replacement for the provider streaming function. Embedders and
    /// tests install one to run the whole loop against a scripted fake model.
    stream_fn: Mutex<Option<StreamFn>>,
}

struct ReasoningRunState {
    decision_model: Option<DecisionModel>,
    lease: crate::decision::ReasoningLease,
    provider: String,
    model_id: String,
    saved_effort: ThinkingLevel,
}

impl ReasoningRunState {
    fn checkpoint(
        &mut self,
        model: &Model,
        saved_effort: ThinkingLevel,
        current_effort: ThinkingLevel,
        decision_model: Option<DecisionModel>,
    ) -> (bool, Option<ThinkingLevel>) {
        let model_changed = self.provider != model.provider || self.model_id != model.id;
        if model_changed
            || self.saved_effort != saved_effort
            || self.decision_model != decision_model
            || decision_model.is_none()
        {
            self.lease.clear();
        }
        self.decision_model = decision_model;
        self.provider.clone_from(&model.provider);
        self.model_id.clone_from(&model.id);
        self.saved_effort = saved_effort;
        if self
            .lease
            .current()
            .is_some_and(|level| level != current_effort)
        {
            self.lease.clear();
        }
        (model_changed, self.lease.current())
    }
}

impl AgentSession {
    pub(crate) fn emit_iterative(&self, job: crate::iterative::JobId, version: u64) {
        (self.sink)(SessionEvent::Iterative { job, version });
    }

    pub(crate) fn emit_workflow(&self, run: crate::workflows::RunId, version: u64) {
        (self.sink)(SessionEvent::Workflow { run, version });
    }

    pub(crate) fn emit_workflow_outcome(
        &self,
        run: Option<crate::workflows::RunId>,
        name: String,
        status: WorkflowTurnStatus,
    ) {
        (self.sink)(SessionEvent::WorkflowOutcome { run, name, status });
    }

    #[allow(clippy::too_many_arguments)]
    pub fn new(
        manager: SessionManager,
        tools: Vec<DynTool>,
        registry: impl Into<Arc<Registry>>,
        settings: Settings,
        system_prompt: String,
        model: Model,
        thinking: ThinkingLevel,
        api_key_override: Option<(String, String)>,
        sink: SessionEventSink,
    ) -> Arc<Self> {
        Self::new_with_subagents_allowed(
            manager,
            tools,
            registry,
            settings,
            system_prompt,
            model,
            thinking,
            api_key_override,
            sink,
            true,
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub fn new_with_subagents_allowed(
        manager: SessionManager,
        tools: Vec<DynTool>,
        registry: impl Into<Arc<Registry>>,
        settings: Settings,
        system_prompt: String,
        model: Model,
        thinking: ThinkingLevel,
        api_key_override: Option<(String, String)>,
        sink: SessionEventSink,
        subagents_allowed: bool,
    ) -> Arc<Self> {
        let totals = manager.usage_totals();
        let registry: Arc<Registry> = registry.into();
        let advisor = settings
            .advisor_model
            .as_deref()
            .filter(|_| subagents_allowed)
            .and_then(|pattern| Self::resolve_advisor(&registry, pattern));
        let session = Arc::new(AgentSession {
            manager: Mutex::new(manager),
            registry,
            base_tools: Mutex::new(tools.clone()),
            session_tools: Mutex::new(Vec::new()),
            tools: Mutex::new(tools),
            settings: Mutex::new(settings),
            system_prompt: Mutex::new(system_prompt),
            model: Mutex::new(model),
            thinking: Mutex::new(thinking),
            fast_mode: Mutex::new(false),
            steering: Default::default(),
            follow_up: Default::default(),
            pending_user_shell: Default::default(),
            cancel: Mutex::new(CancellationToken::new()),
            cache_warm_cancel: Mutex::new(CancellationToken::new()),
            phase: Mutex::new(RunPhase::Idle),
            idle: tokio::sync::Notify::new(),
            totals: Mutex::new(totals),
            context_usage_cache: Default::default(),
            context_file: Mutex::new(None),
            api_key_override,
            sink,
            subagents_allowed,
            subagents: OnceLock::new(),
            workflows: OnceLock::new(),
            iterative: OnceLock::new(),
            workflow_approver: Mutex::new(None),
            advisor: Mutex::new(advisor),
            advisor_tool: OnceLock::new(),
            advisor_calls: Default::default(),
            stream_fn: Mutex::new(None),
        });
        if subagents_allowed {
            let tool: DynTool = Arc::new(AdvisorTool(Arc::downgrade(&session)));
            assert!(session.advisor_tool.set(tool).is_ok());
            let runtime = SubagentRuntime::new(Arc::downgrade(&session));
            assert!(session.subagents.set(runtime).is_ok());
            let workflows = WorkflowRuntime::new(Arc::downgrade(&session));
            assert!(session.workflows.set(workflows).is_ok());
            let iterative = IterativeRuntime::new(Arc::downgrade(&session));
            assert!(session.iterative.set(iterative).is_ok());
        }
        session.rebuild_tools();
        session
    }

    /// Replace the provider streaming function used by every later request.
    ///
    /// Pass `None` to restore the real provider. Child sessions created after
    /// this call inherit the override so a scripted fake model also serves
    /// subagents and compaction.
    pub fn set_stream_fn(&self, stream_fn: Option<StreamFn>) {
        *self.stream_fn.lock().unwrap() = stream_fn;
    }

    pub fn model(&self) -> Model {
        self.model.lock().unwrap().clone()
    }

    /// The model and thinking level the `advisor` tool asks, if one is set.
    pub fn advisor(&self) -> Option<(Model, ThinkingLevel)> {
        self.advisor.lock().unwrap().clone()
    }

    /// Set or clear the advisor for the next request. A child session or a
    /// session without tools has no advisor, so this does nothing there.
    /// Saving the choice is the caller's job (`Settings::advisor_model`).
    pub fn set_advisor(&self, advisor: Option<(Model, ThinkingLevel)>) {
        if self.advisor_tool.get().is_none() {
            return;
        }
        *self.advisor.lock().unwrap() = advisor.map(|(model, thinking)| {
            let thinking = model.clamp_thinking_level(thinking);
            (model, thinking)
        });
        self.rebuild_tools();
    }

    /// The advisor for a pattern such as `anthropic/claude-opus-5-5` or
    /// `openai/gpt-5.2:xhigh`. Without a thinking suffix the advisor thinks
    /// at `high`, or the nearest level the model supports.
    pub fn resolve_advisor(registry: &Registry, pattern: &str) -> Option<(Model, ThinkingLevel)> {
        let (model, thinking) = registry.resolve(pattern, None)?;
        let thinking = model.clamp_thinking_level(thinking.unwrap_or(ThinkingLevel::High));
        Some((model, thinking))
    }

    /// Advisor calls allowed for each user prompt.
    pub fn advisor_max_uses(&self) -> u32 {
        self.settings
            .lock()
            .unwrap()
            .advisor_max_uses
            .unwrap_or(DEFAULT_ADVISOR_MAX_USES)
    }

    /// The system prompt this session was given, without the blocks that
    /// `build_context_for` adds per request.
    pub(crate) fn base_system_prompt(&self) -> String {
        self.system_prompt.lock().unwrap().clone()
    }

    pub fn thinking_level(&self) -> ThinkingLevel {
        *self.thinking.lock().unwrap()
    }

    pub fn fast_mode(&self) -> bool {
        *self.fast_mode.lock().unwrap()
    }

    pub fn set_fast_mode(&self, enabled: bool) {
        *self.fast_mode.lock().unwrap() = enabled;
        self.prefix_changed();
    }

    /// The next request will not start with the prefix that cache warming
    /// keeps warm or that the context-usage estimate measured: the model,
    /// reasoning, tools, history or session changed.
    fn prefix_changed(&self) {
        *self.context_usage_cache.lock().unwrap() = None;
        self.cache_warm_cancel.lock().unwrap().cancel();
    }

    pub fn totals(&self) -> Usage {
        *self.totals.lock().unwrap()
    }

    /// Count usage that no session message carries (side requests, child
    /// agents, cache warming, remote compaction) and save it, so a resumed
    /// session shows the same totals.
    /// Save the usage of a side request made for session `session_id`. It
    /// is dropped when that session was replaced meanwhile, so it never
    /// counts against another session.
    pub(crate) fn record_usage(
        &self,
        session_id: &str,
        kind: &str,
        provider: &str,
        model: &str,
        usage: Usage,
    ) {
        if usage == Usage::default() {
            return;
        }
        let mut manager = self.manager.lock().unwrap();
        if manager.session_id() != session_id {
            return;
        }
        self.totals.lock().unwrap().add(&usage);
        let saved = manager.append_usage(kind, provider, model, usage, None);
        drop(manager);
        self.report_save(saved);
    }

    pub fn run_phase(&self) -> RunPhase {
        *self.phase.lock().unwrap()
    }

    pub fn is_running(&self) -> bool {
        self.run_phase() != RunPhase::Idle
    }

    /// Resolve once no run or manual compaction is active.
    pub async fn wait_idle(&self) {
        loop {
            // This future receives wakeups from here on, so none can be lost
            // between the check and the wait.
            let idle = self.idle.notified();
            if !self.is_running() {
                return;
            }
            idle.await;
        }
    }

    /// Move an idle session into `phase`. Every run and manual compaction
    /// starts here, so each one gets a live cancel token.
    fn activate(self: &Arc<Self>, current: &mut RunPhase, phase: RunPhase) -> ActivePhase {
        *current = phase;
        let cancel = CancellationToken::new();
        *self.cancel.lock().unwrap() = cancel.clone();
        ActivePhase {
            session: self.clone(),
            cancel,
            armed: true,
        }
    }

    pub fn settings(&self) -> Settings {
        self.settings.lock().unwrap().clone()
    }

    pub fn update_settings(&self, settings: Settings) {
        let was_enabled = self.subagents_enabled();
        let workflows_were_enabled = self.workflows_enabled();
        let cache_warming = settings.cache_warming;
        *self.settings.lock().unwrap() = settings;
        if cache_warming == CacheWarmingMode::Off
            || (cache_warming == CacheWarmingMode::Streaming && !self.is_running())
        {
            self.cache_warm_cancel.lock().unwrap().cancel();
        }
        let is_enabled = self.subagents_enabled();
        if was_enabled && !is_enabled {
            self.stop_child_work();
        } else if workflows_were_enabled && !self.workflows_enabled() {
            self.stop_workflows();
        }
        self.rebuild_tools();
    }

    /// Replace resources used by the next model request.
    pub fn reload_runtime(&self, settings: Settings, system_prompt: String, tools: Vec<DynTool>) {
        *self.system_prompt.lock().unwrap() = system_prompt;
        *self.base_tools.lock().unwrap() = tools;
        self.update_settings(settings);
        self.prefix_changed();
    }

    /// Add or replace one tool for this process without changing saved tool configuration.
    pub fn install_session_tool(&self, tool: DynTool) {
        let mut tools = self.session_tools.lock().unwrap();
        if let Some(existing) = tools
            .iter_mut()
            .find(|existing| existing.name() == tool.name())
        {
            *existing = tool;
        } else {
            tools.push(tool);
        }
        drop(tools);
        self.rebuild_tools();
    }

    /// Remove one process-only tool. Return true when a tool was removed.
    pub fn remove_session_tool(&self, name: &str) -> bool {
        let mut tools = self.session_tools.lock().unwrap();
        let previous_len = tools.len();
        tools.retain(|tool| tool.name() != name);
        let removed = tools.len() != previous_len;
        drop(tools);
        if removed {
            self.rebuild_tools();
        }
        removed
    }

    /// Stop every child agent and workflow run this session started.
    fn stop_child_work(&self) {
        if let Some(runtime) = self.subagents.get() {
            runtime.interrupt_all();
        }
        if let Some(runtime) = self.workflows.get() {
            runtime.stop_all();
        }
        if let Some(runtime) = self.iterative.get() {
            runtime.stop_all();
        }
    }

    fn stop_workflows(&self) {
        if let Some(runtime) = self.workflows.get() {
            runtime.stop_all();
        }
    }

    fn subagents_enabled(&self) -> bool {
        self.subagents_allowed && self.settings.lock().unwrap().subagents.enabled
    }

    /// Dynamic workflows are built on child agents, so they need subagents on
    /// as well as their own setting.
    pub fn workflows_enabled(&self) -> bool {
        self.subagents_enabled() && self.settings.lock().unwrap().workflows.enabled
    }

    /// Select workflow mode from one submitted user prompt.
    pub fn prompt_mode_for(&self, text: &str) -> PromptMode {
        if self.workflows_enabled()
            && self.settings.lock().unwrap().workflows.keyword_trigger
            && crate::workflows::workflow_trigger(text).is_some()
        {
            PromptMode::Workflow
        } else {
            PromptMode::Ordinary
        }
    }

    pub fn workflows(&self) -> Option<Arc<WorkflowRuntime>> {
        self.workflows.get().cloned()
    }

    pub fn iterative_jobs(&self) -> Option<Arc<IterativeRuntime>> {
        self.iterative.get().cloned()
    }

    /// Install the callback that asks the user to approve a run.
    pub fn set_workflow_approver(&self, approver: WorkflowApprover) {
        *self.workflow_approver.lock().unwrap() = Some(approver);
    }

    pub(crate) fn workflow_approver(&self) -> Option<WorkflowApprover> {
        self.workflow_approver.lock().unwrap().clone()
    }

    fn rebuild_tools(&self) {
        let mut tools = self.base_tools.lock().unwrap().clone();
        for session_tool in self.session_tools.lock().unwrap().iter().cloned() {
            if let Some(existing) = tools
                .iter_mut()
                .find(|existing| existing.name() == session_tool.name())
            {
                *existing = session_tool;
            } else {
                tools.push(session_tool);
            }
        }
        if self.subagents_enabled()
            && let Some(runtime) = self.subagents.get()
        {
            tools.extend(runtime.control_tools());
        }
        if self.advisor.lock().unwrap().is_some()
            && let Some(tool) = self.advisor_tool.get()
        {
            tools.push(tool.clone());
        }
        // Only a different tool list changes the request prefix; a settings
        // change that keeps the tools must not stop cache warming.
        let signature = |tools: &[DynTool]| {
            tools
                .iter()
                .map(|tool| (tool.name().to_string(), tool.description()))
                .collect::<Vec<_>>()
        };
        let mut current = self.tools.lock().unwrap();
        let changed = signature(&current) != signature(&tools);
        *current = tools;
        drop(current);
        if changed {
            self.prefix_changed();
        }
    }

    fn tools_for(&self, mode: PromptMode) -> Vec<DynTool> {
        let mut tools = self.tools.lock().unwrap().clone();
        if mode == PromptMode::Workflow
            && self.workflows_enabled()
            && let Some(runtime) = self.workflows.get()
        {
            tools.push(runtime.tool());
        }
        tools
    }

    /// The system prompt and each active tool's name and description.
    pub fn system_context(&self) -> (String, Vec<(String, String)>) {
        let tools = self
            .tools
            .lock()
            .unwrap()
            .iter()
            .map(|tool| (tool.name().to_string(), tool.description()))
            .collect();
        (self.system_prompt.lock().unwrap().clone(), tools)
    }

    pub fn available_tool_names(&self) -> Vec<String> {
        self.tools
            .lock()
            .unwrap()
            .iter()
            .map(|tool| tool.name().to_string())
            .collect()
    }

    #[cfg(test)]
    fn available_tool_names_for(&self, mode: PromptMode) -> Vec<String> {
        self.tools_for(mode)
            .iter()
            .map(|tool| tool.name().to_string())
            .collect()
    }

    /// Switch the active session without appending synthetic history.
    /// Refuse replacement while a prompt or compaction is active.
    /// Start a sibling session after admission, preserving the storage mode.
    pub fn new_session(&self) -> anyhow::Result<()> {
        self.replace_manager(SessionManager::create_sibling)
    }

    /// Replace the session with the one `create` builds from the current
    /// one. `create` runs only once the session is idle, so a refused
    /// replacement writes no new session file.
    pub fn replace_manager(
        &self,
        create: impl FnOnce(&SessionManager) -> anyhow::Result<SessionManager>,
    ) -> anyhow::Result<()> {
        let mut current = self.manager.lock().unwrap();
        let phase = self.phase.lock().unwrap();
        if *phase != RunPhase::Idle {
            anyhow::bail!(
                "The session has an active prompt. Abort it or wait for it to finish before you replace the session."
            );
        }
        let manager = create(&current)?;
        if let Some(runtime) = self.subagents.get() {
            runtime.reset();
        }
        if let Some(runtime) = self.workflows.get() {
            runtime.stop_all();
        }
        if let Some(runtime) = self.iterative.get() {
            runtime.stop_all();
        }
        let totals = manager.usage_totals();
        let (saved_model, saved_thinking) = manager.session_settings();
        if let Some((provider, model_id)) = saved_model
            && let Some((model, _)) = self.registry.resolve(&model_id, Some(&provider))
        {
            *self.model.lock().unwrap() = model;
        }
        if let Some(thinking) = saved_thinking {
            *self.thinking.lock().unwrap() = thinking;
        }
        *current = manager;
        drop(phase);
        drop(current);
        // Context-file sync takes this lock before the manager's, so clear it
        // only after both are released.
        *self.context_file.lock().unwrap() = None;
        self.prefix_changed();
        *self.totals.lock().unwrap() = totals;
        Ok(())
    }

    /// Save the result of a shell command the user ran in session
    /// `session_id`. It is dropped when that session was replaced meanwhile.
    /// While a run is active it waits and is saved when the run ends, so it
    /// never lands between a tool call and its result.
    pub fn record_user_shell(&self, session_id: &str, message: AgentMessage) {
        let mut manager = self.manager.lock().unwrap();
        if manager.session_id() != session_id {
            return;
        }
        if *self.phase.lock().unwrap() != RunPhase::Idle {
            self.pending_user_shell
                .lock()
                .unwrap()
                .push((session_id.to_string(), message));
            return;
        }
        let saved = manager.append_message(message);
        drop(manager);
        self.report_save(saved);
    }

    /// Save the user shell results that waited for a run to end. A run that
    /// started again in the meantime keeps them waiting.
    fn flush_user_shell(&self) {
        // This runs from ActivePhase's drop. During a panic a poisoned lock
        // would panic again and abort the process, so do nothing then.
        if std::thread::panicking()
            || self
                .pending_user_shell
                .lock()
                .map_or(true, |pending| pending.is_empty())
        {
            return;
        }
        let mut manager = self.manager.lock().unwrap();
        if *self.phase.lock().unwrap() != RunPhase::Idle {
            return;
        }
        let pending = std::mem::take(&mut *self.pending_user_shell.lock().unwrap());
        let mut saved = Ok(());
        for (session_id, message) in pending {
            if manager.session_id() == session_id {
                saved = saved.and(manager.append_message(message).map(drop));
            }
        }
        drop(manager);
        self.report_save(saved);
    }

    /// Report a failed session-file write. The manager keeps the entries, so
    /// the run continues with them.
    fn report_save<T>(&self, result: anyhow::Result<T>) {
        let error = match result {
            Err(error) => Some(format!("{error:#}")),
            Ok(_) => self.manager.lock().unwrap().take_save_error(),
        };
        if let Some(error) = error {
            (self.sink)(SessionEvent::SaveFailed { error });
        }
    }

    /// Switch the model. A thinking level the new model does not support
    /// moves to the nearest one it does, and that change is recorded too.
    pub fn set_model(&self, model: Model) {
        let thinking = model.clamp_thinking_level(self.thinking_level());
        self.prefix_changed();
        let saved = self
            .manager
            .lock()
            .unwrap()
            .append_model_change(&model.provider, &model.id);
        self.report_save(saved);
        (self.sink)(SessionEvent::ModelChanged {
            provider: model.provider.clone(),
            model_id: model.id.clone(),
        });
        *self.model.lock().unwrap() = model;
        if thinking != self.thinking_level() {
            self.set_thinking_level(thinking);
        }
    }

    /// Set the thinking level, moved to the nearest level the current model
    /// supports.
    pub fn set_thinking_level(&self, level: ThinkingLevel) {
        let level = self.model().clamp_thinking_level(level);
        let saved = self
            .manager
            .lock()
            .unwrap()
            .append_thinking_level_change(level);
        self.report_save(saved);
        *self.thinking.lock().unwrap() = level;
        self.prefix_changed();
    }

    /// Move the active leaf in the current session tree, with an optional
    /// summary of the branch that is no longer active.
    pub async fn navigate_tree(
        self: &Arc<Self>,
        target_id: &str,
        summarize: bool,
        custom_instructions: Option<String>,
        cancel: CancellationToken,
    ) -> anyhow::Result<TreeNavigationOutcome> {
        // A run appends to the current leaf, so the leaf may move only while
        // the session is idle. The manager lock is taken before the phase
        // lock, so a run cannot start between the check and the move.
        let busy = |_: &SessionManager| {
            (*self.phase.lock().unwrap() != RunPhase::Idle).then(|| {
                anyhow::anyhow!(
                    "The session has an active prompt. Abort it or wait for it to finish before you move in the session tree."
                )
            })
        };
        let (old_leaf, new_leaf, editor_text, abandoned_messages) = {
            let manager = self.manager.lock().unwrap();
            if let Some(error) = busy(&manager) {
                return Err(error);
            }
            let target = manager
                .get_entry(target_id)
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("unknown tree entry {target_id}"))?;
            let old_leaf = manager.leaf_id().map(str::to_string);
            if old_leaf.as_deref() == Some(target_id) {
                return Ok(TreeNavigationOutcome {
                    editor_text: None,
                    summarized: false,
                    cancelled: false,
                });
            }

            let (new_leaf, editor_text) = match &target {
                crate::session::entry::SessionEntry::Message {
                    message: AgentMessage::User(user),
                    ..
                } => (
                    target.parent_id().map(str::to_string),
                    Some(user.content.as_text()),
                ),
                _ => (Some(target_id.to_string()), None),
            };

            let target_ancestors: std::collections::HashSet<String> = new_leaf
                .as_deref()
                .map(|leaf| {
                    manager
                        .branch_entries(Some(leaf))
                        .into_iter()
                        .map(|entry| entry.id().to_string())
                        .collect()
                })
                .unwrap_or_default();
            let common_ancestor = old_leaf.as_deref().and_then(|leaf| {
                manager
                    .branch_entries(Some(leaf))
                    .into_iter()
                    .rev()
                    .find(|entry| target_ancestors.contains(entry.id()))
                    .map(|entry| entry.id().to_string())
            });
            let abandoned_messages = old_leaf
                .as_deref()
                .map(|leaf| manager.branch_messages_after(leaf, common_ancestor.as_deref()))
                .unwrap_or_default();
            (old_leaf, new_leaf, editor_text, abandoned_messages)
        };

        let summary = if summarize && !abandoned_messages.is_empty() {
            let model = self.model();
            let credential = self.resolve_credential(&model.provider).await;
            let serialized = compaction::serialize_agent_messages(&abandoned_messages);
            let result = compaction::generate_summary(
                &self.request_fn(),
                &model,
                credential,
                &serialized,
                None,
                custom_instructions.as_deref(),
                false,
                cancel.clone(),
            )
            .await?;
            if cancel.is_cancelled() {
                return Ok(TreeNavigationOutcome {
                    editor_text: None,
                    summarized: false,
                    cancelled: true,
                });
            }
            Some(result)
        } else {
            None
        };

        let mut manager = self.manager.lock().unwrap();
        if let Some(error) = busy(&manager) {
            return Err(error);
        }
        if manager.leaf_id() != old_leaf.as_deref() {
            anyhow::bail!("the session tree changed during navigation");
        }
        let summarized = if let Some(summary) = summary {
            let (read, modified) = extract_file_ops(&abandoned_messages);
            if let Some(usage) = &summary.usage {
                self.totals.lock().unwrap().add(usage);
            }
            manager.branch_with_summary(
                new_leaf.as_deref(),
                old_leaf.as_deref().unwrap_or(target_id),
                summary.summary,
                summary.usage,
                Some(file_ops_details(&read, &modified)),
            )?;
            true
        } else {
            if let Some(new_leaf) = new_leaf.as_deref() {
                manager.branch(new_leaf)?;
            } else {
                manager.reset_leaf();
            }
            false
        };
        drop(manager);
        self.prefix_changed();

        Ok(TreeNavigationOutcome {
            editor_text,
            summarized,
            cancelled: false,
        })
    }

    pub fn queue_steering(&self, message: AgentMessage) {
        self.enqueue(&self.steering, message);
    }

    pub fn queue_follow_up(&self, message: AgentMessage) {
        self.enqueue(&self.follow_up, message);
    }

    /// Queue under the phase lock, as [`ActivePhase::take_queued`] reads the
    /// queues under it, so a message cannot slip in after a finishing run
    /// found the queues empty.
    fn enqueue(&self, queue: &Mutex<VecDeque<QueuedPrompt>>, message: AgentMessage) {
        let phase = self.phase.lock().unwrap();
        queue.lock().unwrap().push_back(QueuedPrompt {
            message,
            mode: PromptMode::Ordinary,
        });
        drop(phase);
        self.emit_queues();
    }

    /// Give the next model request a message without starting one. A busy
    /// session delivers it as a follow-up. An idle session adds it to history,
    /// so that a later prompt sees it without an extra model turn.
    pub fn add_for_next_turn(&self, message: AgentMessage) {
        // The manager lock comes before the phase lock, and holding both
        // keeps a run from starting between the check and the append.
        let mut manager = self.manager.lock().unwrap();
        let phase = self.phase.lock().unwrap();
        if *phase != RunPhase::Idle {
            self.follow_up.lock().unwrap().push_back(QueuedPrompt {
                message,
                mode: PromptMode::Ordinary,
            });
            drop(phase);
            drop(manager);
            self.emit_queues();
            return;
        }
        drop(phase);
        let saved = manager.append_message(message);
        drop(manager);
        self.report_save(saved);
    }

    /// Drain both queues back to the caller (Escape restores to editor).
    pub fn reclaim_queued(&self) -> Vec<AgentMessage> {
        let mut out: Vec<AgentMessage> = self
            .steering
            .lock()
            .unwrap()
            .drain(..)
            .map(|prompt| prompt.message)
            .collect();
        out.extend(
            self.follow_up
                .lock()
                .unwrap()
                .drain(..)
                .map(|prompt| prompt.message),
        );
        self.emit_queues();
        out
    }

    pub fn abort(&self) {
        self.cancel.lock().unwrap().cancel();
    }

    fn emit_queues(&self) {
        let preview = |q: &VecDeque<QueuedPrompt>| {
            q.iter()
                .map(|prompt| match &prompt.message {
                    AgentMessage::User(u) => u.content.as_text().chars().take(80).collect(),
                    other => other.role().to_string(),
                })
                .collect::<Vec<String>>()
        };
        (self.sink)(SessionEvent::QueueUpdate {
            steering: preview(&self.steering.lock().unwrap()),
            follow_up: preview(&self.follow_up.lock().unwrap()),
        });
    }

    async fn resolve_credential(&self, provider: &str) -> Option<kiss_ai::ResolvedCredential> {
        if let Some((override_provider, key)) = &self.api_key_override
            && override_provider == provider
        {
            return Some(kiss_ai::ResolvedCredential::api_key(key));
        }
        let credential_provider = self.registry.credential_provider(provider);
        kiss_ai::auth::resolve_credential_async(credential_provider, &self.registry.declared_keys)
            .await
            .ok()
            .flatten()
    }

    fn loop_config(
        &self,
        session_arc: &Arc<Self>,
        active_prompt_mode: Arc<Mutex<PromptMode>>,
    ) -> AgentLoopConfig {
        let mut config = AgentLoopConfig::new(self.model());
        config.thinking_level = self.thinking_level();
        config.fast_mode = self.fast_mode();
        config.session_id = Some(self.manager.lock().unwrap().session_id().to_string());
        let settings = self.settings();
        config.transport = settings.transport;
        let registry = self.registry.clone();
        let declared = self.registry.declared_keys.clone();
        let api_key_override = self.api_key_override.clone();
        config.get_credential = Some(Arc::new(move |provider| {
            let registry = registry.clone();
            let declared = declared.clone();
            let api_key_override = api_key_override.clone();
            Box::pin(async move {
                if let Some((override_provider, key)) = api_key_override
                    && override_provider == provider
                {
                    return Ok(Some(kiss_ai::ResolvedCredential::api_key(key)));
                }
                let credential_provider = registry.credential_provider(&provider).to_string();
                kiss_ai::auth::resolve_credential_async(&credential_provider, &declared)
                    .await
                    .map_err(|error| {
                        format!("could not get the {credential_provider} credential: {error:#}")
                    })
            })
        }));

        let steering = self.steering.clone();
        let steering_for_mode = self.steering.clone();
        let steering_mode = settings.steering_mode;
        let session_for_queues = session_arc.clone();
        config.get_steering_messages = Some(Arc::new(move || {
            let drained = drain_queue(&steering, steering_mode);
            session_for_queues.emit_queues();
            Box::pin(async move { drained })
        }));
        let follow_up = self.follow_up.clone();
        let follow_up_for_mode = self.follow_up.clone();
        let follow_up_mode = settings.follow_up_mode;
        let session_for_queues = session_arc.clone();
        config.get_follow_up_messages = Some(Arc::new(move || {
            let drained = drain_queue(&follow_up, follow_up_mode);
            session_for_queues.emit_queues();
            Box::pin(async move { drained })
        }));
        let reasoning_run = Arc::new(Mutex::new(ReasoningRunState {
            decision_model: settings.reasoning_effort.mode.decision_model(),
            lease: crate::decision::ReasoningLease::default(),
            provider: config.model.provider.clone(),
            model_id: config.model.id.clone(),
            saved_effort: config.thinking_level,
        }));
        let session_for_reasoning = session_arc.clone();
        let reasoning_for_generation = reasoning_run.clone();
        let prompt_mode_for_reasoning = active_prompt_mode.clone();
        config.prepare_generation = Some(Arc::new(move |current_effort| {
            let session = session_for_reasoning.clone();
            let reasoning_run = reasoning_for_generation.clone();
            let active_prompt_mode = prompt_mode_for_reasoning.clone();
            Box::pin(async move {
                let settings = session.settings();
                let context_file_changed = session.sync_context_file();
                let model = session.model();
                let saved_effort = session.thinking_level();
                let decision_model = settings
                    .reasoning_effort
                    .mode
                    .decision_model()
                    .filter(|_| model.reasoning);
                let (model_changed, leased) = {
                    let mut run = reasoning_run.lock().unwrap();
                    run.checkpoint(&model, saved_effort, current_effort, decision_model)
                };
                let mut update = TurnUpdate {
                    context: (model_changed
                        || context_file_changed
                        || settings.experimental_context_file)
                        .then(|| session.build_context_for(*active_prompt_mode.lock().unwrap())),
                    model: model_changed.then(|| model.clone()),
                    thinking_level: Some(saved_effort),
                    fast_mode: Some(session.fast_mode()),
                    ..Default::default()
                };
                let Some(decision_model) = decision_model else {
                    return Some(update);
                };
                if let Some(level) = leased {
                    update.thinking_level = Some(level);
                    return Some(update);
                }

                let selected = async {
                    let client = crate::decision::DecisionClient::connect(
                        decision_model,
                        &session.registry.declared_keys,
                    )
                    .await?;
                    let messages = session
                        .manager
                        .lock()
                        .unwrap()
                        .build_session_context()
                        .messages;
                    let cancel = session.cancel.lock().unwrap().clone();
                    client.select_reasoning(&messages, &model, cancel).await
                }
                .await
                .map_err(|error: anyhow::Error| error.to_string());
                let latest_model = session.model();
                let latest_effort = session.thinking_level();
                if latest_model.provider != model.provider
                    || latest_model.id != model.id
                    || latest_effort != saved_effort
                    || session.settings().reasoning_effort.mode != settings.reasoning_effort.mode
                {
                    reasoning_run.lock().unwrap().lease.clear();
                    update.context =
                        Some(session.build_context_for(*active_prompt_mode.lock().unwrap()));
                    update.model = Some(latest_model);
                    update.thinking_level = Some(latest_effort);
                    return Some(update);
                }
                match selected {
                    Ok(selection) => {
                        update.thinking_level = Some(selection.level);
                        if selection.level != current_effort {
                            let sink = session.sink.clone();
                            let level = selection.level;
                            let generations = selection.generations;
                            update.on_applied = Some(Box::new(move |_, applied| {
                                if applied == level {
                                    sink(SessionEvent::ReasoningEffortChanged {
                                        decision_model: decision_model.label().into(),
                                        level,
                                        generations,
                                    });
                                }
                            }));
                        }
                        reasoning_run.lock().unwrap().lease.install(&selection);
                    }
                    Err(reason) => {
                        let sink = session.sink.clone();
                        let reason = reason.chars().take(300).collect();
                        update.on_applied = Some(Box::new(move |_, applied| {
                            sink(SessionEvent::ReasoningEffortFallback {
                                level: applied,
                                reason,
                            });
                        }));
                    }
                }
                Some(update)
            })
        }));
        let session_for_compaction = session_arc.clone();
        config.prepare_next_turn = Some(Arc::new(move |turn| {
            let has_tool_results = !turn.tool_results.is_empty();
            let tool_failed = turn.tool_results.iter().any(|result| result.is_error);
            let will_continue = turn.will_continue;
            let session = session_for_compaction.clone();
            let active_prompt_mode = active_prompt_mode.clone();
            let steering_for_mode = steering_for_mode.clone();
            let follow_up_for_mode = follow_up_for_mode.clone();
            let reasoning_run = reasoning_run.clone();
            Box::pin(async move {
                let has_queued_user_input = !steering_for_mode.lock().unwrap().is_empty()
                    || (!will_continue && !follow_up_for_mode.lock().unwrap().is_empty());
                let queued_mode = queued_mode(&steering_for_mode, steering_mode)
                    .or_else(|| queued_mode(&follow_up_for_mode, follow_up_mode));
                let mode_changed = {
                    let mut active = active_prompt_mode.lock().unwrap();
                    let before = *active;
                    if let Some(queued_mode) = queued_mode {
                        *active = queued_mode;
                    } else if *active == PromptMode::Workflow && !has_tool_results {
                        // A workflow turn remains active across its tool call.
                        // Its final assistant response has no tool results, so
                        // the next queued or follow-up turn is ordinary again.
                        *active = PromptMode::Ordinary;
                    }
                    before != *active
                };

                let context_file_changed = session.sync_context_file();
                let mut context_changed = mode_changed || context_file_changed;
                if has_tool_results {
                    let settings = session.settings();
                    let cancel = session.cancel.lock().unwrap().clone();
                    let model = session.model();
                    let revision_before = {
                        let manager = session.manager.lock().unwrap();
                        let context = manager.build_session_context();
                        if !auto_compaction_needed(
                            &settings,
                            &context.messages,
                            &model,
                            cancel.is_cancelled(),
                        ) {
                            None
                        } else {
                            Some(manager.context_revision())
                        }
                    };

                    if let Some(revision_before) = revision_before {
                        session.auto_compact().await;
                        let revision_after = session.manager.lock().unwrap().context_revision();
                        context_changed |= revision_after != revision_before;
                    }
                }

                {
                    let mut run = reasoning_run.lock().unwrap();
                    run.lease.consume_generation();
                    if tool_failed || has_queued_user_input {
                        run.lease.clear();
                    }
                }

                Some(TurnUpdate {
                    context: context_changed
                        .then(|| session.build_context_for(*active_prompt_mode.lock().unwrap())),
                    fast_mode: Some(session.fast_mode()),
                    ..Default::default()
                })
            })
        }));
        config.stream_fn = self.request_fn();
        config
    }

    /// How this session sends every model request (turns, summaries, cache
    /// warming): the installed transport, or the provider, with account
    /// failover on rate limits.
    fn request_fn(&self) -> StreamFn {
        let transport = self
            .stream_fn
            .lock()
            .unwrap()
            .clone()
            .unwrap_or_else(|| Arc::new(kiss_ai::stream_simple));
        let sink = self.sink.clone();
        crate::account_failover::wrap(
            transport,
            self.registry.clone(),
            Arc::new(move |switch| sink(SessionEvent::AccountSwitched(switch))),
        )
    }

    fn sync_context_file(&self) -> bool {
        let mut file = self.context_file.lock().unwrap();
        if !self.settings().experimental_context_file {
            return file.take().is_some();
        }
        let mut manager = self.manager.lock().unwrap();
        let result = (|| {
            if file.is_none() {
                *file = Some(crate::context_file::ContextFile::new().context(
                    "Cannot create a context file. A writable temporary directory is required. Set TMPDIR to a writable directory and restart KISS",
                )?);
            }
            file.as_mut().unwrap().sync(&mut manager)
        })();
        match result {
            Ok(changed) => {
                if changed {
                    self.prefix_changed();
                }
                changed
            }
            Err(error) => {
                let message = AgentMessage::Custom(kiss_agent::CustomMessage {
                    custom_type: "context_file_error".into(),
                    content: kiss_ai::UserContent::Text(format!("{error:#}")),
                    display: true,
                    details: None,
                    timestamp: kiss_ai::now_ms(),
                });
                if manager.append_message(message.clone()).is_err() {
                    return false;
                }
                drop(manager);
                drop(file);
                (self.sink)(SessionEvent::Agent(Box::new(AgentEvent::MessageEnd {
                    message,
                })));
                *self.context_usage_cache.lock().unwrap() = None;
                true
            }
        }
    }

    fn build_context(&self) -> AgentContext {
        self.build_context_for(PromptMode::Ordinary)
    }

    fn build_context_for(&self, prompt_mode: PromptMode) -> AgentContext {
        let model = self.model();
        let manager = self.manager.lock().unwrap();
        let (openai_responses_input, mut messages) = if !self.settings().experimental_context_file
            && kiss_ai::api::openai_compaction::supports_remote_compaction(&model)
            && let Some(remote) = manager.build_openai_compaction_context(&model)
        {
            (Some(remote.replacement_history), remote.messages)
        } else {
            (None, manager.build_session_context().messages)
        };
        drop(manager);
        // A failed answer is not sent back: a retry, or a context rebuilt
        // for another model, asks again from the turn before it.
        while matches!(
            messages.last(),
            Some(AgentMessage::Assistant(assistant)) if assistant.stop_reason == StopReason::Error
        ) {
            messages.pop();
        }
        let mut system_prompt = self.system_prompt.lock().unwrap().clone();
        let tools = self.tools_for(prompt_mode);
        // Cursor also advertises its native Shell tool; it must use the registered shell's syntax.
        if model.provider == "cursor"
            && let Some(shell) = tools.iter().find(|tool| tool.name() == "bash")
        {
            system_prompt.push_str("\n\nKISS shell tool (bash): ");
            system_prompt.push_str(&shell.description());
        }
        if self.settings().experimental_context_file
            && let Some(file) = self.context_file.lock().unwrap().as_ref()
        {
            system_prompt.push_str(&file.instructions(model.context_window));
        }
        if self.subagents_enabled() {
            system_prompt.push_str("\n\n");
            system_prompt.push_str(SUBAGENT_SYSTEM_PROMPT);
        }
        if tools
            .iter()
            .any(|tool| tool.name() == crate::advisor::ADVISOR_TOOL_NAME)
        {
            system_prompt.push_str("\n\n");
            system_prompt.push_str(&advisor_system_prompt(self.advisor_max_uses()));
        }
        if prompt_mode == PromptMode::Workflow
            && self.workflows_enabled()
            && self.workflows.get().is_some()
        {
            let limits = kiss_workflow::Limits::default();
            let size = self.settings.lock().unwrap().workflows.size;
            system_prompt.push_str("\n\n");
            system_prompt.push_str(&crate::workflows::authoring_prompt(
                size,
                limits.max_agents,
                limits.max_fanout,
            ));
        }
        AgentContext {
            system_prompt,
            openai_responses_input,
            messages,
            tools,
        }
    }

    pub(crate) fn create_subagent_session(
        self: &Arc<Self>,
        task_name: &str,
        canonical_path: &str,
        fork_turns: ForkTurns,
        model_pattern: Option<&str>,
        reasoning_effort: Option<&str>,
    ) -> anyhow::Result<Arc<Self>> {
        let (model, suggested_thinking) = match model_pattern {
            Some(pattern) => self
                .registry
                .resolve(pattern, None)
                .with_context(|| format!("no model matches subagent model '{pattern}'"))?,
            None => (self.model(), None),
        };
        let thinking = match reasoning_effort {
            Some(level) => ThinkingLevel::parse(level)
                .with_context(|| format!("unknown subagent reasoning_effort '{level}'"))?,
            None => suggested_thinking.unwrap_or_else(|| self.thinking_level()),
        };
        let (mut manager, parent_messages, parent_id) = {
            let parent = self.manager.lock().unwrap();
            (
                parent.create_child()?,
                if fork_turns == ForkTurns::None {
                    Vec::new()
                } else {
                    parent.build_session_context().messages
                },
                parent.session_id().to_string(),
            )
        };
        for message in fork_messages(&parent_messages, fork_turns) {
            manager.append_message(message)?;
        }
        manager.append_custom(
            "subagent",
            Some(serde_json::json!({
                "taskName": task_name,
                "canonicalPath": canonical_path,
                "parentSessionId": parent_id,
            })),
        )?;

        let mut settings = self.settings();
        settings.subagents.enabled = false;
        let mut system_prompt = self.system_prompt.lock().unwrap().clone();
        system_prompt.push_str(&format!(
            "\n\nYou are child agent {canonical_path}. Complete only the assigned task. Return a concise result to the parent agent."
        ));

        let child = Self::new_with_subagents_allowed(
            manager,
            self.base_tools.lock().unwrap().clone(),
            self.registry.clone(),
            settings,
            system_prompt,
            model,
            thinking,
            self.api_key_override.clone(),
            Arc::new(|_| {}),
            false,
        );
        child.set_stream_fn(self.stream_fn.lock().unwrap().clone());
        Ok(child)
    }

    /// Send one request with no history to `model` and return its answer.
    /// A model other than the session's runs without fast mode.
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn run_ephemeral(
        self: &Arc<Self>,
        kind: &str,
        (model, thinking): (Model, ThinkingLevel),
        system_prompt: String,
        prompt: String,
        tools: Vec<DynTool>,
        max_tokens: Option<u64>,
        cancel: CancellationToken,
    ) -> anyhow::Result<EphemeralResponse> {
        let session_id = self.manager.lock().unwrap().session_id().to_string();
        let mut config = self.loop_config(self, Arc::new(Mutex::new(PromptMode::Ordinary)));
        if config.model.provider != model.provider || config.model.id != model.id {
            config.fast_mode = false;
        }
        config.model = model.clone();
        config.thinking_level = thinking;
        config.max_tokens = max_tokens;
        config.session_id = Some(format!("ephemeral-{}", uuid::Uuid::new_v4()));
        config.get_steering_messages = None;
        config.get_follow_up_messages = None;
        config.prepare_next_turn = None;
        config.prepare_generation = None;

        let context = AgentContext {
            system_prompt,
            openai_responses_input: None,
            messages: Vec::new(),
            tools,
        };
        let sink: EventSink = Arc::new(|_| {});
        let messages = kiss_agent::run_agent_loop(
            vec![AgentMessage::user(prompt)],
            context,
            config,
            cancel.clone(),
            sink,
        )
        .await;
        let mut usage = Usage::default();
        for message in &messages {
            if let AgentMessage::Assistant(assistant) = message {
                usage.add(&assistant.usage);
            }
        }
        self.record_usage(&session_id, kind, &model.provider, &model.id, usage);
        if cancel.is_cancelled() {
            anyhow::bail!("request cancelled");
        }
        let assistant = messages.iter().rev().find_map(|message| match message {
            AgentMessage::Assistant(assistant) => Some(assistant),
            _ => None,
        });
        let Some(assistant) = assistant else {
            anyhow::bail!("the provider returned no answer");
        };
        if assistant.stop_reason == StopReason::Error {
            anyhow::bail!(
                "{}",
                assistant
                    .error_message
                    .as_deref()
                    .unwrap_or("the provider request failed")
            );
        }
        let text = assistant.text();
        if text.trim().is_empty() {
            anyhow::bail!("the provider returned an empty answer");
        }
        Ok(EphemeralResponse { text, usage })
    }

    /// Answer a short side question without changing the active session.
    pub async fn answer_btw(
        self: &Arc<Self>,
        question: &str,
        cancel: CancellationToken,
    ) -> anyhow::Result<EphemeralResponse> {
        let messages = self
            .manager
            .lock()
            .unwrap()
            .build_session_context()
            .messages;
        let transcript = transcript_excerpt(&messages, 4, 4_000);
        let prompt = if transcript.is_empty() {
            format!("Side question:\n{question}")
        } else {
            format!("Recent session context:\n{transcript}\n\nSide question:\n{question}")
        };
        let read_tools = self
            .tools
            .lock()
            .unwrap()
            .iter()
            .filter(|tool| tool.name() == "read")
            .cloned()
            .collect();
        self.run_ephemeral(
            "btw",
            (self.model(), ThinkingLevel::Off),
            "Answer the side question from the supplied session context. This is a read-only request. Use the read tool only when a file is needed. Do not propose or perform edits. Give no more than 150 words or 600 characters. Use no more than five bullets. Return only the answer.".into(),
            prompt,
            read_tools,
            Some(500),
            cancel,
        )
        .await
    }

    /// Generate a short session title without changing conversation history.
    pub async fn generate_session_title(
        self: &Arc<Self>,
        prompt: &str,
        cancel: CancellationToken,
    ) -> anyhow::Result<String> {
        let prompt = bounded_session_title_prompt(prompt);
        if prompt.is_empty() {
            anyhow::bail!("the session prompt is empty");
        }
        let response = self
            .run_ephemeral(
                "title",
                (self.model(), ThinkingLevel::Off),
                format!(
                    "Write a one-line title for this task, no more than {SESSION_TITLE_MAX_CHARS} characters. Aim for fewer than five words and start with an imperative verb. Keep ticket IDs and code terms unchanged. Match the user's language. Return the title in sentence case without quotes, Markdown, or ending punctuation."
                ),
                format!("User prompt:\n{prompt}"),
                Vec::new(),
                Some(64),
                cancel,
            )
            .await?;
        normalize_session_title(&response.text)
            .context("the provider returned an invalid session title")
    }

    /// Create a one-line recap without changing the active session.
    pub async fn generate_recap(
        self: &Arc<Self>,
        previous_recap: Option<&str>,
        cancel: CancellationToken,
    ) -> anyhow::Result<EphemeralResponse> {
        let messages = self
            .manager
            .lock()
            .unwrap()
            .build_session_context()
            .messages;
        let transcript = transcript_excerpt(&messages, 12, 12_000);
        if transcript.is_empty() {
            anyhow::bail!("the session has no conversation to recap");
        }
        let previous = previous_recap
            .filter(|recap| !recap.trim().is_empty())
            .map(|recap| format!("\n\nPrevious recap:\n{recap}"))
            .unwrap_or_default();
        self.run_ephemeral(
            "recap",
            (self.model(), ThinkingLevel::Off),
            "Summarize the supplied coding session in one plain-text line of at most 120 characters. State what was done and the next action when one is clear. Do not use a prefix, Markdown, or a newline. Return only the recap.".into(),
            format!("Session transcript:\n{transcript}{previous}"),
            Vec::new(),
            Some(160),
            cancel,
        )
        .await
    }

    /// Run one prompt to completion, including retry and auto-compaction.
    ///
    /// The session decides during this call whether the prompt starts or joins
    /// the steering queue, before the returned future is first polled.
    pub fn prompt(
        self: &Arc<Self>,
        prompts: Vec<AgentMessage>,
    ) -> impl Future<Output = ()> + Send + 'static + use<> {
        self.prompt_with_mode(prompts, PromptMode::Ordinary)
    }

    /// Run one prompt with tools and instructions selected for this turn.
    pub fn prompt_with_mode(
        self: &Arc<Self>,
        prompts: Vec<AgentMessage>,
        prompt_mode: PromptMode,
    ) -> impl Future<Output = ()> + Send + 'static + use<> {
        let run = self.submit(prompts, prompt_mode, QueueKind::Steering);
        async move {
            if let Some(run) = run {
                run.await;
            }
        }
    }

    /// Start a run, or return `None` when the session is busy.
    pub fn try_start_prompt(
        self: &Arc<Self>,
        message: AgentMessage,
        prompt_mode: PromptMode,
    ) -> Option<BoxFuture<'static, ()>> {
        self.cache_warm_cancel.lock().unwrap().cancel();
        let active = {
            let mut phase = self.phase.lock().unwrap();
            (*phase == RunPhase::Idle).then(|| self.activate(&mut phase, RunPhase::Streaming))?
        };
        Some(
            self.clone()
                .run_prompts(active, vec![message], prompt_mode)
                .boxed(),
        )
    }

    /// Start a run, or queue the prompts when the session is busy. The caller
    /// must drive a returned future to run them.
    pub fn submit(
        self: &Arc<Self>,
        prompts: Vec<AgentMessage>,
        prompt_mode: PromptMode,
        kind: QueueKind,
    ) -> Option<BoxFuture<'static, ()>> {
        self.cache_warm_cancel.lock().unwrap().cancel();
        let mut phase = self.phase.lock().unwrap();
        if *phase != RunPhase::Idle {
            let queue = match kind {
                QueueKind::Steering => &self.steering,
                QueueKind::FollowUp => &self.follow_up,
            };
            queue
                .lock()
                .unwrap()
                .extend(prompts.into_iter().map(|message| QueuedPrompt {
                    message,
                    mode: prompt_mode,
                }));
            drop(phase);
            self.emit_queues();
            return None;
        }
        let active = self.activate(&mut phase, RunPhase::Streaming);
        drop(phase);
        Some(
            self.clone()
                .run_prompts(active, prompts, prompt_mode)
                .boxed(),
        )
    }

    async fn run_prompts(
        self: Arc<Self>,
        active: ActivePhase,
        prompts: Vec<AgentMessage>,
        prompt_mode: PromptMode,
    ) {
        let cancel = active.cancel.clone();
        let saved: anyhow::Result<()> = {
            let mut manager = self.manager.lock().unwrap();
            prompts
                .into_iter()
                .try_for_each(|prompt| manager.append_message(prompt).map(drop))
        };
        self.report_save(saved);
        // The prompts were persisted above. Context includes them, so run as a
        // continuation without a second prompt list.
        self.run_turns(None, prompt_mode, &cancel).await;
        self.run_queued_then_idle(active, true).await;
    }

    /// Run whatever arrived after the agent loop last read the queues, then
    /// leave the active phase. Announces the end when any prompt ran.
    async fn run_queued_then_idle(self: &Arc<Self>, mut active: ActivePhase, mut prompt_ran: bool) {
        let cancel = active.cancel.clone();
        while let Some((prompts, prompt_mode)) = active.take_queued() {
            prompt_ran = true;
            self.run_turns(Some(prompts), prompt_mode, &cancel).await;
        }
        if self.settings.lock().unwrap().cache_warming == CacheWarmingMode::Streaming {
            self.cache_warm_cancel.lock().unwrap().cancel();
        }
        if prompt_ran {
            (self.sink)(SessionEvent::PromptEnd {
                cancelled: cancel.is_cancelled(),
            });
        }
    }

    /// Compact inside an active run.
    async fn auto_compact(self: &Arc<Self>) {
        *self.phase.lock().unwrap() = RunPhase::Compacting;
        self.run_compaction(None, true).await;
        *self.phase.lock().unwrap() = RunPhase::Streaming;
    }

    /// Run the agent loop with retry and automatic compaction. `new_prompts`
    /// are messages that the loop must announce and persist itself.
    async fn run_turns(
        self: &Arc<Self>,
        mut new_prompts: Option<Vec<AgentMessage>>,
        prompt_mode: PromptMode,
        cancel: &CancellationToken,
    ) {
        let session = self.clone();
        let sink: EventSink = Arc::new(move |event: AgentEvent| {
            session.on_agent_event(&event);
            (session.sink)(SessionEvent::Agent(Box::new(event)));
        });

        // The advisor limit counts per user prompt; retries share it.
        self.advisor_calls
            .store(0, std::sync::atomic::Ordering::SeqCst);
        let active_prompt_mode = Arc::new(Mutex::new(prompt_mode));
        let mut config = self.loop_config(self, active_prompt_mode.clone());
        let mut context = self.build_context_for(prompt_mode);

        let mut attempt: u32 = 0;
        loop {
            let messages = match new_prompts.take() {
                Some(prompts) => {
                    kiss_agent::run_agent_loop(
                        prompts,
                        context,
                        config.clone(),
                        cancel.clone(),
                        sink.clone(),
                    )
                    .await
                }
                None => {
                    kiss_agent::run_agent_loop_continue(
                        context,
                        config.clone(),
                        cancel.clone(),
                        sink.clone(),
                    )
                    .await
                }
            };

            // Retry on transient error stops.
            let last_error = messages.iter().rev().find_map(|m| match m {
                AgentMessage::Assistant(a) if a.stop_reason == StopReason::Error => {
                    Some(a.error_message.clone().unwrap_or_default())
                }
                _ => None,
            });
            let settings = self.settings();
            let retry = &settings.retry;
            if let Some(error) = last_error
                && retry.enabled
                && attempt < retry.max_retries
                && is_transient(&error)
                && !cancel.is_cancelled()
            {
                attempt += 1;
                let delay = retry
                    .base_delay_ms
                    .saturating_mul(1u64.checked_shl(attempt - 1).unwrap_or(u64::MAX))
                    .min(retry.max_agent_delay_ms);
                *self.phase.lock().unwrap() = RunPhase::RetryWait;
                (self.sink)(SessionEvent::Retry {
                    attempt,
                    max: retry.max_retries,
                    delay_ms: delay,
                    error,
                });
                tokio::select! {
                    _ = tokio::time::sleep(std::time::Duration::from_millis(delay)) => {}
                    _ = cancel.cancelled() => break,
                }
                *self.phase.lock().unwrap() = RunPhase::Streaming;
                context = self.build_context_for(*active_prompt_mode.lock().unwrap());
                config.model = self.model();
                config.thinking_level = self.thinking_level();
                config.fast_mode = self.fast_mode();
                continue;
            }

            // Auto-compaction check after a completed run.
            let ctx = self.manager.lock().unwrap().build_session_context();
            if auto_compaction_needed(
                &settings,
                &ctx.messages,
                &self.model(),
                cancel.is_cancelled(),
            ) {
                self.auto_compact().await;
            }
            break;
        }
    }

    fn on_agent_event(self: &Arc<Self>, event: &AgentEvent) {
        match event {
            AgentEvent::MessageEnd { message } => {
                // Save answers, tool results and queued prompts. A run's
                // first prompts were saved before it and do not arrive here.
                let persist = match message {
                    AgentMessage::Assistant(a) => {
                        self.schedule_cache_warming(a);
                        let mut totals = self.totals.lock().unwrap();
                        totals.add(&a.usage);
                        true
                    }
                    AgentMessage::ToolResult(_)
                    | AgentMessage::User(_)
                    | AgentMessage::Custom(_) => true,
                    _ => false,
                };
                if persist {
                    let saved = self.manager.lock().unwrap().append_message(message.clone());
                    self.report_save(saved);
                }
            }
            AgentEvent::AgentEnd { .. } => {}
            _ => {}
        }
    }

    fn schedule_cache_warming(self: &Arc<Self>, assistant: &kiss_ai::AssistantMessage) {
        let settings = self.settings();
        let model = self.model();
        let Some(cache) = model.prompt_cache else {
            return;
        };
        let Some(short_ttl) = cache.short else {
            return;
        };
        let reasoning = self.thinking_level();
        let fast_mode = self.fast_mode();
        if settings.cache_warming == CacheWarmingMode::Off
            || matches!(
                assistant.stop_reason,
                StopReason::Error | StopReason::Aborted
            )
            || (assistant.usage.input + assistant.usage.cache_read + assistant.usage.cache_write
                == 0)
            || (reasoning != ThinkingLevel::Off
                && model.api == "anthropic-messages"
                && !model
                    .compat
                    .as_ref()
                    .and_then(|compat| compat.force_adaptive_thinking)
                    .unwrap_or(false))
        {
            return;
        }
        let ttl = std::time::Duration::from_secs(short_ttl);
        let Some(delay) = cache_warming_delay(ttl) else {
            return;
        };
        let prompt_tokens =
            assistant.usage.input + assistant.usage.cache_read + assistant.usage.cache_write;
        let agent_context = self.build_context();
        let context = kiss_ai::Context {
            system_prompt: Some(agent_context.system_prompt),
            openai_responses_input: agent_context.openai_responses_input,
            messages: kiss_agent::convert_to_llm(&agent_context.messages),
            tools: agent_context
                .tools
                .iter()
                .map(|tool| tool.to_def())
                .collect(),
        };
        let cancel = CancellationToken::new();
        {
            let mut current = self.cache_warm_cancel.lock().unwrap();
            current.cancel();
            *current = cancel.clone();
        }
        let session = self.clone();
        tokio::spawn(async move {
            let started = tokio::time::Instant::now();
            loop {
                let scheduled = tokio::time::Instant::now();
                if tokio::select! {
                    _ = tokio::time::sleep(delay) => false,
                    _ = cancel.cancelled() => true,
                } {
                    return;
                }
                if cache_refresh_deadline_missed(scheduled.elapsed(), ttl, delay) {
                    return;
                }
                let idle = !session.is_running();
                let max_age = if idle {
                    std::time::Duration::from_secs(30 * 60)
                } else {
                    std::time::Duration::from_secs(60 * 60)
                };
                if started.elapsed() > max_age {
                    return;
                }
                let priced = |input: u64, output: u64, cache_read: u64, cache_write: u64| {
                    let mut usage = Usage {
                        input,
                        output,
                        cache_read,
                        cache_write,
                        ..Default::default()
                    };
                    kiss_ai::api::finalize_cost(&mut usage, &model);
                    usage.cost.total
                };
                let hit_cost = priced(0, 0, prompt_tokens, 0);
                let miss_cost = if model.cost.cache_write > 0.0 {
                    priced(0, 0, 0, prompt_tokens)
                } else {
                    priced(prompt_tokens, 0, 0, 0)
                };
                let warm_cost = priced(0, 1, prompt_tokens, 0);
                let probability = if idle { 0.15 } else { 1.0 };
                if probability * (miss_cost - hit_cost).max(0.0) - warm_cost < 0.05 {
                    return;
                }
                let Some(credential) = session.resolve_credential(&model.provider).await else {
                    return;
                };
                let session_id = session.manager.lock().unwrap().session_id().to_string();
                let options = kiss_ai::StreamOptions {
                    credential: Some(credential),
                    max_tokens: Some(1),
                    reasoning,
                    fast_mode,
                    session_id: Some(session_id.clone()),
                    transport: settings.transport,
                    cancel: cancel.clone(),
                    // A replay must not replace the conversation's Anthropic thread.
                    disable_continuation: true,
                    ..Default::default()
                };
                let warmed = session.request_fn()(&model, &context, &options)
                    .result()
                    .await;
                if matches!(warmed.stop_reason, StopReason::Error | StopReason::Aborted) {
                    return;
                }
                session.record_usage(
                    &session_id,
                    "cache_warm",
                    &warmed.provider,
                    warmed.response_model.as_deref().unwrap_or(&warmed.model),
                    warmed.usage,
                );
            }
        });
    }

    /// Compact on request. The session refuses during this call when a prompt
    /// or another compaction is active. Prompts that arrive during compaction
    /// run when it ends.
    pub fn compact(
        self: &Arc<Self>,
        custom_instructions: Option<String>,
    ) -> impl Future<Output = Result<(), String>> + Send + 'static + use<> {
        let session = self.clone();
        let active = {
            let mut phase = self.phase.lock().unwrap();
            (*phase == RunPhase::Idle).then(|| self.activate(&mut phase, RunPhase::Compacting))
        };
        async move {
            let Some(active) = active else {
                let error = "cannot compact while a prompt or another compaction is running. Wait for it to finish or abort it, then compact again".to_string();
                (session.sink)(SessionEvent::CompactionEnd {
                    summary: String::new(),
                    tokens_before: 0,
                    error: Some(error.clone()),
                });
                return Err(error);
            };
            session.run_compaction(custom_instructions, false).await;
            session.run_queued_then_idle(active, false).await;
            Ok(())
        }
    }

    async fn run_compaction(self: &Arc<Self>, custom_instructions: Option<String>, auto: bool) {
        (self.sink)(SessionEvent::CompactionStart { auto });
        let ctx = self.manager.lock().unwrap().build_session_context();
        let previous_summary = ctx.messages.iter().rev().find_map(|m| match m {
            AgentMessage::CompactionSummary(c) => Some(c.summary.clone()),
            _ => None,
        });
        let settings = self.settings();
        let model = self.model();
        let keep_recent_tokens = settings.compaction.keep_recent_tokens_for(&model);
        let reserve_tokens = settings.compaction.reserve_tokens_for(&model);
        let plan = plan_compaction(&ctx.messages, keep_recent_tokens);
        if plan.to_summarize.is_empty() && plan.turn_prefix.is_empty() {
            (self.sink)(SessionEvent::CompactionEnd {
                summary: String::new(),
                tokens_before: plan.tokens_before,
                error: Some("Nothing to compact".into()),
            });
            return;
        }

        if let Some(decision_model) = settings.compaction.mode.decision_model()
            && let Ok(client) = crate::decision::DecisionClient::connect(
                decision_model,
                &self.registry.declared_keys,
            )
            .await
        {
            let cancel = self.cancel.lock().unwrap().clone();
            let pinned_start = ctx.messages.len().saturating_sub(plan.kept.len());
            if let Ok(result) = client.compact(&ctx.messages, pinned_start, cancel).await {
                let estimated_before: u64 = ctx
                    .messages
                    .iter()
                    .map(compaction::estimate_message_tokens)
                    .sum();
                let estimated_after: u64 = result
                    .messages
                    .iter()
                    .map(compaction::estimate_message_tokens)
                    .sum();
                let removed = estimated_before.saturating_sub(estimated_after);
                let tokens_after = plan.tokens_before.saturating_sub(removed);
                let useful =
                    estimated_after.saturating_mul(4) <= estimated_before.saturating_mul(3);
                let resolved_auto_threshold =
                    !auto || !should_compact(tokens_after, model.context_window, reserve_tokens);
                if useful && resolved_auto_threshold {
                    let summary = format!(
                        "{} kept {}, truncated {}, and removed {} of {} older tool interactions",
                        decision_model.label(),
                        result.stats.kept,
                        result.stats.truncated,
                        result.stats.dropped,
                        result.stats.eligible,
                    );
                    let details = serde_json::json!({
                        "mode": settings.compaction.mode,
                        "provider": decision_model.provider(),
                        "model": decision_model.id(),
                        "stats": result.stats,
                        "estimatedTokensBefore": plan.tokens_before,
                        "estimatedTokensAfter": tokens_after,
                    });
                    let mut manager = self.manager.lock().unwrap();
                    let append = manager.append_compaction(
                        String::new(),
                        plan.tokens_before,
                        result.messages,
                        None,
                        Some(details),
                    );
                    drop(manager);
                    let error = append.err().map(|error| format!("{error:#}"));
                    (self.sink)(SessionEvent::CompactionEnd {
                        summary: if error.is_none() {
                            summary
                        } else {
                            String::new()
                        },
                        tokens_before: plan.tokens_before,
                        error,
                    });
                    return;
                }
            }
        }

        let credential = self.resolve_credential(&model.provider).await;

        let mut serialized = compaction::serialize_agent_messages(&plan.to_summarize);
        if plan.is_split_turn {
            serialized.push_str(
                "\n\n[The following is the earlier part of the still-active task turn:]\n\n",
            );
            serialized.push_str(&compaction::serialize_agent_messages(&plan.turn_prefix));
        }

        let summary_cancel = self.cancel.lock().unwrap().clone();
        // Remote compaction calls OpenAI directly. A session with its own
        // transport (an embedder or a scripted test) gets only the local
        // summary, which goes through that transport.
        let remote_request = if self.stream_fn.lock().unwrap().is_none()
            && kiss_ai::api::openai_compaction::supports_remote_compaction(&model)
        {
            let context = self.build_context();
            Some((
                kiss_ai::Context {
                    system_prompt: Some(context.system_prompt),
                    openai_responses_input: context.openai_responses_input,
                    messages: kiss_agent::convert_to_llm(&context.messages),
                    tools: context.tools.iter().map(|tool| tool.to_def()).collect(),
                },
                kiss_ai::StreamOptions {
                    credential: credential.clone(),
                    reasoning: self.thinking_level(),
                    fast_mode: self.fast_mode(),
                    session_id: Some(self.manager.lock().unwrap().session_id().to_string()),
                    cancel: summary_cancel.clone(),
                    ..Default::default()
                },
            ))
        } else {
            None
        };
        let request = self.request_fn();
        let local_future = compaction::generate_summary(
            &request,
            &model,
            credential.clone(),
            &serialized,
            previous_summary.as_deref(),
            custom_instructions.as_deref(),
            plan.is_split_turn,
            summary_cancel.clone(),
        );
        let remote_model = model.clone();
        let remote_future = async move {
            match remote_request {
                Some((context, options)) => Some(
                    kiss_ai::api::openai_compaction::compact(&remote_model, &context, &options)
                        .await,
                ),
                None => None,
            }
        };
        let (local_outcome, remote_outcome) = tokio::join!(local_future, remote_future);

        match select_compaction_outcome(&model, local_outcome, remote_outcome) {
            Ok(result) => {
                let mut summarized_all = plan.to_summarize.clone();
                summarized_all.extend(plan.turn_prefix.clone());
                let (read, modified) = extract_file_ops(&summarized_all);
                // The compaction entry saves the local summary's usage.
                if let Some(u) = &result.local_usage {
                    self.totals.lock().unwrap().add(u);
                }
                let details = merge_compaction_details(
                    file_ops_details(&read, &modified),
                    result.remote_details,
                );
                let saved = self.manager.lock().unwrap().append_compaction(
                    result.summary.clone(),
                    plan.tokens_before,
                    plan.kept.clone(),
                    result.local_usage,
                    Some(details),
                );
                self.report_save(saved);
                if let Some(usage) = result.remote_usage {
                    let session_id = self.manager.lock().unwrap().session_id().to_string();
                    self.record_usage(
                        &session_id,
                        "remote_compaction",
                        &model.provider,
                        &model.id,
                        usage,
                    );
                }
                (self.sink)(SessionEvent::CompactionEnd {
                    summary: result.summary,
                    tokens_before: plan.tokens_before,
                    error: None,
                });
            }
            Err(error) => {
                (self.sink)(SessionEvent::CompactionEnd {
                    summary: String::new(),
                    tokens_before: plan.tokens_before,
                    error: Some(format!("{error:#}")),
                });
            }
        }
    }

    /// Current estimated context tokens and window fraction.
    pub fn context_usage(&self) -> (u64, u64) {
        let manager = self.manager.lock().unwrap();
        let revision = manager.context_revision();
        let used = if let Some((cached_revision, tokens)) =
            *self.context_usage_cache.lock().unwrap()
            && cached_revision == revision
        {
            tokens
        } else {
            let tokens = compaction::estimate_request_tokens(
                manager.context_messages(),
                self.settings().experimental_context_file,
            );
            *self.context_usage_cache.lock().unwrap() = Some((revision, tokens));
            tokens
        };
        drop(manager);
        (used, self.model().context_window)
    }
}

struct SelectedCompaction {
    summary: String,
    local_usage: Option<Usage>,
    remote_usage: Option<Usage>,
    remote_details: Option<serde_json::Value>,
}

fn select_compaction_outcome(
    model: &Model,
    local: anyhow::Result<compaction::SummaryOutcome>,
    remote: Option<anyhow::Result<kiss_ai::api::openai_compaction::RemoteCompactionResult>>,
) -> anyhow::Result<SelectedCompaction> {
    match (local, remote) {
        (Ok(local), Some(Ok(remote))) => Ok(SelectedCompaction {
            summary: local.summary,
            local_usage: local.usage,
            remote_usage: remote.usage,
            remote_details: Some(
                kiss_ai::api::openai_compaction::build_remote_compaction_details(model, &remote),
            ),
        }),
        (Ok(local), Some(Err(_)) | None) => Ok(SelectedCompaction {
            summary: local.summary,
            local_usage: local.usage,
            remote_usage: None,
            remote_details: None,
        }),
        (Err(_), Some(Ok(remote))) => Ok(SelectedCompaction {
            summary: format!(
                "OpenAI server-side compaction was applied for {}/{}. The provider-native context is stored in this session, and this notice keeps the compaction boundary readable for other providers.",
                model.provider, model.id
            ),
            local_usage: None,
            remote_usage: remote.usage,
            remote_details: Some(
                kiss_ai::api::openai_compaction::build_remote_compaction_details(model, &remote),
            ),
        }),
        (Err(local), Some(Err(remote))) => anyhow::bail!(
            "local compaction failed: {local:#}. OpenAI remote compaction failed: {remote:#}"
        ),
        (Err(error), None) => Err(error),
    }
}

fn merge_compaction_details(
    mut local: serde_json::Value,
    remote: Option<serde_json::Value>,
) -> serde_json::Value {
    let Some(remote) = remote else {
        return local;
    };
    let Some(local_object) = local.as_object_mut() else {
        return remote;
    };
    if let Some(remote_object) = remote.as_object() {
        for (key, value) in remote_object {
            local_object.insert(key.clone(), value.clone());
        }
    }
    local
}

fn transcript_excerpt(messages: &[AgentMessage], max_messages: usize, max_chars: usize) -> String {
    let mut entries = messages
        .iter()
        .rev()
        .filter_map(|message| match message {
            AgentMessage::User(user) => Some(("User", user.content.as_text())),
            AgentMessage::Assistant(assistant) => Some(("Assistant", assistant.text())),
            _ => None,
        })
        .filter(|(_, text)| !text.trim().is_empty())
        .take(max_messages)
        .collect::<Vec<_>>();
    entries.reverse();
    let transcript = entries
        .into_iter()
        .map(|(role, text)| format!("{role}: {}", text.trim()))
        .collect::<Vec<_>>()
        .join("\n\n");
    let count = transcript.chars().count();
    if count <= max_chars {
        return transcript;
    }
    let omitted = count - max_chars;
    let tail = transcript.chars().skip(omitted).collect::<String>();
    format!("[earlier text omitted]\n{tail}")
}

fn drain_queue(queue: &Arc<Mutex<VecDeque<QueuedPrompt>>>, mode: QueueMode) -> Vec<AgentMessage> {
    let mut q = queue.lock().unwrap();
    match mode {
        QueueMode::All => q.drain(..).map(|prompt| prompt.message).collect(),
        QueueMode::OneAtATime => q
            .pop_front()
            .map(|prompt| prompt.message)
            .into_iter()
            .collect(),
    }
}

fn queued_mode(queue: &Arc<Mutex<VecDeque<QueuedPrompt>>>, mode: QueueMode) -> Option<PromptMode> {
    let queue = queue.lock().unwrap();
    match mode {
        QueueMode::All => queue
            .iter()
            .any(|prompt| prompt.mode == PromptMode::Workflow)
            .then_some(PromptMode::Workflow)
            .or_else(|| (!queue.is_empty()).then_some(PromptMode::Ordinary)),
        QueueMode::OneAtATime => queue.front().map(|prompt| prompt.mode),
    }
}

fn auto_compaction_needed(
    settings: &Settings,
    messages: &[AgentMessage],
    model: &Model,
    cancelled: bool,
) -> bool {
    settings.compaction.enabled
        && !cancelled
        && model.context_window > 0
        && should_compact(
            compaction::estimate_request_tokens(
                messages.iter(),
                settings.experimental_context_file,
            ),
            model.context_window,
            settings.compaction.reserve_tokens_for(model),
        )
}

fn cache_warming_delay(ttl: std::time::Duration) -> Option<std::time::Duration> {
    (ttl > std::time::Duration::from_secs(10))
        .then(|| std::cmp::min(ttl.mul_f64(0.9), ttl - std::time::Duration::from_secs(10)))
}

fn cache_refresh_deadline_missed(
    elapsed: std::time::Duration,
    ttl: std::time::Duration,
    delay: std::time::Duration,
) -> bool {
    elapsed > delay + ttl.saturating_sub(delay) / 2
}

fn is_transient(error: &str) -> bool {
    let e = error.to_lowercase();
    if let Some(value) = error
        .find('{')
        .and_then(|start| serde_json::from_str::<serde_json::Value>(&error[start..]).ok())
    {
        let detail = value.get("error").unwrap_or(&value);
        // Zhipu/Z.ai code 1308 is the plan's usage limit, sent as HTTP 429.
        // It resets after hours, not seconds.
        if detail
            .get("code")
            .is_some_and(|code| *code == 1308 || *code == "1308")
            || detail
                .get("isRetryable")
                .and_then(serde_json::Value::as_bool)
                == Some(false)
            || detail
                .get("details")
                .and_then(serde_json::Value::as_array)
                .is_some_and(|details| {
                    details.iter().any(|detail| {
                        detail
                            .pointer("/debug/details/isRetryable")
                            .and_then(serde_json::Value::as_bool)
                            == Some(false)
                    })
                })
        {
            return false;
        }
    }
    // ChatGPT usage limits, including a Business workspace limit that only
    // the workspace admin can raise.
    if e.contains("subscription_sharing_usage_limit_exceeded")
        || e.contains("subscription_sharing_workplace_individual_usage_limit_exceeded")
    {
        return false;
    }
    let transient_status = [429, 500, 502, 503, 504, 520, 524, 529]
        .into_iter()
        .any(|status| kiss_ai::api::mentions_status(error, status));
    transient_status
        || [
            "overloaded",
            "currently experiencing high demand",
            "rate limit",
            "timeout",
            "timed out",
            "connection reset",
            "connection refused",
            "connection closed",
            "connection aborted",
            "connection error",
            "failed to connect",
            "network error",
            "stream error",
            "subscription_sharing_usage_unavailable",
            "subscription_sharing_user_unavailable",
            "server error",
            "server_busy",
            "servers are currently busy",
            "model is at capacity",
            "service unavailable",
            "service is temporarily unavailable",
            "provider returned error",
            "provider returned an error",
            "unable to forward upstream",
            "upstream request failed",
            "upstream_stream_read_error",
            "stream was interrupted",
            "stream disconnected before completion",
            "pending stream has been canceled",
            // A gateway relaying a dropped upstream Anthropic stream.
            "cut short after the reply had begun",
            "http2 stream failed",
            "http/2 stream failed",
        ]
        .iter()
        .any(|needle| e.contains(needle))
}

#[cfg(test)]
mod ephemeral_tests {
    use super::*;
    use std::collections::BTreeMap;
    use std::path::Path;

    fn openai_model() -> Model {
        Model {
            id: "gpt-test".into(),
            name: "GPT test".into(),
            api: "openai-responses".into(),
            provider: "openai".into(),
            base_url: "https://api.openai.com/v1".into(),
            reasoning: true,
            input: vec!["text".into()],
            cost: Default::default(),
            prompt_cache: None,
            context_window: 100_000,
            max_tokens: 1_000,
            compat: None,
            thinking_level_map: BTreeMap::new(),
            headers: BTreeMap::new(),
            sampling_params: Default::default(),
            sampling_params_by_thinking_level: Default::default(),
        }
    }

    #[tokio::test]
    async fn prompt_end_follows_retries_and_reports_cancellation() {
        for cancel_retry in [false, true] {
            let events = Arc::new(Mutex::new(Vec::new()));
            let saved_events = events.clone();
            let session = AgentSession::new(
                SessionManager::in_memory(Path::new("/test")),
                Vec::new(),
                Registry::load(None),
                Settings {
                    compaction: crate::settings::CompactionSettings {
                        enabled: false,
                        ..Default::default()
                    },
                    retry: crate::settings::RetrySettings {
                        base_delay_ms: if cancel_retry { 60_000 } else { 0 },
                        ..Default::default()
                    },
                    ..Default::default()
                },
                "test".into(),
                openai_model(),
                ThinkingLevel::Off,
                None,
                Arc::new(move |event| saved_events.lock().unwrap().push(event)),
            );
            let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
            let observed_calls = calls.clone();
            session.set_stream_fn(Some(Arc::new(move |_, _, _| {
                let step = observed_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let mut message =
                    kiss_ai::AssistantMessage::empty("openai-responses", "openai", "gpt-test");
                let (sink, stream) = kiss_ai::EventStream::channel();
                if step == 0 {
                    message.stop_reason = StopReason::Error;
                    message.error_message = Some("HTTP 503".into());
                    sink.error(message);
                } else {
                    message.stop_reason = StopReason::Stop;
                    sink.done(message);
                }
                stream
            })));
            let task_session = session.clone();
            let task = tokio::spawn(async move {
                task_session
                    .prompt(vec![AgentMessage::user("finish")])
                    .await;
            });
            if cancel_retry {
                tokio::time::timeout(std::time::Duration::from_secs(5), async {
                    loop {
                        if events
                            .lock()
                            .unwrap()
                            .iter()
                            .any(|event| matches!(event, SessionEvent::Retry { .. }))
                        {
                            break;
                        }
                        tokio::task::yield_now().await;
                    }
                })
                .await
                .unwrap();
                // Intermediate agent completion must not announce a finished prompt.
                assert!(
                    !events
                        .lock()
                        .unwrap()
                        .iter()
                        .any(|event| matches!(event, SessionEvent::PromptEnd { .. }))
                );
                session.abort();
            }
            tokio::time::timeout(std::time::Duration::from_secs(5), task)
                .await
                .unwrap()
                .unwrap();
            let events = events.lock().unwrap();
            assert!(
                matches!(events.last(), Some(SessionEvent::PromptEnd { cancelled }) if *cancelled == cancel_retry)
            );
            assert_eq!(
                events
                    .iter()
                    .filter(|event| matches!(event, SessionEvent::PromptEnd { .. }))
                    .count(),
                1
            );
            assert_eq!(events.iter().filter(|event| matches!(event, SessionEvent::Agent(agent) if matches!(agent.as_ref(), AgentEvent::AgentEnd { .. }))).count(), if cancel_retry { 1 } else { 2 });
            assert_eq!(
                calls.load(std::sync::atomic::Ordering::SeqCst),
                if cancel_retry { 1 } else { 2 }
            );
            assert!(!session.is_running());
        }
    }

    #[tokio::test]
    async fn the_advisor_tool_asks_the_advisor_model_with_the_transcript() {
        let session = AgentSession::new(
            SessionManager::in_memory(Path::new("/test")),
            Vec::new(),
            Registry::from_builtin(),
            Settings {
                advisor_max_uses: Some(1),
                ..Default::default()
            },
            "root prompt".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
        );
        let mut advisor = openai_model();
        advisor.provider = "anthropic".into();
        advisor.id = "advisor-test".into();
        advisor.max_tokens = 128_000;
        session.set_advisor(Some((advisor, ThinkingLevel::High)));
        type Request = (String, ThinkingLevel, Option<u64>, kiss_ai::Context);
        let requests: Arc<Mutex<Vec<Request>>> = Arc::default();
        let saved = requests.clone();
        session.set_stream_fn(Some(Arc::new(move |model, context, options| {
            let mut requests = saved.lock().unwrap();
            requests.push((
                model.id.clone(),
                options.reasoning,
                options.max_tokens,
                context.clone(),
            ));
            let main_turns = requests.iter().filter(|(id, ..)| id == "gpt-test").count();
            let mut message =
                kiss_ai::AssistantMessage::empty("openai-responses", &model.provider, &model.id);
            message.stop_reason = StopReason::Stop;
            // Main turns: call, call again over the limit, finish; then a
            // new prompt calls once more and finishes.
            message.content = vec![if model.id == "advisor-test" {
                message.usage.input = 50;
                kiss_ai::ContentBlock::text("PLAN\nuse approach B")
            } else if matches!(main_turns, 1 | 2 | 4) {
                message.stop_reason = StopReason::ToolUse;
                kiss_ai::ContentBlock::ToolCall(kiss_ai::ToolCall {
                    id: format!("call_{main_turns}"),
                    name: "advisor".into(),
                    arguments: serde_json::json!({}),
                    thought_signature: None,
                    arguments_error: None,
                })
            } else {
                kiss_ai::ContentBlock::text("done")
            }];
            let (sink, stream) = kiss_ai::EventStream::channel();
            sink.done(message);
            stream
        })));

        session
            .prompt(vec![AgentMessage::user("solve the hard problem")])
            .await;

        let advisor_result = |context: &kiss_ai::Context, id: &str| {
            context.messages.iter().find_map(|message| match message {
                kiss_ai::Message::ToolResult(result) if result.tool_call_id == id => Some((
                    result.is_error,
                    kiss_ai::UserContent::Blocks(result.content.clone()).as_text(),
                )),
                _ => None,
            })
        };
        {
            let requests = requests.lock().unwrap();
            let ids: Vec<&str> = requests.iter().map(|(id, ..)| id.as_str()).collect();
            assert_eq!(ids, ["gpt-test", "advisor-test", "gpt-test", "gpt-test"]);
            let (_, _, _, first) = &requests[0];
            assert!(first.tools.iter().any(|tool| tool.name == "advisor"));
            let system_prompt = first.system_prompt.as_deref().unwrap();
            assert!(system_prompt.contains("at most 1 times"));
            assert!(system_prompt.contains("After STOP"));
            let (_, thinking, max_tokens, advice) = &requests[1];
            assert_eq!(*thinking, ThinkingLevel::High);
            assert_eq!(*max_tokens, Some(2_048 + 32_768));
            assert!(advice.tools.is_empty());
            assert!(advice.system_prompt.as_deref().unwrap().contains("PLAN - "));
            assert_eq!(advice.messages.len(), 1);
            let kiss_ai::Message::User(question) = &advice.messages[0] else {
                panic!("the advisor request must be one user message");
            };
            let question = question.content.as_text();
            assert!(question.contains("root prompt"));
            assert!(question.contains("[User]: solve the hard problem"));
            assert!(question.contains("advisor()"));
            let (_, _, _, last) = &requests[3];
            let (failed, text) = advisor_result(last, "call_1").unwrap();
            assert!(!failed);
            assert!(text.starts_with("PLAN\nuse approach B"), "{text}");
            assert!(text.ends_with("[Advisor call 1 of 1 for this user request.]"));
            let (failed, text) = advisor_result(last, "call_2").unwrap();
            assert!(failed);
            assert!(text.contains("limit of 1 calls"), "{text}");
        }
        assert_eq!(session.totals().input, 50);

        // A new user prompt has its own limit.
        session.prompt(vec![AgentMessage::user("next task")]).await;
        let requests = requests.lock().unwrap();
        let ids: Vec<&str> = requests.iter().map(|(id, ..)| id.as_str()).collect();
        assert_eq!(&ids[4..], ["gpt-test", "advisor-test", "gpt-test"]);
    }

    #[test]
    fn only_a_root_session_with_an_advisor_has_the_tool() {
        let session = settings_test_session(Settings::default(), true);
        assert!(session.available_tool_names().is_empty());
        let advisor =
            AgentSession::resolve_advisor(&session.registry, "anthropic/claude-opus-4-5").unwrap();
        assert_eq!(
            advisor.1,
            advisor.0.clamp_thinking_level(ThinkingLevel::High)
        );
        session.set_advisor(Some(advisor.clone()));
        assert_eq!(session.available_tool_names(), ["advisor"]);
        assert!(session.build_context().system_prompt.contains("Advisor:"));
        let child = session
            .create_subagent_session("task", "/root/task", ForkTurns::None, None, None)
            .unwrap();
        assert!(child.advisor().is_none());
        assert!(child.available_tool_names().is_empty());
        session.set_advisor(None);
        assert!(session.available_tool_names().is_empty());
        assert!(!session.build_context().system_prompt.contains("Advisor:"));

        let settings = Settings {
            advisor_model: Some("anthropic/claude-opus-4-5:low".into()),
            ..Default::default()
        };
        let configured = settings_test_session(settings.clone(), true);
        assert_eq!(
            configured
                .advisor()
                .map(|(model, thinking)| (model.id, thinking)),
            Some(("claude-opus-4-5".into(), ThinkingLevel::Low))
        );
        let without_tools = settings_test_session(settings, false);
        without_tools.set_advisor(Some(advisor));
        assert!(without_tools.available_tool_names().is_empty());
    }

    /// A transient stream error can end a response after it starts a tool call.
    /// The retry and every later prompt must still send each call with exactly
    /// one result; OpenAI rejects the request otherwise.
    #[tokio::test]
    async fn side_request_usage_survives_a_resume() {
        let session = AgentSession::new(
            SessionManager::in_memory(Path::new("/test")),
            Vec::new(),
            Registry::load(None),
            Settings::default(),
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
        );
        session.set_stream_fn(Some(Arc::new(|_, _, _| {
            let mut message =
                kiss_ai::AssistantMessage::empty("openai-responses", "openai", "gpt-test");
            message.content = vec![kiss_ai::ContentBlock::text("an answer")];
            message.usage.input = 120;
            message.usage.output = 7;
            message.stop_reason = StopReason::Stop;
            let (sink, stream) = kiss_ai::EventStream::channel();
            sink.done(message);
            stream
        })));
        session
            .answer_btw("what changed?", CancellationToken::new())
            .await
            .unwrap();
        assert_eq!(session.totals().input, 120);
        assert_eq!(
            session.manager.lock().unwrap().usage_totals(),
            session.totals()
        );
    }

    #[tokio::test]
    async fn a_retried_stream_error_leaves_no_unanswered_tool_call() {
        let session = AgentSession::new(
            SessionManager::in_memory(Path::new("/test")),
            Vec::new(),
            Registry::load(None),
            Settings {
                compaction: crate::settings::CompactionSettings {
                    enabled: false,
                    ..Default::default()
                },
                retry: crate::settings::RetrySettings {
                    base_delay_ms: 0,
                    ..Default::default()
                },
                ..Default::default()
            },
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
        );
        let requests: Arc<Mutex<Vec<Vec<kiss_ai::Message>>>> = Arc::default();
        let saved_requests = requests.clone();
        session.set_stream_fn(Some(Arc::new(move |_, context, _| {
            let mut requests = saved_requests.lock().unwrap();
            requests.push(context.messages.clone());
            let mut message =
                kiss_ai::AssistantMessage::empty("openai-responses", "openai", "gpt-test");
            let (sink, stream) = kiss_ai::EventStream::channel();
            if requests.len() == 1 {
                message.content = vec![kiss_ai::ContentBlock::ToolCall(kiss_ai::ToolCall {
                    id: "call_partial|fc_partial".into(),
                    name: "bash".into(),
                    arguments: serde_json::json!({}),
                    thought_signature: None,
                    arguments_error: None,
                })];
                message.stop_reason = StopReason::Error;
                message.error_message = Some("stream error: error decoding response body".into());
                sink.error(message);
            } else {
                message.content = vec![kiss_ai::ContentBlock::text("done")];
                message.stop_reason = StopReason::Stop;
                sink.done(message);
            }
            stream
        })));
        session.prompt(vec![AgentMessage::user("first")]).await;
        session.prompt(vec![AgentMessage::user("second")]).await;

        let requests = requests.lock().unwrap().clone();
        assert_eq!(requests.len(), 3, "error, retry, and the second prompt");
        for (index, messages) in requests.iter().enumerate() {
            let mut pending: Vec<&str> = Vec::new();
            for message in messages {
                match message {
                    kiss_ai::Message::Assistant(assistant) => {
                        pending.extend(assistant.tool_calls().map(|call| call.id.as_str()));
                    }
                    kiss_ai::Message::ToolResult(result) => {
                        let before = pending.len();
                        pending.retain(|id| *id != result.tool_call_id);
                        assert_eq!(
                            pending.len() + 1,
                            before,
                            "request {index} has a result without one matching call: {messages:#?}"
                        );
                    }
                    kiss_ai::Message::User(_) => {}
                }
            }
            assert!(
                pending.is_empty(),
                "request {index} has a tool call without a result: {messages:#?}"
            );
        }
    }

    /// A session whose first model request waits until `release` fires and then
    /// fails with a permanent error. Later requests succeed at once.
    fn session_with_blocked_failure(
        release: Arc<tokio::sync::Notify>,
    ) -> (
        Arc<AgentSession>,
        Arc<std::sync::atomic::AtomicUsize>,
        Arc<Mutex<Vec<SessionEvent>>>,
    ) {
        let events = Arc::new(Mutex::new(Vec::new()));
        let saved_events = events.clone();
        let session = AgentSession::new(
            SessionManager::in_memory(Path::new("/test")),
            Vec::new(),
            Registry::load(None),
            Settings {
                compaction: crate::settings::CompactionSettings {
                    enabled: false,
                    ..Default::default()
                },
                ..Default::default()
            },
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(move |event| saved_events.lock().unwrap().push(event)),
        );
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let observed_calls = calls.clone();
        session.set_stream_fn(Some(Arc::new(move |_, _, _| {
            let step = observed_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            let mut message =
                kiss_ai::AssistantMessage::empty("openai-responses", "openai", "gpt-test");
            let (sink, stream) = kiss_ai::EventStream::channel();
            if step == 0 {
                let release = release.clone();
                tokio::spawn(async move {
                    release.notified().await;
                    message.stop_reason = StopReason::Error;
                    message.error_message = Some("HTTP 400 bad request".into());
                    sink.error(message);
                });
            } else {
                message.stop_reason = StopReason::Stop;
                sink.done(message);
            }
            stream
        })));
        (session, calls, events)
    }

    async fn wait_for_first_request(calls: &std::sync::atomic::AtomicUsize) {
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while calls.load(std::sync::atomic::Ordering::SeqCst) == 0 {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the first model request starts");
    }

    #[tokio::test]
    async fn a_message_queued_during_a_failed_turn_runs_after_it() {
        let release = Arc::new(tokio::sync::Notify::new());
        let (session, calls, _events) = session_with_blocked_failure(release.clone());
        let task_session = session.clone();
        let task = tokio::spawn(async move {
            task_session.prompt(vec![AgentMessage::user("first")]).await;
        });
        wait_for_first_request(&calls).await;

        session.queue_steering(AgentMessage::user("second"));
        release.notify_one();
        tokio::time::timeout(std::time::Duration::from_secs(5), task)
            .await
            .unwrap()
            .unwrap();

        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2);
        assert!(session.reclaim_queued().is_empty());
        let context = session.manager.lock().unwrap().build_session_context();
        assert!(context.messages.iter().any(
            |message| matches!(message, AgentMessage::User(user) if user.content.as_text() == "second")
        ));
    }

    #[test]
    fn a_refused_replacement_writes_no_session_file() {
        let directory = tempfile::tempdir().unwrap();
        let sessions = directory.path().join("sessions");
        let session = AgentSession::new(
            SessionManager::create(directory.path(), Some(sessions.clone())).unwrap(),
            Vec::new(),
            Registry::load(None),
            Settings::default(),
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
        );
        session
            .manager
            .lock()
            .unwrap()
            .append_message(AgentMessage::user("saved"))
            .unwrap();
        fn count(path: &Path) -> usize {
            std::fs::read_dir(path).map_or(0, |entries| {
                entries
                    .flatten()
                    .map(|entry| match entry.file_type() {
                        Ok(kind) if kind.is_dir() => count(&entry.path()),
                        _ => 1,
                    })
                    .sum()
            })
        }
        let files = || count(&sessions);
        let fork = |current: &SessionManager| current.fork_active_branch(current.leaf_id(), true);
        let before = files();
        *session.phase.lock().unwrap() = RunPhase::Streaming;
        assert!(session.replace_manager(fork).is_err());
        assert_eq!(files(), before, "a refused fork must not write a file");
        *session.phase.lock().unwrap() = RunPhase::Idle;
        session.replace_manager(fork).unwrap();
        assert_eq!(files(), before + 1);
    }

    #[tokio::test]
    async fn a_user_shell_result_during_a_run_is_saved_after_it() {
        let release = Arc::new(tokio::sync::Notify::new());
        let (session, calls, _events) = session_with_blocked_failure(release.clone());
        let session_id = session.manager.lock().unwrap().session_id().to_string();
        let task_session = session.clone();
        let task = tokio::spawn(async move {
            task_session.prompt(vec![AgentMessage::user("first")]).await;
        });
        wait_for_first_request(&calls).await;

        let shell = AgentMessage::BashExecution(kiss_agent::BashExecutionMessage {
            command: "ls".into(),
            output: "a".into(),
            exit_code: Some(0),
            cancelled: false,
            truncated: false,
            full_output_path: None,
            exclude_from_context: false,
            timestamp: kiss_ai::now_ms(),
        });
        session.record_user_shell(&session_id, shell);
        let is_shell = |message: &AgentMessage| matches!(message, AgentMessage::BashExecution(_));
        assert!(
            !session
                .manager
                .lock()
                .unwrap()
                .build_session_context()
                .messages
                .iter()
                .any(is_shell),
            "a shell result must not be saved in the middle of a run"
        );
        release.notify_one();
        tokio::time::timeout(std::time::Duration::from_secs(5), task)
            .await
            .unwrap()
            .unwrap();
        let messages = session
            .manager
            .lock()
            .unwrap()
            .build_session_context()
            .messages;
        assert!(messages.last().is_some_and(is_shell), "{messages:?}");
    }

    #[tokio::test]
    async fn manual_compaction_is_refused_during_a_prompt() {
        let release = Arc::new(tokio::sync::Notify::new());
        let (session, calls, events) = session_with_blocked_failure(release.clone());
        let task_session = session.clone();
        let task = tokio::spawn(async move {
            task_session.prompt(vec![AgentMessage::user("first")]).await;
        });
        wait_for_first_request(&calls).await;

        let refused = session.compact(None).await.unwrap_err();
        assert!(refused.contains("cannot compact"), "{refused}");
        assert!(
            !events
                .lock()
                .unwrap()
                .iter()
                .any(|event| { matches!(event, SessionEvent::CompactionStart { .. }) })
        );
        release.notify_one();
        task.await.unwrap();
    }

    #[tokio::test]
    async fn a_message_queued_before_a_compaction_runs_when_it_ends() {
        let release = Arc::new(tokio::sync::Notify::new());
        release.notify_one();
        let (session, calls, _events) = session_with_blocked_failure(release);
        session.queue_steering(AgentMessage::user("queued"));
        session.compact(None).await.unwrap();
        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
        assert!(!session.is_running());
    }

    #[tokio::test]
    async fn a_message_added_while_idle_costs_no_extra_model_request() {
        let session = settings_test_session(
            Settings {
                compaction: crate::settings::CompactionSettings {
                    enabled: false,
                    ..Default::default()
                },
                ..Default::default()
            },
            false,
        );
        let contexts = Arc::new(Mutex::new(Vec::new()));
        let saved_contexts = contexts.clone();
        session.set_stream_fn(Some(Arc::new(move |_, context, _| {
            saved_contexts
                .lock()
                .unwrap()
                .push(format!("{:?}", context.messages));
            let mut message =
                kiss_ai::AssistantMessage::empty("openai-responses", "openai", "gpt-test");
            message.stop_reason = StopReason::Stop;
            let (sink, stream) = kiss_ai::EventStream::channel();
            sink.done(message);
            stream
        })));

        session.add_for_next_turn(AgentMessage::user("the workflow report"));
        session.compact(None).await.unwrap();
        assert!(contexts.lock().unwrap().is_empty());

        session.prompt(vec![AgentMessage::user("next")]).await;
        let contexts = contexts.lock().unwrap();
        assert_eq!(contexts.len(), 1);
        assert!(contexts[0].contains("the workflow report"));
    }

    #[test]
    fn session_replacement_refuses_active_work() {
        let session = settings_test_session(Settings::default(), false);
        let original = session.manager.lock().unwrap().session_id().to_string();
        let run = session.prompt(vec![AgentMessage::user("first")]);
        session
            .replace_manager(|_| Ok(SessionManager::in_memory(Path::new("/other"))))
            .unwrap_err();
        assert_eq!(session.manager.lock().unwrap().session_id(), original);
        drop(run);
        session
            .replace_manager(|_| Ok(SessionManager::in_memory(Path::new("/other"))))
            .unwrap();
        assert_ne!(session.manager.lock().unwrap().session_id(), original);
    }

    #[test]
    fn a_refused_new_session_does_not_create_a_file() {
        let directory = tempfile::tempdir().unwrap();
        let mut manager =
            SessionManager::create(directory.path(), Some(directory.path().join("sessions")))
                .unwrap();
        manager.append_message(AgentMessage::user("saved")).unwrap();
        let files = manager
            .session_file()
            .unwrap()
            .parent()
            .unwrap()
            .to_path_buf();
        let original = manager.session_id().to_string();
        let session = settings_test_session(Settings::default(), false);
        session.replace_manager(|_| Ok(manager)).unwrap();
        let before = std::fs::read_dir(&files).unwrap().count();
        let run = session.prompt(vec![AgentMessage::user("first")]);
        session.new_session().unwrap_err();
        assert_eq!(std::fs::read_dir(&files).unwrap().count(), before);
        assert_eq!(session.manager.lock().unwrap().session_id(), original);
        drop(run);
        session.new_session().unwrap();
        // A saved session writes its file only when its first entry is added.
        assert_eq!(std::fs::read_dir(&files).unwrap().count(), before);
        session
            .manager
            .lock()
            .unwrap()
            .append_message(AgentMessage::user("new saved entry"))
            .unwrap();
        assert_eq!(std::fs::read_dir(&files).unwrap().count(), before + 1);
        assert_ne!(session.manager.lock().unwrap().session_id(), original);
    }

    #[test]
    fn a_dropped_prompt_future_returns_the_session_to_idle() {
        let session = settings_test_session(Settings::default(), false);
        let run = session.prompt(vec![AgentMessage::user("first")]);
        let cancel = session.cancel.lock().unwrap().clone();
        assert_eq!(session.run_phase(), RunPhase::Streaming);
        drop(run);
        assert_eq!(session.run_phase(), RunPhase::Idle);
        assert!(
            cancel.is_cancelled(),
            "dropping a run must cancel its provider work"
        );
    }

    #[tokio::test]
    async fn context_file_bash_edits_reach_next_request_and_survive_resume() {
        let directory = tempfile::tempdir().unwrap();
        let mut manager =
            SessionManager::create(directory.path(), Some(directory.path().join("sessions")))
                .unwrap();
        manager
            .append_message(AgentMessage::user("obsolete output"))
            .unwrap();
        manager
            .append_compaction(
                "portable summary".into(),
                100,
                vec![AgentMessage::user("obsolete output")],
                None,
                Some(serde_json::json!({"remoteCompaction": {
                    "version": 2,
                    "provider": "openai-responses-compaction",
                    "modelKey": "openai:openai-responses:gpt-test",
                    "replacementHistory": [{"type": "compaction", "encrypted_content": "opaque"}]
                }})),
            )
            .unwrap();
        assert!(
            manager
                .build_openai_compaction_context(&openai_model())
                .is_some()
        );
        let session_path = manager.session_file().unwrap().to_path_buf();
        let settings = Settings {
            experimental_context_file: true,
            compaction: crate::settings::CompactionSettings {
                enabled: false,
                ..Default::default()
            },
            retry: crate::settings::RetrySettings {
                base_delay_ms: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let events = Arc::new(Mutex::new(Vec::new()));
        let saved_events = events.clone();
        let session = AgentSession::new(
            manager,
            vec![Arc::new(kiss_agent::tools::bash::BashTool::new(
                directory.path().to_path_buf(),
            ))],
            Registry::load(None),
            settings,
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(move |event| saved_events.lock().unwrap().push(event)),
        );
        let weak = Arc::downgrade(&session);
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let observed_calls = calls.clone();
        session.set_stream_fn(Some(Arc::new(move |_, context, _| {
            let step = observed_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            let prompt = context.system_prompt.as_ref().unwrap();
            let path = prompt.lines().find_map(|line| line.strip_prefix("Experimental context file: ")).unwrap();
            assert!(Path::new(path).is_file());
            assert!(context.openai_responses_input.is_none());
            let mut message = kiss_ai::AssistantMessage::empty("openai-responses", "openai", "gpt-test");
            let replacement = match step {
                0 => {
                    weak.upgrade().unwrap().queue_steering(AgentMessage::user("new instruction"));
                    Some(r#"[{"role":"user","content":"saved notes","timestamp":0}]"#)
                }
                1 => {
                    let text = serde_json::to_string(&context.messages).unwrap();
                    assert!(text.contains("saved notes"));
                    assert!(!text.contains("obsolete output"));
                    assert!(text.contains("new instruction"));
                    assert!(context.messages.iter().any(|message| matches!(message, kiss_ai::Message::ToolResult(result) if result.tool_call_id == "edit_0" && !result.is_error)));
                    Some("invalid JSON")
                }
                2 => {
                    let text = serde_json::to_string(&context.messages).unwrap();
                    assert!(text.contains("saved notes"));
                    assert!(text.contains("Repair the file"));
                    assert_eq!(std::fs::read_to_string(path).unwrap(), "invalid JSON");
                    Some(r#"[{"role":"user","content":"repaired notes","timestamp":0}]"#)
                }
                3 => {
                    let text = serde_json::to_string(&context.messages).unwrap();
                    assert!(text.contains("repaired notes"));
                    assert!(context.messages.iter().any(|message| matches!(message, kiss_ai::Message::User(user) if user.content.as_text() == "repaired notes")));
                    assert!(context.messages.iter().any(|message| matches!(message, kiss_ai::Message::ToolResult(result) if result.tool_call_id == "edit_2" && !result.is_error)));
                    None
                }
                4 => {
                    assert!(!matches!(context.messages.last(), Some(kiss_ai::Message::Assistant(assistant)) if assistant.stop_reason == StopReason::Error));
                    assert!(context.messages.iter().any(|message| matches!(message, kiss_ai::Message::User(user) if user.content.as_text() == "repaired notes")));
                    None
                }
                _ => panic!("unexpected model request"),
            };
            if let Some(replacement) = replacement {
                let quoted_path = path.replace('\'', "'\\''");
                message.content.push(kiss_ai::ContentBlock::ToolCall(kiss_ai::ToolCall {
                    id: format!("edit_{step}"),
                    name: "bash".into(),
                    arguments: serde_json::json!({"command": format!("printf '%s' '{replacement}' > '{quoted_path}'")}),
                    thought_signature: None,
                    arguments_error: None,
                }));
                message.stop_reason = StopReason::ToolUse;
            } else if step == 3 {
                message.stop_reason = StopReason::Error;
                message.error_message = Some("HTTP 503".into());
            } else {
                message.content.push(kiss_ai::ContentBlock::text("done"));
                message.stop_reason = StopReason::Stop;
            }
            let (sink, stream) = kiss_ai::EventStream::channel();
            sink.send(kiss_ai::AssistantEvent::Start { partial: message.clone() });
            if message.stop_reason == StopReason::Error {
                sink.error(message);
            } else {
                sink.done(message);
            }
            stream
        })));
        session
            .prompt(vec![AgentMessage::user("manage context")])
            .await;
        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 5);
        let context = session
            .manager
            .lock()
            .unwrap()
            .build_session_context()
            .messages;
        assert_eq!(
            SessionManager::open(&session_path)
                .unwrap()
                .build_session_context()
                .messages,
            context
        );
        assert!(
            std::fs::read_to_string(session_path)
                .unwrap()
                .contains("obsolete output")
        );
        let errors = events.lock().unwrap().iter().filter(|event| matches!(event, SessionEvent::Agent(event) if matches!(event.as_ref(), AgentEvent::MessageEnd { message: AgentMessage::Custom(custom) } if custom.custom_type == "context_file_error"))).count();
        assert_eq!(errors, 1);
    }

    #[test]
    fn context_file_lifecycle_is_opt_in_and_session_local() {
        let session = AgentSession::new(
            SessionManager::in_memory(Path::new("/test")),
            Vec::new(),
            Registry::load(None),
            Settings::default(),
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
        );
        assert!(!session.sync_context_file());
        assert!(session.context_file.lock().unwrap().is_none());
        assert_eq!(session.build_context().system_prompt, "test");
        let mut settings = session.settings();
        settings.experimental_context_file = true;
        session.update_settings(settings);
        session.sync_context_file();
        let path = session
            .context_file
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .path
            .to_path_buf();
        let child = session
            .create_subagent_session("task", "/root/task", ForkTurns::None, None, None)
            .unwrap();
        child.sync_context_file();
        let child_path = child
            .context_file
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .path
            .to_path_buf();
        assert_ne!(path, child_path);
        assert!(child_path.is_file());
        session
            .replace_manager(|_| Ok(SessionManager::in_memory(Path::new("/other"))))
            .unwrap();
        assert!(!path.exists());
        assert!(session.context_file.lock().unwrap().is_none());
        session.sync_context_file();
        let new_path = session
            .context_file
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .path
            .to_path_buf();
        assert_ne!(new_path, child_path);
        let mut settings = session.settings();
        settings.experimental_context_file = false;
        session.update_settings(settings);
        assert!(session.sync_context_file());
        assert!(!new_path.exists());
        assert_eq!(session.build_context().system_prompt, "test");
    }

    #[test]
    fn reasoning_lease_ends_when_model_saved_effort_or_applied_effort_changes() {
        let model = openai_model();
        let mut run = ReasoningRunState {
            decision_model: Some(DecisionModel::Jev),
            lease: crate::decision::ReasoningLease::default(),
            provider: model.provider.clone(),
            model_id: model.id.clone(),
            saved_effort: ThinkingLevel::Medium,
        };
        let selection = crate::decision::ReasoningSelection {
            level: ThinkingLevel::High,
            generations: 5,
        };
        run.lease.install(&selection);
        assert_eq!(
            run.checkpoint(
                &model,
                ThinkingLevel::Medium,
                ThinkingLevel::High,
                Some(DecisionModel::Jev)
            ),
            (false, Some(ThinkingLevel::High))
        );

        assert_eq!(
            run.checkpoint(
                &model,
                ThinkingLevel::Medium,
                ThinkingLevel::High,
                Some(DecisionModel::D1Ultra)
            ),
            (false, None)
        );
        run.lease.install(&selection);

        let mut other_model = model.clone();
        other_model.provider = "other".into();
        assert_eq!(
            run.checkpoint(
                &other_model,
                ThinkingLevel::Medium,
                ThinkingLevel::High,
                Some(DecisionModel::Jev)
            ),
            (true, None)
        );
        run.lease.install(&selection);
        assert_eq!(
            run.checkpoint(
                &other_model,
                ThinkingLevel::Low,
                ThinkingLevel::High,
                Some(DecisionModel::Jev)
            ),
            (false, None)
        );
        run.lease.install(&selection);
        assert_eq!(
            run.checkpoint(
                &other_model,
                ThinkingLevel::Low,
                ThinkingLevel::Low,
                Some(DecisionModel::Jev)
            ),
            (false, None)
        );
        run.lease.install(&selection);
        assert_eq!(
            run.checkpoint(&other_model, ThinkingLevel::Low, ThinkingLevel::High, None),
            (false, None)
        );
    }

    #[test]
    fn late_cache_refreshes_are_skipped_before_the_cache_expires() {
        let ttl = std::time::Duration::from_secs(300);
        let delay = cache_warming_delay(ttl).unwrap();
        assert!(!cache_refresh_deadline_missed(
            std::time::Duration::from_secs(284),
            ttl,
            delay,
        ));
        assert!(cache_refresh_deadline_missed(
            std::time::Duration::from_secs(286),
            ttl,
            delay,
        ));
    }

    #[test]
    fn transient_errors_require_a_status_or_specific_network_failure() {
        assert!(is_transient("request failed with HTTP 503"));
        assert!(is_transient("HTTP 529: Overloaded"));
        assert!(is_transient(r#"provider error {"status": 503}"#));
        assert!(is_transient("Cloudflare returned HTTP 520"));
        assert!(is_transient("Azure is currently experiencing high demand"));
        assert!(is_transient("connection reset by peer"));
        assert!(is_transient("rate limit exceeded"));
        assert!(!is_transient("model has a 500 token limit"));
        assert!(!is_transient("connection settings are invalid"));
        assert!(!is_transient(
            "HTTP 429: subscription_sharing_usage_limit_exceeded"
        ));
        assert!(is_transient(
            "Selected model is at capacity. Please try again"
        ));
        assert!(is_transient(r#"{"error":{"code":"server_busy"}}"#));
        assert!(is_transient("The service is temporarily unavailable"));
        assert!(is_transient("The upstream provider returned an error"));
        assert!(is_transient("Unable to forward upstream stream"));
        assert!(is_transient(
            "stream disconnected before completion: stream closed before response.completed"
        ));
        assert!(is_transient("The pending stream has been canceled"));
        assert!(is_transient(
            "Claude Code's stream from Anthropic was cut short after the reply had begun"
        ));
        assert!(is_transient("Provider finish_reason: error (server error)"));
        assert!(!is_transient(
            "request req_5030 failed: invalid tool schema"
        ));
        assert!(!is_transient(
            "subscription_sharing_workplace_individual_usage_limit_exceeded"
        ));
        assert!(!is_transient(
            r#"HTTP 429: {"error":{"code":"1308","message":"Usage limit reached for 5 hour."}}"#
        ));
        assert!(!is_transient(r#"HTTP 429: {"code":1308,"msg":"limit"}"#));
        assert!(is_transient(r#"HTTP 429: {"error":{"code":"13080"}}"#));
        assert!(is_transient("subscription_sharing_usage_unavailable"));
        assert!(is_transient("subscription_sharing_user_unavailable"));
        assert!(!is_transient(
            r#"Cursor request failed: stream ended: {"error":{"code":"not_found","message":"Model name is not valid: auto"}}"#
        ));
        assert!(!is_transient(
            r#"Cursor request failed: stream ended: {"error":{"code":"internal","message":"KISS does not run Cursor-native tools"}}"#
        ));
        assert!(!is_transient(
            r#"Cursor request failed: stream ended: {"error":{"details":[{"debug":{"details":{"isRetryable":false,"detail":"rate limit exceeded"}}}]}}"#
        ));
    }

    #[test]
    fn session_tools_install_replace_and_remove_without_changing_base_tools() {
        let registry = Registry::load(None);
        let session = AgentSession::new(
            SessionManager::in_memory(std::path::Path::new("/test")),
            Vec::new(),
            registry,
            Settings::default(),
            "test".into(),
            openai_model(),
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
        );
        let tool = || {
            Arc::new(crate::tools::grep::GrepTool {
                cwd: std::path::PathBuf::from("/test"),
            }) as DynTool
        };

        assert!(!session.available_tool_names().contains(&"grep".into()));
        session.install_session_tool(tool());
        session.install_session_tool(tool());
        assert_eq!(
            session
                .available_tool_names()
                .iter()
                .filter(|name| name.as_str() == "grep")
                .count(),
            1
        );
        assert!(session.remove_session_tool("grep"));
        assert!(!session.remove_session_tool("grep"));
        assert!(!session.available_tool_names().contains(&"grep".into()));
    }

    fn remote_result() -> kiss_ai::api::openai_compaction::RemoteCompactionResult {
        kiss_ai::api::openai_compaction::RemoteCompactionResult {
            replacement_history: vec![serde_json::json!({
                "type": "compaction",
                "encrypted_content": "opaque"
            })],
            usage: Some(Usage {
                input: 10,
                output: 2,
                total_tokens: 12,
                ..Default::default()
            }),
        }
    }

    fn settings_test_session(settings: Settings, subagents_allowed: bool) -> Arc<AgentSession> {
        let registry = Registry::from_builtin();
        let model = registry.all().first().expect("built-in model").clone();
        AgentSession::new_with_subagents_allowed(
            SessionManager::in_memory(std::path::Path::new("/test")),
            Vec::new(),
            registry,
            settings,
            "root prompt".into(),
            model,
            ThinkingLevel::Off,
            None,
            Arc::new(|_| {}),
            subagents_allowed,
        )
    }

    #[test]
    fn cursor_native_shell_receives_registered_shell_instructions() {
        let session = settings_test_session(Settings::default(), false);
        let mut model = session.model();
        model.provider = "cursor".into();
        session.set_model(model);
        for (path, syntax) in [
            ("pwsh.exe", "PowerShell 7"),
            ("powershell.exe", "Windows PowerShell 5.1"),
            ("cmd.exe", "cmd.exe syntax"),
            ("bash", "POSIX shell syntax"),
        ] {
            let mut tool = kiss_agent::tools::bash::BashTool::new("/test".into());
            tool.shell_path = Some(path.into());
            session.install_session_tool(Arc::new(tool));
            let context = session.build_context();
            assert!(
                context.system_prompt.contains(path),
                "{}",
                context.system_prompt
            );
            assert!(
                context.system_prompt.contains(syntax),
                "{}",
                context.system_prompt
            );
            assert!(context.tools[0].to_def().description.contains(syntax));
        }
        assert!(session.remove_session_tool("bash"));
        assert_eq!(session.build_context().system_prompt, "root prompt");
    }

    fn benchmark_tools() -> Vec<DynTool> {
        let cwd = std::path::PathBuf::from("/synthetic");
        vec![
            Arc::new(kiss_agent::tools::read::ReadTool { cwd: cwd.clone() }),
            Arc::new(kiss_agent::tools::write::WriteTool { cwd: cwd.clone() }),
            Arc::new(kiss_agent::tools::edit::EditTool { cwd: cwd.clone() }),
            Arc::new(kiss_agent::tools::bash::BashTool::new(cwd)),
        ]
    }

    #[test]
    fn subagent_tools_follow_settings_and_command_line_authority() {
        let session = settings_test_session(Settings::default(), true);
        assert!(session.available_tool_names().is_empty());
        assert!(
            !session
                .build_context()
                .system_prompt
                .contains("Subagent coordination")
        );

        let mut enabled = session.settings();
        enabled.subagents.enabled = true;
        session.update_settings(enabled.clone());
        assert_eq!(
            session.available_tool_names(),
            [
                "spawn_agent",
                "send_message",
                "followup_task",
                "wait_agent",
                "list_agents",
                "interrupt_agent"
            ]
        );
        assert!(
            session
                .build_context()
                .system_prompt
                .contains("Subagent coordination")
        );

        enabled.subagents.enabled = false;
        session.update_settings(enabled);
        assert!(session.available_tool_names().is_empty());

        let mut blocked_settings = Settings::default();
        blocked_settings.subagents.enabled = true;
        let blocked = settings_test_session(blocked_settings, false);
        assert!(blocked.available_tool_names().is_empty());
        assert!(
            !blocked
                .build_context()
                .system_prompt
                .contains("Subagent coordination")
        );
    }

    #[test]
    fn the_workflow_tool_appears_only_in_workflow_prompt_mode() {
        let mut settings = Settings::default();
        settings.subagents.enabled = true;
        let session = settings_test_session(settings, true);

        // Subagents on, workflow not armed: an ordinary coding turn pays
        // nothing for the feature.
        assert!(
            !session
                .available_tool_names()
                .contains(&"run_workflow".into())
        );
        assert_eq!(
            session.prompt_mode_for("run a dynamic workflow for this task"),
            PromptMode::Workflow
        );
        assert_eq!(
            session.prompt_mode_for("fix this small function"),
            PromptMode::Ordinary
        );
        assert!(
            !session
                .build_context()
                .system_prompt
                .contains("Writing a dynamic workflow")
        );

        assert!(
            session
                .available_tool_names_for(PromptMode::Workflow)
                .contains(&"run_workflow".into())
        );
        assert!(
            session
                .build_context_for(PromptMode::Workflow)
                .system_prompt
                .contains("Writing a dynamic workflow")
        );
        assert!(
            !session
                .available_tool_names()
                .contains(&"run_workflow".into())
        );
    }

    #[test]
    fn workflow_prompt_mode_does_nothing_while_subagents_are_off() {
        // Workflows are built on child agents, so the subagent setting is the
        // authority for both.
        let session = settings_test_session(Settings::default(), true);
        assert!(!session.workflows_enabled());
        assert!(
            !session
                .available_tool_names_for(PromptMode::Workflow)
                .contains(&"run_workflow".into())
        );

        let mut settings = session.settings();
        settings.subagents.enabled = true;
        settings.workflows.enabled = false;
        session.update_settings(settings.clone());
        assert!(!session.workflows_enabled());
        assert!(
            !session
                .available_tool_names_for(PromptMode::Workflow)
                .contains(&"run_workflow".into())
        );

        settings.workflows.enabled = true;
        session.update_settings(settings);
        assert!(session.workflows_enabled());
        assert!(
            session
                .available_tool_names_for(PromptMode::Workflow)
                .contains(&"run_workflow".into())
        );
    }

    #[test]
    fn one_at_a_time_queues_keep_each_prompts_mode() {
        let queue = Arc::new(Mutex::new(VecDeque::from([
            QueuedPrompt {
                message: AgentMessage::user("ordinary"),
                mode: PromptMode::Ordinary,
            },
            QueuedPrompt {
                message: AgentMessage::user("workflow"),
                mode: PromptMode::Workflow,
            },
        ])));

        assert_eq!(
            queued_mode(&queue, QueueMode::OneAtATime),
            Some(PromptMode::Ordinary)
        );
        assert_eq!(drain_queue(&queue, QueueMode::OneAtATime).len(), 1);
        assert_eq!(
            queued_mode(&queue, QueueMode::OneAtATime),
            Some(PromptMode::Workflow)
        );
    }

    #[test]
    fn a_session_without_subagent_authority_has_no_workflow_runtime() {
        let mut settings = Settings::default();
        settings.subagents.enabled = true;
        let child = settings_test_session(settings, false);
        assert!(child.workflows().is_none());
        assert!(!child.workflows_enabled());
    }

    #[test]
    fn child_session_has_safe_forked_context_without_control_tools() {
        let mut settings = Settings::default();
        settings.subagents.enabled = true;
        let parent = settings_test_session(settings, true);
        parent
            .manager
            .lock()
            .unwrap()
            .append_message(AgentMessage::user("parent context"))
            .unwrap();

        let child = parent
            .create_subagent_session("inspect", "/root/inspect", ForkTurns::All, None, None)
            .unwrap();
        assert!(Arc::ptr_eq(&parent.registry, &child.registry));
        assert!(child.available_tool_names().is_empty());
        let context = child.manager.lock().unwrap().build_session_context();
        assert!(matches!(
            context.messages.as_slice(),
            [AgentMessage::User(user)] if user.content.as_text() == "parent context"
        ));
    }

    #[test]
    fn child_without_forked_turns_does_not_copy_parent_context() {
        let parent = settings_test_session(Settings::default(), true);
        parent
            .manager
            .lock()
            .unwrap()
            .append_message(AgentMessage::user("parent context"))
            .unwrap();

        let child = parent
            .create_subagent_session("inspect", "/root/inspect", ForkTurns::None, None, None)
            .unwrap();
        assert!(
            child
                .manager
                .lock()
                .unwrap()
                .build_session_context()
                .messages
                .is_empty()
        );
    }

    #[test]
    #[ignore = "release-mode performance benchmark"]
    fn benchmark_performance_subagent_overhead() {
        let registry = Registry::from_builtin();
        let model = registry.all().first().expect("built-in model").clone();
        let tools = benchmark_tools();
        let make_session = |enabled: bool| {
            let mut settings = Settings::default();
            settings.subagents.enabled = enabled;
            AgentSession::new_with_subagents_allowed(
                SessionManager::in_memory(std::path::Path::new("/synthetic")),
                tools.clone(),
                registry.clone(),
                settings,
                "benchmark root prompt".into(),
                model.clone(),
                ThinkingLevel::Off,
                None,
                Arc::new(|_| {}),
                true,
            )
        };

        kiss_bench::measure_pair(
            (
                "agent_session_create_subagents_off",
                "agent_session_create_subagents_on",
            ),
            21,
            500,
            (
                "new_root_session_4_base_tools_0_control_tools",
                "new_root_session_4_base_tools_6_control_tools",
            ),
            || make_session(false),
            || make_session(true),
        );

        let off = make_session(false);
        let on = make_session(true);
        kiss_bench::measure_pair(
            (
                "agent_context_build_subagents_off",
                "agent_context_build_subagents_on",
            ),
            21,
            10_000,
            (
                "empty_session_4_base_tools_0_control_tools",
                "empty_session_4_base_tools_6_control_tools",
            ),
            || off.build_context(),
            || on.build_context(),
        );
    }

    #[test]
    #[ignore = "release-mode performance benchmark"]
    fn benchmark_performance_workflow_tool_exposure() {
        // An ordinary coding turn must pay nothing for dynamic workflows. The
        // disarmed session is the baseline. The armed one carries the extra
        // tool and the authoring instructions.
        let registry = Registry::from_builtin();
        let model = registry.all().first().expect("built-in model").clone();
        let tools = benchmark_tools();
        let make_session = || {
            let mut settings = Settings::default();
            settings.subagents.enabled = true;
            AgentSession::new_with_subagents_allowed(
                SessionManager::in_memory(std::path::Path::new("/synthetic")),
                tools.clone(),
                registry.clone(),
                settings,
                "benchmark root prompt".into(),
                model.clone(),
                ThinkingLevel::Off,
                None,
                Arc::new(|_| {}),
                true,
            )
        };

        let ordinary = make_session();
        let workflow = make_session();
        kiss_bench::measure_pair(
            (
                "agent_context_build_workflow_disarmed",
                "agent_context_build_workflow_armed",
            ),
            21,
            10_000,
            (
                "empty_session_subagents_on_workflow_disarmed",
                "empty_session_subagents_on_workflow_armed",
            ),
            || ordinary.build_context_for(PromptMode::Ordinary),
            || workflow.build_context_for(PromptMode::Workflow),
        );
    }

    #[test]
    fn transcript_excerpt_keeps_only_recent_user_and_assistant_text() {
        let messages = vec![
            AgentMessage::user("old"),
            AgentMessage::BashExecution(kiss_agent::BashExecutionMessage {
                command: "pwd".into(),
                output: "ignored".into(),
                exit_code: Some(0),
                cancelled: false,
                truncated: false,
                full_output_path: None,
                exclude_from_context: false,
                timestamp: 1,
            }),
            AgentMessage::user("new"),
        ];
        let excerpt = transcript_excerpt(&messages, 1, 100);
        assert_eq!(excerpt, "User: new");
    }

    #[test]
    fn transcript_excerpt_enforces_character_budget_from_the_tail() {
        let excerpt = transcript_excerpt(&[AgentMessage::user("abcdefghij")], 4, 5);
        assert!(excerpt.ends_with("fghij"));
        assert!(excerpt.starts_with("[earlier text omitted]"));
    }

    #[test]
    fn hybrid_compaction_keeps_local_summary_and_remote_details() {
        let selected = select_compaction_outcome(
            &openai_model(),
            Ok(compaction::SummaryOutcome {
                summary: "portable".into(),
                usage: None,
            }),
            Some(Ok(remote_result())),
        )
        .unwrap();
        assert_eq!(selected.summary, "portable");
        assert_eq!(selected.remote_usage.unwrap().input, 10);
        assert_eq!(
            selected.remote_details.unwrap()["remoteCompaction"]["version"],
            2
        );
    }

    #[test]
    fn remote_failure_falls_back_to_local_compaction() {
        let selected = select_compaction_outcome(
            &openai_model(),
            Ok(compaction::SummaryOutcome {
                summary: "portable".into(),
                usage: None,
            }),
            Some(Err(anyhow::anyhow!("remote unavailable"))),
        )
        .unwrap();
        assert_eq!(selected.summary, "portable");
        assert!(selected.remote_details.is_none());
    }

    #[test]
    fn remote_success_survives_local_summary_failure() {
        let selected = select_compaction_outcome(
            &openai_model(),
            Err(anyhow::anyhow!("summary unavailable")),
            Some(Ok(remote_result())),
        )
        .unwrap();
        assert!(
            selected
                .summary
                .contains("server-side compaction was applied")
        );
        assert!(selected.remote_details.is_some());
    }

    #[test]
    fn details_merge_keeps_file_operations_and_remote_artifact() {
        let merged = merge_compaction_details(
            serde_json::json!({"readFiles": ["a.rs"], "modifiedFiles": []}),
            Some(serde_json::json!({"remoteCompaction": {"version": 2}})),
        );
        assert_eq!(merged["readFiles"][0], "a.rs");
        assert_eq!(merged["remoteCompaction"]["version"], 2);
    }

    #[test]
    fn auto_compaction_guard_checks_settings_threshold_and_cancel() {
        let mut settings = Settings::default();
        settings.compaction.reserve_tokens = 20;
        let messages = vec![AgentMessage::user("x".repeat(360))];
        let mut model = openai_model();
        model.context_window = 100;
        assert!(auto_compaction_needed(&settings, &messages, &model, false));
        assert!(!auto_compaction_needed(&settings, &messages, &model, true));
        settings.compaction.enabled = false;
        assert!(!auto_compaction_needed(&settings, &messages, &model, false));
        settings.compaction.enabled = true;
        let mut assistant = kiss_ai::AssistantMessage::empty("test", "test", "test");
        assistant.usage.input = 1000;
        assistant
            .content
            .push(kiss_ai::ContentBlock::text("short note"));
        let edited = vec![
            AgentMessage::user("task"),
            AgentMessage::Assistant(assistant),
        ];
        assert!(auto_compaction_needed(&settings, &edited, &model, false));
        settings.experimental_context_file = true;
        assert!(!auto_compaction_needed(&settings, &edited, &model, false));
    }

    #[test]
    fn session_title_normalization_is_safe_and_bounded() {
        assert_eq!(
            normalize_session_title("  `Fix AUTH-123 login flow!`  \nignored").as_deref(),
            Some("Fix AUTH-123 login flow")
        );
        assert_eq!(normalize_session_title("\n\t"), None);
        assert_eq!(
            normalize_session_title("🚀".repeat(50).as_str())
                .unwrap()
                .chars()
                .count(),
            SESSION_TITLE_MAX_CHARS
        );
    }

    #[test]
    fn session_title_prompt_is_utf8_safe_and_bounded() {
        let prompt = "🚀".repeat(SESSION_TITLE_PROMPT_MAX_BYTES);
        let bounded = bounded_session_title_prompt(&prompt);
        assert!(bounded.len() <= SESSION_TITLE_PROMPT_MAX_BYTES);
        assert!(std::str::from_utf8(bounded.as_bytes()).is_ok());
    }

    #[test]
    fn cache_warming_uses_ninety_percent_with_ten_second_margin() {
        assert_eq!(
            cache_warming_delay(std::time::Duration::from_secs(300)),
            Some(std::time::Duration::from_secs(270))
        );
        assert_eq!(
            cache_warming_delay(std::time::Duration::from_secs(60)),
            Some(std::time::Duration::from_secs(50))
        );
        assert_eq!(
            cache_warming_delay(std::time::Duration::from_secs(10)),
            None
        );
    }
}