//! The agent loop: stream a model response, run tool calls (with permission
//! checks), feed results back, repeat until the model stops.

use crate::app::{App, PermissionReply};
use crate::permission::{Decision, Evaluator, bash_prefix, split_command};
use crate::prompts::{self, Agent};
use crate::provider::{ChatRequest, Provider, StreamEvent};
use crate::store::UsageRecord;
use crate::tools::{self, Tool, ToolCtx};
use crate::types::{Event, Message, Part, Role, Session, Usage, new_id, now_ms};
use anyhow::{Result, anyhow, bail};
use futures::future::join_all;
use serde_json::Value;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

const MAX_STEPS: usize = 250;
/// Modes with enforced reviews work through their todo list (e.g. overnight).
const MAX_STEPS_ENFORCED: usize = 2_000;
const MAX_SUBAGENT_DEPTH: usize = 3;

/// What happened to a prompt.
pub enum Submitted {
    Started,
    Queued(crate::store::QueueItem),
}

fn emit_queue(app: &App, session_id: &str) {
    let items = app.store.queue_list(session_id).unwrap_or_default();
    app.emit(Event::QueueUpdated { session_id: session_id.into(), items: serde_json::json!(items) });
}

/// Take all queued messages (one user message's worth of parts).
fn take_queue(app: &App, session_id: &str) -> Vec<Part> {
    let items = app.store.queue_take(session_id).unwrap_or_default();
    if !items.is_empty() {
        emit_queue(app, session_id);
    }
    items.into_iter().flat_map(|i| i.parts).collect()
}

/// Start processing a user prompt. If the session is busy, the prompt (with
/// attachments) is queued and delivered at the agent's next step.
pub fn start_prompt(app: Arc<App>, session_id: String, parts: Vec<Part>) -> Result<()> {
    submit(app, session_id, parts, "user").map(|_| ())
}

pub fn submit(app: Arc<App>, session_id: String, parts: Vec<Part>, source: &str) -> Result<Submitted> {
    // Queueing and the end-of-run check run under the same runtime lock, so a
    // message is never lost or delivered twice.
    let cancel = CancellationToken::new();
    let queued = app.with_runtime(&session_id, |r| -> Result<Option<crate::store::QueueItem>> {
        if r.cancel.is_some() {
            let item = crate::store::QueueItem { id: new_id(), parts: parts.clone(), source: source.into(), created_at: now_ms() };
            app.store.queue_add(&session_id, &item)?;
            Ok(Some(item))
        } else {
            r.cancel = Some(cancel.clone());
            Ok(None)
        }
    })?;
    if let Some(item) = queued {
        emit_queue(&app, &session_id);
        return Ok(Submitted::Queued(item));
    }
    let text_only: String = parts
        .iter()
        .filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None })
        .collect::<Vec<_>>()
        .join("\n");
    app.emit(Event::SessionStatus { session_id: session_id.clone(), busy: true });
    let source = source.to_string();

    tokio::spawn(async move {
        let res = async {
            let mut parts = parts;
            if source == "user" {
                let session = app.session(&session_id)?;
                let sandbox = crate::sandbox::spec_for(&app, session.owner_id.as_deref());
                let payload = serde_json::json!({"session_id": session_id, "cwd": session.cwd, "prompt": text_only});
                let o = crate::hooks::run(&app, "user_prompt", None, payload, Path::new(&session.cwd), sandbox.as_ref()).await;
                if let Some(reason) = o.blocked {
                    anyhow::bail!("Prompt blocked by a hook: {}", reason.trim());
                }
                if !o.context.trim().is_empty() {
                    parts.push(Part::Text { text: format!("<hook-context event=\"user_prompt\">\n{}\n</hook-context>", o.context.trim()) });
                }
            }
            if source == "user" {
                parts.extend(crate::mentions::expand(&app, &session_id, &text_only).await);
            }
            // Empty parts: continue from the existing history (regenerate).
            if !parts.is_empty() {
                let msg = Message::new(Role::User, parts);
                app.store.add_message(&session_id, &msg)?;
                app.emit(Event::MessageAdded { session_id: session_id.clone(), message: msg });
                maybe_generate_title(app.clone(), &session_id, &text_only);
            }
            run_session(app.clone(), &session_id, cancel.clone(), 0).await
        }
        .await;
        if let Err(e) = res {
            tracing::warn!("session {session_id}: {e:#}");
            app.emit(Event::Error { session_id: session_id.clone(), message: format!("{e:#}") });
        }
        // Messages queued after the last step start a new turn – unless the user
        // stopped the agent: then they stay queued until sent explicitly.
        let leftover = app.with_runtime(&session_id, |r| {
            r.cancel = None;
            r.effort = None;
            if cancel.is_cancelled() { Vec::new() } else { take_queue(&app, &session_id) }
        });
        app.emit(Event::SessionStatus { session_id: session_id.clone(), busy: false });
        if !leftover.is_empty() {
            let _ = start_prompt(app, session_id, leftover);
        }
    });
    Ok(Submitted::Started)
}

/// Send queued messages now (after a stop or a server restart).
pub fn flush_queue(app: Arc<App>, session_id: &str) -> Result<bool> {
    if app.is_busy(session_id) {
        return Ok(false);
    }
    let parts = take_queue(&app, session_id);
    if parts.is_empty() {
        return Ok(false);
    }
    start_prompt(app, session_id.to_string(), parts)?;
    Ok(true)
}

pub fn abort(app: &App, session_id: &str) {
    app.with_runtime(session_id, |r| {
        if let Some(c) = &r.cancel {
            c.cancel();
        }
    });
}

/// Messages since the last compaction summary. Tool calls left without a
/// result (run interrupted, e.g. by a server restart) get a synthetic result:
/// providers reject histories with unanswered tool calls.
fn active_history(all: Vec<Message>) -> Vec<Message> {
    let start = all
        .iter()
        .rposition(|m| m.parts.iter().any(|p| matches!(p, Part::Summary { .. })))
        .unwrap_or(0);
    let msgs: Vec<Message> = all.into_iter().skip(start).collect();
    let mut out: Vec<Message> = Vec::with_capacity(msgs.len());
    for (i, m) in msgs.iter().enumerate() {
        out.push(m.clone());
        if m.role != Role::Assistant {
            continue;
        }
        let answered: std::collections::HashSet<&str> = msgs
            .get(i + 1)
            .filter(|n| n.role == Role::User)
            .map(|n| {
                n.parts
                    .iter()
                    .filter_map(|p| if let Part::ToolResult { call_id, .. } = p { Some(call_id.as_str()) } else { None })
                    .collect()
            })
            .unwrap_or_default();
        let missing: Vec<Part> = m
            .parts
            .iter()
            .filter_map(|p| match p {
                Part::ToolCall { id, name, .. } if !answered.contains(id.as_str()) => {
                    Some(tool_result(id, name, "Not executed: the run was interrupted (e.g. server restart).".into(), true, None))
                }
                _ => None,
            })
            .collect();
        if missing.is_empty() {
            continue;
        }
        // Results must directly follow the call: merge into the next user message or insert one.
        match msgs.get(i + 1) {
            Some(n) if n.role == Role::User => {
                // The next message is pushed in the next iteration; prepend there.
                let mut next = n.clone();
                let mut parts = missing;
                parts.extend(next.parts);
                next.parts = parts;
                // Replace by pushing now and skipping the original.
                out.push(next);
            }
            _ => out.push(Message::new(Role::User, missing)),
        }
    }
    // Remove duplicates created by the merge above (same id pushed twice: keep the merged one).
    let mut dedup: Vec<Message> = Vec::with_capacity(out.len());
    for m in out {
        if let Some(last) = dedup.last() {
            if last.id == m.id {
                continue;
            }
        }
        dedup.push(m);
    }
    dedup
}

fn normalize_tool_name(s: &str) -> String {
    s.chars().filter(|c| c.is_ascii_alphanumeric()).collect::<String>().to_lowercase()
}

pub fn available_tools(app: &App, agent: &Agent, owner: Option<&crate::store::User>) -> Vec<Arc<dyn Tool>> {
    let cfg = app.config();
    let mut all = tools::builtin_tools();
    if !cfg.images.enabled {
        all.retain(|t| t.name() != "generate_image");
    }
    let (direct, deferred) = app.mcp.tools(&cfg, owner);
    all.extend(direct);
    if !deferred.is_empty() {
        all.push(Arc::new(crate::mcp::McpSearch));
        all.push(Arc::new(crate::mcp::McpCall));
    }
    all.into_iter()
        .filter(|t| !agent.read_only || t.read_only())
        // Craft tools only where they are useful.
        .filter(|t| {
            agent.name == "craft"
                || agent.tools.is_some()
                || (agent.reverse && matches!(t.name(), "binary_info" | "decompile"))
                || !matches!(t.name(), "browser_snapshot" | "image_compare" | "binary_info" | "decompile" | "craft_phase")
        })
        .filter(|t| match &agent.tools {
            None => true,
            Some(list) => {
                let n = normalize_tool_name(t.name());
                list.iter().any(|a| {
                    let a = normalize_tool_name(a);
                    a == n || (t.name().starts_with("mcp") && a == "mcp")
                })
            }
        })
        .collect()
}

