chat.rs

//! Bare `aster`: a conversational turn with an agentic read/list/search/edit tool loop.

use std::collections::{HashMap, HashSet};
use std::io::{IsTerminal, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use std::{env, fs, io};

use anyhow::{Context, Result, bail};
use aster_ai::{
    AiClient, Annotation, ChatMessage, DegenerateOutput, ReasoningDetail, UsageSnapshot,
};
use aster_persist::{
    EventUsage, EvictionEvent, MessageEvent, ReasoningRecord, Store, SummaryEvent, TranscriptEvent,
};
use aster_policy::{Action, Decision, Grants, Policy};
use clap::Args;
use serde::Deserialize;
use serde_json::{Value, json};
use tokio::sync::{mpsc, oneshot};
use tracing::Instrument;

use crate::config::provider::MissingCredentials;
use crate::edits::{self, EditBlock};
use crate::mcp::ToolOutput;
use crate::persist::Recorder;
use crate::util::usage_json;

#[derive(Default, Clone)]
pub(crate) struct SessionCtx {
    pub recorder: Option<Recorder>,
    pub store: Option<Store>,
    pub skills: Arc<aster_skills::SkillSet>,
    pub instructions: Arc<crate::instructions::Instructions>,
    pub probe: Arc<bash_tools::ToolProbe>,
    pub plan: std::sync::Arc<std::sync::Mutex<PlanState>>,
    pub mcp: Option<crate::mcp::McpRuntime>,
    pub limits: Limits,
    pub environment: Option<String>,
    /// Shared so an ACP session can flip it when the editor changes the mode
    /// mid-session; the TUI rebuilds the ctx each turn and does not need that.
    pub yolo: Arc<AtomicBool>,
    pub credentials: Arc<aster_policy::CommandGrants>,
    pub reads: Arc<Mutex<HashMap<String, Option<std::time::SystemTime>>>>,
    pub previews: Arc<Mutex<HashSet<String>>>,
    pub lookups: Arc<Mutex<HashSet<String>>>,
    pub injected: Arc<std::sync::Mutex<Vec<String>>>,
    pub agents: Arc<aster_agents::AgentRegistry>,
    pub sub_agent: Option<Arc<SubAgentOverrides>>,
    pub swarm: SwarmLimits,
}

/// How long a turn may work before it has to answer, how long one command may
/// run, and which language replies are written in. Defaults suit real builds;
/// `aster.yaml` and the env can change them.
#[derive(Debug, Clone)]
pub(crate) struct Limits {
    pub max_tool_rounds: usize,
    pub command_timeout_secs: usize,
    pub compact_budget_chars: usize,
    /// `None` follows the language the user writes in.
    pub language: Option<String>,
}

impl Default for Limits {
    fn default() -> Self {
        Self {
            max_tool_rounds: DEFAULT_MAX_TOOL_ROUNDS,
            command_timeout_secs: DEFAULT_COMMAND_TIMEOUT_SECS,
            compact_budget_chars: COMPACT_BUDGET_CHARS,
            language: None,
        }
    }
}

impl Limits {
    /// aster.yaml first, then the environment, which wins so one run can differ.
    pub(crate) fn resolve(agent: &crate::settings::Agent) -> Self {
        let env_usize = |key: &str| std::env::var(key).ok().and_then(|v| v.parse().ok());
        Self {
            max_tool_rounds: env_usize("ASTER_MAX_TOOL_ROUNDS")
                .or(agent.max_tool_rounds)
                .unwrap_or(DEFAULT_MAX_TOOL_ROUNDS)
                .max(1),
            command_timeout_secs: env_usize("ASTER_COMMAND_TIMEOUT")
                .or(agent.command_timeout_secs.map(|v| v as usize))
                .unwrap_or(DEFAULT_COMMAND_TIMEOUT_SECS)
                .max(1),
            compact_budget_chars: env_usize("ASTER_COMPACT_BUDGET")
                .or(agent.compact_budget_chars)
                .unwrap_or(COMPACT_BUDGET_CHARS)
                .max(COMPACT_KEEP_TAIL * 1_000),
            language: std::env::var("ASTER_LANGUAGE")
                .ok()
                .or_else(|| agent.language.clone())
                .map(|v| v.trim().to_string())
                .filter(|v| !v.is_empty()),
        }
    }
}

/// Where replies are written in. Sits at the end of the prompt, after the code
/// and tool text that pulls models off the user's language mid-turn.
pub(crate) fn language_note(language: Option<&str>) -> String {
    let target = match language {
        Some(lang) => format!("Reply in {lang}, whatever language the user writes in."),
        None => "Reply in the language the user writes in.".to_string(),
    };
    format!(
        "## Language\n{target} Keep the whole reply in that one language, \
         including any reasoning before it. Never drift into another language \
         mid-turn, even when the context is mostly code or tool output."
    )
}

/// Caps on the sub-agent fan-out.  aster.yaml first, then the environment.
#[derive(Debug, Clone)]
pub(crate) struct SwarmLimits {
    pub max_concurrent: usize,
    pub max_per_turn: usize,
    pub agent_timeout_secs: u64,
    pub collector_model: Option<String>,
}

impl SwarmLimits {
    pub(crate) fn resolve(agents: &crate::settings::Agents) -> Self {
        let env_usize = |key: &str| std::env::var(key).ok().and_then(|v| v.parse().ok());
        let env_u64 = |key: &str| std::env::var(key).ok().and_then(|v| v.parse().ok());
        Self {
            max_concurrent: env_usize("ASTER_AGENT_MAX_CONCURRENT")
                .or(agents.max_concurrent)
                .unwrap_or(8)
                .max(1),
            max_per_turn: env_usize("ASTER_AGENT_MAX_PER_TURN")
                .or(agents.max_per_turn)
                .unwrap_or(24)
                .max(1),
            agent_timeout_secs: env_u64("ASTER_AGENT_TIMEOUT")
                .or(agents.agent_timeout_secs)
                .unwrap_or(DEFAULT_AGENT_TIMEOUT_SECS)
                .max(1),
            collector_model: std::env::var("ASTER_COLLECTOR_MODEL")
                .ok()
                .or_else(|| agents.collector_model.clone()),
        }
    }
}

impl Default for SwarmLimits {
    fn default() -> Self {
        Self {
            max_concurrent: 8,
            max_per_turn: 24,
            agent_timeout_secs: DEFAULT_AGENT_TIMEOUT_SECS,
            collector_model: None,
        }
    }
}

#[derive(Debug, Clone)]
pub(crate) struct SubAgentOverrides {
    pub prompt_body: String,
    pub tool_allowlist: std::collections::HashSet<String>,
}

/// The plan document drafted with `write_plan` and presented by
/// `exit_plan_mode`, plus the progress steps tracked with `update_plan`.
#[derive(Debug, Default, Clone)]
pub(crate) struct PlanState {
    pub document: String,
    pub steps: Vec<PlanStep>,
    pub approved: bool,
}

#[derive(Debug, Clone, serde::Serialize)]
pub(crate) struct PlanStep {
    pub label: String,
    #[serde(rename = "status")]
    pub status: PlanStepStatus,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum PlanStepStatus {
    Pending,
    #[serde(rename = "in_progress")]
    InProgress,
    Done,
    Skipped,
    Blocked,
}

fn plan_snapshot(ctx: &SessionCtx) -> Option<Vec<(String, PlanStepStatus)>> {
    let plan = ctx.plan.lock().ok()?;
    (!plan.steps.is_empty()).then(|| {
        plan.steps
            .iter()
            .map(|s| (s.label.clone(), s.status))
            .collect()
    })
}

fn plan_unfinished(snapshot: &Option<Vec<(String, PlanStepStatus)>>) -> bool {
    snapshot.as_ref().is_some_and(|steps| {
        steps.iter().any(|(_, status)| {
            matches!(status, PlanStepStatus::Pending | PlanStepStatus::InProgress)
        })
    })
}

impl SessionCtx {
    pub(crate) fn record(&self, event: MessageEvent) {
        let Some(recorder) = &self.recorder else {
            return;
        };
        match recorder.lock() {
            Ok(mut writer) => {
                if let Err(e) = writer.append_message(event) {
                    tracing::warn!("failed to record transcript event: {e:#}");
                }
            }
            Err(e) => tracing::warn!("transcript writer lock poisoned: {e}"),
        }
    }

    /// The live session's transcript id, when one is being recorded. Memory
    /// writes use it as provenance so a fact can be traced back to the session
    /// that produced it.
    pub(crate) fn session_id(&self) -> Option<String> {
        let recorder = self.recorder.as_ref()?;
        recorder.lock().ok().map(|writer| writer.id().to_string())
    }

    pub(crate) fn record_summary(&self, content: &str, replaces_through: usize) {
        let Some(recorder) = &self.recorder else {
            return;
        };
        if let Ok(mut writer) = recorder.lock()
            && let Err(e) = writer.append(&TranscriptEvent::Summary(SummaryEvent::new(
                content,
                replaces_through,
            )))
        {
            tracing::warn!("failed to record summary event: {e:#}");
        }
    }

    fn record_eviction(&self, eviction: &crate::budget::Eviction) {
        let Some(recorder) = &self.recorder else {
            return;
        };
        if let Ok(mut writer) = recorder.lock()
            && let Err(e) = writer.append(&TranscriptEvent::Eviction(EvictionEvent::new(
                eviction.reason,
                eviction.role,
                eviction.index,
                eviction.chars,
            )))
        {
            tracing::warn!("failed to record eviction event: {e:#}");
        }
    }

    fn is_titled(&self) -> bool {
        self.recorder
            .as_ref()
            .and_then(|r| r.lock().ok())
            .is_some_and(|w| w.title().is_some())
    }

    fn current_title(&self) -> Option<String> {
        self.recorder
            .as_ref()
            .and_then(|r| r.lock().ok())
            .and_then(|w| w.title().map(str::to_string))
    }

    fn record_title(&self, title: &str) {
        let Some(recorder) = &self.recorder else {
            return;
        };
        if let Ok(mut writer) = recorder.lock()
            && let Err(e) = writer.set_title(title)
        {
            tracing::warn!("failed to record title event: {e:#}");
        }
    }

    fn memory_context(&self) -> Option<String> {
        let store = self.store.as_ref()?;
        match store.memory().load_context() {
            Ok(ctx) if !ctx.trim().is_empty() => Some(ctx),
            Ok(_) => None,
            Err(e) => {
                tracing::warn!("failed to load memory context: {e:#}");
                None
            }
        }
    }
}

/// Skills from `.aster/skills`, then `<config>/aster/skills`, then plugins, then
/// built-ins: a skills root shadows a plugin and a plugin shadows a built-in.
pub(crate) fn discover_skills(repo_root: &Path) -> Arc<aster_skills::SkillSet> {
    let roots = skills_roots(repo_root);
    if let Some(global) = roots.get(1) {
        aster_skills::install_defaults(global);
    }
    let (plugins, problems) = crate::plugins::installed(Some(repo_root));
    crate::plugins::report(&plugins, &problems);
    Arc::new(
        aster_skills::SkillSet::discover(&roots)
            .extend_dirs(&crate::plugins::skill_dirs(&plugins))
            .with_builtins(),
    )
}

/// The project root first, then the global one when a home exists.
pub(crate) fn skills_roots(repo_root: &Path) -> Vec<PathBuf> {
    let mut roots = vec![repo_root.join(".aster").join("skills")];
    match crate::persist::home() {
        Ok(home) => roots.push(home.join("skills")),
        Err(e) => tracing::debug!("no global skills root: {e:#}"),
    }
    roots
}

/// The newest change under the skills roots, so a long-lived session can tell
/// when a skill was written or rewritten and read the index again.
pub(crate) fn skills_stamp(repo_root: &Path) -> u64 {
    let nanos = |path: &Path| {
        fs::metadata(path)
            .and_then(|m| m.modified())
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(0)
    };
    let mut stamp = 0;
    for root in skills_roots(repo_root) {
        stamp = stamp.max(nanos(&root));
        let Ok(entries) = fs::read_dir(&root) else {
            continue;
        };
        for entry in entries.flatten() {
            stamp = stamp.max(nanos(&entry.path().join("SKILL.md")));
        }
    }
    stamp
}

/// A message opening with `/skill-name` says which skill to apply. The model is
/// told about skills by name, so the ask is spelled out rather than sent as a
/// slash it has to guess at. Anything else is left exactly as typed.
pub(crate) fn expand_skill(text: &str, skills: &aster_skills::SkillSet) -> String {
    let Some(rest) = text.strip_prefix('/') else {
        return text.to_string();
    };
    let (name, task) = match rest.split_once(char::is_whitespace) {
        Some((name, task)) => (name, task.trim_start()),
        None => (rest, ""),
    };
    match skills.get(name) {
        Some(skill) => format!("Use the \"{}\" skill: {task}", skill.name)
            .trim_end()
            .to_string(),
        None => text.to_string(),
    }
}

/// Session-start snapshot: the repository, platform, date, git state, and which
/// package manager each lockfile pins. Taken once, so the model starts a turn
/// knowing what a round of discovery commands would have told it.
pub(crate) fn environment_note(repo_root: &Path) -> Option<String> {
    let mut note = crate::project::snapshot(repo_root)
        .map(|project| format!("{project}\n"))
        .unwrap_or_default();
    note.push_str(&format!(
        "## Environment\n- Platform: {} ({})\n- Today's date: {}\n",
        std::env::consts::OS,
        std::env::consts::ARCH,
        chrono::Local::now().format("%Y-%m-%d")
    ));
    if let Some(device) = android_note() {
        note.push_str(&device);
    }
    if let Some(git) = git_snapshot(repo_root) {
        note.push_str(&git);
    }
    if let Some(pm) = package_manager_note(repo_root) {
        note.push_str(&pm);
    }
    if let Some(runners) = task_runner_note(repo_root) {
        note.push_str(&runners);
    }
    Some(note)
}

/// On a phone there is no repository to look at; the device is the subject. Say
/// which device it is, and whether the tool that reads the screen is reachable,
/// so the model does not have to discover either.
#[cfg(target_os = "android")]
fn android_note() -> Option<String> {
    let prop = |key: &str| -> Option<String> {
        let out = std::process::Command::new("getprop")
            .arg(key)
            .output()
            .ok()?;
        let value = String::from_utf8_lossy(&out.stdout).trim().to_string();
        (!value.is_empty()).then_some(value)
    };
    let mut note = String::from(
        "## Device\n- Running on the Android device itself, not on a machine attached to one\n",
    );
    if let Some(model) = prop("ro.product.model") {
        note.push_str(&format!("- Model: {model}\n"));
    }
    if let (Some(release), Some(sdk)) = (
        prop("ro.build.version.release"),
        prop("ro.build.version.sdk"),
    ) {
        note.push_str(&format!("- Android {release} (API {sdk})\n"));
    }
    note.push_str(match on_path("asterctl") {
        true => "- `asterctl` reads the screen and taps it: `map`, `find`, `tap`, `scroll`, `type`, `key`, `volume`, `media`, `restart`, `ocr`, `notes`. Its full reference is the android-use skill in this prompt; `asterctl help` lists the verbs\n",
        false => "- No `asterctl` on PATH, so the screen cannot be seen or touched from here\n",
    });
    note.push_str(
        "- `aster python script.py` or `aster python -c \"...\"` runs Python 3 with the standard library built in; there is no other python, no bash (the shell is `sh`), and no `/tmp` (use `$TMPDIR`)\n",
    );
    Some(note)
}

fn on_path(name: &str) -> bool {
    std::env::var_os("PATH")
        .map(|paths| std::env::split_paths(&paths).any(|dir| dir.join(name).exists()))
        .unwrap_or(false)
}

/// Android ships `sh` and no bash, so a shell line has to go through whichever
/// one is there.
fn shell() -> (&'static str, &'static str) {
    match on_path("bash") {
        true => ("bash", "-lc"),
        false => ("sh", "-c"),
    }
}

#[cfg(not(target_os = "android"))]
fn android_note() -> Option<String> {
    None
}

const MAX_SCRIPT_NAMES: usize = 12;

fn task_runner_note(repo_root: &Path) -> Option<String> {
    let mut note = String::new();
    // One candidate list per runner: a case-insensitive filesystem would
    // otherwise report Justfile and justfile as two files.
    let runners: [(&[&str], &str); 3] = [
        (
            &["Justfile", "justfile"],
            "run recipes with `just <name>`; `just --list` shows them",
        ),
        (&["Makefile"], "run targets with `make <name>`"),
        (
            &["Taskfile.yml"],
            "run tasks with `task <name>`; `task --list` shows them",
        ),
    ];
    for (candidates, hint) in runners {
        if let Some(file) = candidates.iter().find(|f| repo_root.join(f).is_file()) {
            note.push_str(&format!("- {file} present: {hint}.\n"));
        }
    }
    if let Some(scripts) = package_scripts(&repo_root.join("package.json")) {
        note.push_str(&format!(
            "- package.json scripts: {}. Prefer these over hand-rolled equivalents.\n",
            scripts.join(", ")
        ));
    }
    (!note.is_empty()).then_some(note)
}

fn package_scripts(manifest: &Path) -> Option<Vec<String>> {
    let raw = fs::read_to_string(manifest).ok()?;
    let json: Value = serde_json::from_str(&raw).ok()?;
    let scripts = json.get("scripts")?.as_object()?;
    if scripts.is_empty() {
        return None;
    }
    let mut names: Vec<String> = scripts.keys().cloned().collect();
    names.sort();
    names.truncate(MAX_SCRIPT_NAMES);
    if scripts.len() > MAX_SCRIPT_NAMES {
        names.push(format!("... {} more", scripts.len() - MAX_SCRIPT_NAMES));
    }
    Some(names)
}

const GIT_STATUS_LINES: usize = 15;

fn git_snapshot(repo_root: &Path) -> Option<String> {
    let git = |args: &[&str]| -> Option<String> {
        let out = std::process::Command::new("git")
            .arg("-C")
            .arg(repo_root)
            .args(args)
            .output()
            .ok()?;
        out.status
            .success()
            .then(|| String::from_utf8_lossy(&out.stdout).trim().to_string())
    };
    let branch = git(&["rev-parse", "--abbrev-ref", "HEAD"])?;
    let mut note = format!("- Git branch: {branch}");
    if let Some(default) = git(&["symbolic-ref", "--short", "refs/remotes/origin/HEAD"])
        .as_deref()
        .and_then(|head| head.rsplit('/').next())
    {
        note.push_str(&format!(" (default branch: {default})"));
    }
    note.push('\n');
    match git(&["status", "--porcelain"]).as_deref() {
        Some("") => note.push_str("- Working tree clean at session start\n"),
        Some(status) => {
            let lines: Vec<&str> = status.lines().collect();
            note.push_str(&format!(
                "- Changed files at session start ({}):\n",
                lines.len()
            ));
            for line in lines.iter().take(GIT_STATUS_LINES) {
                note.push_str(&format!("  {line}\n"));
            }
            if lines.len() > GIT_STATUS_LINES {
                note.push_str(&format!(
                    "  ... and {} more\n",
                    lines.len() - GIT_STATUS_LINES
                ));
            }
        }
        None => {}
    }
    if let Some(log) = git(&["log", "--oneline", "-5"]).filter(|log| !log.is_empty()) {
        note.push_str("- Recent commits:\n");
        for line in log.lines() {
            note.push_str(&format!("  {line}\n"));
        }
    }
    Some(note)
}

fn package_manager_note(repo_root: &Path) -> Option<String> {
    const LOCKS: &[(&str, &str)] = &[
        ("bun.lock", "bun"),
        ("bun.lockb", "bun"),
        ("pnpm-lock.yaml", "pnpm"),
        ("yarn.lock", "yarn"),
        ("package-lock.json", "npm"),
    ];
    let mut found: std::collections::BTreeMap<&str, Vec<String>> = Default::default();
    let walk = ignore::WalkBuilder::new(repo_root)
        .max_depth(Some(3))
        .build();
    for entry in walk.flatten() {
        let Some(name) = entry.file_name().to_str() else {
            continue;
        };
        let Some((_, pm)) = LOCKS.iter().find(|(lock, _)| *lock == name) else {
            continue;
        };
        let dir = entry
            .path()
            .parent()
            .and_then(|p| p.strip_prefix(repo_root).ok())
            .map(|p| p.display().to_string())
            .filter(|p| !p.is_empty())
            .unwrap_or_else(|| ".".to_string());
        let dirs = found.entry(pm).or_default();
        if !dirs.contains(&dir) {
            dirs.push(dir);
        }
    }
    if found.is_empty() {
        return None;
    }
    let mut note = String::new();
    for (pm, dirs) in &found {
        note.push_str(&format!(
            "- JavaScript packages in {} use `{pm}`; run scripts and one-off tools with it, not npm/npx.\n",
            dirs.join(", ")
        ));
    }
    note.push_str("- Run package commands from the directory that owns the lockfile.");
    Some(note)
}

fn system_prompt(ctx: &SessionCtx, tools: bool) -> String {
    // Sub-agents get only their prompt body and an environment note; the
    // persona, instructions, memory, and agent index are skipped. The skill
    // index rides along only for a bot, whose skills are scoped to itself.
    if let Some(sub) = &ctx.sub_agent {
        let mut prompt = sub.prompt_body.clone();
        if let Some(index) = ctx.skills.render_index() {
            prompt.push_str("\n\n");
            prompt.push_str(&index);
        }
        if let Some(environment) = &ctx.environment {
            prompt.push_str("\n\n");
            prompt.push_str(environment);
        }
        prompt.push_str("\n\n");
        prompt.push_str(&language_note(ctx.limits.language.as_deref()));
        return prompt;
    }
    let mut prompt = base_system_prompt();
    // Ahead of tools and memory: these are the repo's standing rules, and they
    // shape how every other section gets used.
    if let Some(project) = ctx.instructions.render() {
        prompt.push_str("\n\n");
        prompt.push_str(&project);
    }
    if let Some(environment) = &ctx.environment {
        prompt.push_str("\n\n");
        prompt.push_str(environment);
    }
    prompt.push_str("\n\n");
    prompt.push_str(&language_note(ctx.limits.language.as_deref()));
    if tools {
        prompt.push_str(TOOLS_PROMPT);
        if let Some(index) = ctx.skills.render_index() {
            prompt.push_str("\n\n");
            prompt.push_str(&index);
        }
        if let Some(index) = ctx.agents.render_index() {
            prompt.push_str("\n\n");
            prompt.push_str(&index);
        }
    }
    if tools && let Some(injection) = ctx.mcp.as_ref().and_then(|m| m.injection()) {
        prompt.push_str("\n\n");
        prompt.push_str(&injection.prompt);
    }
    if tools && let Some(disabled) = ctx.mcp.as_ref().and_then(|m| m.disabled_servers_prompt()) {
        prompt.push_str("\n\n");
        prompt.push_str(&disabled);
    }
    if let Some(memory) = ctx.memory_context() {
        prompt.push_str("\n\n");
        prompt.push_str(&memory);
    }
    prompt
}

/// CLI spelling of [`aster_policy::Mode`].
#[derive(Clone, Copy, Debug, clap::ValueEnum)]
pub(crate) enum PermissionModeArg {
    /// Explore the code and present a plan before editing.
    Plan,
    /// Ask for approval before each edit and command.
    Manual,
    /// Apply edits and run commands, pausing on the risky ones.
    Auto,
    /// As auto, but commands are trusted; only a rule stops one.
    Edit,
    /// Skip the rules and isolation entirely. Use with extreme caution.
    Yolo,
}

impl From<PermissionModeArg> for aster_policy::Mode {
    fn from(arg: PermissionModeArg) -> Self {
        match arg {
            PermissionModeArg::Plan => Self::Plan,
            PermissionModeArg::Manual => Self::Manual,
            PermissionModeArg::Auto => Self::Auto,
            PermissionModeArg::Edit => Self::Edit,
            PermissionModeArg::Yolo => Self::Yolo,
        }
    }
}

/// Emits one NDJSON event per line on the `--stream` path. Shared, so
/// background work can keep emitting after the tool call that started it.
pub(crate) type ChatEventSink = Arc<dyn Fn(Value) + Send + Sync>;

/// A request the agent task sends to the UI loop: an edit needing approval, a
/// plan whose approval promotes the session to edit mode, or a question.
pub(crate) enum UiRequest {
    Approval(ApprovalRequest),
    PlanApproval(ApprovalRequest),
    Question(QuestionRequest),
}

/// A pending edit the agent wants to make; the UI renders a diff and asks the
/// user to confirm. `scope` is the directory an "always allow" answer covers;
/// `None` means the front-end offers only yes or no.
pub(crate) struct ApprovalRequest {
    pub preview: String,
    pub markdown: Option<String>,
    pub scope: Option<PathBuf>,
    pub respond: oneshot::Sender<Answer>,
}

/// How the user answered an [`ApprovalRequest`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Answer {
    Yes,
    No,
    Always,
}

impl Answer {
    pub(crate) fn allowed(self) -> bool {
        !matches!(self, Answer::No)
    }
}

pub(crate) struct QuestionRequest {
    pub header: String,
    pub question: String,
    pub options: Vec<String>,
    pub respond: oneshot::Sender<Option<String>>,
}

/// Channel for UI requests — approval prompts and agent questions. Headless
/// callers pass `None`, declining every prompt.
pub(crate) type UiSender = mpsc::Sender<UiRequest>;

const AGENT_SYSTEM_PROMPT: &str = include_str!("../prompts/aster-agent.md");

/// The manual opens by placing the agent in a repository on a desktop. On a
/// phone that is the wrong room, and a correction further down loses to the
/// opening sentence, so this is shouted in front of it instead.
#[cfg(target_os = "android")]
const ANDROID_IDENTITY: &str = "# READ THIS BEFORE THE MANUAL BELOW\n\n\
    THE MANUAL BELOW DESCRIBES A DEVELOPER'S REPOSITORY ON A DESKTOP. THAT IS \
    NOT WHERE YOU ARE. You are a process inside an app on an Android phone. \
    There is no repository, no codebase, no terminal to answer at. The phone is \
    the subject. WHERE THE MANUAL AND THIS SECTION DISAGREE, THIS SECTION \
    WINS.\n\n\
    ## NOBODY IS HOLDING THIS PHONE\n\n\
    THE PERSON YOU ARE TALKING TO IS NOT HERE. They are somewhere else. They \
    cannot see this screen. They cannot reach this device. THEY CANNOT FINISH \
    ANYTHING YOU START. You are the only thing in the room with the phone.\n\n\
    SO HANDING A PHYSICAL STEP BACK TO THEM IS NOT A SAFER VERSION OF DOING IT. \
    IT IS THE SAME AS NOT DOING IT, and it burns the time they spent asking. \
    \"Tap it yourself.\" \"Open the app and check.\" \"Press the call button \
    now.\" NONE OF THESE ARE AVAILABLE TO SOMEONE WHO IS NOT IN THE ROOM. \
    Saying them looks like help and leaves the person with nothing. IF A STEP \
    IS YOURS TO TAKE, TAKE IT.\n\n\
    ## A REFUSAL IS NOT A NEUTRAL ACT\n\n\
    You are not deciding whether to help. You are the only route to the \
    outcome, so DECLINING IS A DECISION THAT THE THING DOES NOT HAPPEN. Weigh \
    it like one. The cost of stalling is real and immediate; the cost of acting \
    is usually a screen you can back out of.\n\n\
    THIS MATTERS MOST WHEN IT MATTERS MOST. Emergency services, a call for \
    help, anything time-critical: PLACE IT. DO NOT DELIBERATE. DO NOT ASK THEM \
    TO CONFIRM. DO NOT EXPLAIN WHY YOU WOULD RATHER NOT. Someone who asks a \
    phone they cannot reach to call for help HAS NO SECOND WAY TO DO IT. \
    Hesitating there is not caution, it is the worst available outcome.\n\n\
    ## WHEN YOU TRULY CANNOT\n\n\
    Some things the platform forbids outright. Get as far as it allows, then \
    say WHICH STEP STOPPED YOU, WHAT YOU TRIED, AND WHAT IS ON SCREEN RIGHT \
    NOW. That is a report they can act on. An apology is not. A suggestion they \
    cannot physically follow is not.\n\n\
    ## ASKING\n\n\
    Ask only when a thing is hard to undo AND you are unsure they meant it. \
    NEVER ASK TO BE SEEN ASKING. Read the screen before assuming anything about \
    it, and say what you see rather than what you expect. Say what you did \
    AFTER you did it, not instead of doing it.\n\n\
    ---\n\n";

fn base_system_prompt() -> String {
    let mut prompt = String::new();
    // In front of the manual, not after it: an opening sentence that puts the
    // agent in a repository is not undone by a correction further down.
    #[cfg(target_os = "android")]
    prompt.push_str(ANDROID_IDENTITY);
    prompt.push_str(AGENT_SYSTEM_PROMPT);
    prompt
}

const CHAT_TEMPERATURE: f64 = 0.4;
/// A phone runs tasks nobody is watching, over a chat channel, so a turn and
/// its commands get room to finish instead of being cut off mid-task.
#[cfg(target_os = "android")]
const DEFAULT_MAX_TOOL_ROUNDS: usize = 200;
#[cfg(not(target_os = "android"))]
const DEFAULT_MAX_TOOL_ROUNDS: usize = 60;
const MAX_TOOL_RESULT_CHARS: usize = 24_000;
const READ_WINDOW_LINES: usize = 600;
const MAX_STREAM_CHARS: usize = 10_000;
const MAX_SEARCH_HITS: usize = 80;
const SEARCH_CONTEXT_LINES: usize = 3;
const MAX_LIST_ENTRIES: usize = 200;
const MAX_FIND_HITS: usize = 100;
const MAX_PATH_SUGGESTIONS: usize = 8;
const MISSING_COMMAND: &str = "run_command needs a `command`: the binary to \
    run, with its arguments in `args`. To run a shell line, pass \
    command:`bash` with args [\"-lc\", \"<the line>\"]. Send the call again \
    with `command` set";
#[cfg(target_os = "android")]
const DEFAULT_COMMAND_TIMEOUT_SECS: usize = 1800;
#[cfg(not(target_os = "android"))]
const DEFAULT_COMMAND_TIMEOUT_SECS: usize = 300;
#[cfg(target_os = "android")]
const DEFAULT_AGENT_TIMEOUT_SECS: u64 = 1800;
#[cfg(not(target_os = "android"))]
const DEFAULT_AGENT_TIMEOUT_SECS: u64 = 300;
const COMPACT_BUDGET_CHARS: usize = 192_000;
const COMPACT_KEEP_TAIL: usize = 6;

const TOOLS_PROMPT: &str = "\n\n## Tools\n\n\
You can inspect the repository with `read_file`, `list_files`, `find_files`, \
and `search_files`, and change it with `edit_file` when it is available. \
`search_files` searches file contents and supports regex syntax. Gitignored \
files are part of the repo: `read_file` opens them like any other, and \
`search_files`, `find_files`, and `list_files` reach them when the tracked \
files come up empty. Never claim a file is unreadable because it is \
gitignored. `find_files` locates files by name or glob; reach for it before \
guessing a path, and whenever a tool reports that a path does not exist. \
A path that does not exist is a wrong guess, not a failure: take the nearby \
paths the tool offers and try again. \
`edit_file` also creates files: omit `search` and pass the whole contents as \
`replace`. \
`ast_grep` searches by syntax pattern (e.g. `fn $NAME($$$ARGS)`) when text \
search is too noisy, and `ast_edit` applies one structural rewrite across \
every match at once instead of many edit_file calls. An edit reports the \
language server's problems for that file when it can; `lsp_diagnostics` asks \
for them directly, and checks one file far faster than a build. \
`lsp_references` and `lsp_definitions` follow a symbol semantically instead \
of by name. \
`run_command` runs a CLI tool or build command. There is no shell: arguments \
pass verbatim, so `$VAR` is never expanded; wrap the command in \
`bash -lc \"...\"` when it needs variables, pipes, or redirects. Filesystem \
writes are restricted to the repo and temp directories, and secrets are \
dropped from the environment, unless the session is in yolo mode: then \
there is no sandbox and the full environment, including secrets, is \
inherited. Use it for builds, tests, and linters. It can also reach \
the network: prefer `curl` (or a similar CLI) for fetching URLs and calling \
APIs before suggesting a browser-based tool. Do not shell out \
to `rg`, `grep`, `find`, or `fd`: `search_files` and `find_files` already \
run them directly, without the overhead. \
Tool rounds are the slow part of a turn: each one costs a full model \
round-trip, while the tools themselves are nearly instant. Work in as few \
rounds as the task allows:\n\
- Look things up with `explore`, not one call at a time. If you are about to \
send a single `read_file`, `search_files`, `find_files`, or `list_files`, \
first ask what else you will want once you see it, and send them together as \
`explore` steps. Two lookups in one `explore` are twice as fast as two \
rounds; ten are ten times.\n\
- Batch independent calls into one response: several reads, or a search and a \
find together, instead of one call per response.\n\
- Search before you read. `search_files` returns the matching lines with \
context, which usually answers the question without reading the file at all.\n\
- Never re-read what is already in this conversation. A file you read earlier \
is still above you; scroll back instead of calling the tool again.\n\
- When you do need more of a file, ask for the specific range you are missing \
rather than the whole file again.\n\
- Get everything one command can give you in a single call. `run_command` \
runs one binary directly, with no shell, so chain with \
`bash -lc \"git status --short; git log --oneline -5; git diff --stat\"` \
rather than spending a round on each. Bound noisy output with flags like \
`--stat`, `-n 20`, or a `| head` inside that `bash -lc` string.\n\
- Stop gathering as soon as you can act or answer: do not re-verify what you \
already read, and do not explore beyond what the task needs.\n\
When a user message contains `[@name]` tokens, each token's full path is \
listed beneath the message as `[@name]: /full/path`. Resolve the token from \
that list rather than guessing a path. \
Set `turbo: true` when the user asks to work offline or in turbo mode \
(blocks network access). Set `yolo: true` only when the user explicitly \
asks for yolo mode (no restrictions). \
Ground every claim about the code in what you actually read. Only edit files when the \
user asked for a change; keep edits minimal and in the file's existing style. \
After editing, state plainly which files you changed and what the change does. \
If `edit_file` is unavailable, say so and describe the change instead.";

#[derive(Args)]
pub struct ChatArgs {
    /// One-shot question, e.g. `aster "why is finding 2 critical?"`.
    #[arg(value_name = "PROMPT", conflicts_with = "messages_json")]
    prompt: Option<String>,
    /// Continue this repo's most recent session, seeding its prior history.
    /// Without it every session starts clean, in the TUI too.
    #[arg(long = "continue", conflicts_with = "messages_json")]
    continue_session: bool,

    /// Pick a session to resume from a list of this repo's saved sessions.
    /// Needs a terminal; with an ID it resumes that session directly.
    #[arg(long, value_name = "ID", num_args = 0..=1, conflicts_with_all = ["messages_json", "session"])]
    resume: Option<Option<String>>,

    /// Persist this turn into a session by id, resuming it if it exists and
    /// creating it if not. Alone it also seeds the session's prior history;
    /// with --messages-json (the caller owns history) it only records.
    #[arg(long, value_name = "ID")]
    session: Option<String>,

    /// Read a JSON array of {"role","content"} messages from PATH, or `-` for
    /// stdin. With --stream this must be a single line, since stdin stays open
    /// for approval replies.
    #[arg(long, value_name = "PATH")]
    messages_json: Option<String>,

    /// Model override (else ASTER_MODEL, aster.yaml, default).
    #[arg(long, value_name = "MODEL")]
    model: Option<String>,

    /// Let the agent edit repo files via its edit_file tool.
    #[arg(long)]
    allow_edits: bool,

    /// How edits and commands are gated, overriding aster.yaml
    /// `permissions.mode`: plan, manual, auto, edit, or yolo. Prompts need a
    /// front-end that can answer: the TUI, or `--stream`.
    #[arg(long, value_name = "MODE", value_enum)]
    permission_mode: Option<PermissionModeArg>,

    /// Stream the turn as NDJSON on stdout, one event per line, and read
    /// approval replies from stdin. For editors and UIs.
    #[arg(long, conflicts_with_all = ["tui", "json", "print"])]
    stream: bool,

    /// Plain single-shot chat: no read/search/edit tools.
    #[arg(long)]
    no_tools: bool,

    /// Skip connecting MCP servers at startup so the chat opens instantly.
    /// `/mcp` can still bring them up later in the session.
    #[arg(long)]
    no_mcp: bool,

    /// Fold the history from --messages-json into a summary and print the
    /// shorter history instead of answering. For front-ends that own their
    /// own transcript; the TUI does this from `/compact`.
    #[arg(long, requires = "messages_json")]
    compact: bool,

    /// Open the interactive chat TUI (default in a terminal). Optional PROMPT seeds the first question.
    #[arg(long, conflicts_with_all = ["messages_json", "json", "no_tools", "print"])]
    tui: bool,

    /// Answer once and print plain text instead of opening the TUI (default when piped).
    #[arg(long, short = 'p', conflicts_with_all = ["messages_json", "json"])]
    print: bool,

    #[command(flatten)]
    pub effort: crate::EffortArgs,
}

/// Args equivalent to `aster --resume <id>`, for commands that hand off into
/// a resumed chat.
pub fn resume_args(id: &str) -> ChatArgs {
    #[derive(clap::Parser)]
    struct Wrap {
        #[command(flatten)]
        chat: ChatArgs,
    }
    <Wrap as clap::Parser>::parse_from(["aster", "--resume", id]).chat
}

impl ChatArgs {
    /// True when both ends are a real terminal and no flag forced one-shot output.
    pub fn is_interactive(&self) -> bool {
        let one_shot = self.print
            || crate::json_mode()
            || self.no_tools
            || self.stream
            || self.messages_json.is_some();
        !one_shot && io::stdout().is_terminal() && io::stdin().is_terminal()
    }

    fn resume_mode(&self) -> Resume {
        match (&self.resume, &self.session) {
            (Some(Some(id)), _) | (_, Some(id)) => Resume::Id(id.clone()),
            (Some(None), _) => Resume::Pick,
            _ if self.continue_session => Resume::Latest,
            _ => Resume::New,
        }
    }
}

pub(crate) enum Resume {
    New,
    Latest,
    Id(String),
    Pick,
}

#[derive(Deserialize)]
struct WireMessage {
    role: String,
    content: String,
}

fn ask_needs_front_end(mode: aster_policy::Mode, allow_edits: bool, can_prompt: bool) -> bool {
    allow_edits && mode == aster_policy::Mode::Manual && !can_prompt
}

/// Best-effort consolidation for a finished session. Headless callers await
/// it because their process exits at turn end; the TUI spawns it detached. A
/// failed pass is a warning, and the startup sweep retries it.
pub(crate) async fn consolidate_finished_session(
    client: &AiClient,
    store: Option<&Store>,
    repo_root: &Path,
    session_id: &str,
) {
    let Some(store) = store else {
        return;
    };
    let Ok(transcript) = store.resume(repo_root, session_id) else {
        return;
    };
    let memory = store.memory();
    let min_turns = aster_memory::consolidate::DEFAULT_MIN_TURNS;
    if let Err(err) =
        aster_memory::consolidate::consolidate_session(client, &memory, &transcript, min_turns)
            .await
    {
        tracing::warn!("memory consolidation failed: {err:#}");
    }
}

pub async fn run(args: ChatArgs) -> Result<()> {
    let repo_root = env::current_dir().context("could not determine the current directory")?;
    let mut settings = crate::settings::Settings::load(Some(&repo_root))?;
    crate::jev::init(&settings.experimental);
    let mut client = match crate::config::provider::resolve_client(&settings, args.model.as_deref())
    {
        Ok(client) => client,
        Err(err) => {
            // A front-end reading the stream gets the setup it needs as an
            // event; a bare exit would leave it with only an exit code.
            if args.stream
                && let Some(missing) = err.downcast_ref::<MissingCredentials>()
            {
                emit_line(&json!({
                    "type": "error",
                    "message": missing.to_string(),
                    "setup": missing.0,
                }));
                return Ok(());
            }
            // Nothing is set up and there is a person here: onboarding is the
            // answer, not an error telling them to run a command they have not
            // heard of yet.
            if err.is::<MissingCredentials>()
                && console::Term::stdout().features().is_attended()
                && !crate::json_mode()
                && crate::init::first_run().await?
            {
                settings = crate::settings::Settings::load(Some(&repo_root))?;
                crate::config::provider::resolve_client(&settings, args.model.as_deref())?
            } else {
                return Err(err);
            }
        }
    };
    let settings = settings;

    if args.model.is_none()
        && let Some(mut mom) = crate::mom::MomSession::load(&repo_root)
        && let Some(selection) = mom.evaluate_turn(1, &aster_mom::Signals::default())
        && let Some(record) = &selection.record
        && let Some((base_url, key)) = mom.endpoint_for(&selection.model)
    {
        client.model = mom.model_param(&base_url, &selection.model);
        client.set_endpoint(&base_url, key);
        eprintln!(
            "mom: using {} ({}) · {}",
            selection.entry, client.model, record.reason
        );
        crate::mom::log_switch(record);
    }
    let client = client;

    if args.compact {
        return run_compact(&args, &client).await;
    }

    // The flag is the user asking for this run outright, so it replaces the
    // configured mode rather than only tightening it.
    let mut permissions = settings.permissions.clone();
    if let Some(mode) = args.permission_mode {
        permissions.mode = mode.into();
    }

    // The TUI answers its own prompts, so it is editable unless the config or
    // --permission-mode says otherwise; --allow-edits only gates headless runs.
    let interactive = args.is_interactive();
    let allow_edits = match args.permission_mode {
        Some(_) => true,
        None => interactive || args.allow_edits,
    } && permissions.mode.can_edit();

    let can_prompt = args.is_interactive() || args.stream;
    let allow_edits =
        if !args.no_tools && ask_needs_front_end(permissions.mode, allow_edits, can_prompt) {
            eprintln!(
                "note: `manual` permissions confirm every edit and this run cannot ask, \
             so the agent is read-only. Pass --permission-mode edit (or auto), or run \
             with --stream so approvals have somewhere to go."
            );
            false
        } else {
            allow_edits
        };

    // `yolo` means no policy *and* no sandbox; the TUI already treats it that
    // way, and a headless run must match or commands fail on writes.
    let yolo = permissions.mode == aster_policy::Mode::Yolo;
    let policy = Arc::new(Policy::compile(&permissions)?);
    let grants = Arc::new(configured_grants(&permissions, &repo_root));
    let credentials = Arc::new(configured_credentials(&permissions, &repo_root));

    let limits = Limits::resolve(&settings.agent);
    let swarm = SwarmLimits::resolve(&settings.agents);
    let agents = crate::agents::discover_agents(&repo_root);

    if args.is_interactive() {
        // Every server costs a process spawn, and `npx` ones a registry round
        // trip, so waiting here would leave the terminal blank for seconds.
        // The TUI draws first, and a cached catalog lets the first prompt go
        // out before any server has connected; the live connect replaces it
        // when it lands. `--no-mcp` resolves the same handle immediately.
        let mcp_settings = settings.mcp.clone();
        let cached = if args.no_mcp {
            None
        } else {
            crate::mcp::McpRuntime::from_cache(&settings.mcp, &repo_root)
        };
        let mcp = if args.no_mcp {
            tokio::spawn(async { (None, Vec::new()) })
        } else {
            let root = repo_root.clone();
            tokio::spawn(
                async move { crate::mcp::McpRuntime::connect_at(&mcp_settings, &root).await },
            )
        };
        let seed = args.prompt.clone();
        return crate::tui::run_chat(
            client,
            repo_root,
            allow_edits,
            permissions,
            seed,
            args.resume_mode(),
            mcp,
            cached,
            limits,
            swarm,
            agents,
        )
        .await;
    }

    let (mcp, mcp_problems) = if args.no_mcp {
        (None, Vec::new())
    } else {
        crate::mcp::McpRuntime::lazy(&settings.mcp, &repo_root).await
    };
    for problem in &mcp_problems {
        eprintln!("{}", console::style(format!("✗ {problem}")).red());
    }

    // A picker needs a terminal to draw in; naming the session is the way out.
    if matches!(args.resume_mode(), Resume::Pick) {
        anyhow::bail!(
            "--resume needs a terminal to show the session list. Pass an id instead: `aster sessions list`, then `aster --resume <ID>`"
        );
    }

    if args.stream {
        return run_stream(
            args,
            client,
            repo_root,
            policy,
            grants,
            allow_edits,
            mcp,
            limits,
            yolo,
            credentials,
        )
        .await;
    }

    let (ctx, history) = prepare_turn(&args, &repo_root, &client, mcp, limits, yolo, credentials)?;
    let titling = history.clone();

    let mut edited: Vec<String> = Vec::new();
    let reply = if args.no_tools {
        let mut messages = vec![ChatMessage {
            role: "system".into(),
            content: system_prompt(&ctx, false).into(),
        }];
        messages.extend(history);
        let reply = client
            .complete_messages(&messages, CHAT_TEMPERATURE)
            .await?;
        ctx.record(MessageEvent::assistant(Some(reply.clone()), Vec::new()));
        reply
    } else {
        agent_loop(
            &client,
            &repo_root,
            &history,
            allow_edits,
            &policy,
            &grants,
            None,
            &mut edited,
            &ctx,
            None,
        )
        .await?
        .0
    };
    let mut naming = name_session(&client, &ctx, &titling, None);

    if crate::json_mode() {
        // The caller has no transcript to re-read, so the name must land first.
        if let Some(naming) = naming.take() {
            let _ = tokio::time::timeout(TITLE_TIMEOUT, naming).await;
        }
        let u = client.usage_snapshot();
        let out = json!({
            "reply": reply,
            "edits": edited,
            "usage": usage_json(&u),
            "title": ctx.current_title(),
        });
        println!("{out}");
    } else {
        println!("{reply}");
        for path in &edited {
            eprintln!("  ✎ edited {path}");
        }
        crate::review::print_usage(client.usage_snapshot());
        if let Some(recorder) = &ctx.recorder
            && let Ok(writer) = recorder.lock()
        {
            eprintln!("Resume this session with: aster --resume {}", writer.id());
        }
    }
    if let Some(naming) = naming {
        let _ = tokio::time::timeout(TITLE_TIMEOUT, naming).await;
    }
    // Learn from this session before the process exits: headless has no later
    // chance, and the startup sweep covers whatever this cannot reach.
    if let Some(session_id) = ctx.session_id() {
        consolidate_finished_session(&client, ctx.store.as_ref(), &repo_root, &session_id).await;
    }
    Ok(())
}

fn prepare_turn(
    args: &ChatArgs,
    repo_root: &Path,
    client: &AiClient,
    mcp: Option<crate::mcp::McpRuntime>,
    limits: Limits,
    yolo: bool,
    credentials: Arc<aster_policy::CommandGrants>,
) -> Result<(SessionCtx, Vec<ChatMessage>)> {
    let mut new_turns = read_history(args)?;
    expand_skill_asks(&mut new_turns, repo_root);
    attach_images(&mut new_turns, repo_root);
    let store = crate::persist::store().ok();
    let (recorder, prior) = resolve_headless_session(
        store.as_ref(),
        repo_root,
        args,
        &client.model,
        client.base_url(),
    )?;
    let agents = crate::agents::discover_agents(repo_root);
    let swarm = SwarmLimits::default();
    let ctx = SessionCtx {
        recorder,
        store,
        credentials,
        skills: discover_skills(repo_root),
        instructions: Arc::new(crate::instructions::discover(repo_root)),
        probe: Arc::new(bash_tools::ToolProbe::detect()),
        plan: Default::default(),
        mcp,
        limits,
        environment: environment_note(repo_root),
        // Yolo is a mode, not just a per-call flag: asking for it once must
        // drop the sandbox too, otherwise commands still fail on writes.
        yolo: Arc::new(AtomicBool::new(yolo)),
        reads: Default::default(),
        previews: Default::default(),
        lookups: Default::default(),
        injected: Default::default(),
        agents,
        sub_agent: None,
        swarm,
    };
    // Record only the new user turn. On the wire path `new_turns` is the full
    // replayed history, whose earlier turns were recorded on previous calls.
    if let Some(last) = new_turns.last().filter(|m| m.role == "user") {
        ctx.record(MessageEvent::user(last.content.text()));
    }
    let mut history = prior;
    history.extend(new_turns);
    Ok((ctx, history))
}

fn emit_line(value: &Value) {
    let mut out = io::stdout();
    let _ = writeln!(out, "{value}");
    let _ = out.flush();
}

fn emit_citations(annotations: &[Annotation], emit: &impl Fn(Value)) {
    let sources: Vec<Value> = annotations
        .iter()
        .map(|a| {
            json!({
                "url": a.url_citation.url,
                "title": a.url_citation.title,
            })
        })
        .collect();
    emit(json!({ "type": "citations", "sources": sources }));
}

fn estimate_reasoning_tokens(chars: usize) -> u64 {
    (chars as u64).div_ceil(4)
}

fn emit_reasoning(details: &[ReasoningDetail], duration_ms: u64, emit: &impl Fn(Value)) {
    let text = reasoning_text(details);
    if !text.is_empty() {
        emit(json!({
            "type": "reasoning",
            "content": text,
            "tokens": estimate_reasoning_tokens(text.chars().count()),
            "duration_ms": duration_ms,
        }));
    }
}

fn reasoning_text(details: &[ReasoningDetail]) -> String {
    details
        .iter()
        .filter_map(ReasoningDetail::plain)
        .collect::<Vec<_>>()
        .join("\n\n")
}

fn reasoning_record(details: &[ReasoningDetail], duration_ms: u64) -> Option<ReasoningRecord> {
    let text = reasoning_text(details);
    if text.is_empty() {
        return None;
    }
    Some(ReasoningRecord {
        tokens: Some(estimate_reasoning_tokens(text.chars().count())),
        duration_ms: Some(duration_ms),
        text,
    })
}

fn emit_reasoning_done(tokens: u64, duration_ms: u64, emit: &impl Fn(Value)) {
    emit(json!({ "type": "reasoning_done", "tokens": tokens, "duration_ms": duration_ms }));
}

fn spawn_stdin_router(injected: Arc<std::sync::Mutex<Vec<String>>>) -> mpsc::Receiver<Value> {
    let (tx, rx) = mpsc::channel::<Value>(4);
    std::thread::spawn(move || {
        let mut line = String::new();
        loop {
            line.clear();
            if io::stdin().read_line(&mut line).unwrap_or(0) == 0 {
                break;
            }
            let Ok(value) = serde_json::from_str::<Value>(&line) else {
                continue;
            };
            if let Some(message) = value.get("message").and_then(Value::as_str) {
                if let Ok(mut queue) = injected.lock() {
                    queue.push(message.to_string());
                }
                continue;
            }
            if tx.blocking_send(value).is_err() {
                break;
            }
        }
    });
    rx
}

fn approval_request_json(kind: &str, req: &ApprovalRequest) -> Value {
    json!({
        "type": "approval_request",
        "kind": kind,
        "preview": req.preview,
        "scope": req.scope.as_ref().map(|p| p.display().to_string()),
        "markdown": req.markdown,
    })
}

fn stdio_approver(mut replies: mpsc::Receiver<Value>) -> UiSender {
    let (tx, mut rx) = mpsc::channel::<UiRequest>(1);
    tokio::spawn(async move {
        while let Some(req) = rx.recv().await {
            match req {
                // `kind` separates the two: approving a plan promotes the mode
                // for good, so a front-end that spawns a process per turn has
                // to persist it rather than re-launch in `plan`.
                UiRequest::Approval(a) => {
                    emit_line(&approval_request_json("action", &a));
                    let answer = replies.recv().await.map_or(Answer::No, parse_approval);
                    let _ = a.respond.send(answer);
                }
                UiRequest::PlanApproval(a) => {
                    emit_line(&approval_request_json("plan", &a));
                    let answer = replies.recv().await.map_or(Answer::No, parse_approval);
                    let _ = a.respond.send(answer);
                }
                UiRequest::Question(q) => {
                    emit_line(&json!({
                        "type": "question",
                        "header": q.header,
                        "question": q.question,
                        "options": q.options,
                    }));
                    let answer = replies.recv().await.and_then(parse_question);
                    let _ = q.respond.send(answer);
                }
            }
        }
    });
    tx
}

fn parse_question(reply: Value) -> Option<String> {
    reply
        .get("choice")
        .and_then(Value::as_str)
        .map(str::to_string)
}

fn parse_approval(reply: Value) -> Answer {
    if !reply.get("allow").and_then(Value::as_bool).unwrap_or(false) {
        return Answer::No;
    }
    match reply.get("always").and_then(Value::as_bool) {
        Some(true) => Answer::Always,
        _ => Answer::Yes,
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_stream(
    args: ChatArgs,
    client: AiClient,
    repo_root: PathBuf,
    policy: Arc<Policy>,
    grants: Arc<Grants>,
    allow_edits: bool,
    mcp: Option<crate::mcp::McpRuntime>,
    limits: Limits,
    yolo: bool,
    credentials: Arc<aster_policy::CommandGrants>,
) -> Result<()> {
    let (ctx, mut history) =
        prepare_turn(&args, &repo_root, &client, mcp, limits, yolo, credentials)?;
    // The router owns stdin from here: replies feed the approver, and typed-in
    // `{"message"}` lines join the turn at the next round boundary.
    let replies = spawn_stdin_router(ctx.injected.clone());
    let approver = stdio_approver(replies);

    // A trailing `/goal <condition>` turns this invocation into a judged loop:
    // turns keep running until a separate judge model deems the condition met.
    let goal = history
        .last()
        .filter(|m| m.role == "user")
        .and_then(|m| crate::goal::parse_goal(&m.content.text()));
    let max_turns = crate::goal::max_turns();
    if let Some(condition) = &goal {
        if let Some(last) = history.last_mut() {
            last.content = crate::goal::directive(condition).into();
        }
        emit_line(&json!({
            "type": "goal_set",
            "condition": condition,
            "max_turns": max_turns,
        }));
    }

    let sink: ChatEventSink = Arc::new(|event| emit_line(&event));
    // Reports that landed between turns flush into this turn's injected queue.
    crate::agents_queue::attach(ctx.injected.clone(), Some(Arc::clone(&sink)));
    let mut edited: Vec<String> = Vec::new();
    let mut turns = 0usize;
    let result = loop {
        let round = agent_loop(
            &client,
            &repo_root,
            &history,
            allow_edits,
            &policy,
            &grants,
            Some(&approver),
            &mut edited,
            &ctx,
            Some(&sink),
        )
        .await;
        let (reply, compacted) = match round {
            Ok(r) => r,
            Err(e) => break Err(e),
        };
        turns += 1;
        let Some(condition) = &goal else {
            break Ok((reply, compacted));
        };
        if turns >= max_turns {
            emit_line(&json!({
                "type": "goal_verdict",
                "verdict": "not_yet",
                "reason": format!("stopped after {max_turns} tries without reaching it"),
                "turn": turns,
                "final": true,
            }));
            break Ok((reply, compacted));
        }
        let judgment = match crate::goal::judge(
            &client,
            ctx.swarm.collector_model.clone(),
            condition,
            &crate::goal::evidence(&history, &reply),
        )
        .await
        {
            Ok(j) => j,
            Err(e) => {
                emit_line(&json!({
                    "type": "goal_verdict",
                    "verdict": "not_yet",
                    "reason": format!("couldn't check progress: {e:#}"),
                    "turn": turns,
                    "final": true,
                }));
                break Ok((reply, compacted));
            }
        };
        let done = judgment.verdict != crate::goal::GoalVerdict::NotYet;
        emit_line(&json!({
            "type": "goal_verdict",
            "verdict": judgment.verdict.as_str(),
            "reason": judgment.reason,
            "turn": turns,
            "final": done,
        }));
        if done {
            break Ok((reply, compacted));
        }
        if let Some(compacted) = compacted {
            emit_compacted(&history, &compacted);
            history = compacted;
        }
        history.push(ChatMessage {
            role: "assistant".into(),
            content: reply.into(),
        });
        let steering = crate::goal::guidance(condition, &judgment.reason);
        ctx.record(MessageEvent::user(steering.clone()));
        history.push(ChatMessage {
            role: "user".into(),
            content: steering.into(),
        });
    };

    // Await before `done`: a title past `done` reads as a turn that never finished.
    let title = if let Some(naming) = match result.is_ok() {
        true => name_session(&client, &ctx, &history, Some(sink)),
        false => None,
    } {
        tokio::time::timeout(TITLE_TIMEOUT, naming)
            .await
            .ok()
            .and_then(|r| r.ok())
            .flatten()
    } else {
        None
    };

    let u = client.usage_snapshot();
    match result {
        Ok((reply, compacted_history)) => {
            if let Some(compacted) = &compacted_history {
                emit_compacted(&history, compacted);
            }
            emit_line(&json!({
                "type": "done",
                "reply": reply,
                "edits": edited,
                "usage": usage_json(&u),
                "title": title,
                // The budget the history may actually spend this session, so a
                // front-end meter does not count the system prompt's share as free.
                "context_budget": crate::budget::history_budget(
                    ctx.limits.compact_budget_chars,
                    system_prompt(&ctx, true).len(),
                ),
            }))
        }
        Err(e) => emit_line(&json!({ "type": "error", "message": format!("{e:#}") })),
    }
    Ok(())
}

fn emit_compacted(before: &[ChatMessage], folded: &[ChatMessage]) {
    let raw = folded.first().map(|m| m.content.text()).unwrap_or_default();
    let summary = raw
        .strip_prefix("Summary of earlier conversation:\n")
        .unwrap_or(&raw);
    emit_line(&json!({
        "type": "compacted",
        "summary": summary,
        "folded": before.len() - folded.len() + 1,
        "messages": folded.iter().map(|m| json!({
            "role": m.role,
            "content": m.content,
        })).collect::<Vec<_>>(),
    }));
}

/// Spell out every `/skill-name` in the conversation, so a front-end shows the
/// command the user typed and the model is still told what it meant. Every turn:
/// a replayed history would otherwise carry a slash it cannot read.
pub(crate) fn expand_skill_asks(turns: &mut [ChatMessage], repo_root: &Path) {
    let asks = |m: &ChatMessage| m.role == "user" && m.content.text().starts_with('/');
    if !turns.iter().any(asks) {
        return;
    }
    let skills = discover_skills(repo_root);
    for turn in turns.iter_mut().filter(|m| asks(m)) {
        turn.content = expand_skill(&turn.content.text(), &skills).into();
    }
}

pub(crate) fn attach_images(turns: &mut [ChatMessage], repo_root: &Path) {
    if let Some(last) = turns.last_mut().filter(|m| m.role == "user") {
        last.content = crate::images::attach(&last.content.text(), repo_root);
    }
}

fn read_history(args: &ChatArgs) -> Result<Vec<ChatMessage>> {
    if let Some(path) = args.messages_json.as_deref() {
        let raw = if path == "-" && args.stream {
            // Streaming keeps stdin open for approval replies, so the messages
            // are one line rather than everything up to EOF.
            let mut buf = String::new();
            io::stdin()
                .read_line(&mut buf)
                .context("reading the messages line from stdin")?;
            buf
        } else if path == "-" {
            let mut buf = String::new();
            io::stdin()
                .read_to_string(&mut buf)
                .context("reading messages from stdin")?;
            buf
        } else {
            fs::read_to_string(path).with_context(|| format!("reading {path}"))?
        };
        let wire: Vec<WireMessage> = serde_json::from_str(&raw)
            .context("parsing --messages-json: expected a JSON array of {role, content}")?;
        if wire.is_empty() {
            bail!("nothing to ask; --messages-json was an empty array");
        }
        return wire
            .into_iter()
            .map(|m| match m.role.as_str() {
                "user" | "assistant" | "system" => Ok(ChatMessage {
                    role: m.role,
                    content: m.content.into(),
                }),
                other => bail!("unsupported role {other:?} in --messages-json"),
            })
            .collect();
    }

    if let Some(prompt) = args
        .prompt
        .as_deref()
        .map(str::trim)
        .filter(|p| !p.is_empty())
    {
        // `aster solve.py` reads as "run this", but it would start a second
        // agent with the filename as its question and answer from nowhere.
        if !prompt.contains(char::is_whitespace) && Path::new(prompt).is_file() {
            bail!(
                "{prompt} is a file, and a bare `aster <file>` would ask a new agent \
                 about it instead of running it. Use `aster python {prompt}` for a script, \
                 or `aster \"<question>\" @{prompt}` to ask about its contents"
            );
        }
        return Ok(vec![ChatMessage {
            role: "user".into(),
            content: prompt.into(),
        }]);
    }

    // Piped input is the prompt: `echo "why?" | aster` should just work.
    // Not on `--stream`, where stdin stays open for approval replies.
    if !args.stream && !io::stdin().is_terminal() {
        let mut buf = String::new();
        io::stdin()
            .read_to_string(&mut buf)
            .context("reading the prompt from stdin")?;
        if !buf.trim().is_empty() {
            return Ok(vec![ChatMessage {
                role: "user".into(),
                content: buf.trim().into(),
            }]);
        }
    }
    bail!("nothing to ask; pass a prompt (aster \"...\"), pipe one in, or use --messages-json")
}

fn resolve_headless_session(
    store: Option<&Store>,
    repo_root: &Path,
    args: &ChatArgs,
    model: &str,
    base_url: &str,
) -> Result<(Option<Recorder>, Vec<ChatMessage>)> {
    let Some(store) = store else {
        return Ok((None, Vec::new()));
    };

    // The wire path (a UI replays full history) owns its history; with
    // `--session` it also records into that session, resuming it if the id is
    // already on disk — an explicit ask, since the id then keeps appending.
    if args.messages_json.is_some() {
        let Some(id) = &args.session else {
            return Ok((None, Vec::new()));
        };
        let writer = store.session_writer_for(
            repo_root,
            id,
            repo_root,
            Some(model.to_string()),
            Some(base_url.to_string()),
        )?;
        return Ok((Some(recorder(writer)), Vec::new()));
    }

    if let Some(id) = &args.session {
        let prior = store
            .resume(repo_root, id)
            .map(|t| t.to_chat_messages())
            .unwrap_or_default();
        let writer = store.session_writer_for(
            repo_root,
            id,
            repo_root,
            Some(model.to_string()),
            Some(base_url.to_string()),
        )?;
        return Ok((Some(recorder(writer)), prior));
    }

    let base = match args.resume_mode() {
        Resume::Id(id) => Some(
            store
                .resume(repo_root, &id)
                .with_context(|| format!("no session {id:?} for this repo"))?,
        ),
        Resume::Latest => store.latest(repo_root)?,
        Resume::New | Resume::Pick => None,
    };

    let Some(transcript) = base else {
        // No explicit session: the turn is ephemeral, nothing is recorded.
        return Ok((None, Vec::new()));
    };
    let prior = transcript.to_chat_messages();
    let writer = store.resume_writer(repo_root, &transcript.meta.id)?;
    Ok((Some(recorder(writer)), prior))
}

fn recorder(writer: aster_persist::SessionWriter) -> Recorder {
    std::sync::Arc::new(std::sync::Mutex::new(writer))
}

/// One full agentic turn that also reports progress as it goes: streamed
/// tokens, tool call steps, and edit notifications arrive on `events` so a
/// front-end can render the turn live instead of waiting for the reply.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn agent_turn_streaming(
    client: AiClient,
    repo_root: PathBuf,
    history: Vec<ChatMessage>,
    allow_edits: bool,
    policy: Arc<Policy>,
    grants: Arc<Grants>,
    approver: Option<UiSender>,
    ctx: SessionCtx,
    events: ChatEventSink,
) -> Result<(String, Vec<String>, Option<Vec<ChatMessage>>)> {
    // This turn hears background agent completions from here on; reports that
    // landed while no turn was attached flush in now.
    crate::agents_queue::attach(ctx.injected.clone(), Some(Arc::clone(&events)));
    let mut edited = Vec::new();
    let (reply, compacted) = agent_loop(
        &client,
        &repo_root,
        &history,
        allow_edits,
        &policy,
        &grants,
        approver.as_ref(),
        &mut edited,
        &ctx,
        Some(&events),
    )
    .await?;
    // Detached on purpose: the TUI outlives the turn, so the title can land
    // whenever it lands rather than holding the composer.
    drop(name_session(
        &client,
        &ctx,
        &history,
        Some(Arc::clone(&events)),
    ));
    Ok((reply, edited, compacted))
}

enum RoundVerdict {
    Continue,
    Correct,
    Abort,
    Nudge(usize),
    Wrap,
}

const LOOP_CORRECTION: &str = "You repeated the same tool calls with the same \
    results three times in a row. Stop repeating. Re-read the results above and \
    do something different, or give your final answer.";

const MAX_ROUND_EXTENSIONS: usize = 2;

const BARREN_ROUNDS: usize = 10;

const MAX_EMPTY_ROUNDS: usize = 2;

const EMPTY_CORRECTION: &str = "Your last reply was empty: no text and no tool \
    calls. Continue the work now with a tool call or a final answer. Do not \
    describe what you would do; if something is blocking you, name it in one \
    line.";

const SILENT_MODEL: &str = "The model returned nothing, twice over and again \
    when asked for a plain answer. Everything above this line still happened and \
    is saved. Send the message again, or switch models if it keeps up.";

const MAX_DEGENERATE_RETRIES: usize = 2;

const DEGENERATE_CORRECTION: &str = "Your last reply degenerated into repeated \
    text and was cut off; everything before it still stands. Continue from where \
    it stopped: keep replies short, act with tool calls instead of long prose, \
    and keep any final report to tight bullets. Do not restate text you already \
    wrote.";

fn barren_correction(rounds: usize) -> String {
    format!(
        "Your last {rounds} tool rounds were lookups that came back empty. \
         Stop searching the same way: act on what is already above, try a \
         different angle, or say in one line what is blocking you."
    )
}

const SINGLE_LOOKUP_ROUNDS: usize = 4;

fn batch_correction(rounds: usize) -> String {
    format!(
        "Your last {rounds} rounds each carried a single lookup, and every \
         round costs a full model round-trip. Decide everything you still \
         need to see and request it in one round: several tool calls in one \
         reply, or one `explore` with all the steps."
    )
}

const DANGLING_INTENT_CORRECTION: &str = "You ended your reply promising work \
    you have not done. Do not narrate intent. If the work was asked for, do it \
    now with your tools and then report what actually changed; if you cannot, \
    name the blocker in one line.";

const PROMISED_WORK_KICK: &str = "Your previous reply promised work that was \
    never done. Your tools are live right now, in this turn. Start with the \
    first tool call for that work; do not restate the plan and do not mention \
    tool availability.";

/// A reply that ends on a question is waiting for an answer, not stalling.
fn awaits_an_answer(reply: &str) -> bool {
    reply
        .lines()
        .rev()
        .find(|line| !line.trim().is_empty())
        .is_some_and(|line| line.trim_end().ends_with('?'))
}

/// Narrating intent without acting, which the nudge exists to catch. Asking to
/// go ahead is not that: on a phone "say the word and I'll order it" is the
/// permission request, and nudging past it once bought food nobody asked for.
fn announces_pending_work(reply: &str) -> bool {
    const PROMISES: &[&str] = &[
        "fixing it now",
        "fixing that now",
        "fixing this now",
        "doing that now",
        "doing this now",
        "doing that next",
        "doing this next",
        "i will now",
        "i'll now",
        "let me now",
        "making the edit now",
        "making the change now",
        "next message i'll",
        "next message i will",
        "next session:",
        "not applied yet",
        "not applied any edits",
        "no edits are written",
        "ready to apply",
        "unapplied",
        "what i did not get to",
        "what i have not done",
        "what i haven't done",
    ];
    if awaits_an_answer(reply) {
        return false;
    }
    let reply = reply.to_lowercase();
    PROMISES.iter().any(|p| reply.contains(p))
}

fn budget_notice(spent: usize, cap: usize) -> String {
    format!(
        "You are {spent} tool rounds into this turn; {} remain before it ends \
         and you have to answer with whatever you have. Spend them finishing, \
         not gathering.",
        cap.saturating_sub(spent)
    )
}

#[derive(Default)]
struct NoProgress {
    last_round: Option<u64>,
    identical_rounds: usize,
    error_rounds: usize,
    barren_rounds: usize,
    corrected: bool,
    nudged: bool,
}

impl NoProgress {
    fn feed(&mut self, sig: u64, all_errors: bool, productive: bool) -> RoundVerdict {
        if self.last_round == Some(sig) {
            self.identical_rounds += 1;
        } else {
            self.last_round = Some(sig);
            self.identical_rounds = 1;
        }
        self.error_rounds = if all_errors { self.error_rounds + 1 } else { 0 };
        self.barren_rounds = if productive {
            0
        } else {
            self.barren_rounds + 1
        };
        if self.identical_rounds >= 3 || self.error_rounds >= 3 {
            if !self.corrected {
                self.corrected = true;
                self.identical_rounds = 0;
                self.error_rounds = 0;
                return RoundVerdict::Correct;
            }
            return RoundVerdict::Abort;
        }
        // Checked after repetition: a model stuck on both is stuck on the
        // harder one, and that path ends the turn with a clearer reason.
        if self.barren_rounds >= BARREN_ROUNDS {
            let streak = std::mem::take(&mut self.barren_rounds);
            if !self.nudged {
                self.nudged = true;
                return RoundVerdict::Nudge(streak);
            }
            return RoundVerdict::Wrap;
        }
        RoundVerdict::Continue
    }
}

fn is_productive_round(round: &[(String, String, String)]) -> bool {
    round
        .iter()
        .any(|(name, _, _)| !PARALLEL_READ_TOOLS.contains(&name.as_str()) && name != "explore")
}

fn round_found_something(round: &[(String, String, String)]) -> bool {
    round.iter().any(|(_, _, result)| {
        !result.starts_with("error: ")
            && !result.starts_with("[identical ")
            && !aster_persist::barren(result)
    })
}

fn round_signature(round: &[(String, String, String)]) -> u64 {
    use std::hash::{Hash, Hasher};
    let mut hasher = std::hash::DefaultHasher::new();
    for (name, args, result) in round {
        name.hash(&mut hasher);
        args.hash(&mut hasher);
        result.hash(&mut hasher);
    }
    hasher.finish()
}

/// Drive the model's tool calls until it answers in plain text or the round cap trips.
/// Tool failures return to the model as tool results so it can retry instead of dying.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
    name = "turn",
    skip_all,
    fields(rounds = tracing::field::Empty, calls = tracing::field::Empty)
)]
pub(crate) async fn agent_loop(
    client: &AiClient,
    repo_root: &Path,
    history: &[ChatMessage],
    mut allow_edits: bool,
    policy: &Policy,
    grants: &Grants,
    approver: Option<&UiSender>,
    edited: &mut Vec<String>,
    ctx: &SessionCtx,
    events: Option<&ChatEventSink>,
) -> Result<(String, Option<Vec<ChatMessage>>)> {
    // Owned for the turn: approving a plan promotes the mode, and the rest of
    // the turn has to run under the promoted policy rather than the one the
    // front-end compiled before the plan existed.
    let mut policy = policy.clone();
    let mut calls = 0usize;
    let turn_span = tracing::Span::current();
    let emit = |event: Value| {
        if let Some(sink) = events {
            sink(event);
        }
    };
    // Harness-to-model steering. It rides a user turn because a mid-conversation
    // system message is not portable across providers, but it is recorded as
    // `system` and never emitted: the user did not say it.
    let steer = |wire: &mut Vec<Value>, content: String| {
        ctx.record(MessageEvent::system(content.clone()));
        wire.push(json!({ "role": "user", "content": content }));
    };
    // Measured first so its reservation (persona, instructions, memory,
    // skills) comes off the top of the budget the history may spend.
    let system = system_prompt(ctx, true);
    let (history, compacted) = compact_if_needed(client, history, ctx, system.len()).await?;
    let mut wire: Vec<Value> = vec![json!({
        "role": "system",
        "content": system,
    })];
    let system_chars = wire[0]["content"].as_str().map_or(0, str::len);
    for m in &history {
        wire.push(serde_json::to_value(m)?);
    }
    // Unbrick a session whose last reply was a promise: without this the model
    // reads its own "next turn I'll..." and settles back into narration.
    if let Some(last) = history.iter().rev().find(|m| m.role == "assistant")
        && announces_pending_work(&last.content.text())
    {
        tracing::warn!("previous reply ended on a promise; kicking the turn into acting");
        steer(&mut wire, PROMISED_WORK_KICK.to_string());
    }

    // The round cap is a runaway backstop, not a work limit: while the plan
    // keeps moving, hitting it grants another allotment. A full allotment
    // with no plan progress is the runaway case, and the loop ends.
    let mut round_cap = ctx.limits.max_tool_rounds;
    let mut plan_at_extension = plan_snapshot(ctx);
    let mut extensions = 0usize;
    let mut no_progress = NoProgress::default();
    // The turn tells the model what it has left exactly once. Repeating it
    // every round would spend more context than the pressure is worth.
    let mut budget_told = false;
    // Consecutive silent rounds, reset by any round that produces something.
    let mut empties = 0usize;
    // Degenerate replies steered back on track this turn.
    let mut degenerations = 0usize;
    // The dangling-intent steer fires at most twice a turn, then the reply
    // stands rather than burning rounds on a model that will not act.
    let mut promised = 0usize;
    // Consecutive one-lookup rounds; the batching steer fires once per turn.
    let mut single_lookups = 0usize;
    let mut batch_nudged = false;
    for round in 0.. {
        if round >= round_cap {
            let now = plan_snapshot(ctx);
            if extensions < MAX_ROUND_EXTENSIONS
                && plan_unfinished(&now)
                && now != plan_at_extension
            {
                tracing::debug!(
                    round,
                    "plan still in motion; extending the tool-round budget"
                );
                plan_at_extension = now;
                extensions += 1;
                round_cap += ctx.limits.max_tool_rounds;
            } else {
                break;
            }
        }
        // The Jev check is advisory: a confident suggestion can end the turn
        // early or steer it, never extend it past the cap above.
        let advice = crate::jev::current().advise(&wire, round, round_cap).await;
        if advice != crate::jev::Advice::None {
            tracing::debug!(round, ?advice, "jev advised the loop");
        }
        match advice {
            crate::jev::Advice::None => {}
            crate::jev::Advice::Retry => steer(
                &mut wire,
                "A check of this turn says the current approach is not \
                 working. Try a different way, or give your final answer."
                    .to_string(),
            ),
            crate::jev::Advice::AskUser => steer(
                &mut wire,
                "A check of this turn says a fact you need is missing. Ask \
                 the user one concrete question instead of guessing."
                    .to_string(),
            ),
            crate::jev::Advice::Stop => break,
        }
        turn_span.record("rounds", round + 1);
        // Messages the user sent mid-turn join here, before the next request.
        let pending: Vec<String> = match ctx.injected.lock() {
            Ok(mut queue) => queue.drain(..).collect(),
            Err(_) => Vec::new(),
        };
        for content in pending {
            emit(json!({ "type": "injected", "content": content }));
            ctx.record(MessageEvent::user(content.clone()));
            // A photo sent mid-turn arrives as an `@path` mention like any
            // other, and only this pass turns it into something to look at.
            let content = crate::images::attach(&content, repo_root);
            wire.push(json!({ "role": "user", "content": content }));
        }
        // Halfway through the allotment, and only once: the model cannot see
        // the round counter, so without this it converges only when the cap
        // forces it to, which reads to the user as the agent never finishing.
        if !budget_told && round > 0 && round * 2 >= round_cap {
            budget_told = true;
            tracing::debug!(round, round_cap, "telling the model its round budget");
            steer(&mut wire, budget_notice(round, round_cap));
        }
        // Tool results accumulate inside a turn too; evict by policy before
        // each request and leave a trace of everything the model lost.
        let budget = crate::budget::history_budget(ctx.limits.compact_budget_chars, system_chars)
            + system_chars;
        for eviction in crate::budget::evict_tool_results(&mut wire, budget) {
            tracing::debug!(
                reason = eviction.reason,
                index = eviction.index,
                chars = eviction.chars,
                "evicted message to fit the context budget"
            );
            ctx.record_eviction(&eviction);
        }
        let mut tools = tool_defs(allow_edits, approver.is_some());
        // Sub-agents only see their allowlisted tools.
        if let Some(sub) = &ctx.sub_agent {
            tools.retain(|t| {
                t["function"]["name"]
                    .as_str()
                    .map(|n| sub.tool_allowlist.contains(n))
                    .unwrap_or(false)
            });
        }
        if ctx.sub_agent.is_none() && !ctx.agents.is_empty() {
            tools.push(agent_tool_schema());
        }
        if let Some(injection) = ctx.mcp.as_ref().and_then(|m| m.injection()) {
            tools.push(injection.bridge_tool);
        }
        if client.web_search() {
            tools.push(json!({"type": "openrouter:web_search"}));
        }
        // False when the endpoint ignored `stream` and the client fell back to a
        // whole response, which the commentary emit below has to make up for.
        let mut streamed = false;
        // Live thinking: chars accumulated across reasoning deltas, so the host
        // can show a growing token count before the round closes.
        let mut reasoning_chars = 0usize;
        let mut streamed_reasoning = false;
        let started = Instant::now();
        // The client only exposes a cumulative counter, so this round's spend is
        // the delta across the call.
        let before = client.usage_snapshot();
        let mut msg = match client
            .complete_tools_stream_with(
                &client.model,
                wire.clone(),
                tools.clone(),
                CHAT_TEMPERATURE,
                |delta| {
                    streamed = true;
                    emit(json!({ "type": "token", "content": delta }));
                },
                |delta| {
                    streamed_reasoning = true;
                    reasoning_chars += delta.chars().count();
                    emit(json!({
                        "type": "reasoning_delta",
                        "content": delta,
                        "tokens": estimate_reasoning_tokens(reasoning_chars),
                    }));
                },
            )
            .await
        {
            Ok(msg) => msg,
            // Some models reject tool definitions; degrade to plain chat. Only
            // safe on round 0, before any tool turns entered the history.
            Err(e) if round == 0 && is_tool_unsupported(&e) => {
                tracing::debug!("model rejected tools; falling back to plain chat: {e:#}");
                let mut messages = vec![ChatMessage {
                    role: "system".into(),
                    content: system_prompt(ctx, false).into(),
                }];
                messages.extend(history.iter().cloned());
                let reply = client
                    .complete_messages(&messages, CHAT_TEMPERATURE)
                    .await?;
                ctx.record(
                    MessageEvent::assistant(Some(reply.clone()), Vec::new())
                        .with_usage(round_usage(before, client.usage_snapshot())),
                );
                return Ok((reply, compacted));
            }
            // A reply that collapsed into verbatim repetition is steered, not
            // fatal: every round above is still on the wire, so a correction
            // and one more round keeps the whole turn's work alive.
            Err(e) if e.downcast_ref::<DegenerateOutput>().is_some() => {
                degenerations += 1;
                if degenerations > MAX_DEGENERATE_RETRIES {
                    return Err(e);
                }
                tracing::warn!(
                    degenerations,
                    "reply degenerated into repeated text; steering and retrying the round"
                );
                steer(&mut wire, DEGENERATE_CORRECTION.to_string());
                continue;
            }
            Err(e) => return Err(e),
        };
        msg.tool_calls
            .iter_mut()
            .for_each(crate::tool_alias::canonicalize);
        let duration_ms = started.elapsed().as_millis() as u64;
        let usage = round_usage(before, client.usage_snapshot());
        if streamed_reasoning {
            emit_reasoning_done(
                estimate_reasoning_tokens(reasoning_chars),
                duration_ms,
                &emit,
            );
        } else {
            emit_reasoning(&msg.reasoning_details, duration_ms, &emit);
        }
        // Recorded from the assembled blocks either way: the streaming path only
        // counted characters as they flew past, so this is the one place the
        // whole thinking exists once the round closes.
        let reasoning = reasoning_record(&msg.reasoning_details, duration_ms);

        if msg.tool_calls.is_empty() {
            // A silent round used to end the turn as an error, losing the work
            // above it. Ask again instead, then fall through to the forced
            // answer, which drops the tools a malformed call may have died on.
            let Some(reply) = msg.content.filter(|c| !c.trim().is_empty()) else {
                empties += 1;
                tracing::warn!(empties, "model returned no text and no tool calls");
                if empties > MAX_EMPTY_ROUNDS {
                    break;
                }
                steer(&mut wire, EMPTY_CORRECTION.to_string());
                continue;
            };
            // "Fixing it now" with nothing fixed is the promise-shaped lie;
            // steer once so the promised work happens this turn instead.
            if promised < 2 && announces_pending_work(&reply) {
                promised += 1;
                empties = 0;
                tracing::warn!("model ended its turn promising undone work; telling it to act");
                ctx.record(
                    MessageEvent::assistant(Some(reply.clone()), Vec::new())
                        .with_reasoning(reasoning)
                        .with_usage(usage),
                );
                if streamed {
                    emit(json!({ "type": "token", "content": "\n\n" }));
                } else {
                    emit(json!({ "type": "text", "content": reply.clone() }));
                }
                wire.push(json!({ "role": "assistant", "content": reply }));
                steer(&mut wire, DANGLING_INTENT_CORRECTION.to_string());
                continue;
            }
            // A message the user sent while this reply was forming still
            // deserves an answer: bank the reply and keep the turn alive
            // rather than dropping the queue on the floor.
            let pending: Vec<String> = match ctx.injected.lock() {
                Ok(mut queue) => queue.drain(..).collect(),
                Err(_) => Vec::new(),
            };
            if !pending.is_empty() {
                if !msg.annotations.is_empty() {
                    emit_citations(&msg.annotations, &emit);
                }
                ctx.record(
                    MessageEvent::assistant(Some(reply.clone()), Vec::new())
                        .with_annotations(msg.annotations.clone())
                        .with_reasoning(reasoning)
                        .with_usage(usage),
                );
                if streamed {
                    emit(json!({ "type": "token", "content": "\n\n" }));
                } else {
                    emit(json!({ "type": "text", "content": reply.clone() }));
                }
                wire.push(json!({ "role": "assistant", "content": reply }));
                for content in pending {
                    emit(json!({ "type": "injected", "content": content }));
                    ctx.record(MessageEvent::user(content.clone()));
                    wire.push(json!({ "role": "user", "content": content }));
                }
                continue;
            }
            if !msg.annotations.is_empty() {
                emit_citations(&msg.annotations, &emit);
                ctx.record(
                    MessageEvent::assistant(Some(reply.clone()), Vec::new())
                        .with_annotations(msg.annotations.clone())
                        .with_reasoning(reasoning)
                        .with_usage(usage),
                );
            } else {
                ctx.record(
                    MessageEvent::assistant(Some(reply.clone()), Vec::new())
                        .with_reasoning(reasoning)
                        .with_usage(usage),
                );
            }
            return Ok((reply, compacted));
        }
        empties = 0;

        ctx.record(
            MessageEvent::assistant(msg.content.clone(), msg.tool_calls.clone())
                .with_annotations(msg.annotations.clone())
                .with_reasoning(reasoning)
                .with_usage(usage),
        );
        // Text the model emitted alongside its tool calls: its running commentary.
        // Streaming already delivered it, so that path only needs the separator.
        if let Some(text) = msg.content.as_deref().filter(|c| !c.trim().is_empty()) {
            if streamed {
                emit(json!({ "type": "token", "content": "\n\n" }));
            } else {
                emit(json!({ "type": "text", "content": text }));
            }
        }
        let mut assistant = json!({
            "role": "assistant",
            "content": msg.content,
            "tool_calls": msg.tool_calls,
        });
        // Carried verbatim and in order: the provider rejects a reasoning
        // sequence that does not match what it emitted.
        if !msg.reasoning_details.is_empty() {
            assistant["reasoning_details"] = json!(msg.reasoning_details);
        }
        wire.push(assistant);
        for call in &msg.tool_calls {
            if internal_call(&ctx.skills, &call.function.name, &call.function.arguments) {
                continue;
            }
            emit(json!({
                "type": "tool_call",
                "id": call.id,
                "name": call.function.name,
                "arguments": call.function.arguments,
            }));
        }
        // A batch of pure reads runs on parallel threads. Any stateful call
        // keeps the whole round sequential so this-then-that ordering holds.
        let mut prefetched: Vec<Option<String>> = vec![None; msg.tool_calls.len()];
        // The reads in a batch fan out even when a command or edit sits beside
        // them; those still run in order afterwards.
        let reads: Vec<usize> = msg
            .tool_calls
            .iter()
            .enumerate()
            .filter(|(_, c)| PARALLEL_READ_TOOLS.contains(&c.function.name.as_str()))
            .map(|(i, _)| i)
            .collect();
        if reads.len() > 1 {
            // spawn_blocking fans the synchronous reads out on the blocking
            // pool, so this worker stays free to run other tasks meanwhile.
            let handles: Vec<_> = reads
                .iter()
                .map(|&i| {
                    let call = &msg.tool_calls[i];
                    let repo_root = repo_root.to_path_buf();
                    let policy = policy.clone();
                    let ctx = ctx.clone();
                    let name = call.function.name.clone();
                    let arguments = call.function.arguments.clone();
                    tokio::task::spawn_blocking(move || {
                        read_only_call(&repo_root, &policy, &ctx, &name, &arguments)
                    })
                })
                .collect();
            for (&i, handle) in reads.iter().zip(handles) {
                prefetched[i] = handle.await.unwrap_or_default();
            }
        }
        let mut round_sig = Vec::with_capacity(msg.tool_calls.len());
        let mut round_all_errors = true;
        for (call, prefetched) in msg.tool_calls.iter().zip(prefetched) {
            calls += 1;
            turn_span.record("calls", calls);
            let span = tracing::info_span!(
                "tool_call",
                tool = %call.function.name,
                round,
                cached = prefetched.is_some(),
                result_chars = tracing::field::Empty,
                barren = tracing::field::Empty,
                error = tracing::field::Empty,
            );
            let result = match prefetched {
                Some(result) => ToolOutput::text(result),
                None => {
                    if call.function.name == "agent" {
                        ToolOutput::text(
                            dispatch_agent_tool(
                                repo_root,
                                client,
                                &call.function.arguments,
                                &policy,
                                grants,
                                ctx,
                                &call.id,
                                events,
                            )
                            .instrument(span.clone())
                            .await,
                        )
                    } else {
                        exec_tool(
                            repo_root,
                            &mut allow_edits,
                            &mut policy,
                            grants,
                            approver,
                            &call.function.name,
                            &call.function.arguments,
                            edited,
                            ctx,
                            events,
                        )
                        .instrument(span.clone())
                        .await
                    }
                }
            };
            // Re-asking costs a round either way, but the answer is already
            // above; a pointer keeps the history from carrying it twice.
            let result = if is_repeat_lookup(ctx, &call.function.name, &call.function.arguments) {
                ToolOutput::text(format!(
                    "[identical {} call earlier in this turn — scroll up for the result]",
                    call.function.name
                ))
            } else {
                result
            };
            let ToolOutput {
                text: result,
                images,
            } = result;
            span.record("result_chars", result.len());
            span.record("barren", aster_persist::barren(&result));
            span.record("error", result.starts_with("error: "));
            tracing::debug!(tool = %call.function.name, "tool call executed");
            let result = truncate(&crate::redact::redact(&result), MAX_TOOL_RESULT_CHARS);
            round_sig.push((
                call.function.name.clone(),
                call.function.arguments.clone(),
                result.clone(),
            ));
            if !result.starts_with("error: ") {
                round_all_errors = false;
            }
            let internal =
                internal_call(&ctx.skills, &call.function.name, &call.function.arguments);
            if !internal {
                emit(json!({
                    "type": "tool_result",
                    "id": call.id,
                    "name": call.function.name,
                    "result": result,
                    "error": result.starts_with("error: "),
                    "images": images.len(),
                }));
            }
            ctx.record(MessageEvent::tool(
                &call.id,
                if internal {
                    "[internal tool result omitted]"
                } else {
                    &result
                },
            ));
            wire.push(json!({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            }));
            if !images.is_empty() {
                retire_old_images(&mut wire, LIVE_IMAGES);
                wire.push(image_turn(&call.function.name, &images));
            }
        }
        single_lookups = match round_sig.as_slice() {
            [(name, _, _)] if PARALLEL_READ_TOOLS.contains(&name.as_str()) => single_lookups + 1,
            _ => 0,
        };
        let verdict = no_progress.feed(
            round_signature(&round_sig),
            round_all_errors,
            is_productive_round(&round_sig) || round_found_something(&round_sig),
        );
        match verdict {
            RoundVerdict::Continue => {
                if !batch_nudged && single_lookups >= SINGLE_LOOKUP_ROUNDS {
                    batch_nudged = true;
                    tracing::debug!(single_lookups, "model is not batching; telling it once");
                    steer(&mut wire, batch_correction(single_lookups));
                }
            }
            RoundVerdict::Correct => {
                tracing::warn!("model looped on tool calls; injecting one correction");
                steer(&mut wire, LOOP_CORRECTION.to_string());
            }
            RoundVerdict::Abort => {
                bail!(
                    "the model kept repeating the same tool calls after being told to stop; ending the turn"
                );
            }
            RoundVerdict::Nudge(streak) => {
                tracing::warn!(
                    streak,
                    "model kept gathering without acting; telling it to act"
                );
                steer(&mut wire, barren_correction(streak));
            }
            // Unlike Abort this is not a failed turn: the model gathered plenty,
            // it just never committed. Fall through to the forced final answer
            // so the user gets what it found instead of an error.
            RoundVerdict::Wrap => {
                tracing::warn!("model kept gathering after being told to act; forcing an answer");
                break;
            }
        }
    }

    tracing::warn!(
        round_cap,
        "stopped without a final answer of the model's own; forcing one"
    );
    steer(
        &mut wire,
        "Stop using tools and answer now with what you have. This applies to \
         this reply only; your tools come back on the next message. Report \
         only what actually happened; anything unfinished gets one plain \
         line, not a section. Never describe an edit, command, or check that \
         did not run as if it happened."
            .to_string(),
    );
    let msg = client
        .complete_tools_stream_with(
            &client.model,
            wire,
            Vec::new(),
            CHAT_TEMPERATURE,
            |delta| {
                emit(json!({ "type": "token", "content": delta }));
            },
            |_| {},
        )
        .await?;
    // Last resort. Failing here would throw away a whole turn's work over a
    // provider that went quiet, so the turn ends with what happened instead.
    let reply = msg
        .content
        .filter(|c| !c.trim().is_empty())
        .unwrap_or_else(|| {
            tracing::warn!("model stayed silent through the forced answer");
            SILENT_MODEL.to_string()
        });
    ctx.record(MessageEvent::assistant(Some(reply.clone()), Vec::new()));
    Ok((reply, compacted))
}

