markon-core 0.15.14

//! Axum handlers for the chat HTTP surface.
//!
//! Routes (all gated by per-workspace `enable_chat`):
//!   - `POST   /api/chat/{workspace_id}` — SSE stream of [`crate::chat::agent::AgentEvent`]
//!   - `GET    /api/chat/{workspace_id}/files?q=<prefix>` — @-mention completions
//!   - `GET    /api/chat/{workspace_id}/threads` — list of threads + summaries
//!   - `POST   /api/chat/{workspace_id}/threads` — create empty thread
//!   - `GET    /api/chat/{workspace_id}/threads/{thread_id}` — thread + messages
//!   - `DELETE /api/chat/{workspace_id}/threads/{thread_id}` — drop thread

use axum::{
    extract::{Path as AxumPath, Query, State},
    http::StatusCode,
    response::{
        sse::{Event, KeepAlive, Sse},
        IntoResponse, Response,
    },
    routing::{get, post},
    Json, Router,
};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use std::convert::Infallible;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

use crate::chat::agent::{auto_title, Agent, AgentEvent, AgentRequest};
use crate::chat::config::ChatRuntimeConfig;
use crate::chat::edits::{Resolution, ResolveError};
use crate::chat::message::{ContentBlock, Message, Role};
use crate::chat::prompt::{build_system_blocks, render_mention_block, PromptInputs};
use crate::chat::provider;
use crate::chat::storage::{ChatStorage, StorageError, Thread};
use crate::chat::tools::{looks_binary, ToolRegistry, MAX_FILE_BYTES};
use crate::server::AppState;
use crate::settings::AppSettings;
use crate::workspace::WorkspaceEntry;

/// Cap how many tool-use rounds the agent will run for a single user turn,
/// independent of the model's `max_tokens`. 8 leaves headroom for "look it
/// up, then verify, then summarize" without letting a confused model burn
/// budget indefinitely.
const MAX_AGENT_STEPS: u8 = 8;
const MAX_TOKENS_PER_TURN: u32 = 4096;

pub(crate) fn router() -> Router<AppState> {
    Router::new()
        .route("/api/chat/{workspace_id}", post(chat_stream_handler))
        .route("/api/chat/{workspace_id}/files", get(list_files_handler))
        .route(
            "/api/chat/{workspace_id}/threads",
            get(list_threads_handler).post(create_thread_handler),
        )
        .route(
            "/api/chat/{workspace_id}/threads/{thread_id}",
            get(get_thread_handler).delete(delete_thread_handler),
        )
        .route(
            "/api/chat/{workspace_id}/edits/{edit_id}/apply",
            post(apply_edit_handler),
        )
        .route(
            "/api/chat/{workspace_id}/edits/{edit_id}/reject",
            post(reject_edit_handler),
        )
        .route(
            "/api/chat/{workspace_id}/edits/undo",
            post(undo_edit_handler),
        )
}

// ── helpers ──────────────────────────────────────────────────────────────────

fn ensure_chat_enabled(
    state: &AppState,
    workspace_id: &str,
) -> Result<Arc<WorkspaceEntry>, ChatHttpError> {
    let ws = state
        .workspace_registry
        .get(workspace_id)
        .ok_or(ChatHttpError::NotFound)?;
    if !ws.enable_chat.load(Ordering::Relaxed) {
        return Err(ChatHttpError::Disabled);
    }
    Ok(ws)
}

fn storage_for(state: &AppState) -> Result<ChatStorage, ChatHttpError> {
    state
        .db
        .clone()
        .map(ChatStorage::new)
        .ok_or_else(|| ChatHttpError::Unavailable("chat persistence not initialized".into()))
}

#[derive(Debug)]
enum ChatHttpError {
    NotFound,
    Disabled,
    BadRequest(String),
    Unavailable(String),
    Storage(StorageError),
}

impl From<StorageError> for ChatHttpError {
    fn from(e: StorageError) -> Self {
        match e {
            StorageError::NotFound => Self::NotFound,
            other => Self::Storage(other),
        }
    }
}

impl IntoResponse for ChatHttpError {
    fn into_response(self) -> Response {
        let (status, body) = match self {
            Self::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()),
            Self::Disabled => (
                StatusCode::FORBIDDEN,
                "chat is not enabled for this workspace".to_string(),
            ),
            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m),
            Self::Unavailable(m) => (StatusCode::SERVICE_UNAVAILABLE, m),
            Self::Storage(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("storage: {e}")),
        };
        (status, body).into_response()
    }
}

/// Fetch a thread and verify it belongs to `workspace_id`. A mismatch is
/// reported as `NotFound`, indistinguishable from a missing thread, so ids
/// can't be probed across workspaces.
async fn thread_in_workspace(
    storage: &ChatStorage,
    thread_id: &str,
    workspace_id: &str,
) -> Result<Thread, ChatHttpError> {
    let thread = storage.get_thread(thread_id).await?;
    if thread.workspace_id != workspace_id {
        return Err(ChatHttpError::NotFound);
    }
    Ok(thread)
}