struct Step {
    message: Message,
    stop_reason: String,
}

/// Boxed with an explicit `Send` bound: subagents recurse into the agent loop,
/// and the declared type breaks the cycle for the compiler.
fn run_session_boxed(
    app: Arc<App>,
    session_id: String,
    cancel: CancellationToken,
    depth: usize,
) -> futures::future::BoxFuture<'static, Result<String>> {
    Box::pin(async move { run_session(app, &session_id, cancel, depth).await })
}

/// Run the loop for a session until the model stops. Returns the final text.
pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationToken, depth: usize) -> Result<String> {
    let mut final_text = String::new();
    let mut stop_hooks = 0;
    let mut refine_done = 0u32;
    let turn_start = now_ms();
    // Compacted for the current step (no second compaction before a request went through).
    let mut compacted_now = false;
    // Mandatory reviews of completed todos: rounds per todo, and where the last one ended.
    let mut todo_reviews: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
    let mut review_since = turn_start;
    // Reminders about open todos at the end of a turn (see MAX_IDLE_CONTINUES).
    let mut last_reminder: Option<Value> = None;
    let mut idle_continues = 0u32;
    for step_no in 0..=MAX_STEPS_ENFORCED {
        if cancel.is_cancelled() {
            break;
        }
        let session = app.session(session_id)?;
        let cwd = PathBuf::from(&session.cwd);
        let roots = app.user_roots(session.owner_id.as_deref());
        let root = app.root_for(session.owner_id.as_deref(), &cwd);
        let mut agent = prompts::find_agent_for(&app, &root, &cwd, &session.agent, session.owner_id.as_deref());
        let clean_room = session.clean_room(app.config().craft.clean_room);
        // The craft prompt follows the session's clean room / deep switch.
        if agent.name == "craft" && agent.source == "builtin" {
            agent.prompt = prompts::craft_prompt(clean_room);
        }
        let model_spec = agent.model.clone().unwrap_or_else(|| session.model.clone());
        let step_limit = if enforced_review(&agent) > 0 { MAX_STEPS_ENFORCED } else { MAX_STEPS };
        if step_no >= step_limit {
            app.emit(Event::Error { session_id: session_id.into(), message: format!("Stopped after {step_limit} steps (limit of a turn) – send a message to continue.") });
            break;
        }
        let (provider, mut model) = app.registry().resolve(&model_spec)?;
        if let Some(effort) = app.with_runtime(session_id, |r| r.effort.clone()) {
            model.effort = Some(effort);
        }

        let history = active_history(app.store.messages_for_model(session_id)?);
        // Providers count input + max_tokens against the context window: the answer needs
        // room too. Estimated input = last request (input + its answer) + what came since.
        let est_input = estimate_input(session.last_usage.as_ref(), &history);
        let want_output = model.max_output.unwrap_or(16_000);
        if let Some(ctx_len) = model.context {
            let threshold = app.config().server.auto_compact_threshold;
            // Input limit: the context window, or the provider's lower input limit.
            let input_limit = model.max_input.map_or(ctx_len, |m| m.min(ctx_len));
            let room = ctx_len.saturating_sub(est_input);
            // Most of the input limit used, or no room left for a reasonable answer.
            if threshold > 0.0 && (est_input as f64 > input_limit as f64 * threshold || room < want_output.min(MIN_ANSWER_ROOM)) && !compacted_now {
                compacted_now = true;
                compact(app.clone(), session_id, cancel.clone(), Some((est_input, input_limit))).await?;
                continue;
            }
        }
        let max_tokens = answer_budget(want_output, model.context, est_input);

        let owner = session.owner_id.as_deref().and_then(|id| app.store.user(id).ok().flatten());
        let tools = available_tools(&app, &agent, owner.as_ref());
        let req = ChatRequest {
            system: prompts::with_chat_instructions(prompts::system_prompt(&app, &agent, &root, &cwd, &model_spec, &session.permission_mode, owner.as_ref()), &session.instructions),
            messages: history,
            tools: tools.iter().map(|t| t.def()).collect(),
            max_tokens,
            reasoning: true,
            session_id: Some(session_id.to_string()),
            model: model.clone(),
        };
        let step = match stream_step(&app, &session, &agent.name, &model_spec, provider, req.clone(), &cancel).await {
            Ok(step) => {
                compacted_now = false;
                step
            }
            // The estimate was off and the provider says the input is too long: compact
            // once and try again instead of stopping.
            Err(e) if is_context_overflow(&e) && !compacted_now && !cancel.is_cancelled() => {
                compacted_now = true;
                app.emit(Event::Error { session_id: session_id.into(), message: format!("context window full ({}) – compacting the conversation and continuing", short_error(&e)) });
                compact(app.clone(), session_id, cancel.clone(), Some((est_input, model.context.unwrap_or(0)))).await?;
                continue;
            }
            // Host down or budget used up: switch to the model's fallback for this step.
            Err(e) if model.fallback.is_some() && !cancel.is_cancelled() && needs_fallback(&e) => {
                let fb = model.fallback.clone().unwrap_or_default();
                let (fb_provider, fb_model) = app.registry().resolve(&fb)?;
                app.emit(Event::Error { session_id: session_id.into(), message: format!("{model_spec} is not available ({}) – using the fallback model {fb}", short_error(&e)) });
                let req = ChatRequest { model: fb_model.clone(), max_tokens: answer_budget(fb_model.max_output.unwrap_or(16_000), fb_model.context, est_input), ..req };
                stream_step(&app, &session, &agent.name, &fb, fb_provider, req, &cancel).await?
            }
            Err(e) => return Err(e),
        };
        final_text = step
            .message
            .parts
            .iter()
            .filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None })
            .collect::<Vec<_>>()
            .join("");

        let calls: Vec<(String, String, Value)> = step
            .message
            .parts
            .iter()
            .filter_map(|p| match p {
                Part::ToolCall { id, name, input, .. } => Some((id.clone(), name.clone(), input.clone())),
                _ => None,
            })
            .collect();

        if cancel.is_cancelled() {
            // Every tool call needs a result, otherwise the history is invalid.
            if !calls.is_empty() {
                let results = calls
                    .iter()
                    .map(|(id, name, _)| tool_result(id, name, "Aborted by user".into(), true, None))
                    .collect();
                save_user_parts(&app, session_id, results)?;
            }
            break;
        }
        if calls.is_empty() && depth == 0 && stop_hooks < 3 && matches!(step.stop_reason.as_str(), "stop" | "end_turn" | "") {
            let payload = serde_json::json!({"session_id": session_id, "cwd": session.cwd, "last_message": final_text});
            let sandbox = crate::sandbox::spec_for(&app, session.owner_id.as_deref());
            let o = crate::hooks::run(&app, "stop", None, payload, &cwd, sandbox.as_ref()).await;
            if let Some(reason) = o.blocked {
                stop_hooks += 1;
                save_user_parts(&app, session_id, vec![Part::Text { text: format!("<hook-feedback event=\"stop\">\n{}\n</hook-feedback>", reason.trim()) }])?;
                continue;
            }
        }
        let normal_end = matches!(step.stop_reason.as_str(), "stop" | "end_turn" | "");
        // Enforced reviews also mean: work until the todo list is done. Ending the turn with
        // open todos sends the agent back (as long as it makes progress).
        if calls.is_empty() && depth == 0 && normal_end && !cancel.is_cancelled() && enforced_review(&agent) > 0 {
            let todos_now = app.session(session_id)?.todos;
            let open = open_todos(&todos_now);
            if !open.is_empty() {
                let progressed = last_reminder.as_ref() != Some(&todos_now);
                idle_continues = if progressed { 0 } else { idle_continues + 1 };
                if idle_continues < MAX_IDLE_CONTINUES {
                    last_reminder = Some(todos_now);
                    save_user_parts(
                        &app,
                        session_id,
                        vec![Part::Text {
                            text: format!(
                                "<todo-reminder open=\"{}\">\n{}\n</todo-reminder>\nThese todos are still open. Continue working until all of them are completed (each completed todo is reviewed). Only if you are truly blocked and need the user, say clearly what you need and stop.",
                                open.len(),
                                open.iter().map(|t| format!("- {t}")).collect::<Vec<_>>().join("\n")
                            ),
                        }],
                    )?;
                    continue;
                }
            }
        }
        if calls.is_empty() && depth == 0 && normal_end && !cancel.is_cancelled() {
            // Refinement loop: a reviewer checks the turn; findings send the agent back to work.
            let rounds = agent.refine.as_ref().map(|r| r.rounds).unwrap_or(0);
            if refine_done < rounds {
                refine_done += 1;
                match review_turn(&app, &session, &agent, &model_spec, turn_start, None, refine_done, rounds, &cancel).await {
                    Ok(Some(feedback)) => {
                        save_user_parts(&app, session_id, vec![Part::Text { text: feedback }])?;
                        continue;
                    }
                    Ok(None) => {}
                    Err(e) => tracing::warn!("review of {session_id} failed: {e:#}"),
                }
            }
            // Craft: a finished check phase means the result is ready.
            if agent.name == "craft" {
                advance_phase(&app, session_id, "ready", Some("check"));
            }
        }
        if calls.is_empty() {
            match step.stop_reason.as_str() {
                "max_tokens" | "length" => app.emit(Event::Error {
                    session_id: session_id.into(),
                    message: "The response hit the output token limit (max_output).".into(),
                }),
                "refusal" | "content_filter" => app.emit(Event::Error {
                    session_id: session_id.into(),
                    message: "The model declined to respond (refusal).".into(),
                }),
                _ => {}
            }
            break;
        }

        let base = ToolCtx {
            app: app.clone(),
            session_id: session_id.to_string(),
            call_id: String::new(),
            cwd: cwd.clone(),
            cancel: cancel.clone(),
            agent: agent.name.clone(),
            roots: roots.clone(),
            sandbox: crate::sandbox::spec_for(&app, session.owner_id.as_deref()),
            // Clean room is a craft rule; other modes with reverse tools may analyse binaries.
            clean_room: clean_room && agent.name == "craft",
            owner_id: None,
        };
        let by_name = |n: &str| tools.iter().find(|t| t.name() == n).cloned();
        let all_read_only = calls.iter().all(|(_, n, _)| by_name(n).is_some_and(|t| t.read_only()));
        let mut results = Vec::new();
        if all_read_only {
            let futs = calls.iter().map(|(id, name, input)| {
                execute_call(&app, &session, &base, by_name(name), id, name, input.clone(), depth)
            });
            results = join_all(futs).await;
        } else {
            for (id, name, input) in &calls {
                results.push(execute_call(&app, &session, &base, by_name(name), id, name, input.clone(), depth).await);
            }
        }
        // Mode with a review: every todo marked completed is reviewed; with findings it is
        // reopened and the agent has to fix them and complete it again (max. `rounds` per todo).
        // The verdict goes into the same message as the tool results.
        let rounds = enforced_review(&agent);
        if rounds > 0 && depth == 0 && !cancel.is_cancelled() && calls.iter().any(|(_, n, _)| n == "todo_write") {
            let todos_now = app.session(session_id)?.todos;
            let due: Vec<String> = newly_completed(&session.todos, &todos_now)
                .into_iter()
                .filter(|t| todo_reviews.get(t).copied().unwrap_or(0) < rounds)
                .collect();
            if !due.is_empty() {
                for t in &due {
                    *todo_reviews.entry(t.clone()).or_default() += 1;
                }
                let round = due.iter().filter_map(|t| todo_reviews.get(t)).copied().max().unwrap_or(1);
                let fresh = app.session(session_id)?;
                let reviewed = review_turn(&app, &fresh, &agent, &model_spec, review_since, Some(&due), round, rounds, &cancel).await;
                review_since = now_ms();
                match reviewed {
                    Ok(Some(feedback)) => {
                        app.set_todos(session_id, reopen_todos(&todos_now, &due))?;
                        results.push(Part::Text { text: feedback });
                    }
                    Ok(None) => {
                        let open = open_todos(&todos_now);
                        let next = if open.is_empty() { "All todos are done – check the whole request once more, then finish." } else { "Continue with the next open todo." };
                        results.push(Part::Text {
                            text: format!("<refine-approved reviewer=\"{}\">\n{}\n</refine-approved>\nReview passed. {next}", review_model(&agent, &model_spec), due.iter().map(|t| format!("- {t}")).collect::<Vec<_>>().join("\n")),
                        });
                    }
                    Err(e) => tracing::warn!("todo review of {session_id} failed: {e:#}"),
                }
            }
        }
        // Deliver messages the user queued while we were working (with attachments).
        results.extend(take_queue(&app, session_id));
        save_user_parts(&app, session_id, results)?;
    }
    Ok(final_text)
}