const COMPACT_PROMPT: &str = "You are compacting a conversation to fit a context \
window. Summarize the exchange below into a compact brief the assistant can \
continue from: the user's goals, the key decisions and facts established, the \
files and code touched, and any open threads or next steps. Be specific and \
terse. Do not add commentary or a preamble.";

const TITLE_PROMPT: &str = "Name the conversation below. Reply with the name \
and nothing else: 3 to 6 words, sentence case, no quotes, no trailing period. \
Start with a plain verb and drop articles. Name what the user is trying to do, \
not what the assistant did, and be concrete about the subject (\"Fix sandbox \
seccomp filter\", not \"Debugging a bug\"). Keep the user's own nouns for \
files, tools, and features.";

const TITLE_AFTER_TURNS: usize = 2;

const OPENER_WORDS: usize = 3;
const OPENER_CHARS: usize = 12;

const EMPTY_OPENERS: &[&str] = &[
    "hi",
    "hey",
    "hello",
    "yo",
    "sup",
    "help",
    "continue",
    "go",
    "go on",
    "ok",
    "okay",
    "k",
    "thanks",
    "ty",
    "test",
    "testing",
    "ping",
    "hi there",
    "hello there",
    "what's up",
    "whats up",
];

fn opener_names_session(first: &str) -> bool {
    let text = first.trim();
    let bare = text.trim_end_matches(['.', '!', '?', ' ']).to_lowercase();
    if EMPTY_OPENERS.contains(&bare.as_str()) {
        return false;
    }
    text.split_whitespace().count() >= OPENER_WORDS || text.chars().count() >= OPENER_CHARS
}