// ── threads ──────────────────────────────────────────────────────────────────

async fn list_threads_handler(
    State(state): State<AppState>,
    AxumPath(workspace_id): AxumPath<String>,
) -> Result<impl IntoResponse, ChatHttpError> {
    ensure_chat_enabled(&state, &workspace_id)?;
    let storage = storage_for(&state)?;
    let summaries = storage.list_thread_summaries(&workspace_id).await?;
    Ok(Json(summaries))
}

#[derive(Debug, Deserialize)]
struct CreateThreadBody {
    #[serde(default)]
    title: String,
}

async fn create_thread_handler(
    State(state): State<AppState>,
    AxumPath(workspace_id): AxumPath<String>,
    Json(body): Json<CreateThreadBody>,
) -> Result<impl IntoResponse, ChatHttpError> {
    ensure_chat_enabled(&state, &workspace_id)?;
    let storage = storage_for(&state)?;
    let title = if body.title.trim().is_empty() {
        "New chat"
    } else {
        body.title.trim()
    };
    let thread = storage.create_thread(&workspace_id, title).await?;
    Ok(Json(thread))
}

#[derive(Debug, Serialize)]
struct ThreadDetail {
    thread: crate::chat::storage::Thread,
    messages: Vec<MessageView>,
}

#[derive(Debug, Serialize)]
struct MessageView {
    seq: i64,
    role: Role,
    content: Vec<crate::chat::message::ContentBlock>,
    created_at: i64,
}

async fn get_thread_handler(
    State(state): State<AppState>,
    AxumPath((workspace_id, thread_id)): AxumPath<(String, String)>,
) -> Result<impl IntoResponse, ChatHttpError> {
    ensure_chat_enabled(&state, &workspace_id)?;
    let storage = storage_for(&state)?;
    let thread = thread_in_workspace(&storage, &thread_id, &workspace_id).await?;
    let messages = storage
        .list_messages(&thread_id)
        .await?
        .into_iter()
        .map(|m| MessageView {
            seq: m.seq,
            role: m.role,
            content: m.content,
            created_at: m.created_at,
        })
        .collect();
    Ok(Json(ThreadDetail { thread, messages }))
}

async fn delete_thread_handler(
    State(state): State<AppState>,
    AxumPath((workspace_id, thread_id)): AxumPath<(String, String)>,
) -> Result<impl IntoResponse, ChatHttpError> {
    ensure_chat_enabled(&state, &workspace_id)?;
    let storage = storage_for(&state)?;
    thread_in_workspace(&storage, &thread_id, &workspace_id).await?;
    storage.delete_thread(&thread_id).await?;
    Ok(StatusCode::NO_CONTENT)
}

// ── pending-edit resolution (edit_file tool) ────────────────────────────────

/// Outcome of a [`drift_guarded_replace`] attempt.
enum ReplaceOutcome {
    /// The file is unreadable or `find` no longer matches exactly once.
    Drifted,
    /// Replacement written; `line` is the 1-based line of the match.
    Applied { line: usize },
    /// The drift check passed but the write-back failed.
    WriteErr(std::io::Error),
}

/// Re-read `abs`, verify `find` occurs *exactly once*, and swap it for
/// `replace`. Shared by the apply and undo paths so both keep identical
/// drift semantics.
fn drift_guarded_replace(abs: &std::path::Path, find: &str, replace: &str) -> ReplaceOutcome {
    let current = match std::fs::read_to_string(abs) {
        Ok(s) => s,
        Err(_) => return ReplaceOutcome::Drifted,
    };
    let mut occurrences = current.match_indices(find).map(|(off, _)| off);
    let offset = match (occurrences.next(), occurrences.next()) {
        (Some(off), None) => off,
        _ => return ReplaceOutcome::Drifted,
    };
    let line = 1 + current[..offset].bytes().filter(|b| *b == b'\n').count();
    let updated = current.replacen(find, replace, 1);
    match std::fs::write(abs, updated.as_bytes()) {
        Ok(()) => ReplaceOutcome::Applied { line },
        Err(e) => ReplaceOutcome::WriteErr(e),
    }
}