/// Review rounds when the mode enforces reviews of completed todos (0 = not enforced).
fn enforced_review(agent: &prompts::Agent) -> u32 {
    agent.refine.as_ref().filter(|r| r.after_todos).map(|r| r.rounds).unwrap_or(0)
}

/// The model that reviews for this mode.
fn review_model(agent: &prompts::Agent, model_spec: &str) -> String {
    agent.refine.as_ref().and_then(|r| r.model.clone()).filter(|m| !m.is_empty()).unwrap_or_else(|| model_spec.to_string())
}

/// Contents of the todos that are not completed yet.
fn open_todos(todos: &Value) -> Vec<String> {
    todos.as_array().into_iter().flatten().filter(|t| t["status"] != "completed").filter_map(|t| t["content"].as_str()).map(String::from).collect()
}

/// Times in a row the agent may end its turn with open todos without making progress.
const MAX_IDLE_CONTINUES: u32 = 10;

/// Contents of the todos that are completed in `after` but were not in `before`.
fn newly_completed(before: &Value, after: &Value) -> Vec<String> {
    let done_before: std::collections::HashSet<&str> = before
        .as_array()
        .into_iter()
        .flatten()
        .filter(|t| t["status"] == "completed")
        .filter_map(|t| t["content"].as_str())
        .collect();
    after
        .as_array()
        .into_iter()
        .flatten()
        .filter(|t| t["status"] == "completed")
        .filter_map(|t| t["content"].as_str())
        .filter(|c| !done_before.contains(c))
        .map(String::from)
        .collect()
}

/// The todo list with the given items open again (in progress).
fn reopen_todos(todos: &Value, items: &[String]) -> Value {
    let mut out = todos.clone();
    for t in out.as_array_mut().into_iter().flatten() {
        if t["content"].as_str().is_some_and(|c| items.iter().any(|i| i == c)) {
            t["status"] = Value::from("in_progress");
        }
    }
    out
}

fn tool_result(call_id: &str, name: &str, content: String, is_error: bool, meta: Option<Value>) -> Part {
    Part::ToolResult { call_id: call_id.into(), name: name.into(), content, images: vec![], is_error, meta }
}

fn save_user_parts(app: &App, session_id: &str, parts: Vec<Part>) -> Result<()> {
    let msg = Message::new(Role::User, parts);
    app.store.add_message(session_id, &msg)?;
    app.emit(Event::MessageAdded { session_id: session_id.into(), message: msg });
    Ok(())
}

