"""
Flask routes for chat API.

Provides endpoints for:
- Chat page rendering
- Session management (create, list, get, archive, delete)
- Message management (send, list)
- Research triggering from chat
"""

import unicodedata
import uuid
from datetime import datetime, timedelta, UTC
from typing import Any, Dict, List
from flask import Blueprint, request, jsonify, session
from loguru import logger
from sqlalchemy import update as sa_update
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

from .service import (
    ArchiveBlockedError,
    AttemptInProgress,
    AttemptNotFound,
    ChatService,
    ChatSessionNotFound,
    DB_EXCEPTIONS,
)
from .context import ChatContextManager
from ..constants import ResearchStatus
from ..database.models import (
    ChatMessage,
    ChatSession,
    ChatSessionStatus,
    ResearchHistory,
    UserActiveResearch,
)
from ..database.session_context import get_user_db_session
from ..exceptions import DuplicateResearchError, SystemAtCapacityError
from ..security.decorators import require_json_body
from ..security.rate_limiter import _get_api_user_key, limiter
from ..settings.manager import SettingsManager
from ..web.auth.decorators import login_required
from ..web.utils.templates import render_template_with_defaults
from ..web.auth.password_utils import (
    get_user_password,
    resolve_user_password,
)
from ..web.routes.globals import (
    cleanup_research,
    is_research_thread_alive,
)
from ..web.services.research_service import (
    run_research_process,
    start_research_process,
)

# Create blueprint
chat_bp = Blueprint("chat", __name__)

# Valid status values for sessions (built from the enum so a typo never
# silently passes validation; the literal "all" sentinel widens the list
# filter to every status without bypassing the whitelist).
VALID_UPDATE_STATUSES = {
    ChatSessionStatus.ACTIVE.value,
    ChatSessionStatus.ARCHIVED.value,
}
VALID_LIST_STATUSES = {*(s.value for s in ChatSessionStatus), "all"}

# Input length limits
MAX_QUERY_LENGTH = 10_000
MAX_TITLE_LENGTH = 500
MAX_MESSAGE_LENGTH = 10_000
# Hard cap on `offset` to prevent server-side DoS: get_session_messages
# fetches `limit + offset` rows from BOTH chat_messages and chat_progress_steps
# tables, so unbounded offset means unbounded SQL LIMIT. With cursor-based
# pagination (`before_created_at`) as the recommended path, offset above a few
# pages is not a normal access pattern.
MAX_OFFSET = 1_000

# Wider exception tuple used by HTTP route handlers (subsumes
# service.DB_EXCEPTIONS plus the attribute/type errors that can escape
# request-shape coercion code). DB_EXCEPTIONS itself is single-sourced
# from chat.service so the two never drift.
ROUTE_EXCEPTIONS = (
    ValueError,
    RuntimeError,
    SQLAlchemyError,
    AttributeError,
    TypeError,
)


def _load_settings(username):
    """Load all settings for a user.

    ``bypass_cache=True`` matches the call pattern in
    ``research_routes.start_research``: a setting changed via the UI
    moments before the user sends a chat message must take effect on the
    next research, not be served from a stale cache.
    """
    with get_user_db_session(username) as db:
        return SettingsManager(db_session=db).get_all_settings(
            bypass_cache=True
        )


def _parse_int_param(
    value: str | None,
    default: int,
    min_val: int = 0,
    max_val: int | None = None,
) -> int:
    """Safely parse an integer parameter with bounds checking."""
    try:
        result = int(value) if value is not None else default
        if result < min_val:
            return min_val
        if max_val is not None and result > max_val:
            return max_val
        return result
    except (ValueError, TypeError):
        return default


_INVISIBLE_UNICODE_CATEGORIES = {"Cf", "Zl", "Zp"}


def _validate_title(title) -> tuple[str, int] | None:
    """Return (error_message, http_status) when *title* is invalid, else None.

    A title is invalid when it is not a non-empty string or exceeds
    ``MAX_TITLE_LENGTH``. Callers that allow ``None`` (e.g. create_session
    where omitting the title is fine) should short-circuit on ``None``
    before calling this helper.

    Strips Unicode format / line-separator characters (``Cf``/``Zl``/``Zp``,
    including zero-width spaces U+200B-U+200D and BOM U+FEFF) before the
    emptiness check so an "invisible" title like 500 zero-width chars is
    rejected instead of saving a session that looks blank in the UI.
    """
    if not isinstance(title, str):
        return ("Title cannot be empty", 400)
    visible = "".join(
        c
        for c in title
        if unicodedata.category(c) not in _INVISIBLE_UNICODE_CATEGORIES
    )
    if not visible.strip():
        return ("Title cannot be empty", 400)
    if len(title) > MAX_TITLE_LENGTH:
        return (
            f"Title too long (max {MAX_TITLE_LENGTH} characters)",
            400,
        )
    return None


def _cleanup_chat_send_rows(
    username, research_id, message_id, session_id, reason: str
) -> None:
    """Undo the user-message + research_history rows committed by send_message
    when ``start_research_process`` rejects the spawn.

    Used by both the ``DuplicateResearchError`` (409) and
    ``SystemAtCapacityError`` (429) paths. Failure to clean up is logged at
    ERROR level so orphan rows + inflated message_count are visible to ops.
    """
    try:
        with get_user_db_session(username) as cleanup_db:
            cleanup_db.query(ResearchHistory).filter_by(id=research_id).delete()
            cleanup_db.query(ChatMessage).filter_by(id=message_id).delete()
            # Drop the per-user-cap tracking row too (the spawn never
            # started, so no live thread owns it). research_id is a fresh
            # UUID, so this only ever matches our own just-inserted row.
            cleanup_db.query(UserActiveResearch).filter_by(
                username=username, research_id=research_id
            ).delete()
            cleanup_db.execute(
                sa_update(ChatSession)
                .where(ChatSession.id == session_id)
                .values(message_count=ChatSession.message_count - 1)
            )
            cleanup_db.commit()
    except DB_EXCEPTIONS:
        logger.exception(
            f"Cleanup after {reason} chat-send rejection FAILED "
            f"for research {research_id[:8]}... in chat "
            f"{session_id[:8]}...; orphan rows + inflated "
            f"message_count may persist until next sweep."
        )