/// Resolve a pending `edit_file` proposal by writing the new content to disk
/// and signalling the awaiting tool.
///
/// Drift defence: re-read the target file and verify `old_string` still
/// matches *exactly once* before writing. If the file has changed between
/// the proposal and this apply (e.g. via the editor save endpoint or a
/// shared-mode broadcast) we send `Resolution::Drifted` instead of writing.
async fn apply_edit_handler(
    State(state): State<AppState>,
    AxumPath((workspace_id, edit_id)): AxumPath<(String, String)>,
) -> Result<impl IntoResponse, ChatHttpError> {
    let ws = ensure_chat_enabled(&state, &workspace_id)?;
    if !ws.enable_edit.load(Ordering::Relaxed) {
        return Err(ChatHttpError::Disabled);
    }

    let snap = ws
        .pending_edits
        .snapshot(&edit_id)
        .ok_or(ChatHttpError::NotFound)?;

    // Re-resolve the path against the workspace root at apply time, exactly as
    // undo_edit_handler does. The proposal-time sandbox check (ToolContext in
    // edit_file) can be invalidated by a symlink swapped in between proposal
    // and apply (TOCTOU); a path that no longer canonicalizes inside the
    // workspace is treated as drift rather than written through.
    let ctx = crate::chat::tools::ToolContext::for_workspace(ws.fs.clone())
        .map_err(|e| ChatHttpError::Unavailable(format!("workspace root: {e}")))?;
    let abs = match ctx.resolve(&snap.path) {
        Ok(p) => p,
        Err(_) => {
            let _ = ws.pending_edits.resolve(&edit_id, Resolution::Drifted);
            return Ok(Json(EditResolutionResponse { status: "drifted" }));
        }
    };
    let line = match drift_guarded_replace(&abs, &snap.old_string, &snap.new_string) {
        ReplaceOutcome::Drifted => {
            // File vanished or changed — surface "drifted" to the model.
            let _ = ws.pending_edits.resolve(&edit_id, Resolution::Drifted);
            return Ok(Json(EditResolutionResponse { status: "drifted" }));
        }
        ReplaceOutcome::WriteErr(e) => {
            return Err(ChatHttpError::Unavailable(format!("write failed: {e}")));
        }
        ReplaceOutcome::Applied { line } => line,
    };

    match ws
        .pending_edits
        .resolve(&edit_id, Resolution::Applied { line })
    {
        Ok(()) => Ok(Json(EditResolutionResponse { status: "applied" })),
        Err(ResolveError::Unknown) => Err(ChatHttpError::NotFound),
        // The model's tool task is gone but the file write already
        // succeeded; report success rather than 500.
        Err(ResolveError::AlreadyResolved | ResolveError::ReceiverGone) => {
            Ok(Json(EditResolutionResponse { status: "applied" }))
        }
    }
}

async fn reject_edit_handler(
    State(state): State<AppState>,
    AxumPath((workspace_id, edit_id)): AxumPath<(String, String)>,
) -> Result<impl IntoResponse, ChatHttpError> {
    let ws = ensure_chat_enabled(&state, &workspace_id)?;
    if !ws.enable_edit.load(Ordering::Relaxed) {
        return Err(ChatHttpError::Disabled);
    }
    match ws.pending_edits.resolve(&edit_id, Resolution::Rejected) {
        Ok(()) => Ok(Json(EditResolutionResponse { status: "rejected" })),
        Err(ResolveError::Unknown) => Err(ChatHttpError::NotFound),
        Err(ResolveError::AlreadyResolved | ResolveError::ReceiverGone) => {
            Ok(Json(EditResolutionResponse { status: "rejected" }))
        }
    }
}

#[derive(Serialize)]
struct EditResolutionResponse {
    status: &'static str,
}

/// Undo a previously-applied `edit_file` by searching for the now-current
/// text and replacing it with the prior text. Stateless: the diff card on
/// the client already holds both sides of the swap, so we don't need a
/// separate "applied history" store on the server. Same drift detection
/// as the apply path — if `find` doesn't appear exactly once we return
/// `drifted` instead of writing.
///
/// Trust model: this endpoint accepts an arbitrary find/replace pair, but
/// the cross-flag gate (`--enable-chat && --enable-edit`) means any caller
/// with reach to it already has `/api/save` available for arbitrary writes,
/// so it doesn't widen the attack surface beyond what the editor already
/// exposes.
#[derive(Deserialize)]
struct UndoEditRequest {
    /// Workspace-relative path. Re-validated through `ToolContext::resolve`.
    path: String,
    /// Current file content to locate (the previous edit's `new_string`).
    find: String,
    /// What to write in its place (the previous edit's `old_string`).
    replace_with: String,
}