#[allow(clippy::too_many_arguments)]
async fn execute_call(
    app: &Arc<App>,
    session: &Session,
    base: &ToolCtx,
    tool: Option<Arc<dyn Tool>>,
    call_id: &str,
    name: &str,
    input: Value,
    depth: usize,
) -> Part {
    let Some(tool) = tool else {
        return tool_result(call_id, name, format!("Unknown or unavailable tool '{name}' in this mode."), true, None);
    };
    if let Some(raw) = input.get("__invalid_json") {
        return tool_result(
            call_id,
            name,
            format!("Tool input was not valid JSON (possibly truncated). Retry with a smaller input.\n{}", raw.as_str().unwrap_or("").chars().take(300).collect::<String>()),
            true,
            None,
        );
    }
    if name == "task" && depth >= MAX_SUBAGENT_DEPTH {
        return tool_result(call_id, name, "Maximum subagent depth reached.".into(), true, None);
    }

    let ctx = ToolCtx {
        app: app.clone(),
        session_id: base.session_id.clone(),
        call_id: call_id.into(),
        cwd: base.cwd.clone(),
        cancel: base.cancel.clone(),
        agent: base.agent.clone(),
        roots: base.roots.clone(),
        sandbox: base.sandbox.clone(),
        clean_room: base.clean_room,
        owner_id: None,
    };
    // mcp_call is judged (and approved) as the MCP tool it calls.
    let call_name = name;
    let (tool, input, name): (Arc<dyn Tool>, Value, String) = if name == "mcp_call" {
        match crate::mcp::resolve_call_target(&ctx, &input) {
            Ok(target) => {
                let args = input.get("arguments").cloned().unwrap_or(serde_json::json!({}));
                let n = target.full_name.clone();
                (target, args, n)
            }
            Err(e) => return tool_result(call_id, call_name, format!("Error: {e:#}"), true, None),
        }
    } else {
        (tool, input, name.to_string())
    };
    let name = name.as_str();

    // Hunks of a file edit the user accepted in the review (None = whole change).
    let mut partial: Option<Vec<usize>> = None;
    // Permission check. Approvals are tracked on the top-level session.
    let root = app.root_session_id(&session.id);
    let cfg = app.config();
    let decision = app.with_runtime(&root, |r| {
        Evaluator { cfg: &cfg.permissions, mode: &session.permission_mode, approved: &r.approved }.evaluate(tool.as_ref(), &input)
    });
    match decision {
        Decision::Deny => {
            return tool_result(call_id, call_name, "Denied by the permission rules in the configuration.".into(), true, None);
        }
        Decision::Ask => {
            let diff = tool.preview(&input, &ctx).await;
            let (reply, message, hunks) = app
                .request_permission(&session.id, name, tool.title(&input), input.clone(), diff, base.cancel.clone())
                .await;
            match reply {
                PermissionReply::Reject => {
                    let mut text = "The user rejected this tool call.".to_string();
                    if let Some(m) = message.filter(|m| !m.trim().is_empty()) {
                        text.push_str(&format!(" User feedback: {m}"));
                    }
                    return tool_result(call_id, call_name, text, true, None);
                }
                PermissionReply::Always => app.with_runtime(&root, |r| {
                    if name == "bash" {
                        let cmd = input["command"].as_str().unwrap_or("");
                        for part in split_command(cmd).unwrap_or_default() {
                            r.approved.insert(format!("bash:{}", bash_prefix(&part)));
                        }
                    } else {
                        r.approved.insert(name.to_string());
                    }
                }),
                PermissionReply::Once => {}
            }
            if let Some(sel) = hunks {
                if sel.is_empty() {
                    return tool_result(call_id, call_name, "The user rejected all parts of this change.".into(), true, None);
                }
                partial = Some(sel);
            }
        }
        Decision::Allow => {}
    }

    // Hooks see the tool input (cloned only when hooks are configured).
    let hook_input = if app.config().hooks.is_empty() { Value::Null } else { input.clone() };
    let hook_payload = |extra: Option<(&str, Value)>| {
        let mut v = serde_json::json!({"session_id": session.id, "cwd": session.cwd, "tool": name, "input": hook_input});
        if let Some((k, x)) = extra {
            v[k] = x;
        }
        v
    };
    let pre = crate::hooks::run(app, "pre_tool", Some(name), hook_payload(None), &ctx.cwd, ctx.sandbox.as_ref()).await;
    if let Some(reason) = pre.blocked {
        return tool_result(call_id, call_name, format!("Blocked by a hook: {}", reason.trim()), true, None);
    }

    // Checkpoint: remember files before the agent changes them (rewind can restore them).
    if tool.is_edit() {
        if let Some(path) = input["path"].as_str().and_then(|p| ctx.resolve(p).ok()) {
            let content = std::fs::read(&path).ok().filter(|c| c.len() <= 5 * 1024 * 1024);
            if path.exists() == content.is_some() {
                let _ = app.store.add_checkpoint(&session.id, &path.display().to_string(), content.as_deref());
            }
        }
    }
    let input_path = input["path"].as_str().map(String::from);
    let input_for_phase = if base.agent == "craft" { input.clone() } else { Value::Null };
    let res = if let (Some(sel), true) = (&partial, matches!(name, "write" | "edit")) {
        apply_partial(&ctx, name, &input, sel).await
    } else if name == "task" {
        // Pass depth through the agent name channel: subagents run with depth+1.
        run_task(app, &ctx, input, depth).await
    } else {
        let moves = name == "bash" && input["command"].as_str().is_some_and(moves_files);
        let out = tool.run(input, &ctx).await;
        if moves {
            // Generated images the command moved or renamed keep their entries.
            let a = app.clone();
            let owner = session.owner_id.clone();
            tokio::task::spawn_blocking(move || crate::images::relocate_missing(&a, owner.as_deref()));
        }
        out
    };
    // Diagnostics: report new errors of the changed file right away.
    let res = match res {
        Ok(mut out) if tool.is_edit() && !out.is_error => {
            if let Some(path) = input_path.as_deref().and_then(|p| ctx.resolve(p).ok()) {
                let wait = std::time::Duration::from_secs(4);
                if let Some(Ok(d)) = crate::lsp::diagnostics(app, &path, &ctx.cwd, ctx.sandbox.as_ref(), wait).await {
                    out.content.push_str(&crate::lsp::format(input_path.as_deref().unwrap_or(""), &d, false));
                }
            }
            Ok(out)
        }
        other => other,
    };
    let res = match res {
        Ok(mut out) => {
            let result = serde_json::json!({"content": out.content.chars().take(20_000).collect::<String>(), "is_error": out.is_error});
            let post = crate::hooks::run(app, "post_tool", Some(name), hook_payload(Some(("result", result))), &ctx.cwd, ctx.sandbox.as_ref()).await;
            if let Some(reason) = post.blocked {
                out.content.push_str(&format!("\n\n<hook-feedback event=\"post_tool\">\n{}\n</hook-feedback>", reason.trim()));
            } else if !post.context.trim().is_empty() {
                out.content.push_str(&format!("\n\n<hook-context event=\"post_tool\">\n{}\n</hook-context>", post.context.trim()));
            }
            Ok(out)
        }
        e => e,
    };
    if matches!(&res, Ok(o) if !o.is_error) && base.agent == "craft" && name != "craft_phase" {
        if let Some(phase) = infer_phase(name, &input_for_phase, session.clean_room == Some(false)) {
            advance_phase(app, &session.id, phase, None);
        }
    }
    match res {
        Ok(out) => Part::ToolResult {
            call_id: call_id.into(),
            name: call_name.into(),
            content: out.content,
            images: out.images,
            is_error: out.is_error,
            meta: out.meta,
        },
        Err(e) => tool_result(call_id, call_name, format!("Error: {e:#}"), true, None),
    }
}

/// A shell command that may move or rename files (`mv`, `git mv`, `rename`).
fn moves_files(cmd: &str) -> bool {
    cmd.split(|c: char| c.is_whitespace() || ";&|()`$".contains(c)).any(|w| w == "mv" || w == "rename")
}

async fn run_task(app: &Arc<App>, ctx: &ToolCtx, input: Value, depth: usize) -> Result<tools::ToolOutput> {
    let prompt = tools::arg_str(&input, "prompt")?.to_string();
    let description = input["description"].as_str().unwrap_or("subagent").to_string();
    let agent_name = input["subagent_type"].as_str().unwrap_or("general").to_string();
    let background = input["background"].as_bool().unwrap_or(false);
    let (child, agent) = create_child(app, ctx, &agent_name, &description, prompt)?;
    let meta = serde_json::json!({
        "session_id": child.id, "agent": agent.name, "model": child.model, "background": background
    });

    if background {
        let app = app.clone();
        let parent_id = ctx.session_id.clone();
        let child_id = child.id.clone();
        let agent_name = agent.name.clone();
        let desc = description.clone();
        // Background tasks outlive the current turn; they are stopped via their own session.
        let cancel = CancellationToken::new();
        tokio::spawn(async move {
            let res = run_child(app.clone(), &child, &agent, cancel, depth + 1).await;
            let body = match res {
                Ok(text) => text,
                Err(e) => format!("The background task failed: {e:#}"),
            };
            let text = format!(
                "<background-task-result agent=\"{agent_name}\" session=\"{child_id}\" description=\"{desc}\">\n{body}\n</background-task-result>"
            );
            // Injected into the running turn, or starts a new turn when the parent is idle.
            if app.store.get_session(&parent_id).ok().flatten().is_some() {
                if let Err(e) = submit(app.clone(), parent_id.clone(), vec![Part::Text { text }], "background") {
                    tracing::warn!("could not deliver background result: {e:#}");
                }
            }
        });
        return Ok(tools::ToolOutput::text(format!(
            "Started background task '{description}' with subagent '{}' (model {}). Its result will arrive \
             automatically as a message when it finishes – continue with other work and do not poll.",
            meta["agent"].as_str().unwrap_or(""),
            meta["model"].as_str().unwrap_or("")
        ))
        .with_meta(meta));
    }

    let text = run_child(app.clone(), &child, &agent, ctx.cancel.child_token(), depth + 1).await?;
    Ok(tools::ToolOutput::text(text).with_meta(meta))
}

/// Entry point used by the task tool when called outside the runner.
pub async fn run_subagent(app: Arc<App>, ctx: &ToolCtx, agent: &str, description: &str, prompt: String) -> Result<(String, String)> {
    let (child, agent) = create_child(&app, ctx, agent, description, prompt)?;
    let text = run_child(app, &child, &agent, ctx.cancel.child_token(), 1).await?;
    Ok((child.id, text))
}