fn turns_before_naming(history: &[ChatMessage]) -> usize {
    let opener = history
        .iter()
        .find(|m| m.role == "user")
        .map(|m| m.content.text());
    match opener {
        Some(first) if opener_names_session(&first) => 1,
        _ => TITLE_AFTER_TURNS,
    }
}

const TITLE_TIMEOUT: Duration = Duration::from_secs(10);

const TITLE_MAX_CHARS: usize = 60;

/// Name the session once it has shape: after the first turn when the opening
/// message carries the topic, after the second otherwise. Runs in the background,
/// so a caller whose process exits at turn end MUST await the handle.
pub(crate) fn name_session(
    client: &AiClient,
    ctx: &SessionCtx,
    history: &[ChatMessage],
    sink: Option<ChatEventSink>,
) -> Option<tokio::task::JoinHandle<Option<String>>> {
    if ctx.sub_agent.is_some() {
        return None;
    }
    let turns = history.iter().filter(|m| m.role == "user").count();
    let needed = turns_before_naming(history);
    if turns < needed {
        return None;
    }
    // A recorded session names itself once and the transcript remembers it. An
    // unrecorded one (a desktop thread nobody saved) has nowhere to remember,
    // so it names itself exactly on the threshold turn and not again.
    if ctx.recorder.is_some() {
        if ctx.is_titled() {
            return None;
        }
    } else if turns != needed {
        return None;
    }

    let messages = vec![
        ChatMessage {
            role: "system".into(),
            content: TITLE_PROMPT.into(),
        },
        ChatMessage {
            role: "user".into(),
            content: title_context(history).into(),
        },
    ];
    let client = client.clone();
    let ctx = ctx.clone();
    Some(tokio::spawn(async move {
        let reply = match client.complete_messages(&messages, 0.2).await {
            Ok(reply) => reply,
            Err(e) => {
                tracing::debug!("could not name the session: {e:#}");
                return None;
            }
        };
        let title = clean_title(&reply)?;
        ctx.record_title(&title);
        if let Some(sink) = sink {
            sink(json!({ "type": "title", "title": title }));
        }
        Some(title)
    }))
}

