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::compaction::{
    self, estimate_context_tokens, 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, 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.
    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;
        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.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 {
    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>>>,
    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>>,
    /// 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 session = Arc::new(AgentSession {
            manager: Mutex::new(manager),
            registry: registry.into(),
            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(),
            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),
            stream_fn: Mutex::new(None),
        });
        if subagents_allowed {
            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()
    }

    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;
    }

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

    pub(crate) fn record_subagent_usage(&self, usage: Usage) {
        self.totals.lock().unwrap().add(&usage);
    }

    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) {
        *self.context_usage_cache.lock().unwrap() = None;
        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.context_usage_cache.lock().unwrap() = None;
        let was_enabled = self.subagents_enabled();
        let workflows_were_enabled = self.workflows_enabled();
        let cache_warming = settings.cache_warming;
        *self.settings.lock().unwrap() = settings;
        *self.system_prompt.lock().unwrap() = system_prompt;
        *self.base_tools.lock().unwrap() = tools;
        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();
    }

    /// 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());
        }
        *self.tools.lock().unwrap() = tools;
    }

    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.
    pub fn replace_manager(&self, manager: SessionManager) -> anyhow::Result<()> {
        self.replace_manager_with(|| Ok(manager))
    }

    /// Start a sibling session after admission, preserving the storage mode.
    pub fn new_session(&self) -> anyhow::Result<()> {
        self.replace_manager_with(|| self.manager.lock().unwrap().create_sibling())
    }

    fn replace_manager_with(
        &self,
        create: impl FnOnce() -> anyhow::Result<SessionManager>,
    ) -> anyhow::Result<()> {
        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()?;
        self.cache_warm_cancel.lock().unwrap().cancel();
        *self.context_file.lock().unwrap() = None;
        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;
        }
        *self.manager.lock().unwrap() = manager;
        *self.context_usage_cache.lock().unwrap() = None;
        *self.totals.lock().unwrap() = totals;
        Ok(())
    }

    pub fn set_model(&self, model: Model) {
        self.cache_warm_cancel.lock().unwrap().cancel();
        {
            let mut m = self.manager.lock().unwrap();
            let _ = m.append_model_change(&model.provider, &model.id);
        }
        (self.sink)(SessionEvent::ModelChanged {
            provider: model.provider.clone(),
            model_id: model.id.clone(),
        });
        *self.model.lock().unwrap() = model;
    }

    pub fn set_thinking_level(&self, level: ThinkingLevel) {
        let _ = self
            .manager
            .lock()
            .unwrap()
            .append_thinking_level_change(level);
        *self.thinking.lock().unwrap() = level;
    }

    /// 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> {
        let (old_leaf, new_leaf, editor_text, abandoned_messages) = {
            let manager = self.manager.lock().unwrap();
            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(
                &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 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
        };

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

    pub fn queue_steering(&self, message: AgentMessage) {
        self.steering.lock().unwrap().push_back(QueuedPrompt {
            message,
            mode: PromptMode::Ordinary,
        });
        self.emit_queues();
    }

    pub fn queue_follow_up(&self, message: AgentMessage) {
        self.follow_up.lock().unwrap().push_back(QueuedPrompt {
            message,
            mode: PromptMode::Ordinary,
        });
        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) {
        let phase = self.phase.lock().unwrap();
        if *phase != RunPhase::Idle {
            self.follow_up.lock().unwrap().push_back(QueuedPrompt {
                message,
                mode: PromptMode::Ordinary,
            });
            drop(phase);
            self.emit_queues();
            return;
        }
        // Other code holds the manager lock while it reads the phase, so this
        // must not hold the phase lock while it takes the manager lock.
        drop(phase);
        let _ = self.manager.lock().unwrap().append_message(message);
    }

    /// 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 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
                    .ok()
                    .flatten()
            })
        }));

        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()
                };
                if settings.experimental_context_file
                    && let Some(context) = &mut update.context
                {
                    // A retry must not put the failed answer back in the
                    // provider request when the file refreshes the context.
                    while matches!(context.messages.last(), Some(AgentMessage::Assistant(assistant)) if assistant.stop_reason == StopReason::Error)
                    {
                        context.messages.pop();
                    }
                }
                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 = match kiss_ai::auth::resolve_api_key_async(
                    decision_model.provider(),
                    &session.registry.declared_keys,
                )
                .await
                {
                    Ok(Some(api_key)) => {
                        let messages = session
                            .manager
                            .lock()
                            .unwrap()
                            .build_session_context()
                            .messages;
                        let cancel = session.cancel.lock().unwrap().clone();
                        async {
                            crate::decision::DecisionClient::new(decision_model, &api_key)?
                                .select_reasoning(&messages, &[], &model, cancel)
                                .await
                        }
                        .await
                        .map_err(|error: anyhow::Error| error.to_string())
                    }
                    Ok(None) => Err(format!(
                        "{} credentials are unavailable; run /login {}",
                        decision_model.label(),
                        decision_model.provider(),
                    )),
                    Err(error) => Err(format!(
                        "{} credentials could not be read: {error}; run /login {}",
                        decision_model.label(),
                        decision_model.provider(),
                    )),
                };
                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()
                })
            })
        }));
        if let Some(stream_fn) = self.stream_fn.lock().unwrap().clone() {
            config.stream_fn = stream_fn;
        }
        let sink = self.sink.clone();
        config.stream_fn = crate::account_failover::wrap(
            config.stream_fn,
            self.registry.clone(),
            Arc::new(move |switch| sink(SessionEvent::AccountSwitched(switch))),
        );
        config
    }

    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.context_usage_cache.lock().unwrap() = None;
                    self.cache_warm_cancel.lock().unwrap().cancel();
                }
                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, 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);
        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(&format!(
                "\n\nExperimental context file: {}\n\
                 This JSON array contains your live KISS conversation, without system instructions. \
                 Use your normal file tools to edit it. Changes apply after the tool batch and before the next model request. \
                 New user messages, your current answer, and tool results are added automatically. \
                 Keep important task instructions, exact facts, progress, and next steps. Replace stale output with useful notes. \
                 Keep complete assistant tool-call/result groups, or replace the whole group with a user note. \
                 A user note has the form {{\"role\":\"user\",\"content\":\"Notes here\",\"timestamp\":0}}. \
                 Preserve fields of messages you keep, including image and reasoning data. \
                 The file must be a valid JSON array at most 16 MiB. Invalid edits leave the live conversation unchanged; repair the file after an error. \
                 Read only the parts you need, because the conversation is already in your context. \
                 Your model context window is {} tokens. Manage the file before it fills; automatic compaction remains an emergency fallback. \
                 Batch edits: changing early text can require the provider to process all later text again. \
                 The full session record remains separate from this editable file.",
                file.path().display(), model.context_window
            ));
        }
        if self.subagents_enabled() {
            system_prompt.push_str("\n\n");
            system_prompt.push_str(SUBAGENT_SYSTEM_PROMPT);
        }
        if prompt_mode == PromptMode::Workflow
            && self.workflows_enabled()
            && let Some(runtime) = self.workflows.get()
        {
            let limits = runtime.limits();
            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)
    }

    async fn run_ephemeral(
        self: &Arc<Self>,
        system_prompt: String,
        prompt: String,
        tools: Vec<DynTool>,
        max_tokens: u64,
        cancel: CancellationToken,
    ) -> anyhow::Result<EphemeralResponse> {
        let mut config = self.loop_config(self, Arc::new(Mutex::new(PromptMode::Ordinary)));
        config.thinking_level = ThinkingLevel::Off;
        config.max_tokens = Some(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;
        if cancel.is_cancelled() {
            anyhow::bail!("request cancelled");
        }

        let mut usage = Usage::default();
        for message in &messages {
            if let AgentMessage::Assistant(assistant) = message {
                usage.add(&assistant.usage);
            }
        }
        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");
        }
        self.totals.lock().unwrap().add(&usage);
        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(
            "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,
            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(
                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(),
                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(
            "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(),
            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 mut manager = self.manager.lock().unwrap();
            for prompt in prompts {
                let _ = manager.append_message(prompt);
            }
        }
        // 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)));
        });

        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());
                // Drop the trailing error assistant message from context.
                while matches!(
                    context.messages.last(),
                    Some(AgentMessage::Assistant(a)) if a.stop_reason == StopReason::Error
                ) {
                    context.messages.pop();
                }
                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 } => {
                // Persist assistant + tool results (prompts persisted earlier;
                // steering/follow-up user messages arrive here too).
                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 {
                    // User prompts were persisted in prompt(). Avoid double
                    // writes by checking the current leaf message identity.
                    let mut manager = self.manager.lock().unwrap();
                    let duplicate = matches!(
                        (manager.entries().last(), message),
                        (Some(crate::session::entry::SessionEntry::Message { message: last, .. }), m) if last == m
                    );
                    if !duplicate {
                        let _ = manager.append_message(message.clone());
                    }
                }
            }
            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 options = kiss_ai::StreamOptions {
                    credential: Some(credential),
                    max_tokens: Some(1),
                    reasoning,
                    fast_mode,
                    session_id: Some(session.manager.lock().unwrap().session_id().to_string()),
                    transport: settings.transport,
                    cancel: cancel.clone(),
                    ..Default::default()
                };
                let stream_fn = session
                    .stream_fn
                    .lock()
                    .unwrap()
                    .clone()
                    .unwrap_or_else(|| Arc::new(kiss_ai::stream_simple));
                let warmed = stream_fn(&model, &context, &options).result().await;
                if matches!(warmed.stop_reason, StopReason::Error | StopReason::Aborted) {
                    return;
                }
                session.totals.lock().unwrap().add(&warmed.usage);
                let _ = session.manager.lock().unwrap().append_usage(
                    "cache_warm",
                    &warmed.provider,
                    warmed.response_model.as_deref().unwrap_or(&warmed.model),
                    warmed.usage,
                    None,
                );
            }
        });
    }

    /// 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 override_settings = settings
            .compaction
            .model_overrides
            .get(&format!("{}/{}", model.provider, model.id));
        let keep_recent_tokens = override_settings
            .and_then(|value| value.keep_recent_tokens)
            .unwrap_or(settings.compaction.keep_recent_tokens);
        let reserve_tokens = override_settings
            .and_then(|value| value.reserve_tokens)
            .unwrap_or(settings.compaction.reserve_tokens);
        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(Some(api_key)) = kiss_ai::auth::resolve_api_key_async(
                decision_model.provider(),
                &self.registry.declared_keys,
            )
            .await
            && let Ok(client) = crate::decision::DecisionClient::new(decision_model, &api_key)
        {
            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();
        let remote_request = if 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 local_future = compaction::generate_summary(
            &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);
                {
                    let mut totals = self.totals.lock().unwrap();
                    if let Some(u) = &result.local_usage {
                        totals.add(u);
                    }
                    if let Some(u) = &result.remote_usage {
                        totals.add(u);
                    }
                }
                let details = merge_compaction_details(
                    file_ops_details(&read, &modified),
                    result.remote_details,
                );
                let mut manager = self.manager.lock().unwrap();
                let _ = manager.append_compaction(
                    result.summary.clone(),
                    plan.tokens_before,
                    plan.kept.clone(),
                    result.local_usage,
                    Some(details),
                );
                (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 messages = manager.context_messages();
            let tokens = if self.settings().experimental_context_file {
                messages
                    .map(|message| compaction::estimate_message_tokens(&message))
                    .sum()
            } else {
                compaction::estimate_context_tokens_iter(messages)
            };
            *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 {
    let reserve_tokens = settings
        .compaction
        .model_overrides
        .get(&format!("{}/{}", model.provider, model.id))
        .and_then(|value| value.reserve_tokens)
        .unwrap_or(settings.compaction.reserve_tokens);
    settings.compaction.enabled
        && !cancelled
        && model.context_window > 0
        && should_compact(
            if settings.experimental_context_file {
                // Usage in retained assistant messages describes the old
                // request, not the conversation after a model-owned edit.
                messages
                    .iter()
                    .map(compaction::estimate_message_tokens)
                    .sum()
            } else {
                estimate_context_tokens(messages)
            },
            model.context_window,
            reserve_tokens,
        )
}

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].iter().any(|status| {
        [
            format!("http {status}"),
            format!("status {status}"),
            format!("status: {status}"),
            format!("status code {status}"),
        ]
        .iter()
        .any(|marker| e.contains(marker))
    });
    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",
            "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());
        }
    }

    /// 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")
        ));
    }

    #[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(SessionManager::in_memory(Path::new("/other")))
            .unwrap_err();
        assert_eq!(session.manager.lock().unwrap().session_id(), original);
        drop(run);
        session
            .replace_manager(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(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(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("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("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
        );
    }
}