async fn undo_edit_handler(
    State(state): State<AppState>,
    AxumPath(workspace_id): AxumPath<String>,
    Json(body): Json<UndoEditRequest>,
) -> Result<impl IntoResponse, ChatHttpError> {
    let ws = ensure_chat_enabled(&state, &workspace_id)?;
    if !ws.enable_edit.load(Ordering::Relaxed) {
        return Err(ChatHttpError::Disabled);
    }
    if body.find.is_empty() {
        return Err(ChatHttpError::BadRequest("find must not be empty".into()));
    }
    if body.find == body.replace_with {
        return Err(ChatHttpError::BadRequest(
            "find and replace_with are identical — no-op".into(),
        ));
    }

    // Re-validate path under the workspace sandbox. ToolContext::resolve
    // rejects `..`, absolute paths, and symlink escapes, then canonicalizes;
    // we reuse it rather than reimplementing the same check here.
    let ctx = match crate::chat::tools::ToolContext::for_workspace(ws.fs.clone()) {
        Ok(c) => c,
        Err(e) => {
            return Err(ChatHttpError::Unavailable(format!("workspace root: {e}")));
        }
    };
    let abs = ctx
        .resolve(&body.path)
        .map_err(|_| ChatHttpError::NotFound)?;

    match drift_guarded_replace(&abs, &body.find, &body.replace_with) {
        ReplaceOutcome::Drifted => Ok(Json(EditResolutionResponse { status: "drifted" })),
        ReplaceOutcome::WriteErr(e) => {
            Err(ChatHttpError::Unavailable(format!("write failed: {e}")))
        }
        ReplaceOutcome::Applied { .. } => Ok(Json(EditResolutionResponse { status: "reverted" })),
    }
}

// ── file autocomplete (@-mention) ────────────────────────────────────────────

#[derive(Debug, Deserialize)]
struct FileQuery {
    #[serde(default)]
    q: String,
    #[serde(default)]
    limit: Option<usize>,
}

#[derive(Debug, Serialize)]
struct FileSuggestion {
    path: String,
    /// Higher = better match. Used for client-side stable sort if the server
    /// later returns multiple sources.
    score: i32,
}

async fn list_files_handler(
    State(state): State<AppState>,
    AxumPath(workspace_id): AxumPath<String>,
    Query(query): Query<FileQuery>,
) -> Result<impl IntoResponse, ChatHttpError> {
    let ws = ensure_chat_enabled(&state, &workspace_id)?;
    let limit = query.limit.unwrap_or(50).clamp(1, 200);
    let needle = query.q.trim().to_lowercase();
    let workspace_fs = ws.fs.clone();

    // The walk and the text-sniffs are blocking filesystem I/O — keep them
    // off the async pool (same policy as `ChatStorage::with_conn`).
    let suggestions = tokio::task::spawn_blocking(move || {
        let mut candidates: Vec<(std::path::PathBuf, FileSuggestion)> = Vec::new();

        for (rel, path) in workspace_fs.content_files(10_000) {
            let rel_str = rel.as_route();

            // Skip likely-binary files cheaply: by extension and by metadata
            // size. The NUL-byte sniff (which has to open the file) waits
            // until after sorting, so only potential winners are opened.
            if BINARY_EXT.iter().any(|ext| rel_str.ends_with(ext)) {
                continue;
            }
            if let Ok(meta) = std::fs::metadata(&path) {
                if meta.len() > MAX_FILE_BYTES {
                    // Big files can be referenced by typing the path explicitly,
                    // but we keep them out of autocomplete to discourage accidents.
                    continue;
                }
            }
            let score = match score_match(&needle, &rel_str) {
                Some(s) => s,
                None => continue,
            };
            candidates.push((
                path,
                FileSuggestion {
                    path: rel_str,
                    score,
                },
            ));
        }

        candidates.sort_by(|a, b| {
            b.1.score
                .cmp(&a.1.score)
                .then_with(|| a.1.path.cmp(&b.1.path))
        });

        // Sniff in sort order and stop once `limit` text files are found, so
        // the open/read work is bounded by `limit`, not by the workspace size.
        let mut suggestions: Vec<FileSuggestion> = Vec::new();
        for (abs, suggestion) in candidates {
            if suggestions.len() >= limit {
                break;
            }
            if is_text_file_quick(&abs) {
                suggestions.push(suggestion);
            }
        }
        suggestions
    })
    .await
    .map_err(|e| ChatHttpError::Unavailable(format!("file walk: {e}")))?;

    Ok(Json(suggestions))
}

const BINARY_EXT: &[&str] = &[
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".ico", ".tiff", ".heic", ".avif", ".mp3",
    ".mp4", ".mov", ".webm", ".ogg", ".wav", ".flac", ".m4a", ".pdf", ".zip", ".tar", ".gz",
    ".bz2", ".7z", ".rar", ".woff", ".woff2", ".ttf", ".otf", ".eot", ".so", ".dylib", ".dll",
    ".exe", ".class", ".jar", ".wasm", ".pyc",
];

/// Rough relevance score. Returns `None` if `needle` doesn't match at all.
/// Higher = better. Scoring tiers:
///   100 — needle == basename
///    80 — basename starts with needle
///    60 — basename contains needle
///    40 — full path contains needle
///     0 — empty needle (everything matches at score 0)
fn score_match(needle: &str, rel: &str) -> Option<i32> {
    if needle.is_empty() {
        return Some(0);
    }
    let lower = rel.to_lowercase();
    let basename = rel.rsplit('/').next().unwrap_or(rel).to_lowercase();
    if basename == needle {
        return Some(100);
    }
    if basename.starts_with(needle) {
        return Some(80);
    }
    if basename.contains(needle) {
        return Some(60);
    }
    if lower.contains(needle) {
        return Some(40);
    }
    None
}