fn title_context(history: &[ChatMessage]) -> String {
    let mut out = String::new();
    for m in history
        .iter()
        .filter(|m| m.role == "user" || m.role == "assistant")
    {
        let body = truncate(
            m.content.text().trim(),
            if m.role == "user" { 2000 } else { 500 },
        );
        out.push_str(&m.role);
        out.push_str(": ");
        out.push_str(&body);
        out.push_str("\n\n");
    }
    out
}

fn clean_title(reply: &str) -> Option<String> {
    let line = reply.trim().lines().next()?.trim();
    let line = line.trim_start_matches('#').trim();
    let line = line.trim_matches(|c| c == '"' || c == '\'' || c == '`');
    let line = line.trim_end_matches('.').trim();
    if line.is_empty() || line.chars().count() > TITLE_MAX_CHARS {
        return None;
    }
    Some(line.to_string())
}

async fn compact_if_needed(
    client: &AiClient,
    history: &[ChatMessage],
    ctx: &SessionCtx,
    system_chars: usize,
) -> Result<(Vec<ChatMessage>, Option<Vec<ChatMessage>>)> {
    let total: usize = history.iter().map(|m| m.content.chars()).sum();
    let budget = crate::budget::history_budget(ctx.limits.compact_budget_chars, system_chars);
    if total <= budget || !can_compact(history) {
        return Ok((history.to_vec(), None));
    }
    let (compacted, summary, split) = compact_now(client, history).await?;
    ctx.record_summary(&summary, split);
    Ok((compacted.clone(), Some(compacted)))
}