class ChatSpawnError(Exception):
    """Raised by ``_spawn_chat_research`` to signal a mapped HTTP failure.

    Carries the HTTP status code, an error string for the response body,
    and optional extra fields merged into the JSON payload (e.g.
    ``active_research_id`` for the 409 path). Both ``send_message`` and
    ``retry_attempt`` catch this and translate to a ``jsonify(...)`` with
    the same shape, keeping the spawn path single-sourced.
    """

    def __init__(self, status_code: int, error: str, **extra):
        super().__init__(error)
        self.status_code = status_code
        self.error = error
        self.extra = extra


def _spawn_chat_research(
    username: str,
    session_id: str,
    content: str,
    settings_snapshot: Dict[str, Any],
    research_context: Dict[str, Any],
    messages: List[Dict[str, Any]],
    service: "ChatService",
    research_mode: str = "quick",
) -> tuple[str, str]:
    """Atomically write the user message + IN_PROGRESS research row, then
    spawn the research worker.

    Single-sources the spawn path shared by ``send_message`` (new chat
    turn) and ``retry_attempt`` (delete failed turn, re-submit same
    content). Returns ``(research_id, message_id)`` on success.

    Pre-requisites the CALLER must satisfy before invoking:
      - Session exists and is active (``ChatService.get_session`` passed).
      - Per-session in-progress guard passed (no other live research).
      - Per-user concurrent cap not exceeded.

    Failure modes (mapped to ``ChatSpawnError``):
      - 400: malformed numeric settings (``iterations`` / ``questions``).
      - 404: session was deleted between caller's existence check and
        this helper's atomic write.
      - 409: race against the partial unique index
        ``ux_research_history_chat_session_in_progress`` (two concurrent
        sends both passed the per-session guard), or
        ``DuplicateResearchError`` from ``start_research_process`` (live
        thread already owns the id).
      - 429: ``SystemAtCapacityError`` from ``start_research_process``
        (cap filled between caller's check and spawn).

    Raises ``ChatSpawnError`` only; never returns ``None``. The caller
    catches once and translates to ``jsonify`` + status.
    """
    # Parse numeric search settings up-front. A malformed value (a
    # non-numeric string in the user's settings DB) must return a clean
    # 400 HERE — before the atomic write below commits the user message
    # + IN_PROGRESS research row. Mirrors the original send_message
    # ordering rationale.
    try:
        iterations = int(
            settings_snapshot.get("search.iterations", {}).get("value", 3)
        )
        questions = int(
            settings_snapshot.get("search.questions_per_iteration", {}).get(
                "value", 1
            )
        )
    except (ValueError, TypeError):
        raise ChatSpawnError(
            400,
            "Invalid numeric value in search settings "
            "(iterations / questions_per_iteration).",
        )

    research_id = str(uuid.uuid4())

    # ---- Atomic write: user message + research row in ONE transaction ----
    # Closes the orphan window: any IntegrityError or concurrent-delete on
    # the research insert rolls back the user message too. The
    # UPDATE...RETURNING inside insert_message_in_db doubles as the
    # authoritative "session-still-exists" check; if the session was
    # deleted between the caller's existence check and now, a ValueError
    # surfaces with "not found" and we map it to 404.
    try:
        with get_user_db_session(username) as db_session:
            message_id = service.insert_message_in_db(
                db_session,
                session_id=session_id,
                role="user",
                content=content,
                message_type="query" if len(messages) == 0 else "followup",
            )

            created_at = datetime.now(UTC).isoformat()
            research_meta = {
                "submission": {
                    "chat_session_id": session_id,
                    "message_id": message_id,
                    "research_mode": research_mode,
                },
            }
            research = ResearchHistory(
                id=research_id,
                query=content,
                mode=research_mode,
                status=ResearchStatus.IN_PROGRESS.value,
                created_at=created_at,
                progress_log=[{"time": created_at, "progress": 0}],
                research_meta=research_meta,
                chat_session_id=session_id,
            )
            db_session.add(research)

            # Count this research toward the per-user concurrent cap.
            # Added in the SAME transaction as the research row so the
            # IntegrityError rollback below undoes both.
            import threading

            db_session.add(
                UserActiveResearch(
                    username=username,
                    research_id=research_id,
                    status=ResearchStatus.IN_PROGRESS,
                    thread_id=str(threading.current_thread().ident),
                    settings_snapshot=settings_snapshot,
                )
            )

            try:
                db_session.commit()
            except IntegrityError:
                # Two near-simultaneous POSTs both passed the per-session
                # guard; the partial unique index on (chat_session_id)
                # WHERE status='in_progress' (migration 0010) catches the
                # loser here. Rolling back the transaction also undoes the
                # user-message INSERT and the message_count increment.
                db_session.rollback()
                logger.warning(
                    f"Concurrent in-progress research race for chat "
                    f"{session_id[:8]}..."
                )
                raise ChatSpawnError(
                    409,
                    "Research already in progress on this chat session.",
                )
    except ValueError as exc:
        # ``insert_message_in_db`` raises ValueError("not found") when
        # the session row was deleted between the existence check and
        # the UPDATE...RETURNING.
        if "not found" in str(exc).lower():
            raise ChatSpawnError(404, "Session not found")
        raise
    # ---- end atomic write ----

    # Get user password for metrics writes inside the worker thread.
    pw = get_user_password(username)

    model_provider = settings_snapshot.get("llm.provider", {}).get("value", "")
    model = settings_snapshot.get("llm.model", {}).get("value", "")
    search_engine = settings_snapshot.get("search.tool", {}).get("value", "")
    custom_endpoint = settings_snapshot.get("llm.openai_endpoint.url", {}).get(
        "value"
    )
    user_strategy = settings_snapshot.get("search.search_strategy", {}).get(
        "value", "langgraph-agent"
    )

    # For follow-up messages, use the contextual follow-up strategy
    # which wraps the user's preferred strategy as a delegate.
    if research_context.get("is_multi_turn"):
        strategy = "enhanced-contextual-followup"
        research_context["delegate_strategy"] = user_strategy
    else:
        strategy = user_strategy

    # Spawn the worker thread. ``DuplicateResearchError`` and
    # ``SystemAtCapacityError`` inherit from ``Exception`` (not
    # RuntimeError) so they are NOT in ROUTE_EXCEPTIONS and would
    # otherwise escape to a generic 500 — leaving the user message +
    # research row we just committed as orphans. Catch them here, undo
    # our side effects via ``_cleanup_chat_send_rows``, and re-raise as
    # ``ChatSpawnError`` for the caller to translate.
    try:
        start_research_process(
            research_id,
            content,
            research_mode,
            run_research_process,
            username=username,
            user_password=pw,
            model_provider=model_provider,
            model=model,
            search_engine=search_engine,
            custom_endpoint=custom_endpoint,
            strategy=strategy,
            iterations=iterations,
            questions_per_iteration=questions,
            research_context=research_context,
            chat_session_id=session_id,
            chat_message_id=message_id,
            settings_snapshot=settings_snapshot,
        )
    except DuplicateResearchError:
        logger.warning(
            f"DuplicateResearchError on chat spawn for "
            f"{research_id[:8]}... (chat {session_id[:8]}...)"
        )
        _cleanup_chat_send_rows(
            username, research_id, message_id, session_id, "duplicate"
        )
        raise ChatSpawnError(
            409, "Research already in progress on this chat session."
        )
    except SystemAtCapacityError:
        logger.warning(
            f"SystemAtCapacityError on chat spawn for "
            f"{research_id[:8]}... (chat {session_id[:8]}...)"
        )
        _cleanup_chat_send_rows(
            username, research_id, message_id, session_id, "capacity"
        )
        raise ChatSpawnError(
            429, "Server is at research capacity. Please retry shortly."
        )

    logger.info(
        f"Started chat research {research_id[:8]}... for chat "
        f"{session_id[:8]}..."
    )
    return research_id, message_id