/// Create the subagent session with its first message.
fn create_child(app: &Arc<App>, ctx: &ToolCtx, agent_name: &str, description: &str, prompt: String) -> Result<(Session, Agent)> {
    let parent = app.session(&ctx.session_id)?;
    let root = ctx.root();
    let parent_agent = prompts::find_agent_for(app, &root, &ctx.cwd, &parent.agent, parent.owner_id.as_deref());
    let agents = prompts::agents(app, &root, &ctx.cwd);
    let mut agent = agents
        .iter()
        .find(|a| a.name == agent_name && a.mode != "primary")
        .cloned()
        .ok_or_else(|| {
            let names: Vec<_> = agents.iter().filter(|a| a.mode != "primary").map(|a| a.name.as_str()).collect();
            anyhow!("unknown subagent '{agent_name}'. Available: {}", names.join(", "))
        })?;
    // Read-only parents may only spawn read-only subagents.
    if parent_agent.read_only && !agent.read_only {
        agent = agents.into_iter().find(|a| a.name == "explore").unwrap_or(agent);
    }
    if let Some(m) = &agent.model {
        app.registry().resolve(m).map_err(|e| anyhow!("subagent '{}' has an unusable model: {e:#}", agent.name))?;
    }
    let now = now_ms();
    let child = Session {
        id: new_id(),
        title: description.to_string(),
        cwd: parent.cwd.clone(),
        model: agent.model.clone().unwrap_or_else(|| parent.model.clone()),
        agent: agent.name.clone(),
        permission_mode: parent.permission_mode.clone(),
        parent_id: Some(parent.id.clone()),
        owner_id: parent.owner_id.clone(),
        created_at: now,
        updated_at: now,
        todos: Value::Null,
        last_usage: None,
        clean_room: parent.clean_room,
        phase: None,
        worktree: None,
        pinned: false,
        archived: false,
        folder: None,
        instructions: String::new(),
        telegram_quiet: false,
    };
    let child = app.save_session(child)?;
    let msg = Message::new(Role::User, vec![Part::Text { text: prompt }]);
    app.store.add_message(&child.id, &msg)?;
    app.emit(Event::MessageAdded { session_id: child.id.clone(), message: msg });
    Ok((child, agent))
}

/// Run a subagent session to completion, honouring the subagent limits.
async fn run_child(app: Arc<App>, child: &Session, agent: &Agent, cancel: CancellationToken, depth: usize) -> Result<String> {
    // Registered like a normal run so the user can stop it from its own session.
    app.with_runtime(&child.id, |r| r.cancel = Some(cancel.clone()));
    app.emit(Event::SessionStatus { session_id: child.id.clone(), busy: true });
    // Only first-level subagents count against the limits: nested subagents would
    // otherwise deadlock waiting for a slot their own parent holds.
    let _slot = if depth <= 1 {
        let (slot, waited) = app.subagent_slot(&agent.name, agent.max_parallel, &child.id, &cancel).await;
        if waited {
            app.emit(Event::Waiting { session_id: child.id.clone(), reason: None });
        }
        slot
    } else {
        None
    };
    let res = if cancel.is_cancelled() {
        Err(anyhow!("aborted"))
    } else {
        run_session_boxed(app.clone(), child.id.clone(), cancel, depth).await
    };
    app.with_runtime(&child.id, |r| r.cancel = None);
    app.emit(Event::SessionStatus { session_id: child.id.clone(), busy: false });
    let text = res?;
    Ok(if text.trim().is_empty() { "(subagent returned no text)".into() } else { text })
}

/// Rough prompt size (≈ 3.8 characters per token, ~1,000 tokens per image).
fn estimate_prompt_tokens(req: &ChatRequest) -> u64 {
    let mut chars = req.system.len();
    let mut images = 0u64;
    for m in &req.messages {
        for p in &m.parts {
            match p {
                Part::Text { text } | Part::Reasoning { text, .. } | Part::Summary { text, .. } => chars += text.len(),
                Part::ToolCall { input, .. } => chars += input.to_string().len(),
                Part::ToolResult { content, images: imgs, .. } => {
                    chars += content.len();
                    images += imgs.len() as u64;
                }
                Part::Image { .. } => images += 1,
            }
        }
    }
    for t in &req.tools {
        chars += t.description.len() + t.input_schema.to_string().len() + t.name.len();
    }
    (chars as f64 / 3.8) as u64 + images * 1000
}

/// Errors after which the fallback model takes over: the host is unreachable
/// (connection, timeout, gateway errors) or the provider's budget is used up.
/// Room an answer gets at least before the conversation is compacted.
const MIN_ANSWER_ROOM: u64 = 32_000;

/// Estimated input tokens of the next request: the last request (its input and answer)
/// plus the messages that came after that answer (tool results, user messages), at about
/// 3 characters per token; without usage (fresh or just compacted) the whole history.
fn estimate_input(last: Option<&crate::types::Usage>, history: &[Message]) -> u64 {
    let chars_of = |m: &Message| -> u64 {
        m.parts
            .iter()
            .map(|p| match p {
                Part::Text { text } | Part::Reasoning { text, .. } | Part::Summary { text, .. } => text.len() as u64,
                Part::ToolCall { input, .. } => input.to_string().len() as u64,
                Part::ToolResult { content, images, .. } => content.len() as u64 + images.len() as u64 * 4_500,
                Part::Image { .. } => 4_500,
            })
            .sum()
    };
    match last {
        Some(u) => {
            let after = history.iter().rposition(|m| m.role == Role::Assistant).map(|i| &history[i + 1..]).unwrap_or(&[]);
            u.context_tokens() + after.iter().map(chars_of).sum::<u64>() / 3
        }
        None => history.iter().map(chars_of).sum::<u64>() / 3 + 8_000,
    }
}

/// max_tokens for a request: the model's maximum, but no more than the window leaves
/// (with a margin for the estimate), and never below a small floor.
fn answer_budget(max_output: u64, context: Option<u64>, est_input: u64) -> u64 {
    let Some(ctx) = context else { return max_output };
    let room = ctx.saturating_sub(est_input + est_input / 50 + 2_000);
    max_output.min(room).max(4_096.min(max_output))
}

/// The provider rejected the request because input + output exceed the context window.
fn is_context_overflow(e: &anyhow::Error) -> bool {
    let s = format!("{e:#}").to_lowercase();
    ["input too long", "context length", "context_length", "maximum context", "context window", "prompt is too long", "too many tokens", "reduce the length"].iter().any(|k| s.contains(k))
}

fn needs_fallback(e: &anyhow::Error) -> bool {
    let s = format!("{e:#}").to_lowercase();
    ["connect", "timed out", "timeout", "http 502", "http 503", "http 504", "dns", "budget"].iter().any(|k| s.contains(k))
}

fn short_error(e: &anyhow::Error) -> String {
    format!("{e:#}").lines().next().unwrap_or("").chars().take(120).collect()
}

fn is_retryable(e: &anyhow::Error) -> bool {
    let s = format!("{e:#}").to_lowercase();
    ["http 429", "http 5", "overloaded", "connect", "timed out", "timeout"].iter().any(|k| s.contains(k))
}