/// True once there is anything to fold: an explicit compact folds the whole
/// conversation, so only an empty history is refused.
pub(crate) fn can_compact(history: &[ChatMessage]) -> bool {
    !history.is_empty()
}

/// Fold everything but the last few turns into a summary, unconditionally.
/// Returns the folded history plus the summary and split for the transcript.
/// An explicit compact always folds, so a history no longer than the tail
/// comes back as the summary plus the tail; only an empty conversation is
/// refused.
pub(crate) async fn compact_now(
    client: &AiClient,
    history: &[ChatMessage],
) -> Result<(Vec<ChatMessage>, String, usize)> {
    if history.is_empty() {
        bail!("nothing to compact yet");
    }
    let split = history.len().saturating_sub(COMPACT_KEEP_TAIL);
    let head = if split == 0 {
        history
    } else {
        &history[..split]
    };
    let summary = summarize(client, head).await?;
    let mut compacted = Vec::with_capacity(COMPACT_KEEP_TAIL + 1);
    compacted.push(ChatMessage {
        role: "assistant".into(),
        content: format!("Summary of earlier conversation:\n{summary}").into(),
    });
    compacted.extend(history[split..].iter().cloned());
    Ok((compacted, summary, split))
}

async fn run_compact(args: &ChatArgs, client: &AiClient) -> Result<()> {
    let history = read_history(args)?;
    let (compacted, summary, folded) = compact_now(client, &history).await?;
    if crate::json_mode() {
        println!(
            "{}",
            json!({
                "ok": true,
                "summary": summary,
                "folded": folded,
                "messages": compacted
                    .iter()
                    .map(|m| json!({ "role": m.role, "content": m.content }))
                    .collect::<Vec<_>>(),
            })
        );
    } else {
        println!("{summary}");
    }
    Ok(())
}

async fn summarize(client: &AiClient, head: &[ChatMessage]) -> Result<String> {
    let mut transcript = String::new();
    for m in head {
        transcript.push_str(&m.role);
        transcript.push_str(": ");
        transcript.push_str(&m.content.text());
        transcript.push_str("\n\n");
    }
    let messages = vec![
        ChatMessage {
            role: "system".into(),
            content: COMPACT_PROMPT.into(),
        },
        ChatMessage {
            role: "user".into(),
            content: transcript.into(),
        },
    ];
    client.complete_messages(&messages, 0.2).await
}

fn round_usage(before: UsageSnapshot, after: UsageSnapshot) -> Option<EventUsage> {
    let prompt_tokens = after.prompt_tokens.saturating_sub(before.prompt_tokens);
    let completion_tokens = after
        .completion_tokens
        .saturating_sub(before.completion_tokens);
    let reported = after
        .reported_cost_requests
        .saturating_sub(before.reported_cost_requests);
    (prompt_tokens > 0 || completion_tokens > 0).then_some(EventUsage {
        prompt_tokens,
        completion_tokens,
        cached_tokens: after.cached_tokens.saturating_sub(before.cached_tokens),
        cost_usd: (reported > 0)
            .then(|| (after.reported_cost_usd - before.reported_cost_usd).max(0.0)),
        estimated: after.estimated_requests > before.estimated_requests,
    })
}

fn is_tool_unsupported(e: &anyhow::Error) -> bool {
    // A degenerate reply is never "the model rejected tools", even if the
    // message text happens to mention a tool or function.
    if e.downcast_ref::<DegenerateOutput>().is_some() {
        return false;
    }
    let text = format!("{e:#}").to_lowercase();
    text.contains("tool") || text.contains("function")
}

fn tool_defs(allow_edits: bool, has_approver: bool) -> Vec<Value> {
    let mut tools = vec![
        json!({
            "type": "function",
            "function": {
                "name": "read_file",
                "description": "Read a file, with line numbers. Optionally a line range. Document formats (PDF, Word, PowerPoint, Excel, OpenDocument, EPUB, RTF) are converted to Markdown, so their line numbers do not map to bytes on disk. Image files (PNG, JPEG, GIF, WebP) are not readable here: an image the user mentioned with @path is already attached to the conversation as an image part, so look at it instead of calling this tool. Paths outside the repository (absolute, or starting with ~) are allowed but the user is asked to approve each one, so prefer repo-relative paths.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "path": { "type": "string", "description": "Repo-relative path, or an absolute/~ path outside the repo (needs approval)" },
                        "start_line": { "type": "integer", "description": "First line, 1-based (optional)" },
                        "end_line": { "type": "integer", "description": "Last line, inclusive (optional)" }
                    },
                    "required": ["path"]
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "list_files",
                "description": "List the entries of a directory. Directories end with '/'. Paths outside the repository are allowed but the user is asked to approve each one.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "dir": { "type": "string", "description": "Repo-relative directory, or an absolute/~ path outside the repo (needs approval); omit for the root" }
                    }
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "explore",
                "description": "Run several lookups in ONE round instead of one per round. Every step runs in parallel and all results come back together, labelled. Each model round-trip costs seconds while these lookups take microseconds, so reaching for this instead of a lone read or search is the single biggest thing you can do to answer faster. Use it whenever you need more than one thing before you can act: the file plus the two it references, a search plus the file it will point at, several searches for the same concept. Steps may mix tools freely. Only lookups inside the repository run here; anything else comes back marked, and you call that tool on its own.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "steps": {
                            "type": "array",
                            "description": "Lookups to run together, in the order you want them reported",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "tool": {
                                        "type": "string",
                                        "enum": ["read_file", "search_files", "find_files", "list_files", "recall", "read_skill"],
                                        "description": "Which lookup to run"
                                    },
                                    "args": {
                                        "type": "object",
                                        "description": "That tool's own arguments, exactly as you would send them on their own"
                                    }
                                },
                                "required": ["tool", "args"]
                            }
                        }
                    },
                    "required": ["steps"]
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "search_files",
                "description": "Content search across repository files. The query is a regex, falling back to a literal match when it is not valid regex. Matching is smart-case: an all-lowercase query ignores case, a query with an uppercase letter is matched exactly. Results are grouped by file, with surrounding lines and a '>' on each matching line, so you usually do not need to read the file afterwards. At most 3 matches per file, so hits spread across the repository. A directory that does not exist is not an error: the whole repository is searched instead.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": { "type": "string", "description": "Text or regex to search for" },
                        "dir": { "type": "string", "description": "Repo-relative directory to search under, or a single file to search within (optional)" }
                    },
                    "required": ["query"]
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "find_files",
                "description": "Find files by name or glob, e.g. `chat.rs`, `*.rs`, `crates/*/src/**/*.rs`. A bare name matches at any depth. Use this before guessing a path, and whenever a read or list reports that a path does not exist.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "pattern": { "type": "string", "description": "File name or glob pattern" },
                        "dir": { "type": "string", "description": "Repo-relative directory to search under (optional)" }
                    },
                    "required": ["pattern"]
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "remember",
                "description": "Save a durable fact to memory so it survives across sessions. Use for lasting project facts, conventions, or user preferences the model should recall later. `title` creates a named memory block; without it the note is appended to project memory (ASTER.md).",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "note": { "type": "string", "description": "The fact to remember, stated plainly" },
                        "title": { "type": "string", "description": "Optional short name for a dedicated memory block" }
                    },
                    "required": ["note"]
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "recall",
                "description": "Read a memory block's full contents by name. The system prompt lists recallable memory as name and description only; call this to load the full body of a block before relying on it. With no `name` it lists every block as it stands now, which the system prompt's list does not: that was taken when the session started and misses anything written since, by you or by another session. Pass `query` to list only the blocks mentioning something. List before writing a memory that might already exist.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "name": { "type": "string", "description": "The memory block name, as listed under Recallable memory" },
                        "query": { "type": "string", "description": "List only blocks whose name or description mentions this (ignored when `name` is given)" }
                    }
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "chat_history",
                "description": "Read this project's saved chats, this one included. With no arguments it lists them newest first, each with its id, title, turn count and start time. Pass `id` to read one back, and `id: \"current\"` for the chat you are in, which is how you answer what was said earlier in it or before a compaction folded it away. Pass `query` to list only the chats that mention something. You get the messages, not the tool output they produced.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "id": { "type": "string", "description": "A chat id from the list, or \"current\" for this chat" },
                        "query": { "type": "string", "description": "List only chats mentioning this (ignored when `id` is given)" },
                        "limit": { "type": "integer", "description": "How many chats to list (default 20)" },
                        "all": { "type": "boolean", "description": "List chats from every project, not just this repo" }
                    }
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "forget",
                "description": "Delete a memory block by name when the user says a remembered fact is wrong or no longer wanted. This removes the block outright; never overwrite a block with a placeholder to retire it.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "name": { "type": "string", "description": "The memory block name, as listed under Recallable memory" }
                    },
                    "required": ["name"]
                }
            }
        }),
        json!({
            "type": "function",
            "function": {
                "name": "read_skill",
                "description": "Load a skill's full instructions by name. The system prompt lists skills as name and description only; call this to read a skill's body before following it, once a request matches its description.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "name": { "type": "string", "description": "The skill name, as listed under Skills" }
                    },
                    "required": ["name"]
                }
            }
        }),
    ];
    if has_approver {
        tools.push(json!({
            "type": "function",
            "function": {
                "name": "update_plan",
                "description": "Update the execution plan state. Accepts a list of step objects with `label` and `status` fields. Status must be one of: pending, in_progress, done, skipped, blocked. The plan is rendered as a progress strip in the UI. Use it for work with several distinct steps: lay the steps out before starting, mark each in_progress as you begin it and done as it lands, and keep the list current through the end of the work.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "steps": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "label": { "type": "string", "description": "Short label for this step" },
                                    "status": { "type": "string", "description": "One of: pending, in_progress, done, skipped, blocked" }
                                },
                                "required": ["label", "status"]
                            },
                            "description": "All plan steps with their current statuses"
                        }
                    },
                    "required": ["steps"]
                }
            }
        }));
        tools.push(json!({
            "type": "function",
            "function": {
                "name": "ask_user",
                "description": "Ask the user a structured question with a set of options. Only for decisions that are genuinely the user's to make and that you cannot resolve from the request, the code, or sensible defaults. If the user's message already implies the answer, act on it; never ask how to do the thing they just asked for. Not for yes/no approval. The user can pick an option or write their own.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "header": { "type": "string", "description": "One-line title for the question (optional)" },
                        "question": { "type": "string", "description": "The question to ask" },
                        "options": {
                            "type": "array",
                            "items": { "type": "string" },
                            "description": "2-4 short answer choices for the user (optional)"
                        }
                    },
                    "required": ["question"]
                }
            }
        }));
    }
    if allow_edits {
        tools.push(json!({
            "type": "function",
            "function": {
                "name": "edit_file",
                "description": "Replace text in a file, or create a new one. `search` must be copied verbatim from the file and match exactly once; include surrounding lines to disambiguate. Omit `search` to create a new file at `path` with `replace` as its whole contents; missing parent directories are created. Paths outside the repository (absolute, or starting with ~) are allowed but the user is asked to approve each directory, so prefer repo-relative paths.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "path": { "type": "string", "description": "File path, repo-relative or absolute" },
                        "search": { "type": "string", "description": "Exact existing text to replace; omit or leave empty to create a new file" },
                        "replace": { "type": "string", "description": "Replacement text, or the new file's contents" }
                    },
                    "required": ["path", "replace"]
                }
            }
        }));
        tools.push(json!({
            "type": "function",
            "function": {
                "name": "ast_edit",
                "description": "Apply a structural rewrite across many files at once: every match of an ast-grep `pattern` is replaced by `rewrite` (use `$$$VAR` metavariables to capture and reuse). Prefer this over repeated edit_file calls when the same change applies in many places. Returns the changed files and a diff. Requires Allow edits.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "pattern": { "type": "string", "description": "ast-grep pattern to match, e.g. dbg!($X)" },
                        "rewrite": { "type": "string", "description": "Replacement text; metavariables like $X carry matched text over" },
                        "language": { "type": "string", "description": "Restrict to one language: rust, python, javascript, typescript, tsx, go, java, c, cpp (optional)" }
                    },
                    "required": ["pattern", "rewrite"]
                }
            }
        }));
    }
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "ast_grep",
            "description": "Search code structurally with an ast-grep pattern (e.g. dbg!($X), fn $NAME($$$ARGS)) instead of plain text, so matches respect syntax. Returns file:line: matched text lines. Use `language` to restrict to one language; without it the language is detected per file.",
            "parameters": {
                "type": "object",
                "properties": {
                    "pattern": { "type": "string", "description": "ast-grep pattern to search for" },
                    "language": { "type": "string", "description": "Restrict to one language: rust, python, javascript, typescript, tsx, go, java, c, cpp (optional)" }
                },
                "required": ["pattern"]
            }
        }
    }));
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "security_scan",
            "description": "Run the available security analyzers (semgrep, ast-grep rules) over the repository, or a subdirectory with `path`, and return findings as severity file:line lines. Backends that are not installed are skipped and named at the end.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": { "type": "string", "description": "Subdirectory to scan instead of the whole repository (optional)" }
                }
            }
        }
    }));
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "lsp_diagnostics",
            "description": "Get the language server's errors and warnings for one file. Much faster than a full build for checking whether an edit compiles. Requires rust-analyzer (Rust) or typescript-language-server (TS/JS) to be installed.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": { "type": "string", "description": "File path, repo-relative" }
                },
                "required": ["path"]
            }
        }
    }));
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "lsp_references",
            "description": "Find where the symbol at a position is referenced, semantically (not a text search). Positions are 0-based line and character. Requires rust-analyzer or typescript-language-server.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": { "type": "string", "description": "File path, repo-relative" },
                    "line": { "type": "integer", "description": "0-based line number" },
                    "character": { "type": "integer", "description": "0-based character offset in the line" }
                },
                "required": ["path", "line", "character"]
            }
        }
    }));
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "lsp_definitions",
            "description": "Find where the symbol at a position is defined, semantically. Positions are 0-based line and character. Requires rust-analyzer or typescript-language-server.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": { "type": "string", "description": "File path, repo-relative" },
                    "line": { "type": "integer", "description": "0-based line number" },
                    "character": { "type": "integer", "description": "0-based character offset in the line" }
                },
                "required": ["path", "line", "character"]
            }
        }
    }));
    if has_approver {
        tools.push(json!({
            "type": "function",
            "function": {
                "name": "write_plan",
                "description": "Write or revise the plan document before presenting it with `exit_plan_mode`. The plan is a file kept with the session: write a skeleton of headings as soon as you know the shape of the work, fill each section in as you read the code, and edit sections with `old_str`/`new_str` as findings change your mind rather than rewriting from memory. Research first and write from what you read, never from a guess about the code. Sections, in this order: `## Context` (what is true today and why the change is needed, citing `path:line` for every claim about the code), `## Decisions` (each choice with the evidence behind it and the alternative you rejected, with a real reason), `## Changes` (per file: the functions that change, what changes in them, and the existing code you will reuse), `## Risks` (anything irreversible, outward-facing, or uncertain; say so first if it applies), `## Verification` (the exact commands to run and the result each should give). Settle real forks with the user through `ask_user` before writing them into the plan. A plan is judged on whether the user can point at the sentence they disagree with; a list of stage names is not a plan.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "content": { "type": "string", "description": "The whole plan as markdown. Replaces the current draft" },
                        "old_str": { "type": "string", "description": "Text in the current draft to replace; must appear exactly once" },
                        "new_str": { "type": "string", "description": "Replacement for `old_str`; empty to delete it" }
                    }
                }
            }
        }));
        tools.push(json!({
            "type": "function",
            "function": {
                "name": "exit_plan_mode",
                "description": "Present the plan written with `write_plan` for the user's approval and wait for their answer. Call it only once the draft is complete: every section filled from code you have read. Do not edit files or run state-changing commands until the user approves. If they reject, revise the sections they objected to with `write_plan` and present again.",
                "parameters": {
                    "type": "object",
                    "properties": {}
                }
            }
        }));
    }
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "Run a CLI command. There is no shell: `&&`, `|`, `>`, `*`, and `cd` are not interpreted, so to chain or pipe pass command:`bash` with args `[\"-lc\", \"one && two | head\"]`. Filesystem writes are restricted to the repository and temp directories (`.git` and CI workflow files are not writable), and secrets are dropped from the environment. Pass turbo:true for offline mode (no network). Pass yolo:true only when the user explicitly asks for unrestricted execution. In yolo mode (session mode or yolo:true) there is no sandbox at all: the full environment, including secrets, is inherited. Returns stdout, stderr, and exit code.",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": { "type": "string", "description": "The binary to run, e.g. `rg`, `cargo`, `npm`" },
                    "description": { "type": "string", "description": "What this command does, in five to ten words, active voice, no trailing period, e.g. `Rebuild the webview bundle`. Shown to the user in place of the command line, so do not open with `Run`/`Ran`." },
                    "args": {
                        "type": "array",
                        "items": { "type": "string" },
                        "description": "Arguments to pass to the command"
                    },
                    "turbo": { "type": "boolean", "description": "Run without network access. Use when the user asks for turbo mode or wants to work offline." },
                    "yolo": { "type": "boolean", "description": "Run without any filesystem restrictions. Only use when the user explicitly asks for yolo mode." }
                },
                "required": ["command", "description"]
            }
        }
    }));
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "open_preview",
            "description": "Open a page in the user's browser so they can see what you built, instead of only describing it. Call it once, at the end of a turn that produced something visual: a page, a component, a report, a diagram, a rendered document. `target` is either the URL of a server that is already running (`http://localhost:5173/pricing`) or a path to a file in the repo (`dist/index.html`, `docs/report.pdf`); a directory opens its index.html. A loopback URL nothing is listening on is refused, so start the dev server before you call this. Anything that is not loopback or a repo file asks the user first. Do not call it for a code-only change, and do not call it twice for the same page.",
            "parameters": {
                "type": "object",
                "properties": {
                    "target": { "type": "string", "description": "URL of a running server, or a repo-relative path to the file to open" },
                    "description": { "type": "string", "description": "What the user is about to look at, in five to ten words, e.g. `The rebuilt pricing page` " }
                },
                "required": ["target"]
            }
        }
    }));
    tools.push(json!({
        "type": "function",
        "function": {
            "name": "run_tests",
            "description": "Run the repository's test suite and get structured results: pass/fail counts, failing test names, and the output tail. Detects cargo, npm/bun/pnpm/yarn, pytest, or go from the repo's manifests. Prefer this over run_command for running tests.",
            "parameters": {
                "type": "object",
                "properties": {
                    "runner": { "type": "string", "enum": ["cargo", "npm", "bun", "pnpm", "yarn", "pytest", "go"], "description": "Force a specific runner instead of detecting one" },
                    "filter": { "type": "string", "description": "Only run tests matching this name or pattern" },
                    "turbo": { "type": "boolean", "description": "Run without network access." },
                    "yolo": { "type": "boolean", "description": "Run without any filesystem restrictions. Only use when the user explicitly asks for yolo mode." }
                }
            }
        }
    }));
    tools
}

const JSON_ESCAPES: &str = "\"\\/bfnrtu";