def _chat_spawn_response(exc: ChatSpawnError):
    """Translate a ChatSpawnError into the (jsonify, status) tuple the
    Flask route returns.

    Both ``send_message`` and ``retry_attempt`` shape errors identically,
    so this lives once. ``exc.extra`` (e.g. ``active_research_id``) is
    merged into the JSON body when present.
    """
    payload = {"success": False, "error": exc.error}
    payload.update(exc.extra)
    return jsonify(payload), exc.status_code


# ============================================================================
# Concurrency-guard helpers shared by send_message + retry_attempt
# ============================================================================


def _enforce_chat_session_research_slot(
    cap_db, username: str, session_id: str
) -> tuple[str, int] | None:
    """Reject send/retry when this chat session or user is at capacity.

    Runs inside the caller's ``cap_db`` transaction so the stale-row
    reclaims it performs are visible to the count check below it.

    Returns ``(error_message, http_status)`` when the request should be
    rejected, else ``None``.

    Two checks:
      1. Per-session guard: at most one live research per chat session.
         Mirrors the same check in send_message's inline block
         (chat/routes.py:925-941 in the pre-refactor layout).
      2. Per-user global cap: at most ``app.max_concurrent_researches``
         researches per user across ALL sessions. Mirrors
         research_routes.start_research.

    DOES NOT include the stale-row reclaim sweep — the caller does that
    first via ``reclaim_stale_user_active_research`` so the count check
    sees accurate numbers.
    """
    # Per-session guard. Note: this fires for ANY in-progress research
    # on the session, INCLUDING the target research on a retry path.
    # Retry routes must therefore ensure the target is not in-progress
    # (typically via delete_attempt's AttemptInProgress semantics)
    # BEFORE calling this.
    existing_session_research = (
        cap_db.query(ResearchHistory)
        .filter_by(
            chat_session_id=session_id,
            status=ResearchStatus.IN_PROGRESS,
        )
        .first()
    )
    if existing_session_research:
        return (
            "Research already in progress on this chat session. "
            "Stop it before sending a new message.",
            409,
        )

    active_count = (
        cap_db.query(UserActiveResearch)
        .filter_by(
            username=username,
            status=ResearchStatus.IN_PROGRESS,
        )
        .count()
    )
    max_concurrent = SettingsManager(db_session=cap_db).get_setting(
        "app.max_concurrent_researches", 3
    )
    if active_count >= max_concurrent:
        return (
            f"Concurrent research limit reached ({active_count}/"
            f"{max_concurrent}). Wait for an existing research to finish.",
            429,
        )

    return None


# ============================================================================
# Page Routes
# ============================================================================


@chat_bp.route("/chat/")
@chat_bp.route("/chat/<session_id>")
@login_required
def chat_page(session_id=None):
    """
    Render the chat page.

    Args:
        session_id: Optional session ID to load existing session
    """
    return render_template_with_defaults(
        "pages/chat.html", session_id=session_id
    )


# ============================================================================
# Session API Routes
# ============================================================================