/// One model request with streaming to the UI, usage recording and retries.
async fn stream_step(
    app: &Arc<App>,
    session: &Session,
    agent: &str,
    model_spec: &str,
    provider: Arc<dyn Provider>,
    req: ChatRequest,
    cancel: &CancellationToken,
) -> Result<Step> {
    let provider_name = model_spec.split_once('/').map(|(p, _)| p).unwrap_or(model_spec);
    crate::budget::check(app, provider_name, session.owner_id.as_deref())?;
    let mut attempt = 0;
    loop {
        attempt += 1;
        let msg_id = new_id();
        let (tx, mut rx) = mpsc::channel::<StreamEvent>(256);
        let p = provider.clone();
        let r = req.clone();
        let c = cancel.clone();
        // Wait for a free slot on the model/host; not counted as time to first token.
        let registry = app.registry();
        if registry.is_saturated(model_spec) {
            app.emit(Event::Waiting { session_id: session.id.clone(), reason: Some(format!("model:{model_spec}")) });
        }
        let slot = tokio::select! {
            s = registry.acquire(model_spec) => s,
            _ = cancel.cancelled() => bail!("aborted"),
        };
        app.emit(Event::Waiting { session_id: session.id.clone(), reason: None });
        app.emit(Event::StepStarted {
            session_id: session.id.clone(),
            message_id: msg_id.clone(),
            started_at: now_ms(),
            prompt_tokens: estimate_prompt_tokens(&req),
            estimated: true,
        });
        let handle = tokio::spawn(async move {
            let _slot = slot;
            p.stream(r, tx, c).await
        });

        let started = Instant::now();
        let mut ttft: Option<u128> = None;
        let mut parts: Vec<Part> = Vec::new();
        let mut usage = Usage::default();
        let mut stop_reason = String::new();
        while let Some(ev) = rx.recv().await {
            if ttft.is_none() && !matches!(ev, StreamEvent::Usage(_) | StreamEvent::Done { .. }) {
                ttft = Some(started.elapsed().as_millis());
            }
            match ev {
                StreamEvent::TextDelta(t) => {
                    match parts.last_mut() {
                        Some(Part::Text { text }) => text.push_str(&t),
                        _ => parts.push(Part::Text { text: t.clone() }),
                    }
                    app.emit(Event::TextDelta { session_id: session.id.clone(), message_id: msg_id.clone(), text: t });
                }
                StreamEvent::ReasoningDelta(t) => {
                    match parts.last_mut() {
                        Some(Part::Reasoning { text, signature: None }) => text.push_str(&t),
                        _ => parts.push(Part::Reasoning { text: t.clone(), signature: None }),
                    }
                    app.emit(Event::ReasoningDelta { session_id: session.id.clone(), message_id: msg_id.clone(), text: t });
                }
                StreamEvent::ReasoningSignature(sig) => match parts.last_mut() {
                    Some(Part::Reasoning { signature, .. }) if signature.is_none() => *signature = Some(sig),
                    // Thinking without visible text (display omitted / redacted).
                    _ => parts.push(Part::Reasoning { text: String::new(), signature: Some(sig) }),
                },
                StreamEvent::ToolCall { id, name, input, extra } => {
                    app.emit(Event::ToolStarted {
                        session_id: session.id.clone(),
                        message_id: msg_id.clone(),
                        call_id: id.clone(),
                        name: name.clone(),
                        input: input.clone(),
                    });
                    parts.push(Part::ToolCall { id, name, input, extra });
                }
                StreamEvent::Usage(u) => usage = u,
                StreamEvent::Done { stop_reason: s } => stop_reason = s,
            }
        }
        let result = handle.await.map_err(|e| anyhow!("provider task failed: {e}"))?;
        let duration = started.elapsed().as_millis() as u64;
        let (provider_name, model_id) = model_spec.split_once('/').unwrap_or(("", model_spec));
        let record = |error: bool, usage: &Usage| UsageRecord {
            ts: now_ms(),
            session_id: session.id.clone(),
            root_session_id: app.root_session_id(&session.id),
            provider: provider_name.into(),
            model: model_id.into(),
            agent: agent.into(),
            input_tokens: usage.input_tokens,
            output_tokens: usage.output_tokens,
            cache_read_tokens: usage.cache_read_tokens,
            cache_write_tokens: usage.cache_write_tokens,
            ttft_ms: ttft.unwrap_or(duration as u128) as u64,
            duration_ms: duration,
            cost: req.model.cost(usage),
            error,
            user_id: None,
            allowance: usage.allowance,
        };

        if let Err(e) = result {
            let _ = app.store.add_usage(&record(true, &usage));
            if parts.is_empty() && attempt < 4 && is_retryable(&e) && !cancel.is_cancelled() {
                let wait = std::time::Duration::from_secs(2u64.pow(attempt));
                app.emit(Event::Error {
                    session_id: session.id.clone(),
                    message: format!("{e:#} – retrying in {}s", wait.as_secs()),
                });
                tokio::time::sleep(wait).await;
                continue;
            }
            return Err(e);
        }
        if cancel.is_cancelled() {
            // Keep what was streamed, but drop unfinished tool calls and unsigned thinking.
            parts.retain(|p| matches!(p, Part::Text { .. }));
        }
        if usage.input_tokens + usage.output_tokens > 0 || !cancel.is_cancelled() {
            let _ = app.store.add_usage(&record(false, &usage));
        }
        let mut message = Message::new(Role::Assistant, parts);
        message.id = msg_id;
        message.model = Some(model_spec.to_string());
        message.usage = Some(usage.clone());
        if !message.parts.is_empty() {
            app.store.add_message(&session.id, &message)?;
            app.emit(Event::MessageAdded { session_id: session.id.clone(), message: message.clone() });
        }
        app.update_session(&session.id, |s| s.last_usage = Some(usage))?;
        return Ok(Step { message, stop_reason });
    }
}

/// Collect a complete, tool-free answer (titles, summaries).
async fn complete_text(app: &App, spec: &str, provider: Arc<dyn Provider>, req: ChatRequest, cancel: CancellationToken) -> Result<(String, Usage)> {
    let slot = app.registry().acquire(spec).await;
    let (tx, mut rx) = mpsc::channel(256);
    let handle = tokio::spawn(async move {
        let _slot = slot;
        provider.stream(req, tx, cancel).await
    });
    let mut text = String::new();
    let mut usage = Usage::default();
    while let Some(ev) = rx.recv().await {
        match ev {
            StreamEvent::TextDelta(t) => text.push_str(&t),
            StreamEvent::Usage(u) => usage = u,
            _ => {}
        }
    }
    handle.await??;
    Ok((text, usage))
}

fn small_model(app: &App, session: &Session) -> String {
    let cfg = app.config();
    if !cfg.server.small_model.is_empty() && app.registry().resolve(&cfg.server.small_model).is_ok() {
        cfg.server.small_model.clone()
    } else {
        session.model.clone()
    }
}

fn record_side_usage(app: &App, session: &Session, spec: &str, model: &crate::config::ModelConfig, usage: &Usage, agent: &str) {
    let (p, m) = spec.split_once('/').unwrap_or(("", spec));
    let _ = app.store.add_usage(&UsageRecord {
        ts: now_ms(),
        session_id: session.id.clone(),
        root_session_id: app.root_session_id(&session.id),
        provider: p.into(),
        model: m.into(),
        agent: agent.into(),
        input_tokens: usage.input_tokens,
        output_tokens: usage.output_tokens,
        cache_read_tokens: usage.cache_read_tokens,
        cache_write_tokens: usage.cache_write_tokens,
        ttft_ms: 0,
        duration_ms: 0,
        cost: model.cost(usage),
        error: false,
        user_id: None,
        allowance: usage.allowance,
    });
}

fn maybe_generate_title(app: Arc<App>, session_id: &str, text: &str) {
    let Ok(session) = app.session(session_id) else { return };
    if session.parent_id.is_some() || !session.title.is_empty() {
        return;
    }
    let text: String = text.chars().take(2000).collect();
    tokio::spawn(async move {
        let spec = small_model(&app, &session);
        let Ok((provider, mut model)) = app.registry().resolve(&spec) else { return };
        model.effort = Some("low".into());
        let req = ChatRequest {
            system: "Write a short title (max 6 words) for a coding session that starts with the user's message below. Use the user's language. Reply with the title only, no quotes or punctuation at the end.".into(),
            messages: vec![Message::new(Role::User, vec![Part::Text { text }])],
            tools: vec![],
            max_tokens: 2000,
            reasoning: false,
            session_id: Some(session.id.clone()),
            model: model.clone(),
        };
        match complete_text(&app, &spec, provider, req, CancellationToken::new()).await {
            Ok((title, usage)) => {
                record_side_usage(&app, &session, &spec, &model, &usage, "title");
                let title = title.lines().find(|l| !l.trim().is_empty()).unwrap_or("").trim().trim_matches('"').to_string();
                if !title.is_empty() {
                    let _ = app.update_session(&session.id, |s| {
                        if s.title.is_empty() {
                            s.title = title.chars().take(80).collect();
                        }
                    });
                }
            }
            Err(e) => tracing::debug!("title generation failed: {e:#}"),
        }
    });
}

const COMPACT_PROMPT: &str = "Summarize this coding session so far so that the work can continue seamlessly from the summary alone. Include: the user's goals and explicit requests (verbatim where important), decisions made, files created/changed (with paths) and their current state, commands/tests run and results, open problems, and the exact next steps. Be thorough but concise; use headings and bullet points.";

/// Suggest a title for a session from its conversation, with the given model
/// (or the small model). The title is returned, not saved.
pub async fn suggest_title(app: Arc<App>, session_id: &str, model_spec: Option<String>) -> Result<String> {
    let session = app.session(session_id)?;
    let spec = model_spec.filter(|m| !m.is_empty()).unwrap_or_else(|| small_model(&app, &session));
    let (provider, mut model) = app.registry().resolve(&spec)?;
    model.effort = Some("low".into());
    let history = active_history(app.store.messages(session_id)?);
    let mut transcript = transcript_of(&history);
    if transcript.len() > 12_000 {
        // Beginning (the goal) and end (where it went).
        let head = clip(&transcript, 6_000);
        let mut start = transcript.len() - 5_000;
        while !transcript.is_char_boundary(start) {
            start += 1;
        }
        transcript = format!("{head}\n…\n{}", &transcript[start..]);
    }
    if transcript.trim().is_empty() {
        bail!("the session has no messages yet");
    }
    let req = ChatRequest {
        system: "Write a short, specific title (max 6 words) for the coding session below. Use the language of the user's messages. Reply with the title only, no quotes, no punctuation at the end.".into(),
        messages: vec![Message::new(Role::User, vec![Part::Text { text: format!("<session>\n{transcript}\n</session>") }])],
        tools: vec![],
        max_tokens: 2000,
        reasoning: false,
        session_id: Some(session.id.clone()),
        model: model.clone(),
    };
    let (text, usage) = complete_text(&app, &spec, provider, req, CancellationToken::new()).await?;
    record_side_usage(&app, &session, &spec, &model, &usage, "title");
    let text = strip_think(&text);
    let title = text.lines().find(|l| !l.trim().is_empty()).unwrap_or("").trim().trim_matches(['"', '\'', '*', '#', ' ']).to_string();
    if title.is_empty() {
        bail!("the model returned no title");
    }
    Ok(title.chars().take(80).collect())
}