/// Open the first 4 KiB and check for NUL bytes. Cheap enough for an
/// autocomplete handler that returns up to a few dozen entries.
fn is_text_file_quick(path: &std::path::Path) -> bool {
    use std::io::Read;
    let mut f = match std::fs::File::open(path) {
        Ok(f) => f,
        Err(e) => {
            tracing::warn!("cannot open {} for text-sniff: {e}", path.display());
            return false;
        }
    };
    let mut buf = [0u8; 4096];
    let n = match f.read(&mut buf) {
        Ok(n) => n,
        Err(e) => {
            tracing::warn!("read failed during text-sniff of {}: {e}", path.display());
            return false;
        }
    };
    !looks_binary(&buf[..n])
}

// ── chat stream (SSE) ────────────────────────────────────────────────────────

#[derive(Debug, Deserialize)]
pub(crate) struct ChatStreamRequest {
    /// `None` → start a new thread (server creates and reports the id back via
    /// the first `thread_assigned` SSE event).
    #[serde(default)]
    pub thread_id: Option<String>,
    pub user_message: String,
    /// Optional quoted text from the reading view's selection.
    #[serde(default)]
    pub selection: Option<String>,
    /// Path of the document the user is currently looking at, workspace-relative.
    #[serde(default)]
    pub current_doc: Option<String>,
    /// `@`-mentioned files. Server reads them with the same guards as the
    /// `read_file` tool and inlines into the system prompt.
    #[serde(default)]
    pub mentions: Vec<MentionRef>,
}

#[derive(Debug, Deserialize)]
pub(crate) struct MentionRef {
    pub path: String,
}

async fn chat_stream_handler(
    State(state): State<AppState>,
    AxumPath(workspace_id): AxumPath<String>,
    Json(body): Json<ChatStreamRequest>,
) -> Response {
    let ws = match ensure_chat_enabled(&state, &workspace_id) {
        Ok(w) => w,
        Err(e) => return e.into_response(),
    };
    let storage = match storage_for(&state) {
        Ok(s) => s,
        Err(e) => return e.into_response(),
    };

    // Re-read settings on every request so a key update from the GUI is
    // picked up immediately. Trivial cost compared to the LLM call.
    let app_settings = AppSettings::load();
    let runtime_cfg = match ChatRuntimeConfig::from_settings(&app_settings.chat) {
        Ok(c) => c,
        Err(msg) => return ChatHttpError::BadRequest(msg.to_string()).into_response(),
    };

    let user_text = body.user_message.trim();
    if user_text.is_empty() {
        return ChatHttpError::BadRequest("empty user_message".into()).into_response();
    }

    // Resolve / create thread.
    let thread = match body.thread_id.as_deref() {
        Some(id) => match thread_in_workspace(&storage, id, &workspace_id).await {
            Ok(t) => t,
            Err(e) => return e.into_response(),
        },
        None => {
            let title = auto_title(user_text);
            match storage.create_thread(&workspace_id, &title).await {
                Ok(t) => t,
                Err(e) => return ChatHttpError::Storage(e).into_response(),
            }
        }
    };

    // Load prior history and rehydrate as Message objects.
    let history: Vec<Message> = match storage.list_messages(&thread.id).await {
        Ok(msgs) => msgs
            .into_iter()
            .map(|m| Message {
                role: m.role,
                content: m.content,
            })
            .collect(),
        Err(e) => return ChatHttpError::Storage(e).into_response(),
    };

    // The persisted user turn just carries their text; selection / mentions
    // live in the per-turn system prompt block so re-played history stays
    // clean and small.
    let user_blocks = vec![ContentBlock::Text {
        text: user_text.to_string(),
    }];
    if let Err(e) = storage
        .append_message(&thread.id, Role::User, &user_blocks)
        .await
    {
        return ChatHttpError::Storage(e).into_response();
    }

    // Inline `@`-mentioned files (subject to the read_file size/binary guards).
    let mention_blocks = body
        .mentions
        .iter()
        .filter_map(|m| inline_mention(&ws, &m.path))
        .collect();

    // Workspace outline for tier-2 cache layer — top-level dirs/files.
    let outline = workspace_outline(&ws.fs);
    let workspace_label = display_workspace_label(ws.fs.ambient_root());

    let system = build_system_blocks(&PromptInputs {
        workspace_label,
        workspace_outline: outline,
        current_doc: body.current_doc.clone(),
        selection: body.selection.clone(),
        mention_blocks,
    });

    // Build agent. `edit_file` is registered only when --enable-edit is on
    // for this workspace — the cross-flag gate the design committed to.
    let provider = provider::build(runtime_cfg.clone());
    let tools = Arc::new(ToolRegistry::for_workspace(
        ws.enable_edit.load(Ordering::Relaxed),
    ));
    let agent = Agent::new(provider, tools, storage);

    let agent_req = AgentRequest {
        thread_id: thread.id.clone(),
        thread_title: thread.title.clone(),
        workspace_id: workspace_id.clone(),
        workspace_fs: ws.fs.clone(),
        history,
        user_message: Message {
            role: Role::User,
            content: user_blocks,
        },
        system,
        model: runtime_cfg.model.clone(),
        max_steps: MAX_AGENT_STEPS,
        max_tokens: MAX_TOKENS_PER_TURN,
        pending_edits: ws.pending_edits.clone(),
    };

    let (tx, rx) = mpsc::channel::<AgentEvent>(64);
    tokio::spawn(async move {
        agent.run(agent_req, tx).await;
    });

    let stream = ReceiverStream::new(rx).map(|ev| {
        let event = Event::default().json_data(&ev).unwrap_or_else(|e| {
            Event::default().data(format!(
                "{{\"type\":\"error\",\"message\":\"sse encode: {e}\"}}"
            ))
        });
        Ok::<_, Infallible>(event)
    });

    Sse::new(stream)
        .keep_alive(KeepAlive::default())
        .into_response()
}