@chat_bp.route("/api/chat/sessions", methods=["POST"])
@login_required
# Per-user keying (default is per-IP). Without this, users behind a shared
# NAT/proxy share one bucket and can DoS each other for legitimate chat use.
@limiter.limit("20 per minute", key_func=_get_api_user_key)
@require_json_body(
    error_format="success",
    error_message="Request body must be a JSON object",
)
def create_session():
    """
    Create a new chat session.

    Request body:
    {
        "initial_query": "optional initial question",
        "title": "optional custom title"
    }

    Returns:
    {
        "success": true,
        "session_id": "uuid",
        "session": { session data }
    }
    """
    try:
        username = session.get("username")

        # @require_json_body has already guaranteed a dict body; reach for it
        # directly. Flask caches the parse so this is not a duplicate call.
        data = request.get_json(silent=True)

        # Validate input lengths
        initial_query = data.get("initial_query")
        title = data.get("title")

        # Reject non-string initial_query early so len() / downstream
        # string ops don't raise TypeError → 500.
        if initial_query is not None and not isinstance(initial_query, str):
            return jsonify(
                {
                    "success": False,
                    "error": "initial_query must be a string",
                }
            ), 400

        if initial_query and len(initial_query) > MAX_QUERY_LENGTH:
            return jsonify(
                {
                    "success": False,
                    "error": f"Initial query too long (max {MAX_QUERY_LENGTH} characters)",
                }
            ), 400

        if title is not None:
            err = _validate_title(title)
            if err is not None:
                msg, status = err
                return jsonify({"success": False, "error": msg}), status

        settings_snapshot = _load_settings(username)

        service = ChatService(username)
        session_id = service.create_session(
            initial_query=initial_query,
            title=title,
            settings_snapshot=settings_snapshot,
        )

        # Get the created session
        try:
            session_data = service.get_session(session_id)
        except ChatSessionNotFound:
            # Session was just created in this request — getting "not
            # found" here means a delete-race or storage failure.
            # Don't include session_id in the log message (flagged as
            # sensitive by check-sensitive-logging); the exception's
            # stack trace already carries enough context to diagnose.
            logger.exception("Just-created chat session missing on read-back")
            return jsonify(
                {"success": False, "error": "Failed to load created session"}
            ), 500

        return jsonify(
            {
                "success": True,
                "session_id": session_id,
                "session": session_data,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error creating chat session")
        return jsonify(
            {
                "success": False,
                "error": "Failed to create chat session",
            }
        ), 500


@chat_bp.route(
    "/api/chat/sessions/<session_id>/generate-title", methods=["POST"]
)
@login_required
# Per-user keying + lower limit than create_session because each call is a
# real LLM round-trip on a server-paid endpoint (vs create_session which is
# zero-LLM DB work). Without per-user keying, shared-IP users share the bucket.
@limiter.limit("10 per minute", key_func=_get_api_user_key)
@require_json_body(
    error_format="success",
    error_message="Request body must be a JSON object",
)
def generate_session_title(session_id):
    """
    Regenerate the session title using the configured LLM.

    This is a fire-and-forget endpoint the frontend calls asynchronously
    right after creating a session, so the synchronous POST
    /api/chat/sessions response isn't blocked on an LLM round-trip.

    Request body: {"query": "the initial research query"}

    Returns: {"success": true, "title": "..."} on success,
             {"success": false, "error": "..."} on failure.
    """
    try:
        username = session.get("username")
        # @require_json_body has already guaranteed a dict body.
        data = request.get_json(silent=True)

        query = data.get("query")

        if not query:
            return jsonify(
                {"success": False, "error": "query is required"}
            ), 400
        if not isinstance(query, str) or len(query) > MAX_QUERY_LENGTH:
            return jsonify(
                {
                    "success": False,
                    "error": f"query must be a string up to {MAX_QUERY_LENGTH} chars",
                }
            ), 400

        service = ChatService(username)
        try:
            service.get_session(session_id)
        except ChatSessionNotFound:
            return jsonify(
                {"success": False, "error": "Session not found"}
            ), 404

        settings_snapshot = _load_settings(username)
        new_title = service.regenerate_title_with_llm(
            session_id, query, settings_snapshot
        )
        if not new_title:
            # LLM disabled, or LLM call failed — keep existing fallback title
            return jsonify({"success": False, "title": None}), 200

        return jsonify({"success": True, "title": new_title})

    except ROUTE_EXCEPTIONS:
        logger.exception("Error regenerating chat title")
        return jsonify(
            {"success": False, "error": "Failed to regenerate title"}
        ), 500


@chat_bp.route("/api/chat/sessions", methods=["GET"])
@login_required
def list_sessions():
    """
    List chat sessions for the current user.

    Query params:
    - status: active, archived, deleted, or all (default: active)
    - limit: max sessions to return (default: 20)
    - offset: pagination offset (default: 0)

    Returns:
    {
        "success": true,
        "sessions": [ session data list ]
    }
    """
    try:
        username = session.get("username")
        status = request.args.get("status", ChatSessionStatus.ACTIVE.value)
        # Validate status parameter
        if status not in VALID_LIST_STATUSES:
            status = ChatSessionStatus.ACTIVE.value
        limit = _parse_int_param(
            request.args.get("limit"), 20, min_val=1, max_val=100
        )
        offset = _parse_int_param(
            request.args.get("offset"), 0, min_val=0, max_val=MAX_OFFSET
        )

        service = ChatService(username)
        sessions = service.list_sessions(
            status=status, limit=limit, offset=offset
        )

        return jsonify(
            {
                "success": True,
                "sessions": sessions,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error listing chat sessions")
        return jsonify(
            {
                "success": False,
                "error": "Failed to list chat sessions",
            }
        ), 500


@chat_bp.route("/api/chat/sessions/<session_id>", methods=["GET"])
@login_required
def get_session(session_id):
    """
    Get a specific chat session.

    Returns:
    {
        "success": true,
        "session": { session data }
    }
    """
    try:
        username = session.get("username")
        service = ChatService(username)
        try:
            session_data = service.get_session(session_id)
        except ChatSessionNotFound:
            return jsonify(
                {
                    "success": False,
                    "error": "Session not found",
                }
            ), 404

        return jsonify(
            {
                "success": True,
                "session": session_data,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error getting chat session")
        return jsonify(
            {
                "success": False,
                "error": "Failed to get chat session",
            }
        ), 500


@chat_bp.route("/api/chat/sessions/<session_id>", methods=["PATCH"])
@login_required
# Per-user keying, like the other state-changing chat routes. Without a
# per-route limit, rename/archive was bounded only by the global limiter,
# leaving an uneven abuse surface across the session API.
@limiter.limit("30 per minute", key_func=_get_api_user_key)
@require_json_body(
    error_format="success",
    error_message="Request body must be a JSON object",
)
def update_session(session_id):
    """
    Update a chat session (title, archive, delete).

    Request body:
    {
        "title": "new title",  // optional
        "status": "archived"   // optional: active, archived
    }
    """
    try:
        username = session.get("username")
        # @require_json_body has already guaranteed a dict body.
        data = request.get_json(silent=True)

        # Require at least one valid field
        valid_fields = {"title", "status"}
        if not any(field in data for field in valid_fields):
            return jsonify(
                {
                    "success": False,
                    "error": "Request must include at least one of: title, status",
                }
            ), 400

        service = ChatService(username)

        try:
            service.get_session(session_id)
        except ChatSessionNotFound:
            return jsonify(
                {"success": False, "error": "Session not found"}
            ), 404

        ops_ok = True

        if "title" in data:
            title = data["title"]
            err = _validate_title(title)
            if err is not None:
                msg, status = err
                return jsonify({"success": False, "error": msg}), status
            ops_ok = service.update_session_title(session_id, title) and ops_ok

        if "status" in data:
            new_status = data["status"]
            if new_status not in VALID_UPDATE_STATUSES:
                return jsonify(
                    {"success": False, "error": "Invalid status value"}
                ), 400
            if new_status == ChatSessionStatus.ACTIVE.value:
                ops_ok = service.reactivate_session(session_id) and ops_ok
            elif new_status == ChatSessionStatus.ARCHIVED.value:
                try:
                    ops_ok = service.archive_session(session_id) and ops_ok
                except ArchiveBlockedError:
                    # Symmetric with send-to-archived (also 409): the
                    # client should stop the research or wait for it to
                    # finish before archiving the session.
                    # Hard-coded message — never echo str(exc) here so a
                    # future ArchiveBlockedError raise with interpolated
                    # data can't leak to the response (information
                    # exposure through an exception, CWE-209).
                    return jsonify(
                        {
                            "success": False,
                            "error": "Cannot archive: research in_progress. Stop it first.",
                        }
                    ), 409

        try:
            session_data = service.get_session(session_id)
        except ChatSessionNotFound:
            # Session was deleted by a concurrent request between the
            # update above and this read-back. Treat as 404 rather than
            # returning a partial success with null data.
            return jsonify(
                {"success": False, "error": "Session not found"}
            ), 404

        if not ops_ok:
            # The read-back above succeeded, so the session still exists, yet
            # an update reported failure — a DB write error was swallowed into
            # a False return. Surface it instead of reporting success.
            logger.error(
                f"Chat session update failed at DB layer for "
                f"{session_id[:8]}..."
            )
            return jsonify(
                {"success": False, "error": "Failed to update session"}
            ), 500

        return jsonify(
            {
                "success": True,
                "session": session_data,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error updating chat session")
        return jsonify(
            {
                "success": False,
                "error": "Failed to update chat session",
            }
        ), 500


@chat_bp.route("/api/chat/sessions/<session_id>", methods=["DELETE"])
@login_required
# Per-user keying, like the other state-changing chat routes. Caps bulk
# delete attempts that the global limiter alone left under-constrained.
@limiter.limit("30 per minute", key_func=_get_api_user_key)
def delete_session(session_id):
    """Delete a chat session permanently."""
    try:
        username = session.get("username")
        service = ChatService(username)
        success = service.delete_session(session_id)

        if not success:
            return jsonify(
                {
                    "success": False,
                    "error": "Session not found",
                }
            ), 404

        return jsonify(
            {
                "success": True,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error deleting chat session")
        return jsonify(
            {
                "success": False,
                "error": "Failed to delete chat session",
            }
        ), 500


# ============================================================================
# Message API Routes
# ============================================================================


@chat_bp.route("/api/chat/sessions/<session_id>/messages", methods=["GET"])
@login_required
def get_messages(session_id):
    """
    Get messages for a chat session.

    Query params:
    - limit: max messages to return (default: 50, max: 100)
    - offset: pagination offset into the DESC slice (default: 0)
    - before_created_at: ISO timestamp cursor — return only entries
      strictly older than this. Use the oldest currently-displayed
      ``created_at`` to implement "load older messages".
    - before_id: optional id of the oldest currently-displayed row;
      when paired with `before_created_at` the cursor becomes
      composite, preventing same-millisecond rows at the page boundary
      from being silently dropped.

    Returns:
    {
        "success": true,
        "messages": [ message data list, ASC by created_at ],
        "has_more": bool,
        "in_progress_research_id": str | null
    }
    """
    try:
        username = session.get("username")
        limit = _parse_int_param(
            request.args.get("limit"), 50, min_val=1, max_val=100
        )
        offset = _parse_int_param(
            request.args.get("offset"), 0, min_val=0, max_val=MAX_OFFSET
        )
        before_created_at = request.args.get("before_created_at") or None
        before_id = request.args.get("before_id") or None

        service = ChatService(username)

        try:
            service.get_session(session_id)
        except ChatSessionNotFound:
            return jsonify(
                {"success": False, "error": "Session not found"}
            ), 404

        # Fetch one extra row so we can tell the client whether more
        # older entries exist without a second round-trip.
        peek_limit = limit + 1
        page = service.get_session_messages(
            session_id,
            limit=peek_limit,
            offset=offset,
            before_created_at=before_created_at,
            before_id=before_id,
        )
        has_more = len(page) > limit
        messages = page[-limit:] if has_more else page

        # The client (chat.js loadSession) restores the live "thinking"
        # indicator from this field instead of inferring in-flight state
        # from message metadata. O(1) via the partial-unique index
        # ux_research_history_chat_session_in_progress.
        in_progress_research_id = service.get_in_progress_research_id(
            session_id
        )

        return jsonify(
            {
                "success": True,
                "messages": messages,
                "has_more": has_more,
                "in_progress_research_id": in_progress_research_id,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error getting chat messages")
        return jsonify(
            {
                "success": False,
                "error": "Failed to get chat messages",
            }
        ), 500


@chat_bp.route("/api/chat/sessions/<session_id>/messages", methods=["POST"])
@login_required
# Per-user keying (default is per-IP). send_message launches a full research
# run, so this is the heaviest chat endpoint; shared-IP users sharing the
# bucket would lock each other out.
@limiter.limit("10 per minute", key_func=_get_api_user_key)
@require_json_body(
    error_format="success",
    error_message="Request body must be a JSON object",
)
def send_message(session_id):
    """
    Send a message in a chat session.

    This endpoint:
    1. Adds the user message to the session
    2. Decides if research is needed
    3. If research needed, starts research process
    4. Returns message ID and research ID (if applicable)

    Request body:
    {
        "content": "user message",
        "trigger_research": true  // optional, default true
    }

    Note: Research mode is always "quick" in chat. This is intentional for v1.

    Returns:
    {
        "success": true,
        "message_id": "uuid",
        "research_id": "uuid or null",
        "research_mode": "quick/none"
    }
    """
    try:
        username = session.get("username")
        # @require_json_body has already guaranteed a dict body and rejected
        # non-JSON content types (which also hardens CSRF, matching the other
        # state-changing chat POSTs). Flask caches the parse, so this is free.
        data = request.get_json(silent=True)

        if not data or not data.get("content"):
            return jsonify(
                {
                    "success": False,
                    "error": "Message content is required",
                }
            ), 400

        # Reject non-string content before .strip() raises AttributeError
        # → 500. Mirrors the isinstance guard in _validate_title.
        if not isinstance(data["content"], str):
            return jsonify(
                {
                    "success": False,
                    "error": "content must be a string",
                }
            ), 400

        content = data["content"].strip()

        # Reject whitespace-only content
        if not content:
            return jsonify(
                {
                    "success": False,
                    "error": "Message content is required",
                }
            ), 400

        if len(content) > MAX_MESSAGE_LENGTH:
            return jsonify(
                {
                    "success": False,
                    "error": f"Message too long (max {MAX_MESSAGE_LENGTH} characters)",
                }
            ), 400

        # trigger_research is a strict boolean. Reject non-bool values with a
        # 400 instead of silently coercing them to True: a client sending
        # {"trigger_research": "no"} or {"trigger_research": 0} intends to
        # SUPPRESS research, and coercing the truthy-string to True would
        # launch an unwanted (paid) research run against their intent.
        raw = data.get("trigger_research", True)
        if not isinstance(raw, bool):
            return jsonify(
                {
                    "success": False,
                    "error": "trigger_research must be a boolean",
                }
            ), 400
        trigger_research = raw

        service = ChatService(username)

        # Verify session exists (informational fast-fail; the
        # UPDATE...RETURNING inside insert_message_in_db is the
        # authoritative check that survives a delete-race).
        try:
            session_data = service.get_session(session_id)
        except ChatSessionNotFound:
            return jsonify(
                {
                    "success": False,
                    "error": "Session not found",
                }
            ), 404

        # Reject sends to non-active sessions. Archived/deleted sessions
        # are intentionally read-only — users must reactivate before
        # continuing the conversation.
        if session_data.get("status") != ChatSessionStatus.ACTIVE.value:
            return jsonify(
                {
                    "success": False,
                    "error": "This chat is archived. Reactivate it to continue.",
                }
            ), 409

        # Pre-fetch existing messages for context decisions.
        messages = service.get_session_messages(session_id, limit=20)

        research_id = None
        research_mode = "none"
        message_id = None
        settings_snapshot = None
        research_context = None

        if trigger_research:
            # Always quick mode in chat (intentional v1 scope).
            research_mode = "quick"

            # Verify the DB password is available BEFORE creating any rows
            # or spawning a worker. Chat-triggered research runs on a
            # background thread that writes token/search metrics to the
            # user's encrypted database; without the password every metric
            # write is silently dropped, leaving the metrics dashboard empty
            # while the research still completes (issue #4457). The password
            # is re-fetched at the spawn site below, so discard it here.
            _pw, session_expired = resolve_user_password(username)
            if session_expired:
                return jsonify(
                    {
                        "success": False,
                        "error": "Your session has expired. Please log out "
                        "and log back in to continue.",
                    }
                ), 401

            # ---- Concurrency guards (per-session + global per-user) ----
            # Both guards run in one transaction so a stale-row reclaim
            # is visible to the count check below it.
            #
            # Without the stale-thread sweep, a process crash leaves the
            # ResearchHistory row at IN_PROGRESS forever — every later
            # send_message returns 409 with no in-chat way to recover.
            #
            # Sweep AGE NOTE: a brand-new IN_PROGRESS row briefly exists
            # before its worker registers in `_active_research` (between
            # the DB commit below and the `start_research_process` call).
            # During that window `is_research_thread_alive` would return
            # False even though the thread spawn is in flight. Only reclaim
            # rows older than `_STALE_RESEARCH_GRACE_SECONDS` (default 30s)
            # so we don't kill our own freshly-inserted research from a
            # racing concurrent send.
            _STALE_RESEARCH_GRACE_SECONDS = 30
            grace_cutoff_dt = datetime.now(UTC) - timedelta(
                seconds=_STALE_RESEARCH_GRACE_SECONDS
            )
            # ResearchHistory.created_at is a String column (ISO-8601);
            # UserActiveResearch.started_at is a UtcDateTime column.
            grace_cutoff_iso = grace_cutoff_dt.isoformat()
            with get_user_db_session(username) as cap_db:
                # 1. Reclaim stale chat-session research rows whose
                #    worker thread is dead AND that are older than the
                #    spawn-grace cutoff.
                stale_chat = (
                    cap_db.query(ResearchHistory)
                    .filter(
                        ResearchHistory.chat_session_id == session_id,
                        ResearchHistory.status == ResearchStatus.IN_PROGRESS,
                        ResearchHistory.created_at < grace_cutoff_iso,
                    )
                    .all()
                )
                reclaimed_chat = False
                for row in stale_chat:
                    if not is_research_thread_alive(row.id):
                        logger.warning(
                            f"Reclaiming stale chat research {row.id[:8]}... "
                            f"(thread dead) on chat {session_id[:8]}..."
                        )
                        row.status = ResearchStatus.FAILED
                        cleanup_research(row.id)
                        reclaimed_chat = True
                if reclaimed_chat:
                    cap_db.commit()

                # 2. Per-session guard: at most one live research per chat.
                existing_session_research = (
                    cap_db.query(ResearchHistory)
                    .filter_by(
                        chat_session_id=session_id,
                        status=ResearchStatus.IN_PROGRESS,
                    )
                    .first()
                )
                if existing_session_research:
                    return jsonify(
                        {
                            "success": False,
                            "error": "Research already in progress on this chat session. Stop it before sending a new message.",
                            "active_research_id": existing_session_research.id,
                        }
                    ), 409

                # 3. Reclaim stale UserActiveResearch rows so the count
                #    below isn't inflated by dead threads. Same grace
                #    window applied via started_at to avoid killing a
                #    sibling request's just-spawned thread. Shared with
                #    research_routes.start_research; chat passes a
                #    grace_cutoff_dt because chat send can race with
                #    its own concurrent sibling, research_routes can't.
                from ..web.routes.globals import (
                    reclaim_stale_user_active_research,
                )

                if reclaim_stale_user_active_research(
                    cap_db,
                    username,
                    grace_cutoff_dt=grace_cutoff_dt,
                    logger=logger,
                ):
                    cap_db.commit()

                # 4. Global per-user cap (mirrors
                #    research_routes.start_research). Without this,
                #    multiple chat tabs let a user bypass the cap.
                active_count = (
                    cap_db.query(UserActiveResearch)
                    .filter_by(
                        username=username,
                        status=ResearchStatus.IN_PROGRESS,
                    )
                    .count()
                )
                max_concurrent = SettingsManager(db_session=cap_db).get_setting(
                    "app.max_concurrent_researches", 3
                )
                if active_count >= max_concurrent:
                    return jsonify(
                        {
                            "success": False,
                            "error": (
                                f"Concurrent research limit reached "
                                f"({active_count}/{max_concurrent}). "
                                "Wait for an existing research to finish."
                            ),
                        }
                    ), 429
            # ---- end concurrency guards ----

        # Settings + context (read-only — fine to do after the cap
        # check, before the atomic write).
        if trigger_research:
            settings_snapshot = _load_settings(username)
            context_manager = ChatContextManager(
                session_id,
                messages,
                session_data.get("accumulated_context"),
                settings_snapshot=settings_snapshot,
            )
            # Pass the new user message so prior conversation is condensed
            # into a summary focused on this question (used as the follow-up
            # prompt's "previous findings").
            research_context = context_manager.build_research_context(
                current_query=content
            )

            # Atomically insert user message + IN_PROGRESS research row,
            # then spawn the worker. All failure modes (400 malformed
            # settings, 404 session-deleted-race, 409 concurrent /
            # duplicate, 429 at-capacity) surface as ChatSpawnError and
            # are translated by ``_chat_spawn_response`` so the response
            # shape stays single-sourced across send_message and
            # retry_attempt.
            try:
                research_id, message_id = _spawn_chat_research(
                    username,
                    session_id,
                    content,
                    settings_snapshot,
                    research_context,
                    messages,
                    service,
                    research_mode=research_mode,
                )
            except ChatSpawnError as exc:
                return _chat_spawn_response(exc)
        else:
            # trigger_research=False: persist the user message without
            # spawning a research run. Mirrors the original send_message
            # shape — the message lands even when research is suppressed.
            try:
                with get_user_db_session(username) as db_session:
                    message_id = service.insert_message_in_db(
                        db_session,
                        session_id=session_id,
                        role="user",
                        content=content,
                        message_type=(
                            "query" if len(messages) == 0 else "followup"
                        ),
                    )
                    db_session.commit()
            except ValueError as exc:
                if "not found" in str(exc).lower():
                    return jsonify(
                        {"success": False, "error": "Session not found"}
                    ), 404
                raise
            research_id = None

        return jsonify(
            {
                "success": True,
                "message_id": message_id,
                "session_id": session_id,
                "research_id": research_id,
                "research_mode": research_mode,
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error sending chat message")
        return jsonify(
            {
                "success": False,
                "error": "Failed to send message",
            }
        ), 500


# ============================================================================
# Per-attempt API Routes (delete + retry a single chat turn)
# ============================================================================


@chat_bp.route(
    "/api/chat/sessions/<session_id>/attempts/<research_id>",
    methods=["DELETE"],
)
@login_required
# Per-user keying, like the other state-changing chat routes. Caps bulk
# delete attempts that the global limiter alone left under-constrained.
@limiter.limit("30 per minute", key_func=_get_api_user_key)
def delete_attempt(session_id, research_id):
    """Delete a single chat attempt (user message + research + response).

    Refuses with 409 if the target research is IN_PROGRESS and its worker
    thread is alive — the client must Stop it first (or wait for it to
    fail naturally). Stale IN_PROGRESS rows whose thread is dead are
    reclaimed and deleted.
    """
    try:
        username = session.get("username")
        service = ChatService(username)
        try:
            service.delete_attempt(session_id, research_id)
        except AttemptNotFound:
            return jsonify(
                {"success": False, "error": "Attempt not found"}
            ), 404
        except AttemptInProgress:
            return jsonify(
                {
                    "success": False,
                    "error": (
                        "Research is in progress. Stop it before deleting "
                        "the attempt."
                    ),
                    "active_research_id": research_id,
                }
            ), 409

        return jsonify({"success": True})

    except ROUTE_EXCEPTIONS:
        logger.exception("Error deleting chat attempt")
        return jsonify(
            {"success": False, "error": "Failed to delete attempt"}
        ), 500


@chat_bp.route(
    "/api/chat/sessions/<session_id>/attempts/<research_id>/retry",
    methods=["POST"],
)
@login_required
# Same per-minute cap as send_message: retry spawns a full research run,
# so it shares the spawn budget rather than getting a separate one (else
# alternating send/retry would let a user exceed the intended rate).
@limiter.limit("10 per minute", key_func=_get_api_user_key)
def retry_attempt(session_id, research_id):
    """Retry a chat attempt: delete the old turn, re-submit same content.

    Looks up the original user message content via
    ``ChatService.get_original_attempt_query`` (uses
    ``research_meta.submission.message_id`` first, falls back to a
    ChatMessage query for older rows), deletes the old attempt
    atomically, then runs the same spawn path as ``send_message``.

    Returns the SAME shape as ``send_message`` so the client can
    subscribe to the new research_id via its existing post-send flow.
    """
    try:
        username = session.get("username")
        service = ChatService(username)

        # Verify session exists + is active BEFORE doing anything
        # destructive. The get_session call also seeds session_data for
        # the context-builder below.
        try:
            session_data = service.get_session(session_id)
        except ChatSessionNotFound:
            return jsonify(
                {"success": False, "error": "Session not found"}
            ), 404

        if session_data.get("status") != ChatSessionStatus.ACTIVE.value:
            return jsonify(
                {
                    "success": False,
                    "error": (
                        "This chat is archived. Reactivate it to continue."
                    ),
                }
            ), 409

        # Resolve DB password BEFORE any destructive op: the spawn path
        # needs it for metrics writes, and re-checking AFTER delete_attempt
        # would leave the attempt gone on a session-expired 401.
        _pw, session_expired = resolve_user_password(username)
        if session_expired:
            return jsonify(
                {
                    "success": False,
                    "error": (
                        "Your session has expired. Please log out and "
                        "log back in to continue."
                    ),
                }
            ), 401

        # Per-session + per-user concurrency guards. Same sweep pattern
        # as send_message: reclaim stale rows older than the grace
        # window first so the count check is accurate, then refuse if
        # this session OR the user is at capacity.
        #
        # If the TARGET research is still IN_PROGRESS, the per-session
        # guard catches it here and returns 409 — the user must Stop it
        # first. (delete_attempt would also catch this via
        # AttemptInProgress, but failing earlier — before any rows are
        # touched — keeps the failure cheap.)
        _STALE_RESEARCH_GRACE_SECONDS = 30
        grace_cutoff_dt = datetime.now(UTC) - timedelta(
            seconds=_STALE_RESEARCH_GRACE_SECONDS
        )
        grace_cutoff_iso = grace_cutoff_dt.isoformat()
        with get_user_db_session(username) as cap_db:
            # Reclaim stale chat-session research rows whose worker is
            # dead AND older than the grace cutoff. Mirrors
            # send_message's sweep at chat/routes.py:903-923.
            stale_chat = (
                cap_db.query(ResearchHistory)
                .filter(
                    ResearchHistory.chat_session_id == session_id,
                    ResearchHistory.status == ResearchStatus.IN_PROGRESS,
                    ResearchHistory.created_at < grace_cutoff_iso,
                )
                .all()
            )
            reclaimed_chat = False
            for row in stale_chat:
                if not is_research_thread_alive(row.id):
                    logger.warning(
                        f"Reclaiming stale chat research {row.id[:8]}... "
                        f"(thread dead) on chat {session_id[:8]}..."
                    )
                    row.status = ResearchStatus.FAILED
                    cleanup_research(row.id)
                    reclaimed_chat = True
            if reclaimed_chat:
                cap_db.commit()

            # Reclaim stale UserActiveResearch rows globally for this
            # user, so the per-user cap count below isn't inflated.
            from ..web.routes.globals import (
                reclaim_stale_user_active_research,
            )

            if reclaim_stale_user_active_research(
                cap_db,
                username,
                grace_cutoff_dt=grace_cutoff_dt,
                logger=logger,
            ):
                cap_db.commit()

            cap_error = _enforce_chat_session_research_slot(
                cap_db, username, session_id
            )
            if cap_error is not None:
                msg, status = cap_error
                if status == 409:
                    return jsonify(
                        {
                            "success": False,
                            "error": msg,
                            "active_research_id": research_id,
                        }
                    ), 409
                return jsonify({"success": False, "error": msg}), status

        # Fetch original user content BEFORE deleting the old attempt.
        # After delete_attempt the row is gone, so this lookup must
        # succeed first. Raises AttemptNotFound → 404.
        try:
            original_content = service.get_original_attempt_query(
                session_id, research_id
            )
        except AttemptNotFound:
            return jsonify(
                {"success": False, "error": "Attempt not found"}
            ), 404

        if not original_content.strip():
            # Empty/whitespace-only original — shouldn't happen since
            # send_message rejects these, but guard against a corrupt
            # row before we destroy the only copy.
            return jsonify(
                {
                    "success": False,
                    "error": (
                        "Original attempt has no query content to retry."
                    ),
                }
            ), 400

        # Delete the old attempt. By this point the target is guaranteed
        # not IN_PROGRESS (the per-session guard above would have
        # caught it), so this should not raise AttemptInProgress — but
        # catch it anyway in case of a narrow race.
        try:
            service.delete_attempt(session_id, research_id)
        except AttemptInProgress:
            return jsonify(
                {
                    "success": False,
                    "error": (
                        "Research is in progress. Stop it before retrying."
                    ),
                    "active_research_id": research_id,
                }
            ), 409
        except AttemptNotFound:
            # Concurrent retry already deleted the target.
            return jsonify(
                {"success": False, "error": "Attempt not found"}
            ), 404

        # Re-fetch messages list AFTER the delete so the context
        # builder doesn't see the just-removed attempt (otherwise the
        # new research would treat the retried query as a follow-up to
        # itself).
        messages = service.get_session_messages(session_id, limit=20)

        settings_snapshot = _load_settings(username)
        context_manager = ChatContextManager(
            session_id,
            messages,
            session_data.get("accumulated_context"),
            settings_snapshot=settings_snapshot,
        )
        research_context = context_manager.build_research_context(
            current_query=original_content
        )

        # Spawn the new research via the shared helper. Same failure
        # shape as send_message.
        try:
            new_research_id, new_message_id = _spawn_chat_research(
                username,
                session_id,
                original_content,
                settings_snapshot,
                research_context,
                messages,
                service,
                research_mode="quick",
            )
        except ChatSpawnError as exc:
            return _chat_spawn_response(exc)

        return jsonify(
            {
                "success": True,
                "session_id": session_id,
                "research_id": new_research_id,
                "message_id": new_message_id,
                "research_mode": "quick",
            }
        )

    except ROUTE_EXCEPTIONS:
        logger.exception("Error retrying chat attempt")
        return jsonify(
            {"success": False, "error": "Failed to retry attempt"}
        ), 500