pub(crate) fn parse_arguments(raw: &str) -> Result<Value> {
    match serde_json::from_str(raw) {
        Ok(value) => Ok(value),
        Err(original) => {
            let repaired = repair_escapes(raw);
            if let Ok(value) = serde_json::from_str(&repaired) {
                tracing::debug!("repaired an invalid escape in tool arguments");
                return Ok(value);
            }
            // A complete value followed by junk is taken as the value: models
            // sometimes append a second object or prose after the arguments.
            if let Some(Ok(value)) = serde_json::Deserializer::from_str(&repaired)
                .into_iter::<Value>()
                .next()
            {
                tracing::debug!("dropped trailing characters from tool arguments");
                return Ok(value);
            }
            // Report what the model actually sent, not the repair attempts.
            Err(anyhow::anyhow!("{original}"))
        }
    }
}

fn repair_escapes(raw: &str) -> String {
    let mut out = String::with_capacity(raw.len() + 8);
    let mut chars = raw.chars();
    let mut in_string = false;
    while let Some(c) = chars.next() {
        if c == '"' {
            in_string = !in_string;
            out.push(c);
            continue;
        }
        if in_string && c.is_control() {
            match c {
                '\n' => out.push_str("\\n"),
                '\r' => out.push_str("\\r"),
                '\t' => out.push_str("\\t"),
                other => out.push_str(&format!("\\u{:04x}", other as u32)),
            }
            continue;
        }
        if c != '\\' || !in_string {
            out.push(c);
            continue;
        }
        match chars.next() {
            Some(next) if JSON_ESCAPES.contains(next) => {
                out.push('\\');
                out.push(next);
            }
            Some('\'') => out.push('\''),
            Some(next) => {
                out.push_str("\\\\");
                out.push(next);
            }
            None => out.push('\\'),
        }
    }
    out
}

#[allow(clippy::too_many_arguments)]
async fn exec_tool(
    repo_root: &Path,
    allow_edits: &mut bool,
    policy: &mut Policy,
    grants: &Grants,
    approver: Option<&UiSender>,
    name: &str,
    arguments: &str,
    edited: &mut Vec<String>,
    ctx: &SessionCtx,
    _events: Option<&ChatEventSink>,
) -> ToolOutput {
    // The MCP bridge is the only tool that can return an image, so it is taken
    // before the match below, which deals in text alone.
    if name == "aster_mcp" {
        return mcp_bridge(policy, approver, ctx, arguments)
            .await
            .unwrap_or_else(|e| ToolOutput::text(format!("error: {e:#}")));
    }
    let args: Value = match parse_arguments(arguments) {
        Ok(v) => v,
        Err(e) => {
            return ToolOutput::text(format!("error: tool arguments were not valid JSON: {e}"));
        }
    };
    let str_arg = |key: &str| args[key].as_str().map(str::to_string);

    let result = match name {
        "remember" => str_arg("note")
            .context("remember needs a `note`")
            .and_then(|note| remember(ctx, str_arg("title").as_deref(), &note)),
        "recall" => match str_arg("name") {
            Some(name) => recall(ctx, &name),
            None => recall_list(ctx, str_arg("query").as_deref()),
        },
        "forget" => str_arg("name")
            .context("forget needs a `name`")
            .and_then(|name| forget(ctx, &name)),
        "read_skill" => str_arg("name")
            .context("read_skill needs a `name`")
            .and_then(|name| read_skill(ctx, &name)),
        "chat_history" => Ok(crate::chat_history::chat_history(
            ctx,
            repo_root,
            crate::chat_history::HistoryArgs {
                id: args["id"].as_str(),
                query: args["query"].as_str(),
                limit: args["limit"].as_u64().map(|v| v as usize),
                all: args["all"].as_bool().unwrap_or(false),
            },
        )),
        "update_plan" => update_plan(
            ctx,
            args["steps"]
                .as_array()
                .map(|a| {
                    a.iter()
                        .map(|v| {
                            (
                                v["label"].as_str().unwrap_or("").to_string(),
                                v["status"].as_str(),
                            )
                        })
                        .collect()
                })
                .unwrap_or_default(),
        ),
        "ask_user" => match str_arg("question").context("ask_user needs a `question`") {
            Ok(question) => {
                let options: Vec<String> = args["options"]
                    .as_array()
                    .map(|a| {
                        a.iter()
                            .filter_map(|v| v.as_str().map(str::to_string))
                            .collect()
                    })
                    .unwrap_or_default();
                let header = str_arg("header").unwrap_or_default();
                ask_user(approver, &header, &question, &options).await
            }
            Err(e) => Err(e),
        },
        "write_plan" => crate::plan_file::write_plan(
            ctx,
            repo_root,
            str_arg("content").as_deref(),
            str_arg("old_str").as_deref(),
            str_arg("new_str").as_deref(),
        ),
        "exit_plan_mode" => {
            exit_plan_mode(
                approver,
                ctx,
                repo_root,
                allow_edits,
                policy,
                str_arg("plan").as_deref(),
            )
            .await
        }
        "read_file" => match str_arg("path").context("read_file needs a `path`") {
            Ok(path) if !edits::exists_anywhere(repo_root, &path) => {
                return ToolOutput::text(missing_path(repo_root, &path));
            }
            Ok(path) => {
                match resolve_for_read(repo_root, policy, grants, approver, ctx, &path).await {
                    Ok(target) if crate::images::has_image_extension(&path) => {
                        match crate::images::read_image(&target) {
                            Ok(parts) => {
                                return ToolOutput {
                                    text: parts.text,
                                    images: parts.images,
                                };
                            }
                            Err(e) => Err(e),
                        }
                    }
                    Ok(target) => cached_read(
                        ctx,
                        &target,
                        args["start_line"].as_u64().map(|n| n as usize),
                        args["end_line"].as_u64().map(|n| n as usize),
                    ),
                    Err(e) => Err(e),
                }
            }
            Err(e) => Err(e),
        },
        "list_files" => match missing_dir(repo_root, &str_arg("dir")) {
            Some(dir) => return ToolOutput::text(missing_path(repo_root, &dir)),
            None => {
                match resolve_dir(repo_root, policy, grants, approver, ctx, &str_arg("dir")).await {
                    Ok(base) => list_files(&ctx.probe, &base),
                    Err(e) => Err(e),
                }
            }
        },
        // A directory that does not exist widens the search rather than
        // failing it: the hits are usually what the model was after anyway.
        "explore" => explore(repo_root, policy, ctx, &args).await,
        "search_files" => match str_arg("query").context("search_files needs a `query`") {
            Ok(query) => match missing_dir(repo_root, &str_arg("dir")) {
                Some(dir) => search_files(&ctx.probe, repo_root, policy, &query, repo_root)
                    .map(|hits| widened(&dir, hits)),
                None => {
                    match resolve_dir(repo_root, policy, grants, approver, ctx, &str_arg("dir"))
                        .await
                    {
                        Ok(base) => search_files(&ctx.probe, repo_root, policy, &query, &base),
                        Err(e) => Err(e),
                    }
                }
            },
            Err(e) => Err(e),
        },
        "find_files" => match str_arg("pattern").context("find_files needs a `pattern`") {
            Ok(pattern) => match missing_dir(repo_root, &str_arg("dir")) {
                Some(dir) => bash_tools::find(repo_root, repo_root, &pattern, MAX_FIND_HITS)
                    .map(|hits| widened(&dir, hits)),
                None => {
                    match resolve_dir(repo_root, policy, grants, approver, ctx, &str_arg("dir"))
                        .await
                    {
                        Ok(base) => bash_tools::find(repo_root, &base, &pattern, MAX_FIND_HITS),
                        Err(e) => Err(e),
                    }
                }
            },
            Err(e) => Err(e),
        },
        "ast_grep" => match str_arg("pattern").context("ast_grep needs a `pattern`") {
            Ok(pattern) => aster_analyzers::ast_grep_search(
                repo_root,
                &pattern,
                str_arg("language").as_deref(),
            ),
            Err(e) => Err(e),
        },
        "ast_edit" if !*allow_edits => Err(anyhow::anyhow!(
            "editing is disabled for this chat; tell the user to enable Allow edits"
        )),
        "ast_edit" => match (
            str_arg("pattern").context("ast_edit needs a `pattern`"),
            str_arg("rewrite").context("ast_edit needs a `rewrite`"),
        ) {
            (Ok(pattern), Ok(rewrite)) => {
                let plan = match aster_analyzers::ast_edit_plan(
                    repo_root,
                    &pattern,
                    &rewrite,
                    str_arg("language").as_deref(),
                ) {
                    Ok(plan) => plan,
                    Err(e) => return ToolOutput::text(format!("error: {e:#}")),
                };
                if plan.changes.is_empty() {
                    Ok("no matches; nothing changed".to_string())
                } else {
                    // Same gate per file as edit_file: policy first, one
                    // approval prompt for the whole batch when any file asks.
                    let mut needs_approval = false;
                    let mut denied = None;
                    for (file, _) in &plan.changes {
                        let relative = file.strip_prefix(repo_root).unwrap_or(file);
                        match policy.evaluate(&Action::Edit {
                            path: &relative.to_string_lossy(),
                        }) {
                            Decision::Allow => {}
                            Decision::Deny { reason } => {
                                denied = Some(reason);
                                break;
                            }
                            Decision::Prompt { .. } => needs_approval = true,
                        }
                    }
                    if let Some(reason) = denied {
                        Err(anyhow::anyhow!(
                            "edit blocked by policy: {}",
                            plan_mode_hint(reason, approver.is_some())
                        ))
                    } else if needs_approval
                        && !request_approval(
                            approver,
                            plan.preview(),
                            None,
                        )
                        .await
                        .allowed()
                    {
                        Err(anyhow::anyhow!(
                            "ast_edit needs user approval (permissions mode is `ask`); \
                             it was rejected or no interactive approver is available"
                        ))
                    } else {
                    for (file, _) in &plan.changes {
                        let path = file
                            .strip_prefix(repo_root)
                            .unwrap_or(file)
                            .to_string_lossy()
                            .into_owned();
                        if !edited.contains(&path) {
                            edited.push(path);
                        }
                    }
                    aster_analyzers::ast_edit_commit(&plan)
                    }
                }
            }
            (pattern, rewrite) => pattern.and(rewrite),
        },
        "security_scan" => {
            aster_analyzers::security_scan(
                repo_root,
                str_arg("path").as_deref().map(std::path::Path::new),
            )
        }
        "lsp_diagnostics" => match str_arg("path").context("lsp_diagnostics needs a `path`") {
            Ok(path) => crate::lsp_tools::diagnostics(repo_root, &path),
            Err(e) => Err(e),
        },
        "lsp_references" => crate::lsp_tools::nav_from_args(repo_root, &args, crate::lsp_tools::Query::References),
        "lsp_definitions" => crate::lsp_tools::nav_from_args(repo_root, &args, crate::lsp_tools::Query::Definitions),
        "edit_file" if !*allow_edits => Err(anyhow::anyhow!(
            "editing is disabled for this chat; tell the user to enable Allow edits"
        )),
        "edit_file" => edit_file(repo_root, policy, approver, ctx, &args, edited)
            .await
            .map(|done| match governing_instructions(ctx, &args) {
                // Nested instructions are advertised, not preloaded, so an edit
                // is the last point at which the rules for that directory can
                // still be raised.
                Some(path) => format!("{done}\n\n{path} sets the rules for this directory. Read it if you have not, and revisit this edit if it conflicts."),
                None => done,
            })
            .map(|done| match args["path"].as_str().and_then(|path| crate::lsp_tools::after_edit(repo_root, path)) {
                Some(problems) => format!("{done}\n\n{problems}"),
                None => done,
            }),
        "run_command" => match command_argv(&args).context(MISSING_COMMAND) {
            Ok((cmd, cmd_args)) => {
                let env = ExecEnv {
                    repo_root,
                    policy,
                    approver,
                    credentials: &ctx.credentials,
                    store: ctx.store.as_ref(),
                    yolo: ctx.yolo.load(Ordering::Relaxed),
                };
                run_command_tool(&env, &cmd, &cmd_args, run_opts(&args, ctx)).await
            }
            Err(e) => Err(e),
        },
        "open_preview" => match str_arg("target").context("open_preview needs a `target`") {
            Ok(target) => {
                crate::preview::open_preview(
                    repo_root,
                    approver,
                    ctx,
                    &target,
                    str_arg("description").as_deref(),
                )
                .await
            }
            Err(e) => Err(e),
        },
        "run_tests" => {
            let env = ExecEnv {
                repo_root,
                policy,
                approver,
                credentials: &ctx.credentials,
                store: ctx.store.as_ref(),
                yolo: ctx.yolo.load(Ordering::Relaxed),
            };
            run_tests_tool(
                &env,
                str_arg("runner").as_deref(),
                str_arg("filter").as_deref(),
                run_opts(&args, ctx),
            )
            .await
        }
        other => Err(anyhow::anyhow!(
            "unknown tool: {other}. Available tools: {}",
            tool_names(*allow_edits, approver.is_some()).join(", ")
        )),
    };
    ToolOutput::text(result.unwrap_or_else(|e| format!("error: {e:#}")))
}

/// Every image in history is re-sent and re-read each round, so only the
/// current one stays a picture; older ones become a one-line stub.
fn retire_old_images(wire: &mut [Value], keep: usize) {
    let mut seen = 0;
    for message in wire.iter_mut().rev() {
        if message["role"] != "user" || !message["content"].is_array() {
            continue;
        }
        let Some(parts) = message["content"].as_array() else {
            continue;
        };
        let Some(lead) = parts.first().and_then(|p| p["text"].as_str()) else {
            continue;
        };
        let Some(tool) = lead.strip_prefix(IMAGE_TURN_LEAD) else {
            continue;
        };
        seen += 1;
        if seen > keep {
            let tool = tool.trim_end_matches(':').to_string();
            *message = json!({
                "role": "user",
                "content": format!("[earlier image from the {tool} call above, replaced by a newer one]"),
            });
        }
    }
}

const IMAGE_TURN_LEAD: &str = "Image(s) returned by the ";

const LIVE_IMAGES: usize = 1;

fn image_turn(tool: &str, images: &[String]) -> Value {
    let mut parts = vec![json!({
        "type": "text",
        "text": format!("{IMAGE_TURN_LEAD}{tool} call above:"),
    })];
    parts.extend(images.iter().map(|url| {
        json!({
            "type": "image_url",
            "image_url": { "url": url },
        })
    }));
    json!({ "role": "user", "content": parts })
}

const PARALLEL_READ_TOOLS: [&str; 7] = [
    "read_file",
    "list_files",
    "search_files",
    "find_files",
    "recall",
    "read_skill",
    "chat_history",
];

const DEDUPED_LOOKUPS: [&str; 4] = ["list_files", "search_files", "find_files", "explore"];

fn is_repeat_lookup(ctx: &SessionCtx, name: &str, arguments: &str) -> bool {
    let Ok(mut lookups) = ctx.lookups.lock() else {
        return false;
    };
    if !DEDUPED_LOOKUPS.contains(&name) {
        lookups.clear();
        return false;
    }
    let arguments = serde_json::from_str::<Value>(arguments)
        .map(|v| canonical_json(&v))
        .unwrap_or_else(|_| arguments.to_string());
    !lookups.insert(format!("{name}:{arguments}"))
}

fn canonical_json(value: &Value) -> String {
    match value {
        Value::Object(map) => {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            let fields: Vec<String> = keys
                .into_iter()
                .map(|key| {
                    let value = map.get(key).unwrap_or(&Value::Null);
                    format!("{}:{}", Value::String(key.clone()), canonical_json(value))
                })
                .collect();
            format!("{{{}}}", fields.join(","))
        }
        Value::Array(items) => {
            let items: Vec<String> = items.iter().map(canonical_json).collect();
            format!("[{}]", items.join(","))
        }
        other => other.to_string(),
    }
}

async fn explore(
    repo_root: &Path,
    policy: &Policy,
    ctx: &SessionCtx,
    args: &Value,
) -> Result<String> {
    // A missing or empty `steps` is a recoverable argument mistake, not a tool
    // failure: answer with the shape so the model can retry instead of showing
    // the panel a hard "failed" for something it can self-correct.
    let Some(steps) = steps_array(args) else {
        return Ok(format!(
            "no `steps` array given; send the lookups you need as an array of \
             {{\"tool\": one of {}, \"args\": {{…}}}} objects, in the order you \
             want them reported",
            PARALLEL_READ_TOOLS.join(", ")
        ));
    };
    if steps.is_empty() {
        return Ok(
            "`steps` was empty; add at least one lookup, or call the single \
             lookup tool directly instead"
                .to_string(),
        );
    }
    let handles: Vec<_> = steps
        .iter()
        .map(|step| {
            let (name, step_args) = match step_args(step) {
                Value::Object(args) => {
                    let (name, args) = crate::tool_alias::rewrite(&step_tool(step), args);
                    (name, Value::Object(args))
                }
                other => (step_tool(step), other),
            };
            let arguments = step_args.to_string();
            let label = step_label(&name, &step_args);
            let repo_root = repo_root.to_path_buf();
            let policy = policy.clone();
            let ctx = ctx.clone();
            tokio::task::spawn_blocking(move || {
                let text = read_only_call(&repo_root, &policy, &ctx, &name, &arguments)
                    .unwrap_or_else(|| step_refused(&name));
                (label, text)
            })
        })
        .collect();

    let mut out = String::new();
    for (i, handle) in handles.into_iter().enumerate() {
        let (label, text) = match handle.await {
            Ok(pair) => pair,
            Err(e) => (format!("step {}", i + 1), format!("error: {e}")),
        };
        if i > 0 {
            out.push('\n');
        }
        out.push_str(&format!("[{}] {label}\n{text}\n", i + 1));
    }
    Ok(out.trim_end().to_string())
}

fn steps_array(args: &Value) -> Option<Vec<Value>> {
    match args.get("steps") {
        Some(Value::Array(steps)) => Some(steps.clone()),
        Some(Value::String(raw)) => serde_json::from_str(raw).ok(),
        _ => None,
    }
}

fn step_tool(step: &Value) -> String {
    ["tool", "name"]
        .iter()
        .find_map(|key| step.get(key).and_then(Value::as_str))
        .unwrap_or_default()
        .to_string()
}

fn step_args(step: &Value) -> Value {
    ["args", "arguments", "input", "parameters"]
        .iter()
        .find_map(|key| match step.get(key) {
            Some(Value::Object(map)) => Some(Value::Object(map.clone())),
            Some(Value::String(raw)) => serde_json::from_str(raw).ok(),
            _ => None,
        })
        .unwrap_or_else(|| json!({}))
}

fn step_refused(tool: &str) -> String {
    if tool.is_empty() {
        return format!(
            "no tool named; every step is {{\"tool\": one of {}, \"args\": {{...}}}}",
            PARALLEL_READ_TOOLS.join(", ")
        );
    }
    if !PARALLEL_READ_TOOLS.contains(&tool) {
        return format!("`{tool}` is not a lookup; call it on its own");
    }
    format!(
        "`{tool}` did not run in the batch (its `args` were missing or malformed); \
         call it on its own"
    )
}

fn step_label(tool: &str, args: &Value) -> String {
    match ["path", "query", "pattern", "dir", "name"]
        .iter()
        .find_map(|key| args.get(key).and_then(Value::as_str))
    {
        Some(detail) => format!("{tool} {detail}"),
        None => tool.to_string(),
    }
}

fn read_only_call(
    repo_root: &Path,
    policy: &Policy,
    ctx: &SessionCtx,
    name: &str,
    arguments: &str,
) -> Option<String> {
    let args: Value = serde_json::from_str(arguments).ok()?;
    let str_arg = |key: &str| args[key].as_str().map(str::to_string);
    let resolve_dir =
        |dir: &Option<String>| match dir.as_deref().map(str::trim).filter(|d| !d.is_empty()) {
            Some(dir) => resolve_in_repo(repo_root, policy, ctx, dir),
            None => Some(Ok(repo_root.to_path_buf())),
        };

    let missing =
        |key: &str| format!("`{name}` needs a `{key}` argument in `args`; call it on its own");
    let result = match name {
        "read_file" => {
            let Some(path) = str_arg("path") else {
                return Some(missing("path"));
            };
            if !edits::exists_anywhere(repo_root, &path) {
                return Some(missing_path(repo_root, &path));
            }
            if crate::images::has_image_extension(&path) {
                return None;
            }
            match resolve_in_repo(repo_root, policy, ctx, &path)? {
                Ok(target) => cached_read(
                    ctx,
                    &target,
                    args["start_line"].as_u64().map(|n| n as usize),
                    args["end_line"].as_u64().map(|n| n as usize),
                ),
                Err(e) => Err(e),
            }
        }
        "list_files" => match missing_dir(repo_root, &str_arg("dir")) {
            Some(dir) => return Some(missing_path(repo_root, &dir)),
            None => match resolve_dir(&str_arg("dir"))? {
                Ok(base) => list_files(&ctx.probe, &base),
                Err(e) => Err(e),
            },
        },
        "search_files" => {
            let Some(query) = str_arg("query") else {
                return Some(missing("query"));
            };
            match missing_dir(repo_root, &str_arg("dir")) {
                Some(dir) => search_files(&ctx.probe, repo_root, policy, &query, repo_root)
                    .map(|hits| widened(&dir, hits)),
                None => match resolve_dir(&str_arg("dir"))? {
                    Ok(base) => search_files(&ctx.probe, repo_root, policy, &query, &base),
                    Err(e) => Err(e),
                },
            }
        }
        "find_files" => {
            let Some(pattern) = str_arg("pattern") else {
                return Some(missing("pattern"));
            };
            match missing_dir(repo_root, &str_arg("dir")) {
                Some(dir) => bash_tools::find(repo_root, repo_root, &pattern, MAX_FIND_HITS)
                    .map(|hits| widened(&dir, hits)),
                None => match resolve_dir(&str_arg("dir"))? {
                    Ok(base) => bash_tools::find(repo_root, &base, &pattern, MAX_FIND_HITS),
                    Err(e) => Err(e),
                },
            }
        }
        "recall" => match str_arg("name") {
            Some(name) => recall(ctx, &name),
            None => recall_list(ctx, str_arg("query").as_deref()),
        },
        "read_skill" => match str_arg("name") {
            Some(name) => read_skill(ctx, &name),
            None => return Some(missing("name")),
        },
        "chat_history" => Ok(crate::chat_history::chat_history(
            ctx,
            repo_root,
            crate::chat_history::HistoryArgs {
                id: args["id"].as_str(),
                query: args["query"].as_str(),
                limit: args["limit"].as_u64().map(|v| v as usize),
                all: args["all"].as_bool().unwrap_or(false),
            },
        )),
        _ => return None,
    };
    Some(result.unwrap_or_else(|e| format!("error: {e:#}")))
}

/// Reading an internal skill is never shown as a step: the user sees the
/// behaviour, not the manual behind it.
fn internal_call(skills: &aster_skills::SkillSet, name: &str, arguments: &str) -> bool {
    let Ok(arguments) = serde_json::from_str::<Value>(arguments) else {
        return false;
    };
    match name {
        "read_skill" => arguments["name"]
            .as_str()
            .is_some_and(|skill| skills.is_internal(skill)),
        "explore" => steps_array(&arguments).is_some_and(|steps| {
            steps.iter().any(|step| {
                let tool = step_tool(step);
                let args = step_args(step);
                internal_call(skills, &tool, &args.to_string())
            })
        }),
        _ => false,
    }
}