// ── helpers for chat stream handler ──────────────────────────────────────────

fn display_workspace_label(root: &std::path::Path) -> String {
    if let Some(home) = dirs::home_dir() {
        if let Ok(rel) = root.strip_prefix(&home) {
            return if rel.as_os_str().is_empty() {
                "~".to_string()
            } else {
                format!("~/{}", rel.display())
            };
        }
    }
    root.display().to_string()
}

/// One-level workspace listing used as cacheable system-prompt context.
/// Capped so a giant repo doesn't blow out the prefix.
fn workspace_outline(workspace_fs: &crate::workspace_fs::WorkspaceFs) -> String {
    use std::fmt::Write;
    let mut dirs: Vec<String> = Vec::new();
    let mut files: Vec<String> = Vec::new();
    for (rel, _) in workspace_fs.content_files(10_000) {
        let route = rel.as_route();
        if let Some((dir, _)) = route.split_once('/') {
            dirs.push(format!("{dir}/"));
        } else {
            files.push(route);
        }
    }
    dirs.sort();
    dirs.dedup();
    files.sort();
    let mut out = String::new();
    for d in dirs.iter().take(40) {
        let _ = writeln!(out, "{d}");
    }
    for f in files.iter().take(40) {
        let _ = writeln!(out, "{f}");
    }
    out
}