/// Write only the hunks of a proposed write/edit that the user accepted.
async fn apply_partial(ctx: &ToolCtx, name: &str, input: &Value, selected: &[usize]) -> Result<tools::ToolOutput> {
    let path = ctx.resolve(tools::arg_str(input, "path")?)?;
    let old = tokio::fs::read_to_string(&path).await.unwrap_or_default();
    let proposed = if name == "write" { tools::arg_str(input, "content")?.to_string() } else { tools::fs::apply_edit(&old, input)? };
    let (result, total) = tools::fs::apply_hunks(&old, &proposed, selected);
    let accepted = selected.iter().filter(|i| **i < total).count();
    if let Some(dir) = path.parent() {
        tokio::fs::create_dir_all(dir).await?;
    }
    tokio::fs::write(&path, &result).await?;
    ctx.app.mark_read(&ctx.session_id, &path);
    let rel = path.strip_prefix(&ctx.cwd).map(|p| p.display().to_string()).unwrap_or_else(|_| path.display().to_string());
    let applied = tools::fs::unified_diff(&rel, &old, &result);
    let rejected = tools::fs::unified_diff(&rel, &result, &proposed);
    Ok(tools::ToolOutput::text(format!(
        "The user reviewed your change to {rel} and applied only {accepted} of {total} hunks. \
         The file now contains the accepted parts; the rejected parts were NOT applied – do not re-apply them unless the user asks.\n\n\
         Rejected (what is still missing compared to your proposal):\n{}",
        clip(&rejected, 8000)
    ))
    .with_meta(serde_json::json!({"diff": applied, "path": path, "partial": {"accepted": accepted, "total": total}})))
}

const PHASES: &[&str] = &["study", "specify", "build", "check", "ready"];

/// Craft phase implied by a tool call (fallback when the agent does not report it).
fn infer_phase(tool: &str, input: &Value, deep: bool) -> Option<&'static str> {
    let path = input["path"].as_str().unwrap_or("");
    let label = input["label"].as_str().unwrap_or("");
    let cmd = input["command"].as_str().unwrap_or("");
    Some(match tool {
        "write" | "edit" => {
            let doc = path.contains(".codecraft/spec") || path.starts_with("spec/") || path.contains("/spec/") || path.starts_with("docs/") || path.contains("proposal/");
            if doc && deep { "specify" } else if doc { return None } else { "build" }
        }
        "image_compare" | "openscad" => "check",
        "browser_snapshot" if label.starts_with("build") => "check",
        "bash" if ["test", "pytest", "cargo check", "cargo build", "npm run build", "http.server", "serve", "vite"].iter().any(|k| cmd.contains(k)) => "check",
        "read" | "glob" | "grep" | "list" | "browser_snapshot" | "web_fetch" | "web_search" | "binary_info" | "decompile" => "study",
        _ => return None,
    })
}

/// Move the craft phase forward (never backwards); `only_from` restricts the transition.
fn advance_phase(app: &App, session_id: &str, phase: &str, only_from: Option<&str>) {
    let root = app.root_session_id(session_id);
    let Ok(s) = app.session(&root) else { return };
    let cur = s.phase.as_deref().and_then(|p| PHASES.iter().position(|x| *x == p));
    if let Some(from) = only_from {
        if s.phase.as_deref() != Some(from) {
            return;
        }
    }
    let Some(next) = PHASES.iter().position(|x| *x == phase) else { return };
    // Studying again after building is normal – no step back automatically.
    if cur.is_some_and(|c| c >= next) {
        return;
    }
    // Clean room has no specify phase.
    let phase = if phase == "specify" && s.clean_room != Some(false) { "study" } else { phase };
    if s.phase.as_deref() == Some(phase) {
        return;
    }
    let _ = app.update_session(&root, |s| s.phase = Some(phase.to_string()));
}

/// Review the last turn against the user's request. Returns feedback for the
/// agent, or None when the reviewer approves.
#[allow(clippy::too_many_arguments)]
async fn review_turn(app: &Arc<App>, session: &Session, agent: &prompts::Agent, model_spec: &str, since: i64, todos: Option<&[String]>, round: u32, rounds: u32, cancel: &CancellationToken) -> Result<Option<String>> {
    let refine = agent.refine.clone().unwrap_or_default();
    let spec = refine.model.clone().filter(|m| !m.is_empty()).unwrap_or_else(|| model_spec.to_string());
    let (provider, mut model) = app.registry().resolve(&spec)?;
    model.effort = Some("medium".into());
    app.emit(Event::Waiting { session_id: session.id.clone(), reason: Some(format!("review:{round}/{rounds}")) });
    let result = async {
        let history = active_history(app.store.messages_for_model(&session.id)?);
        // The user's request: the last user message with text that is not feedback/tool output.
        let req_idx = history
            .iter()
            .rposition(|m| matches!(m.role, Role::User) && m.parts.iter().any(|p| matches!(p, Part::Text { text } if !text.starts_with("<refine-feedback") && !text.starts_with("<hook-"))))
            .unwrap_or(0);
        let request: String = history.get(req_idx).map(|m| m.parts.iter().filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None }).collect::<Vec<_>>().join("\n")).unwrap_or_default();
        // A todo review looks at the work since the last review, the end-of-turn review at the whole turn.
        let from = match todos {
            Some(_) => history.iter().position(|m| m.created_at >= since).unwrap_or(history.len()).max(req_idx + 1),
            None => req_idx + 1,
        };
        let mut work = transcript_of(&history[from.min(history.len())..]);
        if work.len() > 30_000 {
            let mut start = work.len() - 30_000;
            while !work.is_char_boundary(start) {
                start += 1;
            }
            work = format!("…{}", &work[start..]);
        }
        // Actual file changes of this turn (checkpoint vs. current content).
        let mut changes = String::new();
        let mut seen = std::collections::HashSet::new();
        for (path, before) in app.store.checkpoints_since(&session.id, since)? {
            if !seen.insert(path.clone()) || changes.len() > 30_000 {
                continue;
            }
            let old = before.map(|b| String::from_utf8_lossy(&b).into_owned()).unwrap_or_default();
            let new = std::fs::read_to_string(&path).unwrap_or_default();
            let diff = similar::TextDiff::from_lines(&old, &new).unified_diff().context_radius(3).header(&path, &path).to_string();
            changes.push_str(&clip(&diff, 12_000));
            changes.push('\n');
        }
        let criteria = refine.criteria.clone().filter(|c| !c.trim().is_empty()).unwrap_or_else(|| prompts::refine_criteria(&agent.name).to_string());
        let system = "You review the work of an AI coding agent for the user. Be strict about what matters for the user's request and ignore style nitpicks. \
            Reply in exactly this format: the first line is APPROVED or CHANGES_NEEDED. For CHANGES_NEEDED, follow with at most 8 concrete, actionable findings as a bullet list \
            (with file:line where possible). If the important work is done and correct, reply APPROVED.";
        let focus = match todos {
            Some(items) => format!(
                "<completed-todos>\n{}\n</completed-todos>\nThe agent just marked these todo items as completed. Review only whether they are really done and correct; other parts of the request may still be in progress.\n\n",
                items.iter().map(|t| format!("- {t}")).collect::<Vec<_>>().join("\n")
            ),
            None => String::new(),
        };
        let body = format!(
            "<request>\n{request}\n</request>\n\n{focus}<criteria>\n{criteria}\n</criteria>\n\n<agent-work mode=\"{}\">\n{work}\n</agent-work>\n\n<file-changes>\n{}\n</file-changes>",
            agent.name,
            if changes.is_empty() { "(no files changed since the last review)" } else { changes.as_str() }
        );
        let req = ChatRequest { system: system.into(), messages: vec![Message::new(Role::User, vec![Part::Text { text: body }])], tools: vec![], max_tokens: 4000, reasoning: false, model: model.clone(), session_id: Some(session.id.clone()) };
        let (text, usage) = complete_text(app, &spec, provider, req, cancel.clone()).await?;
        record_side_usage(app, session, &spec, &model, &usage, "review");
        let text = strip_think(&text);
        let mut lines = text.lines().skip_while(|l| l.trim().is_empty());
        let verdict = lines.next().unwrap_or("").trim().to_uppercase();
        if verdict.starts_with("APPROVED") || text.trim().is_empty() {
            return Ok(None);
        }
        let findings: String = lines.collect::<Vec<_>>().join("\n");
        let findings = if findings.trim().is_empty() { text.clone() } else { findings };
        Ok(Some(match todos {
            Some(items) => format!(
                "<refine-feedback round=\"{round}/{rounds}\" reviewer=\"{spec}\" todos=\"{}\">\n{}\n</refine-feedback>\nThe review of the todo(s) you just completed found problems, so they are open again. Fix these findings (or explain briefly why one does not apply), verify, then mark the todo(s) completed again – that triggers the next review.",
                items.len(),
                findings.trim()
            ),
            None => format!(
                "<refine-feedback round=\"{round}/{rounds}\" reviewer=\"{spec}\">\n{}\n</refine-feedback>\nAddress these review findings (or explain briefly why one does not apply), verify, then finish.",
                findings.trim()
            ),
        }))
    }
    .await;
    app.emit(Event::Waiting { session_id: session.id.clone(), reason: None });
    result
}