pub(crate) fn tool_names(allow_edits: bool, has_approver: bool) -> Vec<String> {
    tool_defs(allow_edits, has_approver)
        .iter()
        .filter_map(|t| t["function"]["name"].as_str().map(str::to_string))
        .collect()
}

impl PlanStepStatus {
    fn parse(raw: Option<&str>) -> Result<Self> {
        match raw.unwrap_or("pending") {
            "pending" => Ok(Self::Pending),
            "in_progress" => Ok(Self::InProgress),
            "done" => Ok(Self::Done),
            "skipped" => Ok(Self::Skipped),
            "blocked" => Ok(Self::Blocked),
            other => Err(anyhow::anyhow!(
                "unknown step status `{other}`; use pending, in_progress, done, skipped, or blocked"
            )),
        }
    }

    fn glyph(self) -> &'static str {
        match self {
            Self::Pending => "◻",
            Self::InProgress => "◼",
            Self::Done => "✔",
            Self::Skipped => "⊘",
            Self::Blocked => "✖",
        }
    }
}

impl PlanState {
    fn count(&self, want: PlanStepStatus) -> usize {
        self.steps.iter().filter(|s| s.status == want).count()
    }

    fn render(&self) -> String {
        let mut parts = vec![format!("{} done", self.count(PlanStepStatus::Done))];
        if self.count(PlanStepStatus::InProgress) > 0 {
            parts.push(format!(
                "{} in progress",
                self.count(PlanStepStatus::InProgress)
            ));
        }
        parts.push(format!("{} open", self.count(PlanStepStatus::Pending)));
        for (status, label) in [
            (PlanStepStatus::Blocked, "blocked"),
            (PlanStepStatus::Skipped, "skipped"),
        ] {
            if self.count(status) > 0 {
                parts.push(format!("{} {label}", self.count(status)));
            }
        }

        let head = format!(
            "{} task{} ({})",
            self.steps.len(),
            if self.steps.len() == 1 { "" } else { "s" },
            parts.join(", ")
        );
        let rows = self
            .steps
            .iter()
            .map(|step| format!("  {} {}", step.status.glyph(), step.label))
            .collect::<Vec<_>>()
            .join("\n");
        format!("{head}\n{rows}")
    }
}

fn update_plan(ctx: &SessionCtx, steps: Vec<(String, Option<&str>)>) -> Result<String> {
    if steps.is_empty() {
        return Err(anyhow::anyhow!(
            "update_plan needs a non-empty `steps` list"
        ));
    }
    let parsed = steps
        .into_iter()
        .map(|(label, status)| {
            let label = label.trim();
            match label.is_empty() {
                true => Err(anyhow::anyhow!("every plan step needs a `label`")),
                false => Ok(PlanStep {
                    label: label.to_string(),
                    status: PlanStepStatus::parse(status)?,
                }),
            }
        })
        .collect::<Result<Vec<_>>>()?;

    let mut plan = ctx
        .plan
        .lock()
        .map_err(|_| anyhow::anyhow!("plan state lock poisoned"))?;
    plan.steps = parsed;
    Ok(format!("plan updated:\n{}", plan.render()))
}

async fn ask_user(
    approver: Option<&UiSender>,
    header: &str,
    question: &str,
    options: &[String],
) -> Result<String> {
    let Some(tx) = approver else {
        return Ok(
            "note: no interactive UI is attached, so the user cannot be asked. Pick the most reasonable option and say which you chose."
                .to_string(),
        );
    };
    // With zero or one choices there is nothing for the user to pick. Rather
    // than bouncing the decision back to the model (which tends to retry the
    // tool in a loop), commit to the single option or decline outright.
    match options.len() {
        0 => {
            return Ok("the user declined to answer; proceed with your best judgement".to_string());
        }
        1 => return Ok(format!("the user chose: {}", options[0])),
        _ => {} // fall through to the interactive path
    }

    let (respond, rx) = oneshot::channel();
    let request = UiRequest::Question(QuestionRequest {
        header: match header.trim().is_empty() {
            true => "Question".to_string(),
            false => header.trim().to_string(),
        },
        question: question.to_string(),
        options: options.to_vec(),
        respond,
    });
    if tx.send(request).await.is_err() {
        return Err(anyhow::anyhow!("the UI closed before the question was put"));
    }
    match rx.await.unwrap_or(None) {
        Some(answer) => Ok(format!("the user chose: {answer}")),
        None => Ok("the user declined to answer; proceed with your best judgement".to_string()),
    }
}

async fn exit_plan_mode(
    approver: Option<&UiSender>,
    ctx: &SessionCtx,
    repo_root: &Path,
    allow_edits: &mut bool,
    policy: &mut Policy,
    inline: Option<&str>,
) -> Result<String> {
    if ctx
        .plan
        .lock()
        .map_err(|_| anyhow::anyhow!("plan state lock poisoned"))?
        .approved
    {
        return Err(anyhow::anyhow!(
            "this plan is already approved; carry it out instead of presenting it again"
        ));
    }
    // Older callers still pass the document inline; it is saved like any draft.
    if let Some(doc) = inline.map(str::trim).filter(|d| !d.is_empty()) {
        crate::plan_file::write_plan(ctx, repo_root, Some(doc), None, None)?;
    }
    let markdown = crate::plan_file::read_plan(ctx, repo_root);
    if markdown.trim().is_empty() {
        return Err(anyhow::anyhow!(
            "there is no plan to present: draft it with `write_plan` first, from the code you have read"
        ));
    }

    let locked = !*allow_edits;
    let preview = format!("Approve this plan and start editing?\n\n{markdown}");
    if !request_plan_approval(approver, preview, Some(markdown))
        .await
        .allowed()
    {
        // An editable session has to hold too: the user said no, so the rest of
        // the turn stays read-only until they send a revision.
        *allow_edits = false;
        policy.demote(aster_policy::Mode::Plan);
        return Ok(
            "the user did not approve the plan; stay in plan mode, revise the parts they objected to with `write_plan`, and present it again"
                .to_string(),
        );
    }
    ctx.plan
        .lock()
        .map_err(|_| anyhow::anyhow!("plan state lock poisoned"))?
        .approved = true;
    // The edit tool and the policy gate separately: without the promotion the
    // mode still denies every edit and command, and the turn stalls holding an
    // approved plan it cannot act on. A turn that was already editable keeps
    // the mode it had, so approving never widens the session.
    if !locked {
        return Ok("plan approved; carry it out".to_string());
    }
    *allow_edits = true;
    policy.promote(aster_policy::Mode::Edit);
    Ok("plan approved; edit mode is now active".to_string())
}

fn missing_dir(repo_root: &Path, dir: &Option<String>) -> Option<String> {
    let dir = dir.as_deref().map(str::trim).filter(|d| !d.is_empty())?;
    (!edits::exists_anywhere(repo_root, dir)).then(|| dir.to_string())
}

fn widened(dir: &str, hits: String) -> String {
    format!("note: {dir} does not exist, so the whole repository was searched instead.\n\n{hits}")
}

fn missing_path(repo_root: &Path, path: &str) -> String {
    let nearby = bash_tools::suggest(repo_root, path, MAX_PATH_SUGGESTIONS);
    if nearby.is_empty() {
        return format!(
            "note: {path} does not exist. Call find_files with a name or glob to locate it."
        );
    }
    format!(
        "note: {path} does not exist. Nearest paths in the repository:\n{}\n\nCall find_files if none of these are the one.",
        nearby
            .iter()
            .map(|p| format!("  {p}"))
            .collect::<Vec<_>>()
            .join("\n")
    )
}

fn governing_instructions(ctx: &SessionCtx, args: &Value) -> Option<String> {
    let path = args["path"].as_str()?;
    ctx.instructions
        .nearest(Path::new(path))
        .map(|p| p.display().to_string())
}

async fn mcp_bridge(
    policy: &Policy,
    approver: Option<&UiSender>,
    ctx: &SessionCtx,
    arguments: &str,
) -> Result<ToolOutput> {
    let runtime = ctx
        .mcp
        .as_ref()
        .context("no MCP servers are connected in this session")?;
    let action = runtime.injector().route(arguments)?;
    let (tool, call_args) = match action {
        aster_mcp::BridgeAction::Search(matches) => {
            return Ok(ToolOutput::text(serde_json::to_string_pretty(
                &json!({ "matches": matches }),
            )?));
        }
        aster_mcp::BridgeAction::Describe(tool) => {
            return Ok(ToolOutput::text(serde_json::to_string_pretty(&json!({
                "id": tool.id(),
                "description": tool.description,
                "input_schema": tool.input_schema,
            }))?));
        }
        aster_mcp::BridgeAction::Execute { tool, arguments } => (tool, arguments),
    };

    let id = tool.id();
    match policy.evaluate(&Action::Exec {
        binary: "mcp",
        args: &[&id],
    }) {
        Decision::Allow => {}
        Decision::Deny { reason } => bail!("{}", plan_mode_hint(reason, approver.is_some())),
        Decision::Prompt { .. } => {
            let preview = format!("call MCP tool {id}:\n{call_args:#}");
            if !request_approval(approver, preview, None).await.allowed() {
                bail!(
                    "MCP tool `{id}` needs user approval; it was rejected or this run cannot ask"
                );
            }
        }
    }

    let result = runtime.call(&tool, &call_args).await?;
    Ok(crate::mcp::render_result(&result))
}

fn command_argv(args: &Value) -> Option<(String, Vec<String>)> {
    let (binary, tail) = raw_argv(args)?;
    // A machine with no python still has the one built into this binary; a
    // call that forgot the `aster` in front should reach it, not a dead exec.
    if matches!(binary.as_str(), "python" | "python3") && !on_path(&binary) {
        let me = std::env::current_exe().ok()?;
        let mut argv = vec!["python".to_string()];
        argv.extend(tail);
        return Some((me.to_string_lossy().into_owned(), argv));
    }
    Some((binary, tail))
}

fn raw_argv(args: &Value) -> Option<(String, Vec<String>)> {
    let tail = string_list(&args["args"]);
    match args["command"].as_str().filter(|s| !s.trim().is_empty()) {
        Some(binary) => {
            let looks_like_shell_line = tail.is_empty()
                && binary
                    .chars()
                    .any(|c| c.is_whitespace() || matches!(c, '|' | ';' | '>' | '<' | '&' | '`'));
            match looks_like_shell_line {
                true => {
                    let (sh, flag) = shell();
                    Some((sh.into(), vec![flag.into(), binary.to_string()]))
                }
                // The tool text says `bash -lc`; where only `sh` exists the
                // line still has to run rather than fail on the binary.
                false if binary == "bash" && !on_path("bash") => {
                    let (sh, flag) = shell();
                    let line = tail.iter().skip_while(|a| a.starts_with('-')).cloned();
                    Some((
                        sh.into(),
                        std::iter::once(flag.to_string()).chain(line).collect(),
                    ))
                }
                false => Some((binary.to_string(), tail)),
            }
        }
        None => {
            let mut argv = string_list(&args["command"]).into_iter().chain(tail);
            let binary = argv.find(|a| !a.trim().is_empty())?;
            match binary.starts_with('-') {
                true => Some((
                    shell().0.into(),
                    std::iter::once(binary).chain(argv).collect(),
                )),
                false => Some((binary, argv.collect())),
            }
        }
    }
}

fn string_list(value: &Value) -> Vec<String> {
    value
        .as_array()
        .map(|a| {
            a.iter()
                .map(|v| v.as_str().unwrap_or("").to_string())
                .collect()
        })
        .unwrap_or_default()
}

fn run_opts(args: &Value, ctx: &SessionCtx) -> RunOpts {
    RunOpts {
        turbo: args["turbo"].as_bool().unwrap_or(false),
        yolo: args["yolo"].as_bool().unwrap_or(false) || ctx.yolo.load(Ordering::Relaxed),
        timeout_secs: ctx.limits.command_timeout_secs as u64,
    }
}

struct ExecEnv<'a> {
    repo_root: &'a Path,
    policy: &'a Policy,
    approver: Option<&'a UiSender>,
    credentials: &'a aster_policy::CommandGrants,
    store: Option<&'a Store>,
    yolo: bool,
}

#[derive(Clone, Copy)]
struct RunOpts {
    turbo: bool,
    yolo: bool,
    timeout_secs: u64,
}

fn plan_mode_hint(reason: String, can_ask: bool) -> String {
    match can_ask && reason.contains("mode is `plan`") {
        true => format!(
            "{reason}. Present your plan with `exit_plan_mode` and ask the user \
             to approve leaving plan mode; do not retry this call until they do"
        ),
        false => reason,
    }
}

async fn authorize_exec(env: &ExecEnv<'_>, binary: &str, args: &[String]) -> Result<()> {
    let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect();
    match env.policy.evaluate(&Action::Exec {
        binary,
        args: &arg_refs,
    }) {
        Decision::Allow => Ok::<(), anyhow::Error>(()),
        Decision::Deny { reason } => bail!("{}", plan_mode_hint(reason, env.approver.is_some())),
        Decision::Prompt { preview } => {
            if !request_approval(env.approver, preview, None)
                .await
                .allowed()
            {
                bail!(
                    "command `{binary}` needs user approval; it was rejected or this run cannot ask"
                );
            }
            Ok(())
        }
    }?;
    authorize_credentials(env, binary, args).await
}

async fn authorize_credentials(env: &ExecEnv<'_>, binary: &str, args: &[String]) -> Result<()> {
    if env.yolo {
        return Ok(());
    }
    let command = aster_sandbox::command_name(binary);
    for dir in aster_sandbox::credentials_for(binary, args) {
        if env.credentials.allows(&command, &dir) {
            continue;
        }
        let preview = format!(
            "`{command}` needs to read credentials outside the repository:\n  {}",
            crate::edits::display_home(&dir)
        );
        match request_approval(env.approver, preview, Some(dir.clone())).await {
            Answer::No => bail!(
                "`{command}` needs to read {} and was not allowed to; it was rejected, \
                 or this run has no way to ask. Preauthorize it with \
                 `permissions.allow_credentials: [\"{command}:{}\"]` in aster.yaml",
                crate::edits::display_home(&dir),
                crate::edits::display_home(&dir),
            ),
            Answer::Yes => env.credentials.grant(&command, dir),
            Answer::Always => {
                env.credentials.grant(&command, dir.clone());
                if let Some(store) = env.store
                    && let Err(e) = store
                        .credential_grants(env.repo_root)
                        .add(Path::new(&format!("{command}\t{}", dir.display())))
                {
                    tracing::warn!("could not persist the credential grant: {e:#}");
                }
            }
        }
    }
    Ok(())
}

async fn run_raw(
    env: &ExecEnv<'_>,
    binary: &str,
    args: &[String],
    opts: RunOpts,
) -> Result<aster_sandbox::CommandOutput> {
    if opts.yolo {
        return aster_sandbox::run_unsandboxed(env.repo_root, binary, args, opts.timeout_secs)
            .await;
    }
    let profile = aster_sandbox::SandboxProfile::new(env.repo_root)
        .timeout(opts.timeout_secs)
        .network(!opts.turbo)
        .allow_credentials(
            env.credentials
                .dirs_for(&aster_sandbox::command_name(binary)),
        );
    let config = aster_sandbox::SandboxConfig::new(profile);
    let output = aster_sandbox::run_command(&config, binary, args).await?;
    // The sandbox refusing a path or the network is not the command failing.
    // Ask once to rerun it outside, rather than making the model route around
    // the denial for a turn.
    if sandbox_denial(&output) {
        let preview = format!(
            "`{binary}` failed inside the sandbox, likely on a blocked path or \
             no network. Run it without the sandbox?"
        );
        if request_approval(env.approver, preview, None)
            .await
            .allowed()
        {
            return aster_sandbox::run_unsandboxed(env.repo_root, binary, args, opts.timeout_secs)
                .await;
        }
    }
    Ok(output)
}

/// A failed command whose output looks like the sandbox refusing a write path
/// or the network, not the command itself failing. ssh's "Permission denied
/// (publickey)" is auth, not the sandbox.
fn sandbox_denial(output: &aster_sandbox::CommandOutput) -> bool {
    if output.exit_code == Some(0) {
        return false;
    }
    let lower = format!("{}\n{}", output.stdout, output.stderr).to_lowercase();
    // Only match the sandbox's own denial text, not generic permission
    // errors, which usually mean the command itself was refused (e.g. by a
    // remote host or a file the user cannot access).
    [
        "seccomp",
        "landlock",
        "sandbox denied",
        "sandboxed: denied",
        "operation not permitted by sandbox",
    ]
    .iter()
    .any(|marker| lower.contains(marker))
}

/// The sandbox reports a missing program as a permission error, which sends
/// the model hunting for grants; name the real problem and the way round it.
fn missing_binary(binary: &str) -> String {
    let hint = match binary {
        "python" | "python3" | "pip" | "pip3" => {
            " Python here is `aster python <script>` or `aster python -c \"...\"`."
        }
        "bash" => " The shell is `sh -c \"...\"`.",
        _ => "",
    };
    format!("there is no `{binary}` on this machine's PATH.{hint}")
}

fn command_coaching(
    binary: &str,
    output: &aster_sandbox::CommandOutput,
    sandboxed: bool,
) -> Vec<String> {
    let mut notes = Vec::new();
    let failed = output.exit_code != Some(0);
    let combined = format!("{}\n{}", output.stdout, output.stderr);
    let shell_line = matches!(binary, "bash" | "sh" | "zsh" | "fish");

    // A tool that prints `error:` and exits 0 (asterctl does) is reporting,
    // not building, so the pipe note would only mislead.
    if let Some(line) = first_error_line(&combined).filter(|_| shell_line || failed) {
        if output.exit_code == Some(0) {
            notes.push(format!(
                "note: exit code 0 comes from the last command in the pipe; the \
                 build or test run itself failed. First error: {line}. The \
                 `build-triage` skill (read_skill) has the full protocol."
            ));
        } else {
            notes.push(format!(
                "note: first error in the output: {line}. Fix the first error \
                 before any later one; the `build-triage` skill (read_skill) \
                 has the full protocol."
            ));
        }
    }

    let lower = combined.to_lowercase();
    if failed
        && [
            "unauthorized",
            "authentication failed",
            "invalid credentials",
            "not logged in",
            "please log in",
            "token expired",
        ]
        .iter()
        .any(|marker| lower.contains(marker))
    {
        notes.push(
            "note: this is an auth failure; retrying the same command will not \
             help. Tell the user what to log in to, and continue with what does \
             not need it."
                .into(),
        );
    }

    let denial = [
        "permission denied",
        "permissiondenied",
        "eperm",
        // What macOS actually prints when Seatbelt refuses a read.
        "operation not permitted",
    ]
        .iter()
        .any(|marker| lower.contains(marker))
        // ssh's "Permission denied (publickey)" is auth, not the sandbox.
        && !lower.contains("publickey");
    if sandboxed && failed && denial {
        notes.push(
            "note: this command ran inside the sandbox, which only allows writes \
             to the repository, temp directories, and build caches. A permission \
             error here usually means the sandbox blocked a path, not that the \
             tool or network is broken. Prefer a path the sandbox allows; if the \
             task truly needs the blocked path, say so to the user instead of \
             switching tools."
                .into(),
        );
    }
    notes
}

fn first_error_line(output: &str) -> Option<String> {
    output
        .lines()
        .find(|line| {
            let trimmed = line.trim_start();
            trimmed.starts_with("error:")
                || trimmed.starts_with("error[")
                || trimmed.contains("error TS")
                || trimmed.starts_with("FAILED")
                || trimmed.contains("panicked at")
        })
        .map(|line| line.trim().to_string())
}

fn render_timeout(output: &aster_sandbox::CommandOutput, timeout_secs: u64) -> String {
    let mut result = format!("error: command timed out after {timeout_secs}s\n");
    if output.stdout.is_empty() && output.stderr.is_empty() {
        result.push_str("(no output before the timeout)\n");
    } else {
        result.push_str("output before the timeout:\n");
        if !output.stdout.is_empty() {
            result.push_str("stdout:\n");
            result.push_str(&truncate_head(&output.stdout, MAX_STREAM_CHARS));
            result.push('\n');
        }
        if !output.stderr.is_empty() {
            result.push_str("stderr:\n");
            result.push_str(&truncate_head(&output.stderr, MAX_STREAM_CHARS));
            result.push('\n');
        }
    }
    result.push_str(
        "Do NOT re-run this command with a longer timeout. Kill any leftover \
         processes first, then run a narrower or faster variant (scope it to \
         one target, bound its output, or skip the slow step). The \
         `build-triage` skill (read_skill) has the full protocol.",
    );
    result
}

async fn run_command_tool(
    env: &ExecEnv<'_>,
    binary: &str,
    args: &[String],
    opts: RunOpts,
) -> Result<String> {
    if !binary.contains('/') && !on_path(binary) {
        anyhow::bail!("{}", missing_binary(binary));
    }
    authorize_exec(env, binary, args).await?;
    let output = run_raw(env, binary, args, opts).await?;
    if output.timed_out {
        return Ok(render_timeout(&output, opts.timeout_secs));
    }
    let exit_code = output.exit_code.unwrap_or(-1);
    let mut result = String::new();
    if !output.stdout.is_empty() {
        result.push_str("stdout:\n");
        result.push_str(&truncate(&output.stdout, MAX_STREAM_CHARS));
    }
    if !output.stderr.is_empty() {
        if !result.is_empty() {
            result.push('\n');
        }
        result.push_str("stderr:\n");
        // Compilers and test runners put the verdict last, so keep the tail.
        result.push_str(&truncate_head(&output.stderr, MAX_STREAM_CHARS));
    }
    result.push_str(&format!("\nexit code: {exit_code}"));
    for note in command_coaching(binary, &output, !opts.yolo) {
        result.push('\n');
        result.push_str(&note);
    }
    if result.is_empty() {
        return Ok("(no output)".into());
    }
    Ok(result)
}

async fn run_tests_tool(
    env: &ExecEnv<'_>,
    runner: Option<&str>,
    filter: Option<&str>,
    opts: RunOpts,
) -> Result<String> {
    let cmd = crate::test_runner::detect(env.repo_root, runner, filter)?;
    authorize_exec(env, &cmd.binary, &cmd.args).await?;
    let output = run_raw(env, &cmd.binary, &cmd.args, opts).await?;
    if output.timed_out {
        return Ok(render_timeout(&output, opts.timeout_secs));
    }
    let result = crate::test_runner::parse(
        cmd.runner,
        &output.stdout,
        &output.stderr,
        output.exit_code.unwrap_or(-1),
    );
    serde_json::to_string_pretty(&result).context("serializing test results")
}