/// Read an `@`-mentioned file with the same guards as the `read_file` tool
/// (workspace containment, size cap, binary sniff) and wrap it in a tagged
/// block. Silently skips files that fail any check — better than a hard
/// failure, since the user just typed `@` and the model can still answer.
fn inline_mention(ws: &WorkspaceEntry, rel_path: &str) -> Option<String> {
    let canon = ws.fs.resolve_content(rel_path).ok()?;
    let meta = std::fs::metadata(&canon).ok()?;
    if !meta.is_file() || meta.len() > MAX_FILE_BYTES {
        return None;
    }
    let bytes = std::fs::read(&canon).ok()?;
    if looks_binary(&bytes) {
        return None;
    }
    let text = String::from_utf8(bytes).ok()?;
    Some(render_mention_block(rel_path, &text))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::chat::storage::ChatStorage;
    use crate::workspace::{WorkspaceConfig, WorkspaceFlags, WorkspaceRegistry};
    use axum::body::{to_bytes, Body};
    use axum::http::{Request, StatusCode};
    use rusqlite::Connection;
    use std::sync::Mutex;
    use tempfile::TempDir;
    use tera::Tera;
    use tokio::sync::broadcast;
    use tower::ServiceExt;

    struct TestEnv {
        _tmp: TempDir,
        _db_tmp: tempfile::NamedTempFile,
        state: AppState,
        workspace_id: String,
        storage: ChatStorage,
    }

    fn build_env(enable_chat: bool) -> TestEnv {
        build_env_with_flags(WorkspaceFlags {
            enable_chat,
            ..Default::default()
        })
    }

    fn build_env_with_flags(flags: WorkspaceFlags) -> TestEnv {
        let tmp = TempDir::new().expect("workspace tmpdir");
        let registry = Arc::new(WorkspaceRegistry::new("salt".into()));
        let workspace_id = registry.add(WorkspaceConfig {
            path: tmp.path().to_path_buf(),
            flags,
            single_file: None,
            collaborator_access_code_hash: String::new(),
            ..Default::default()
        });

        let db_tmp = tempfile::NamedTempFile::new().expect("sqlite tmpfile");
        let conn = Connection::open(db_tmp.path()).expect("open db");
        ChatStorage::init(&conn).expect("init schema");
        let db = Arc::new(Mutex::new(conn));
        let storage = ChatStorage::new(db.clone());

        let (shutdown_tx, _) = mpsc::channel(1);
        let state = AppState {
            theme: Arc::new("dark".into()),
            tera: Arc::new(Tera::default()),
            db: Some(db),
            workspace_registry: registry,
            management_token: Arc::new("token".into()),
            admin_bootstraps: Arc::new(crate::admin_auth::AdminBootstrapStore::new()),
            allowed_hosts: Arc::new(Default::default()),
            save_token: Arc::new("save-token".into()),
            i18n_json: Arc::new("{}".into()),
            i18n_lang: Arc::new("zh".into()),
            shortcuts_json: Arc::new("null".into()),
            styles_css: Arc::new(String::new()),
            default_chat_mode: Arc::new("in_page".into()),
            editor_theme: Arc::new("follow".into()),
            collaborator_access_code_hash: Arc::new(String::new()),
            access_secret: Arc::new("test-salt".into()),
            access_attempts: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
            markdown_diff_cache: Arc::new(Mutex::new(crate::server::MarkdownDiffCache::default())),
            print_collapsed_content: false,
            shutdown_tx,
            #[cfg(debug_assertions)]
            dev_reload_tx: Arc::new(broadcast::channel::<()>(1).0),
        };
        TestEnv {
            _tmp: tmp,
            _db_tmp: db_tmp,
            state,
            workspace_id,
            storage,
        }
    }

    async fn body_json(resp: axum::response::Response) -> serde_json::Value {
        let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
        serde_json::from_slice(&bytes).unwrap_or_else(|_| {
            serde_json::Value::String(String::from_utf8_lossy(&bytes).into_owned())
        })
    }

    fn json_body(value: serde_json::Value) -> Body {
        Body::from(serde_json::to_vec(&value).unwrap())
    }

    #[test]
    fn single_file_prompt_context_excludes_sibling_files() {
        let tmp = TempDir::new().unwrap();
        std::fs::write(tmp.path().join("opened.md"), "visible context\n").unwrap();
        std::fs::write(tmp.path().join("sibling.md"), "secret context\n").unwrap();
        let registry = WorkspaceRegistry::new("single-chat-salt".into());
        let id = registry.add(WorkspaceConfig {
            path: tmp.path().to_path_buf(),
            flags: WorkspaceFlags {
                enable_chat: true,
                ..WorkspaceFlags::default()
            },
            single_file: Some("opened.md".into()),
            collaborator_access_code_hash: String::new(),
            ..Default::default()
        });
        let ws = registry.get(&id).unwrap();

        let outline = workspace_outline(&ws.fs);
        assert!(outline.contains("opened.md"), "outline: {outline}");
        assert!(!outline.contains("sibling.md"), "outline: {outline}");
        assert!(inline_mention(&ws, "opened.md")
            .unwrap()
            .contains("visible context"));
        assert!(inline_mention(&ws, "sibling.md").is_none());
    }

    #[tokio::test]
    async fn list_threads_returns_403_when_chat_disabled() {
        let env = build_env(false);
        let app = router().with_state(env.state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/api/chat/{}/threads", env.workspace_id))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn list_threads_returns_empty_then_populated() {
        let env = build_env(true);
        let app = router().with_state(env.state.clone());

        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri(format!("/api/chat/{}/threads", env.workspace_id))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v = body_json(resp).await;
        assert_eq!(v.as_array().unwrap().len(), 0);

        env.storage
            .create_thread(&env.workspace_id, "t1")
            .await
            .unwrap();
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/api/chat/{}/threads", env.workspace_id))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        let v = body_json(resp).await;
        assert_eq!(v.as_array().unwrap().len(), 1);
        assert_eq!(v[0]["title"], "t1");
    }

    #[tokio::test]
    async fn create_thread_uses_default_title_when_blank() {
        let env = build_env(true);
        let app = router().with_state(env.state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri(format!("/api/chat/{}/threads", env.workspace_id))
                    .header("content-type", "application/json")
                    .body(json_body(serde_json::json!({})))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v = body_json(resp).await;
        assert_eq!(v["title"], "New chat");
        assert_eq!(v["workspace_id"], env.workspace_id);
    }

    #[tokio::test]
    async fn create_thread_trims_custom_title() {
        let env = build_env(true);
        let app = router().with_state(env.state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri(format!("/api/chat/{}/threads", env.workspace_id))
                    .header("content-type", "application/json")
                    .body(json_body(serde_json::json!({"title": "  hello  "})))
                    .unwrap(),
            )
            .await
            .unwrap();
        let v = body_json(resp).await;
        assert_eq!(v["title"], "hello");
    }

    #[tokio::test]
    async fn get_thread_returns_404_for_unknown() {
        let env = build_env(true);
        let app = router().with_state(env.state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/api/chat/{}/threads/nope", env.workspace_id))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn get_thread_returns_messages_in_order() {
        let env = build_env(true);
        let thread = env
            .storage
            .create_thread(&env.workspace_id, "t")
            .await
            .unwrap();
        env.storage
            .append_message(
                &thread.id,
                Role::User,
                &[ContentBlock::Text { text: "hi".into() }],
            )
            .await
            .unwrap();
        env.storage
            .append_message(
                &thread.id,
                Role::Assistant,
                &[ContentBlock::Text {
                    text: "hello".into(),
                }],
            )
            .await
            .unwrap();

        let app = router().with_state(env.state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!(
                        "/api/chat/{}/threads/{}",
                        env.workspace_id, thread.id
                    ))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v = body_json(resp).await;
        let msgs = v["messages"].as_array().unwrap();
        assert_eq!(msgs.len(), 2);
        assert_eq!(msgs[0]["role"], "user");
        assert_eq!(msgs[1]["role"], "assistant");
    }

    #[tokio::test]
    async fn delete_thread_removes_it() {
        let env = build_env(true);
        let thread = env
            .storage
            .create_thread(&env.workspace_id, "t")
            .await
            .unwrap();
        let app = router().with_state(env.state.clone());

        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("DELETE")
                    .uri(format!(
                        "/api/chat/{}/threads/{}",
                        env.workspace_id, thread.id
                    ))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NO_CONTENT);

        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!(
                        "/api/chat/{}/threads/{}",
                        env.workspace_id, thread.id
                    ))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn list_files_returns_only_text_files() {
        let env = build_env(true);
        std::fs::write(env._tmp.path().join("a.md"), "hello").unwrap();
        std::fs::write(env._tmp.path().join("b.txt"), "world").unwrap();
        std::fs::write(env._tmp.path().join("img.png"), [0u8; 16]).unwrap();

        let app = router().with_state(env.state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/api/chat/{}/files", env.workspace_id))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v = body_json(resp).await;
        let paths: Vec<String> = v
            .as_array()
            .unwrap()
            .iter()
            .map(|s| s["path"].as_str().unwrap().to_string())
            .collect();
        assert!(paths.contains(&"a.md".to_string()));
        assert!(paths.contains(&"b.txt".to_string()));
        assert!(!paths.contains(&"img.png".to_string()));
    }

    // ── undo_edit_handler ──────────────────────────────────────────────────

    async fn post_undo(env: &TestEnv, body: serde_json::Value) -> axum::response::Response {
        let app = router().with_state(env.state.clone());
        app.oneshot(
            Request::builder()
                .method("POST")
                .uri(format!("/api/chat/{}/edits/undo", env.workspace_id))
                .header("content-type", "application/json")
                .body(json_body(body))
                .unwrap(),
        )
        .await
        .unwrap()
    }

    #[tokio::test]
    async fn undo_rejects_when_chat_disabled() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: false,
            enable_edit: true,
            ..Default::default()
        });
        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "x.md", "find": "a", "replace_with": "b" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn undo_rejects_when_edit_disabled() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: true,
            enable_edit: false,
            ..Default::default()
        });
        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "x.md", "find": "a", "replace_with": "b" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn undo_replaces_find_with_replace_when_unique() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: true,
            enable_edit: true,
            ..Default::default()
        });
        let path = env._tmp.path().join("doc.md");
        std::fs::write(&path, "hello FOO world").unwrap();

        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "doc.md", "find": "FOO", "replace_with": "foo" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::OK);
        let body = body_json(resp).await;
        assert_eq!(body["status"], "reverted");
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "hello foo world");
    }

    #[tokio::test]
    async fn undo_reports_drift_when_find_missing_and_does_not_write() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: true,
            enable_edit: true,
            ..Default::default()
        });
        let path = env._tmp.path().join("doc.md");
        std::fs::write(&path, "hello world").unwrap();

        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "doc.md", "find": "MISSING", "replace_with": "x" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::OK);
        let body = body_json(resp).await;
        assert_eq!(body["status"], "drifted");
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "hello world");
    }

    #[tokio::test]
    async fn undo_reports_drift_when_find_is_ambiguous() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: true,
            enable_edit: true,
            ..Default::default()
        });
        let path = env._tmp.path().join("doc.md");
        std::fs::write(&path, "foo bar foo baz").unwrap();

        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "doc.md", "find": "foo", "replace_with": "FOO" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::OK);
        let body = body_json(resp).await;
        assert_eq!(body["status"], "drifted");
        // file untouched
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "foo bar foo baz");
    }

    #[tokio::test]
    async fn undo_rejects_path_outside_workspace() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: true,
            enable_edit: true,
            ..Default::default()
        });
        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "../etc/passwd", "find": "x", "replace_with": "y" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn undo_rejects_empty_find_and_noop_swap() {
        let env = build_env_with_flags(WorkspaceFlags {
            enable_chat: true,
            enable_edit: true,
            ..Default::default()
        });
        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "x.md", "find": "", "replace_with": "y" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        let resp = post_undo(
            &env,
            serde_json::json!({ "path": "x.md", "find": "same", "replace_with": "same" }),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }
}