/// Remove a leading `<think>…</think>` block some models put into the text.
fn strip_think(text: &str) -> String {
    let t = text.trim_start();
    if t.starts_with("<think>") {
        if let Some(end) = t.find("</think>") {
            return t[end + 8..].trim().to_string();
        }
        return String::new();
    }
    text.trim().to_string()
}

fn clip(s: &str, max: usize) -> String {
    if s.len() <= max {
        return s.to_string();
    }
    let mut end = max;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    format!("{} […{} more bytes]", &s[..end], s.len() - end)
}

/// The history as readable text for summarizing (tool output shortened).
pub fn transcript_of(history: &[Message]) -> String {
    let mut out = String::new();
    for m in history {
        let role = if matches!(m.role, Role::User) { "USER" } else { "ASSISTANT" };
        let mut body = String::new();
        for p in &m.parts {
            match p {
                Part::Text { text } => body.push_str(&format!("{text}\n")),
                Part::Summary { text, .. } => body.push_str(&format!("[Summary of the earlier conversation]\n{text}\n")),
                Part::ToolCall { name, input, .. } => body.push_str(&format!("[tool call {name}: {}]\n", clip(&input.to_string(), 1500))),
                Part::ToolResult { name, content, is_error, .. } => {
                    body.push_str(&format!("[{} of {name}]\n{}\n", if *is_error { "error" } else { "result" }, clip(content, 3000)))
                }
                Part::Image { .. } => body.push_str("[image]\n"),
                Part::Reasoning { .. } => {}
            }
        }
        if !body.trim().is_empty() {
            out.push_str(&format!("### {role}\n{}\n", body.trim_end()));
        }
    }
    out
}

/// Replace the active history with a summary (the old messages stay visible in the UI).
/// Summarize the history. `auto` = (context tokens, context limit) when triggered by a
/// nearly full context; the chat shows the progress and the reason.
pub async fn compact(app: Arc<App>, session_id: &str, cancel: CancellationToken, auto: Option<(u64, u64)>) -> Result<()> {
    let event = |state: &str| Event::Compaction {
        session_id: session_id.to_string(),
        state: state.into(),
        auto: auto.is_some(),
        tokens: auto.map(|a| a.0),
        limit: auto.map(|a| a.1),
    };
    app.emit(event("started"));
    let res = compact_inner(app.clone(), session_id, cancel, auto).await;
    app.emit(event(if res.is_ok() { "done" } else { "failed" }));
    res
}

async fn compact_inner(app: Arc<App>, session_id: &str, cancel: CancellationToken, auto: Option<(u64, u64)>) -> Result<()> {
    let session = app.session(session_id)?;
    let history = active_history(app.store.messages_for_model(session_id)?);
    if history.is_empty() {
        return Ok(());
    }
    let spec = session.model.clone();
    let (provider, model) = app.registry().resolve(&spec)?;
    // The history goes in as a plain transcript: with real tool calls in the
    // history (and no tools offered) some models answer with a tool call or
    // only think, leaving the summary empty.
    let transcript = transcript_of(&history);
    let mut summary = String::new();
    for attempt in 0..2 {
        let prompt = if attempt == 0 {
            format!("<transcript>\n{transcript}\n</transcript>\n\n{COMPACT_PROMPT}")
        } else {
            format!("<transcript>\n{transcript}\n</transcript>\n\n{COMPACT_PROMPT}\n\nAnswer directly with the summary as plain Markdown text. Do not call tools.")
        };
        let req = ChatRequest {
            system: "You are a precise summarizer of software engineering sessions. You never call tools; you only write the requested summary.".into(),
            messages: vec![Message::new(Role::User, vec![Part::Text { text: prompt }])],
            tools: vec![],
            max_tokens: 16_000,
            reasoning: false,
            session_id: Some(session.id.clone()),
            model: model.clone(),
        };
        let (text, usage) = complete_text(&app, &spec, provider.clone(), req, cancel.clone()).await?;
        record_side_usage(&app, &session, &spec, &model, &usage, "compaction");
        summary = strip_think(&text);
        if !summary.trim().is_empty() {
            break;
        }
        tracing::warn!("compaction of {session_id}: empty summary (attempt {})", attempt + 1);
    }
    if summary.trim().is_empty() {
        bail!("compaction produced an empty summary");
    }
    let tokens = auto.map(|a| a.0).or_else(|| session.last_usage.as_ref().map(|u| u.context_tokens()));
    save_user_parts(&app, session_id, vec![Part::Summary { text: summary, auto: auto.is_some(), tokens }])?;
    app.update_session(session_id, |s| s.last_usage = None)?;
    // Files must be re-read after compaction before editing.
    app.with_runtime(session_id, |r| r.read_files.clear());
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn todo_completion_and_reopening() {
        let before = serde_json::json!([{"content": "A", "status": "in_progress"}, {"content": "B", "status": "completed"}, {"content": "C", "status": "pending"}]);
        let after = serde_json::json!([{"content": "A", "status": "completed"}, {"content": "B", "status": "completed"}, {"content": "C", "status": "pending"}]);
        assert_eq!(newly_completed(&before, &after), vec!["A".to_string()]);
        assert_eq!(newly_completed(&Value::Null, &after), vec!["A".to_string(), "B".to_string()]);
        assert_eq!(open_todos(&after), vec!["C".to_string()]);
        let reopened = reopen_todos(&after, &["A".to_string()]);
        assert_eq!(reopened[0]["status"], "in_progress");
        assert_eq!(reopened[1]["status"], "completed");
    }

    #[test]
    fn answer_room_and_overflow() {
        // The case from production: DeepSeek V4.1 Flash, 1M window, 384K output, 720K input.
        let (ctx, out) = (1_048_576u64, 393_216u64);
        let b = answer_budget(out, Some(ctx), 720_000);
        assert!(720_000 + b <= ctx && b > 300_000, "{b}");
        assert_eq!(answer_budget(out, Some(ctx), 100_000), out);
        assert_eq!(answer_budget(16_000, None, 5_000_000), 16_000);
        assert_eq!(answer_budget(out, Some(ctx), 1_040_000), 4_096);
        assert!(is_context_overflow(&anyhow!("HTTP 400: [invalid_request_error] Input too long: model limit 1048576 tokens, requested 1108739 tokens")));
        assert!(is_context_overflow(&anyhow!("This model's maximum context length is 128000 tokens")));
        assert!(!is_context_overflow(&anyhow!("HTTP 429 rate limited")));
        // Estimate: last request + what came after its answer (about 3 chars per token).
        let u = crate::types::Usage { input_tokens: 1_000, cache_read_tokens: 9_000, output_tokens: 500, ..Default::default() };
        let mut call = Message::new(Role::Assistant, vec![call("c1")]);
        call.id = "a".into();
        let result = Message::new(Role::User, vec![Part::ToolResult { call_id: "c1".into(), name: "bash".into(), content: "x".repeat(3_000), images: vec![], is_error: false, meta: None }]);
        assert_eq!(estimate_input(Some(&u), &[call.clone(), result.clone()]), 10_500 + 1_000);
        assert!(estimate_input(None, &[call, result]) >= 1_000 + 8_000);
    }

    #[test]
    fn detects_commands_that_move_files() {
        assert!(moves_files("mkdir -p art/old && mv art/*.png art/old/"));
        assert!(moves_files("git mv a.png b.png"));
        assert!(moves_files("cd art;mv x y"));
        assert!(!moves_files("ls -la movies && cat remove.txt"));
    }

    fn call(id: &str) -> Part {
        Part::ToolCall { id: id.into(), name: "bash".into(), input: serde_json::json!({}), extra: None }
    }

    #[test]
    fn dangling_tool_calls_get_results() {
        let a = Message::new(Role::Assistant, vec![call("1"), call("2")]);
        let r = Message::new(Role::User, vec![tool_result("1", "bash", "ok".into(), false, None)]);
        let a2 = Message::new(Role::Assistant, vec![call("3")]);
        let h = active_history(vec![Message::new(Role::User, vec![Part::Text { text: "hi".into() }]), a, r, a2]);
        // user, assistant(1,2), user(result 2 + result 1), assistant(3), user(result 3)
        assert_eq!(h.len(), 5);
        let ids = |m: &Message| m.parts.iter().filter_map(|p| if let Part::ToolResult { call_id, .. } = p { Some(call_id.clone()) } else { None }).collect::<Vec<_>>();
        assert_eq!(ids(&h[2]), vec!["2", "1"]);
        assert_eq!(ids(&h[4]), vec!["3"]);
    }
}