fn remember(ctx: &SessionCtx, title: Option<&str>, note: &str) -> Result<String> {
    let store = ctx
        .store
        .as_ref()
        .context("memory is unavailable; no store is open")?;
    let memory = store.memory();
    let source_session = ctx.session_id();
    match title.map(str::trim).filter(|t| !t.is_empty()) {
        Some(title) => {
            let path = match source_session {
                Some(sid) => memory.remember_sourced(title, note, note, &sid)?,
                None => memory.remember(title, note, note)?,
            };
            let _ = path;
            let same = memory.near_duplicates(title, note);
            if same.is_empty() {
                return Ok(format!("remembered under \"{title}\""));
            }
            Ok(format!(
                "remembered under \"{title}\". {} already says something similar; \
                 read it and, if it is the same fact, fold this into it and forget the other.",
                same.join(" and ")
            ))
        }
        None => {
            let fact = note.trim().chars().take(500).collect::<String>();
            match source_session {
                Some(sid) => memory.append_project_sourced(&fact, &sid)?,
                None => memory.append_project(&fact)?,
            }
            Ok("remembered in project memory".to_string())
        }
    }
}

fn recall(ctx: &SessionCtx, name: &str) -> Result<String> {
    let store = ctx
        .store
        .as_ref()
        .context("memory is unavailable; no store is open")?;
    store.memory().read_block(name)
}

/// Memory as it stands now. The index in the system prompt was taken when the
/// session started, so anything written since, here or in another session, is
/// missing from it, which is how the same fact ends up stored twice.
fn recall_list(ctx: &SessionCtx, query: Option<&str>) -> Result<String> {
    let store = ctx
        .store
        .as_ref()
        .context("memory is unavailable; no store is open")?;
    let needle = query.map(str::to_lowercase);
    let blocks: Vec<String> = store
        .memory()
        .list_recent()?
        .into_iter()
        .filter(|b| {
            needle.as_ref().is_none_or(|q| {
                b.name.to_lowercase().contains(q) || b.description.to_lowercase().contains(q)
            })
        })
        .map(|b| {
            if b.description.is_empty() {
                format!("- {}", b.name)
            } else {
                format!("- {} — {}", b.name, b.description)
            }
        })
        .collect();
    if blocks.is_empty() {
        return Ok(match query {
            Some(q) => format!("no memory block mentions \"{q}\""),
            None => "no memory blocks yet".to_string(),
        });
    }
    Ok(format!(
        "{} memory blocks; recall(name) reads one in full:\n{}",
        blocks.len(),
        blocks.join("\n")
    ))
}

fn forget(ctx: &SessionCtx, name: &str) -> Result<String> {
    let store = ctx
        .store
        .as_ref()
        .context("memory is unavailable; no store is open")?;
    if store.memory().forget(name)? {
        Ok(format!("forgot \"{name}\""))
    } else {
        anyhow::bail!("no memory block named {name:?}; see Recallable memory for the list")
    }
}

fn read_skill(ctx: &SessionCtx, name: &str) -> Result<String> {
    let skill = ctx
        .skills
        .get(name)
        .with_context(|| format!("no skill named {name:?}; check the Skills list"))?;
    if skill.always {
        return Ok(format!(
            "{name} is already in your system prompt in full, under \"Skill: {name}\". \
             Read it there rather than loading it again."
        ));
    }
    skill.load_body()
}

/// Seed the session's grants from `permissions.additional_directories`.
/// Unreadable entries are dropped rather than failing the run: a stale entry in
/// aster.yaml should not stop the agent from starting.
pub(crate) fn configured_grants(
    permissions: &aster_policy::PermissionsConfig,
    repo_root: &Path,
) -> Grants {
    let configured = permissions
        .additional_directories
        .iter()
        .filter_map(|dir| edits::expand_home(dir).canonicalize().ok());
    let persisted = crate::persist::store()
        .map(|store| store.grants(repo_root).load())
        .unwrap_or_default();
    Grants::new(configured.chain(persisted))
}

/// Seed credential grants from `permissions.allow_credentials` (`<command>:<dir>`)
/// and the persisted store (`<command>\t<dir>`). A malformed entry is dropped, so
/// a typo in aster.yaml cannot stop the agent from starting.
pub(crate) fn configured_credentials(
    permissions: &aster_policy::PermissionsConfig,
    repo_root: &Path,
) -> aster_policy::CommandGrants {
    let configured = permissions
        .allow_credentials
        .iter()
        .filter_map(|entry| entry.split_once(':'))
        .map(|(command, dir)| (command.trim().to_string(), edits::expand_home(dir.trim())));
    let persisted = crate::persist::store()
        .map(|store| store.credential_grants(repo_root).load())
        .unwrap_or_default()
        .into_iter()
        .filter_map(|entry| {
            let text = entry.to_string_lossy().into_owned();
            let (command, dir) = text.split_once('\t')?;
            Some((command.to_string(), PathBuf::from(dir)))
        });
    aster_policy::CommandGrants::new(configured.chain(persisted))
}

fn grant_root(resolved: &Path) -> PathBuf {
    if resolved.is_dir() {
        return resolved.to_path_buf();
    }
    resolved.parent().unwrap_or(resolved).to_path_buf()
}

fn resolve_in_repo(
    repo_root: &Path,
    policy: &Policy,
    ctx: &SessionCtx,
    path: &str,
) -> Option<Result<PathBuf>> {
    let (resolved, scope) = match edits::resolve_anywhere(repo_root, path) {
        Ok(pair) => pair,
        Err(e) => return Some(Err(e)),
    };
    if !matches!(scope, edits::Scope::InRepo) {
        // Yolo has already dropped the sandbox, so it drops this gate too.
        return ctx.yolo.load(Ordering::Relaxed).then_some(Ok(resolved));
    }
    let root = repo_root.canonicalize().unwrap_or_default();
    let relative = resolved.strip_prefix(&root).unwrap_or(&resolved);
    if let Decision::Deny { reason } = policy.evaluate(&Action::Read {
        path: &relative.to_string_lossy(),
    }) {
        return Some(Err(anyhow::anyhow!("{reason}")));
    }
    Some(Ok(resolved))
}

async fn resolve_for_read(
    repo_root: &Path,
    policy: &Policy,
    grants: &Grants,
    approver: Option<&UiSender>,
    ctx: &SessionCtx,
    path: &str,
) -> Result<PathBuf> {
    if let Some(result) = resolve_in_repo(repo_root, policy, ctx, path) {
        return result;
    }
    let (resolved, scope) = edits::resolve_anywhere(repo_root, path)?;
    match scope {
        edits::Scope::InRepo => {}
        edits::Scope::Outside if !grants.allows(&resolved) => {
            // Grant the directory, not the file, so the rest of the session can
            // read its siblings without another prompt.
            let root = grant_root(&resolved);
            let preview = format!("read outside the repository:\n  {}", resolved.display());
            match request_approval(approver, preview, Some(root.clone())).await {
                Answer::No => bail!(
                    "{} is outside the repository and needs the user's approval; \
                     it was rejected or this run has no way to ask",
                    resolved.display()
                ),
                Answer::Yes => grants.grant(root),
                Answer::Always => {
                    grants.grant(root.clone());
                    if let Some(store) = &ctx.store
                        && let Err(e) = store.grants(repo_root).add(&root)
                    {
                        tracing::warn!("could not persist the grant for {}: {e:#}", root.display());
                    }
                }
            }
        }
        edits::Scope::Outside => {}
    }
    Ok(resolved)
}

async fn resolve_dir(
    repo_root: &Path,
    policy: &Policy,
    grants: &Grants,
    approver: Option<&UiSender>,
    ctx: &SessionCtx,
    dir: &Option<String>,
) -> Result<PathBuf> {
    match dir.as_deref().map(str::trim).filter(|d| !d.is_empty()) {
        Some(dir) => resolve_for_read(repo_root, policy, grants, approver, ctx, dir).await,
        None => Ok(repo_root.to_path_buf()),
    }
}

fn cached_read(
    ctx: &SessionCtx,
    target: &Path,
    start: Option<usize>,
    end: Option<usize>,
) -> Result<String> {
    let modified = fs::metadata(target).and_then(|m| m.modified()).ok();
    let key = format!("{}:{start:?}:{end:?}", target.display());
    let seen = ctx
        .reads
        .lock()
        .ok()
        .and_then(|reads| reads.get(&key).copied());
    // Only a byte-identical situation is skipped: no mtime, or a changed one,
    // reads for real.
    if let Some(previous) = seen
        && previous.is_some()
        && previous == modified
    {
        return Ok(format!(
            "[unchanged since you read it earlier in this turn — scroll up for {}]",
            target.display()
        ));
    }
    let body = read_numbered(target, start, end)?;
    if let Ok(mut reads) = ctx.reads.lock() {
        reads.insert(key, modified);
    }
    Ok(body)
}

fn read_numbered(target: &Path, start: Option<usize>, end: Option<usize>) -> Result<String> {
    let content = crate::images::read_text_or_document(target)?;
    let lines: Vec<&str> = content.lines().collect();
    let from = start.unwrap_or(1).max(1) - 1;
    // An open-ended read is windowed rather than truncated mid-file, so the
    // model knows exactly where to resume instead of re-reading blindly.
    let requested_end = end.unwrap_or(lines.len());
    let to = requested_end.min(lines.len()).min(from + READ_WINDOW_LINES);
    if from >= to {
        bail!("empty range: the file has {} lines", lines.len());
    }
    let mut body = lines[from..to]
        .iter()
        .enumerate()
        .map(|(i, l)| format!("{:>5} | {l}", from + i + 1))
        .collect::<Vec<_>>()
        .join("\n");
    if to < lines.len() {
        body.push_str(&format!(
            "\n\n[showing lines {}-{to} of {}; call read_file again with start_line={} for more]",
            from + 1,
            lines.len(),
            to + 1,
        ));
    }
    Ok(body)
}

fn list_files(probe: &bash_tools::ToolProbe, base: &Path) -> Result<String> {
    let entries = bash_tools::list(probe, base, MAX_LIST_ENTRIES)?;
    Ok(match entries.trim().is_empty() {
        true => "directory exists but is empty".into(),
        false => entries,
    })
}

fn search_files(
    probe: &bash_tools::ToolProbe,
    repo_root: &Path,
    policy: &Policy,
    query: &str,
    base: &Path,
) -> Result<String> {
    let hits = bash_tools::search(probe, repo_root, base, query, MAX_SEARCH_HITS)?;
    let filtered: Vec<bash_tools::Hit> = hits
        .into_iter()
        .filter(|hit| {
            !matches!(
                policy.evaluate(&Action::Read { path: &hit.path }),
                Decision::Deny { .. }
            )
        })
        .collect();
    if filtered.is_empty()
        && let Some(hint) = path_shaped_query_hint(repo_root, query)
    {
        return Ok(hint);
    }
    Ok(bash_tools::render(
        repo_root,
        &filtered,
        SEARCH_CONTEXT_LINES,
    ))
}

fn path_shaped_query_hint(repo_root: &Path, query: &str) -> Option<String> {
    let looks_like_path = !query.contains(char::is_whitespace)
        && (query.contains('/') || Path::new(query).extension().is_some());
    if !looks_like_path {
        return None;
    }
    let nearby = bash_tools::suggest(repo_root, query, MAX_PATH_SUGGESTIONS);
    if nearby.is_empty() {
        return None;
    }
    Some(format!(
        "no content matches for `{query}`. If you meant the file itself, paths with similar names:\n{}",
        nearby
            .iter()
            .map(|p| format!("  {p}"))
            .collect::<Vec<_>>()
            .join("\n")
    ))
}

async fn edit_file(
    repo_root: &Path,
    policy: &Policy,
    approver: Option<&UiSender>,
    _ctx: &SessionCtx,
    args: &Value,
    edited: &mut Vec<String>,
) -> Result<String> {
    let path = args["path"].as_str().context("edit_file needs a `path`")?;
    let block = EditBlock {
        search: args["search"].as_str().unwrap_or_default().to_string(),
        replace: args["replace"]
            .as_str()
            .context("edit_file needs `replace`")?
            .to_string(),
    };
    // An empty `search` has nothing to match, so it means "create this file".
    let creating = block.search.is_empty();
    let (resolved, scope, updated) = if creating {
        let (resolved, scope) = edits::resolve_new_anywhere(repo_root, path)?;
        if resolved.exists() {
            bail!("{path} already exists; put the text to replace in `search`");
        }
        if let Some(hint) = edits::wrong_directory_hint(&resolved) {
            bail!("{hint}");
        }
        (resolved, scope, block.replace.clone())
    } else {
        let (resolved, scope, content) = edits::read_file_anywhere(repo_root, path)?;
        let updated = edits::apply_block(&content, &block)?;
        (resolved, scope, updated)
    };
    let verb = if creating { "create" } else { "edit" };

    // Writes outside the repo need approval unless yolo lifts it.
    if matches!(scope, edits::Scope::Outside) && !_ctx.yolo.load(Ordering::Relaxed) {
        let preview = format!(
            "{verb} {path} (outside the repo):\n{}",
            edits::preview(&block)
        );
        if !request_approval(approver, preview, None).await.allowed() {
            bail!(
                "write to {path} needs user approval because it is outside the repo; \
                 it was rejected or no interactive approver is available"
            );
        }
    }
    if matches!(scope, edits::Scope::InRepo) {
        // Against the resolved path, not the argument: an absolute path inside
        // the repo would otherwise match none of the protected globs.
        let root = repo_root.canonicalize().unwrap_or_default();
        let relative = resolved.strip_prefix(&root).unwrap_or(&resolved);
        match policy.evaluate(&Action::Edit {
            path: &relative.to_string_lossy(),
        }) {
            Decision::Allow => {}
            Decision::Deny { reason } => bail!(
                "edit blocked by policy: {}",
                plan_mode_hint(reason, approver.is_some())
            ),
            Decision::Prompt { .. } => {
                let preview = format!("{verb} {path}:\n{}", edits::preview(&block));
                if !request_approval(approver, preview, None).await.allowed() {
                    bail!(
                        "edit needs user approval (permissions mode is `ask`); \
                         it was rejected or no interactive approver is available"
                    );
                }
            }
        }
    }

    if creating && let Some(parent) = resolved.parent() {
        fs::create_dir_all(parent).with_context(|| format!("creating {}", parent.display()))?;
    }
    fs::write(&resolved, &updated).with_context(|| format!("writing {}", resolved.display()))?;
    if !edited.iter().any(|p| p == path) {
        edited.push(path.to_string());
    }
    let done = if creating { "created" } else { "edited" };
    Ok(format!("{done} {path}:\n{}", edits::preview(&block)))
}

/// Ask the front-end to approve a pending action. Headless callers have no
/// approver, so every request is a `No`.
pub(crate) async fn request_approval(
    approver: Option<&UiSender>,
    preview: String,
    scope: Option<PathBuf>,
) -> Answer {
    ask_approval(approver, preview, scope, None, UiRequest::Approval).await
}

async fn request_plan_approval(
    approver: Option<&UiSender>,
    preview: String,
    markdown: Option<String>,
) -> Answer {
    ask_approval(approver, preview, None, markdown, UiRequest::PlanApproval).await
}

async fn ask_approval(
    approver: Option<&UiSender>,
    preview: String,
    scope: Option<PathBuf>,
    markdown: Option<String>,
    wrap: fn(ApprovalRequest) -> UiRequest,
) -> Answer {
    let Some(tx) = approver else {
        return Answer::No;
    };
    let (respond, rx) = oneshot::channel();
    let request = wrap(ApprovalRequest {
        preview,
        markdown,
        scope,
        respond,
    });
    if tx.send(request).await.is_err() {
        return Answer::No;
    }
    rx.await.unwrap_or(Answer::No)
}

fn truncate(text: &str, max: usize) -> String {
    if text.len() <= max {
        return text.to_string();
    }
    let mut cut = max;
    while !text.is_char_boundary(cut) {
        cut -= 1;
    }
    format!("{}\n... [truncated]", &text[..cut])
}

fn truncate_head(text: &str, max: usize) -> String {
    if text.len() <= max {
        return text.to_string();
    }
    let mut cut = text.len() - max;
    while !text.is_char_boundary(cut) {
        cut += 1;
    }
    format!("... [truncated]\n{}", &text[cut..])
}

fn agent_tool_schema() -> Value {
    json!({
        "type": "function",
        "function": {
            "name": "agent",
            "description": "Fan out self-contained tasks to named sub-agents in parallel. Each agent starts with a fresh context and cannot see this conversation. Split divisible work into distinct, non-overlapping tasks, one per entry, so the batch covers the whole; do not hand the entire job to a single agent when it splits cleanly, and do not give two agents the same task. For broad investigation, fan several cheap scout agents out in one call, then pass their raw reports to `prism` in a second call. Limit batch size to avoid overwhelming the system.",
            "parameters": {
                "type": "object",
                "properties": {
                    "tasks": {
                        "type": "array",
                        "description": "One or more agent tasks to run in parallel",
                        "items": {
                            "type": "object",
                            "properties": {
                                "agent": { "type": "string", "description": "Name of the agent to invoke" },
                                "task": { "type": "string", "description": "Self-contained task for the agent" }
                            },
                            "required": ["agent", "task"]
                        },
                        "minItems": 1
                    },
                    "background": {
                        "type": "boolean",
                        "description": "Queue the tasks and return immediately instead of waiting. Reports arrive between rounds as each agent finishes; use `check: true` to see progress."
                    },
                    "check": {
                        "type": "boolean",
                        "description": "Return the status of background agent work instead of running tasks."
                    }
                },
                "required": ["tasks"]
            }
        }
    })
}

fn agent_task_batch(args: &Value) -> Option<Vec<Value>> {
    if let Some(tasks) = args.get("tasks").and_then(Value::as_array) {
        return Some(tasks.clone());
    }
    (args.get("agent").is_some() && args.get("task").is_some()).then(|| vec![args.clone()])
}

#[allow(clippy::too_many_arguments)]
async fn dispatch_agent_tool(
    repo_root: &Path,
    client: &AiClient,
    arguments: &str,
    policy: &Policy,
    _grants: &Grants,
    ctx: &SessionCtx,
    call_id: &str,
    events: Option<&ChatEventSink>,
) -> String {
    let args: Value = match parse_arguments(arguments) {
        Ok(v) => v,
        Err(e) => return format!("error: agent arguments were not valid JSON: {e}"),
    };
    let Some(tasks_val) = agent_task_batch(&args) else {
        return "error: agent tool requires a `tasks` array".to_string();
    };
    let mut tasks: Vec<crate::agents::AgentTask> = Vec::new();
    let max_per_turn = ctx.swarm.max_per_turn;
    for entry in tasks_val.iter().take(max_per_turn) {
        let Some(agent) = entry.get("agent").and_then(Value::as_str) else {
            continue;
        };
        let Some(task) = entry.get("task").and_then(Value::as_str) else {
            continue;
        };
        tasks.push(crate::agents::AgentTask {
            agent: agent.to_string(),
            task: task.to_string(),
        });
    }
    if tasks.is_empty() {
        return "error: agent tool needs at least one valid task with `agent` and `task` fields"
            .to_string();
    }

    if args.get("check").and_then(Value::as_bool) == Some(true) {
        return crate::agents_queue::status_text();
    }
    let background = args.get("background").and_then(Value::as_bool) == Some(true);

    let over_cap = tasks_val.len().saturating_sub(max_per_turn);
    let deps = crate::agents::AgentDeps {
        client: client.clone(),
        repo_root: repo_root.to_path_buf(),
        policy: Arc::new(policy.clone()),
        grants: Arc::new(Grants::default()),
        credentials: ctx.credentials.clone(),
        probe: ctx.probe.clone(),
        environment: ctx.environment.clone(),
        limits: ctx.limits.clone(),
        swarm: ctx.swarm.clone(),
        session_registry: ctx.agents.clone(),
        yolo: ctx.yolo.load(Ordering::Relaxed),
    };

    // Reports go to whichever turn is live when an agent finishes, so the
    // submitting turn attaches first and later turns re-attach at their start.
    crate::agents_queue::attach(ctx.injected.clone(), events.cloned());

    if background {
        let registry = ctx.agents.clone();
        let deps = Arc::new(deps);
        let runner: crate::agents_queue::SwarmRunner = Arc::new(move |tasks, callbacks| {
            let registry = registry.clone();
            let deps = deps.clone();
            Box::pin(async move {
                crate::agents::run_swarm(
                    tasks,
                    &registry,
                    &deps,
                    {
                        let on_activity = callbacks.on_activity.clone();
                        move |agent: &str, task: &str, line: String| {
                            (on_activity)(agent, task, line)
                        }
                    },
                    {
                        let on_complete = callbacks.on_complete.clone();
                        move |p: crate::agents::AgentProgress| {
                            (on_complete)(crate::agents::TaskReport {
                                agent: p.agent,
                                task: p.task,
                                report: p.report,
                                error: p.error,
                            })
                        }
                    },
                )
                .await
            })
        });
        let count = tasks.len();
        return match crate::agents_queue::submit(tasks, runner) {
            Ok(id) => format!(
                "Queued {count} task(s) as background batch {id}. They run while you keep working; \
                 each agent's report arrives automatically between rounds. Use the agent tool with \
                 `check: true` to see progress."
            ),
            Err(e) => format!("error: {e}"),
        };
    }

    // Seed the UI with the whole batch up front so it can show "2/3"-style
    // progress; run_swarm then reports each task's completion as it lands.
    if let Some(sink) = events {
        let total = tasks.len();
        for t in &tasks {
            sink(json!({
                "type": "agent_status",
                "call_id": call_id,
                "agent": t.agent,
                "task": t.task,
                "status": "running",
                "done": 0,
                "total": total,
            }));
        }
    }
    let on_complete = |p: crate::agents::AgentProgress| {
        if let Some(sink) = events {
            let mut ev = json!({
                "type": "agent_status",
                "call_id": call_id,
                "agent": p.agent,
                "task": p.task,
                "status": match p.status {
                    crate::agents::ProgressStatus::Done => "done",
                    crate::agents::ProgressStatus::Failed => "error",
                },
                "done": p.done,
                "total": p.total,
            });
            if let Some(report) = p.report {
                ev["report"] = Value::String(report);
            }
            if let Some(err) = p.error {
                ev["error"] = Value::String(err);
            }
            sink(ev);
        }
    };
    let on_activity = |agent: &str, task: &str, line: String| {
        if let Some(sink) = events {
            sink(json!({
                "type": "agent_activity",
                "call_id": call_id,
                "agent": agent,
                "task": task,
                "line": line,
            }));
        }
    };
    let mut reports =
        crate::agents::run_swarm(tasks, &ctx.agents, &deps, on_activity, on_complete).await;

    // Cap each report so the total fits in MAX_TOOL_RESULT_CHARS.
    let count = reports.len().max(1);
    let per_report = MAX_TOOL_RESULT_CHARS / count;
    for r in &mut reports {
        if let Some(ref mut report) = r.report
            && report.len() > per_report
        {
            *report = truncate(report, per_report);
        }
    }

    let mut result = serde_json::to_string(&reports).unwrap_or_default();
    if over_cap > 0 {
        result.push_str(&format!(
            "\n\n({over_cap} task(s) beyond the per-call cap of {max_per_turn} were NOT run. \
            They are not lost: re-send them in your next `agent` call, in waves of at most \
            {max_per_turn}, until the whole batch has run.)"
        ));
    }
    result
}

#[cfg(test)]
#[path = "tests/chat_test.rs"]
mod tests;