"""
Service layer for chat functionality.

This service handles the business logic for chat sessions and messages,
including session management, message handling, and context building.
"""

from typing import Dict, Any, List, Optional
from concurrent.futures import (
    ThreadPoolExecutor,
    TimeoutError as FuturesTimeoutError,
)
from datetime import datetime, UTC
import uuid
from loguru import logger
from sqlalchemy import and_, or_, update
from sqlalchemy.exc import SQLAlchemyError

from ..database.models import (
    ChatMessage,
    ChatMessageType,
    ChatProgressStep,
    ChatRole,
    ChatSession,
    ChatSessionStatus,
    ResearchHistory,
    UserActiveResearch,
)
from ..database.session_context import get_user_db_session
from ..web.routes.globals import (
    cleanup_research,
    is_research_thread_alive,
    set_termination_flag,
)
from ..constants import ResearchStatus

# Standard exception tuple for service-layer DB operations
DB_EXCEPTIONS = (ValueError, RuntimeError, SQLAlchemyError)


class ArchiveBlockedError(RuntimeError):
    """Raised when archive_session is called while a research is in_progress.

    Archive flips a session to read-only; allowing it while research is
    still running would leave an orphaned research tied to a session the
    user thinks is frozen. The route layer translates this to HTTP 409.
    """


class ChatSessionNotFound(LookupError):
    """Raised by get_session when no row matches the supplied session_id.

    The route layer translates this to HTTP 404. Distinct from
    ChatRepositoryError so a transient DB failure cannot masquerade as
    "session does not exist".
    """


class ChatRepositoryError(RuntimeError):
    """Raised by get_session when the underlying DB query fails.

    The route layer translates this to HTTP 500. Keeping this separate
    from ChatSessionNotFound prevents false 404s on infrastructure
    errors (locked DB file, encryption key failure, etc.).
    """


class AttemptNotFound(LookupError):
    """Raised when no ResearchHistory row matches the supplied research_id
    inside the session scoped by ``session_id``.

    The route layer translates this to HTTP 404. Distinct from
    ChatSessionNotFound (also 404) so the diagnostic log distinguishes
    "session missing" from "research missing".
    """


class AttemptInProgress(RuntimeError):
    """Raised by delete_attempt when the target research is IN_PROGRESS
    AND its worker thread is alive.

    Deleting a research out from under a live worker would orphan the
    thread (it would keep burning LLM cycles against a ResearchHistory
    row that no longer exists). The route layer translates this to HTTP
    409 and returns ``research_id`` so the client can offer a Stop +
    retry flow.

    A stale IN_PROGRESS row whose thread is dead does NOT raise: the
    sweep logic inside ``delete_attempt`` reclaims it (mirrors
    send_message's stale-row reclaim at chat/routes.py:903-923).
    """


# Title generation should not block the request thread on a slow LLM.
# Wrap the synchronous llm.invoke() call in a worker future with a hard
# wall-clock timeout. The default matches the rest of the codebase's
# 30s LLM-timeout convention; tune via chat.title_llm_timeout_seconds.
_DEFAULT_TITLE_LLM_TIMEOUT_SECONDS = 30.0


def _serialize_dt(value):
    """Return an ISO-8601 string for a datetime, or None."""
    return value.isoformat() if value is not None else None


class ChatService:
    """Service for managing chat conversations and messages."""

    def __init__(self, username: str):
        """
        Initialize the chat service.

        Args:
            username: Username for database access
        """
        self.username = username

    @staticmethod
    def _atomic_increment(db, counter_col, where_clause):
        """Atomically increment an integer counter column by one and return
        its new value.

        Emits a single ``UPDATE <table> SET col = col + 1 WHERE <clause>
        RETURNING col`` so the next sequence number is allocated without a
        read-modify-write race between concurrent writers. ``counter_col`` is
        a mapped column attribute (e.g. ``ChatSession.message_count``); the
        target table and column name are derived from it.

        Returns the post-increment value, or ``None`` when no row matched
        ``where_clause`` — the caller decides whether a miss is a 404/ValueError.
        Extracted from the previously-duplicated counter logic in
        ``insert_message_in_db`` (message_count) and ``add_progress_step``
        (step_count).
        """
        # Key the values() dict by the column object (not its string name)
        # so the table/column are both derived straight from counter_col.
        stmt = (
            update(counter_col.class_)
            .where(where_clause)
            .values({counter_col: counter_col + 1})
            .returning(counter_col)
        )
        return db.execute(stmt).scalar_one_or_none()

    def create_session(
        self,
        initial_query: Optional[str] = None,
        title: Optional[str] = None,
        settings_snapshot: Optional[Dict[str, Any]] = None,
    ) -> str:
        """
        Create a new chat session.

        Args:
            initial_query: Optional initial query (used for title generation)
            title: Optional custom title for the session
            settings_snapshot: Optional settings for LLM title generation

        Returns:
            Session ID (UUID string)
        """
        try:
            session_id = str(uuid.uuid4())

            # Use fast, non-LLM fallback title synchronously. If the caller
            # wants an LLM-generated title they trigger it asynchronously via
            # POST /api/chat/sessions/<id>/generate-title so the creation
            # request isn't blocked on an LLM round-trip.
            resolved_title = title or self._fallback_title(initial_query)

            with get_user_db_session(self.username) as db:
                # created_at is populated by the utcnow() default on the
                # column — no need to pass it explicitly.
                session = ChatSession(
                    id=session_id,
                    title=resolved_title,
                    status=ChatSessionStatus.ACTIVE.value,
                    accumulated_context={
                        "key_entities": [],
                        "topics": [],
                        "summary": "",
                    },
                    message_count=0,
                )
                db.add(session)
                db.commit()

                logger.info(
                    f"Created chat {session_id[:8]}... for user {self.username}"
                )
                return session_id

        except DB_EXCEPTIONS:
            logger.exception("Error creating chat session")
            raise

    def regenerate_title_with_llm(
        self,
        session_id: str,
        query: Optional[str],
        settings_snapshot: Optional[Dict[str, Any]],
    ) -> Optional[str]:
        """
        Regenerate a session's title using the LLM.

        Intended to be called from a dedicated endpoint the frontend fires
        after session creation, so the create request doesn't block on the
        LLM round-trip.

        Idempotency: if the current title no longer matches the non-LLM
        fallback (i.e. the user manually edited it, or a sibling tab's
        LLM-gen already ran), skip the LLM call so we don't spend credits
        only to overwrite the user's deliberate edit on the way back.

        Returns the new title on success, or None on failure / no-op.
        """
        if not query:
            return None
        # Check whether the session still has the fallback title. If the
        # user (or a concurrent generate-title request) has already moved
        # past the fallback, don't burn an LLM call to overwrite their work.
        try:
            current = self.get_session(session_id)
        except ChatSessionNotFound:
            return None
        current_title = (current or {}).get("title") or ""
        fallback = self._fallback_title(query)
        if current_title and current_title != fallback:
            logger.info(
                f"Skipping LLM title gen for {session_id[:8]}...: title "
                f"already set ('{current_title[:30]}...')"
            )
            return None
        new_title = self._generate_title(query, settings_snapshot)
        if not new_title:
            return None
        updated = self.update_session_title(session_id, new_title)
        return new_title if updated else None

    def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        message_type: str,
        research_id: Optional[str] = None,
        allow_archived: bool = False,
    ) -> str:
        """
        Add a durable message (query/followup/response) to a chat session.

        Content is required and stored inline. Step rows live in
        chat_progress_steps and are written via add_progress_step().

        Args:
            session_id: ID of the session to add message to
            role: Message role (user or assistant)
            content: Message content (required, non-empty)
            message_type: Type of message (query, followup, response)
            research_id: Optional ID of associated research
            allow_archived: When True, the atomic-update WHERE clause omits
                the ``status='active'`` filter so a system-written assistant
                response can land even if the session was archived between
                research start and completion. Use ONLY for system writes
                (final assistant response, terminate-partial) — the user
                send-message path MUST keep the default False so that
                archiving a session in one browser tab still blocks a
                concurrent user reply from another tab mid-flight.

        Returns:
            Message ID (UUID string)

        Raises:
            ValueError: if content is None. Validated before opening the DB
                session so callers (e.g. the route layer) can return HTTP 400
                without paying SQLCipher cold-open cost on a doomed request.
        """
        if content is None:
            raise ValueError("content is required for chat messages")
        try:
            with get_user_db_session(self.username) as db:
                message_id = self.insert_message_in_db(
                    db,
                    session_id=session_id,
                    role=role,
                    content=content,
                    message_type=message_type,
                    research_id=research_id,
                    allow_archived=allow_archived,
                )
                db.commit()
                return message_id

        except DB_EXCEPTIONS:
            logger.exception("Error adding message to chat session")
            raise

    def insert_message_in_db(
        self,
        db,
        session_id: str,
        role: str,
        content: str,
        message_type: str,
        research_id: Optional[str] = None,
        allow_archived: bool = False,
    ) -> str:
        """
        Insert a durable chat message in an active SQLAlchemy session WITHOUT
        committing. The caller owns the transaction lifecycle and is
        responsible for commit/rollback.

        This exists so the route layer can atomically commit the user message
        together with the research-history row in a single transaction —
        avoiding the orphan-message bug that occurs if the user-message
        commit succeeds and the research insert later raises.

        Validation and the atomic message_count increment are identical to
        ``add_message``; only the commit responsibility differs.

        Raises:
            ValueError: if role/message_type are invalid, content is None,
                or the session row does not exist.
        """
        # Content is required (NOT NULL on the column).
        # Empty string is permitted by the column (NOT NULL only rejects
        # SQL NULL); reject Python None.
        if content is None:
            raise ValueError("content is required for chat messages")

        # Authoritative validation via the enum constructors — raises
        # ValueError for unknown values, which the route layer maps to HTTP
        # 400 via ROUTE_EXCEPTIONS. Keeps failure fast (before DB hit) and
        # avoids the HTTP-500 regression we'd get from letting SQLAlchemy's
        # StatementError surface at commit time.
        try:
            ChatRole(role)
        except ValueError as exc:
            raise ValueError(f"Invalid role: {role!r}") from exc
        try:
            ChatMessageType(message_type)
        except ValueError as exc:
            raise ValueError(f"Invalid message_type: {message_type!r}") from exc

        message_id = str(uuid.uuid4())
        # Atomic increment-and-return on ChatSession.message_count.
        # By default the WHERE clause re-checks `status='active'` so an
        # archive PATCH racing with a user-message send cannot land on
        # a now-archived session. When ``allow_archived=True`` (system-
        # written assistant responses), the filter is relaxed so a
        # final-report save can complete even if the session flipped to
        # archived mid-research — losing the answer is worse than the
        # "archive means stop" semantic for system writes.
        if allow_archived:
            where_clause = ChatSession.id == session_id
            not_found_msg = f"Chat session {session_id} not found"
        else:
            where_clause = (ChatSession.id == session_id) & (
                ChatSession.status == ChatSessionStatus.ACTIVE.value
            )
            not_found_msg = f"Chat session {session_id} not found or not active"
        sequence = self._atomic_increment(
            db, ChatSession.message_count, where_clause
        )
        if sequence is None:
            raise ValueError(not_found_msg)

        # created_at populated by column default (utcnow()).
        message = ChatMessage(
            id=message_id,
            session_id=session_id,
            research_id=research_id,
            role=role,
            message_type=message_type,
            content=content,
            sequence_number=sequence,
        )
        db.add(message)
        logger.debug(
            f"Staged message {sequence} for chat {session_id[:8]}... (uncommitted)"
        )
        return message_id

    def add_progress_step(
        self,
        session_id: str,
        research_id: str,
        content: str,
        phase: Optional[str] = None,
    ) -> str:
        """
        Add a transient research-progress step for a chat session.

        Step rows live in chat_progress_steps and have their
        own per-research sequence (allocated atomically against
        ResearchHistory.step_count). They do NOT increment the chat
        session's message_count.

        Args:
            session_id: ID of the parent chat session
            research_id: ID of the research producing the step
            content: Rendered step text (e.g. "Searching for ...")
            phase: Optional phase tag from research_service._STEP_PHASES

        Returns:
            Step ID (UUID string)
        """
        if content is None:
            raise ValueError("content is required for progress steps")

        try:
            step_id = str(uuid.uuid4())

            with get_user_db_session(self.username) as db:
                # Atomic increment-and-return on research_history.step_count.
                sequence = self._atomic_increment(
                    db,
                    ResearchHistory.step_count,
                    ResearchHistory.id == research_id,
                )
                if sequence is None:
                    raise ValueError(  # noqa: TRY301
                        f"Research {research_id} not found"
                    )

                step = ChatProgressStep(
                    id=step_id,
                    research_id=research_id,
                    session_id=session_id,
                    phase=phase,
                    content=content,
                    sequence_number=sequence,
                )
                db.add(step)
                db.commit()

                logger.debug(
                    f"Added progress step {sequence} for research "
                    f"{research_id[:8]}... in chat {session_id[:8]}..."
                )
                return step_id

        except DB_EXCEPTIONS:
            logger.exception("Error adding progress step")
            raise

    def get_session(self, session_id: str) -> Dict[str, Any]:
        """
        Get a chat session by ID.

        Args:
            session_id: ID of the session

        Returns:
            Session data dictionary.

        Raises:
            ChatSessionNotFound: if no row matches ``session_id``.
                Route layer maps to HTTP 404.
            ChatRepositoryError: if the DB query itself fails. Route
                layer maps to HTTP 500. Keeping these separate avoids
                masking transient DB errors as "not found".
        """
        try:
            with get_user_db_session(self.username) as db:
                session = db.query(ChatSession).filter_by(id=session_id).first()

                if not session:
                    logger.warning(f"Chat not found: {session_id[:8]}...")
                    # noqa: TRY301 — re-raised by the outer except
                    # ChatSessionNotFound below to propagate as 404.
                    raise ChatSessionNotFound(session_id)  # noqa: TRY301

                return {
                    "id": session.id,
                    "title": session.title,
                    "status": session.status,
                    "message_count": session.message_count,
                    "created_at": _serialize_dt(session.created_at),
                    "accumulated_context": session.accumulated_context,
                }

        except ChatSessionNotFound:
            # Propagate as-is; this is the genuine 404 signal.
            raise
        except DB_EXCEPTIONS as exc:
            logger.exception("Error getting chat session")
            raise ChatRepositoryError(
                f"DB error reading session {session_id[:8]}..."
            ) from exc

    def get_session_messages(
        self,
        session_id: str,
        limit: int = 50,
        offset: int = 0,
        before_created_at: Optional[str] = None,
        before_id: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """
        Get messages for a session, server-side merged with progress steps.

        chat_messages.content is always inline; step rows live in
        chat_progress_steps. This method merges both into a single ordered
        stream by created_at so the client renderer (chat.js) sees a
        unified message list with `message_type='step'` rows interleaved.

        Pagination is SQL-level via per-table LIMIT + Python merge: each
        table fetches at most ``limit`` rows ordered by created_at DESC,
        the two streams are merged on the (timestamp, kind) sort key, and
        the latest ``limit`` are returned in ASC order so the client
        renders oldest→newest as before.

        Pass ``before_created_at`` to fetch the page IMMEDIATELY older
        than the given ISO timestamp (use the oldest currently-displayed
        ``created_at`` to implement a "load older messages" trigger).
        Without the cursor, ``offset`` selects which DESC slice to return
        (offset=0 → newest, offset=limit → next older window, …).

        Args:
            session_id: ID of the session
            limit: Maximum number of (merged) entries to return
            offset: Number of entries to skip (DESC-ordered slice index)
            before_created_at: Optional ISO timestamp cursor — return only
                entries strictly older than this. Useful for cursor-based
                "load older" pagination instead of offset arithmetic.

        Returns:
            List of message + step data dictionaries, ordered by
            created_at ascending.
        """
        try:
            with get_user_db_session(self.username) as db:
                msg_q = db.query(ChatMessage).filter_by(session_id=session_id)
                step_q = db.query(ChatProgressStep).filter_by(
                    session_id=session_id
                )

                if before_created_at:
                    try:
                        cutoff = datetime.fromisoformat(
                            before_created_at.replace("Z", "+00:00")
                        )
                    except ValueError:
                        logger.warning(
                            "Invalid before_created_at cursor: "
                            f"{before_created_at!r} — ignoring."
                        )
                    else:
                        # Composite cursor: when `before_id` is also
                        # supplied, the filter becomes
                        #     created_at < cutoff
                        #     OR (created_at = cutoff AND id < before_id)
                        # which prevents same-millisecond rows at the
                        # page boundary from being silently dropped on
                        # "Load older" pagination. With a bare timestamp
                        # cursor we fall back to strict `<` for
                        # backwards-compat with older clients.
                        if before_id:
                            msg_q = msg_q.filter(
                                or_(
                                    ChatMessage.created_at < cutoff,
                                    and_(
                                        ChatMessage.created_at == cutoff,
                                        ChatMessage.id < before_id,
                                    ),
                                )
                            )
                            # ChatProgressStep ids are integers but
                            # message ids are UUID strings; using the
                            # bare `<` operator on string ids gives a
                            # stable lexicographic tie-break, and
                            # progress-step rows tie-break by their
                            # own integer id (id < int(before_id) is
                            # not safe because before_id is the UUID of
                            # a chat message, not a step). For steps,
                            # drop the equality branch so duplicates
                            # rather than drops occur on tie — the
                            # client-side dedup catches them.
                            step_q = step_q.filter(
                                ChatProgressStep.created_at <= cutoff
                            )
                        else:
                            msg_q = msg_q.filter(
                                ChatMessage.created_at < cutoff
                            )
                            step_q = step_q.filter(
                                ChatProgressStep.created_at < cutoff
                            )

                # Pull at most ``limit`` rows from EACH table in DESC
                # order. The merged window is at most 2 * limit rows
                # (one extreme: all from one table), which we trim to
                # ``limit`` after the Python merge. This bounds the SQL
                # work and avoids the old .all() cliff at large N.
                fetch_n = limit + offset
                # Secondary ORDER BY on sequence_number stabilises rows
                # whose created_at collide at SQLite's millisecond
                # precision (sqlalchemy_utc stores `%Y-%m-%d %H:%M:%S.fff`).
                # Without it, rapid-fire inserts (paste-and-submit,
                # auto-retries) can be returned in arbitrary order even
                # though sequence_number is monotonic.
                messages = (
                    msg_q.order_by(
                        ChatMessage.created_at.desc(),
                        ChatMessage.sequence_number.desc(),
                    )
                    .limit(fetch_n)
                    .all()
                )
                steps = (
                    step_q.order_by(
                        ChatProgressStep.created_at.desc(),
                        ChatProgressStep.sequence_number.desc(),
                    )
                    .limit(fetch_n)
                    .all()
                )

                merged: List[Dict[str, Any]] = []
                for msg in messages:
                    merged.append(
                        {
                            "id": msg.id,
                            "session_id": msg.session_id,
                            "role": msg.role,
                            "message_type": msg.message_type,
                            "content": msg.content,
                            "sequence_number": msg.sequence_number,
                            "research_id": msg.research_id,
                            "created_at": _serialize_dt(msg.created_at),
                        }
                    )
                for step in steps:
                    merged.append(
                        {
                            "id": f"step-{step.id}",
                            "session_id": step.session_id,
                            "role": "assistant",
                            "message_type": "step",
                            "content": step.content,
                            "phase": step.phase,
                            "sequence_number": step.sequence_number,
                            "research_id": step.research_id,
                            "created_at": _serialize_dt(step.created_at),
                        }
                    )

                # Sort DESC by (created_at, sequence_number,
                # step-before-message on tie), take the newest
                # [offset:offset+limit] slice, then flip to ASC so the
                # client still renders oldest→newest. Including
                # sequence_number in the Python tie-break mirrors the
                # SQL ORDER BY above and prevents same-timestamp messages
                # from rendering out of insertion order.
                merged.sort(
                    key=lambda m: (
                        m["created_at"] or "",
                        m.get("sequence_number") or 0,
                        0 if m["message_type"] == "step" else 1,
                    ),
                    reverse=True,
                )
                window = merged[offset : offset + limit]
                window.reverse()
                return window

        except DB_EXCEPTIONS:
            # Re-raise so the route returns HTTP 500 instead of a
            # misleading 200 + []. An empty list here would be
            # indistinguishable from a session that genuinely has no
            # messages, hiding infrastructure failures from the client.
            logger.exception("Error getting chat messages")
            raise

    def get_in_progress_research_id(self, session_id: str) -> Optional[str]:
        """Return the id of the in-progress research for this chat session,
        or ``None`` if no research is currently running.

        Used by the GET messages endpoint so the client can restore the
        live "thinking" indicator on reload without inferring it from
        message metadata (which fails during the wrapper-strategy
        preprocessing window before any progress step has persisted).

        The partial-unique index
        ``ux_research_history_chat_session_in_progress`` (migration 0010)
        guarantees at most one matching row exists and turns this into
        an O(1) index lookup.
        """
        try:
            with get_user_db_session(self.username) as db:
                row = (
                    db.query(ResearchHistory.id)
                    .filter(
                        ResearchHistory.chat_session_id == session_id,
                        ResearchHistory.status == ResearchStatus.IN_PROGRESS,
                    )
                    .first()
                )
                return row[0] if row else None
        except DB_EXCEPTIONS:
            # Re-raise rather than swallow → the route handler can
            # surface a 500 so the client shows an error banner. Returning
            # None here is indistinguishable from "no research running",
            # which leaves the send button enabled and lets the user
            # double-submit into the unique-index guard.
            logger.exception(
                "Error fetching in-progress research id for chat session"
            )
            raise

    def list_sessions(
        self,
        status: str = ChatSessionStatus.ACTIVE.value,
        limit: int = 20,
        offset: int = 0,
    ) -> List[Dict[str, Any]]:
        """
        List chat sessions for the user.

        Args:
            status: Filter by status (active, archived, deleted, or all)
            limit: Maximum number of sessions to return
            offset: Number of sessions to skip

        Returns:
            List of session data dictionaries
        """
        try:
            with get_user_db_session(self.username) as db:
                query = db.query(ChatSession)

                if status != "all":
                    query = query.filter_by(status=status)

                sessions = (
                    query.order_by(ChatSession.created_at.desc())
                    .offset(offset)
                    .limit(limit)
                    .all()
                )

                return [
                    {
                        "id": s.id,
                        "title": s.title,
                        "status": s.status,
                        "message_count": s.message_count,
                        "created_at": _serialize_dt(s.created_at),
                    }
                    for s in sessions
                ]

        except DB_EXCEPTIONS:
            # Re-raise so the route returns HTTP 500. Silently returning
            # [] would make a real DB failure look like a brand-new user
            # with no sessions, hiding the problem from operators and
            # confusing the UI.
            logger.exception("Error listing chat sessions")
            raise

    def update_session_title(self, session_id: str, title: str) -> bool:
        """
        Update the title of a chat session.

        Args:
            session_id: ID of the session
            title: New title

        Returns:
            True if updated successfully
        """
        try:
            with get_user_db_session(self.username) as db:
                session = db.query(ChatSession).filter_by(id=session_id).first()
                if session:
                    session.title = title

                    db.commit()
                    return True
                return False

        except DB_EXCEPTIONS:
            logger.exception("Error updating chat session title")
            return False

    def reactivate_session(self, session_id: str) -> bool:
        """
        Reactivate an archived or deleted chat session.

        Args:
            session_id: ID of the session to reactivate

        Returns:
            True if reactivated successfully
        """
        try:
            with get_user_db_session(self.username) as db:
                session = db.query(ChatSession).filter_by(id=session_id).first()
                if session:
                    session.status = ChatSessionStatus.ACTIVE.value

                    db.commit()
                    logger.info(f"Reactivated chat: {session_id[:8]}...")
                    return True
                return False

        except DB_EXCEPTIONS:
            logger.exception("Error reactivating chat session")
            return False

    def archive_session(self, session_id: str) -> bool:
        """
        Archive a chat session.

        Refuses to archive while a research is still in_progress for the
        session: archive flips the session read-only, and an in-flight
        research would otherwise survive as an orphaned process writing
        back into a session the user believes is frozen. The caller (route layer)
        must stop the research first (or use delete, which terminates
        in-flight research as a side effect).

        Args:
            session_id: ID of the session to archive

        Returns:
            True if archived successfully, False if the session does not
            exist or a DB error occurred.

        Raises:
            ArchiveBlockedError: if the session has an in_progress
                research tied to it. The route layer maps this to HTTP
                409, mirroring the existing send-to-archived 409 rule.
        """
        try:
            with get_user_db_session(self.username) as db:
                session = db.query(ChatSession).filter_by(id=session_id).first()
                if not session:
                    return False

                in_flight = (
                    db.query(ResearchHistory.id)
                    .filter(
                        ResearchHistory.chat_session_id == session_id,
                        ResearchHistory.status == ResearchStatus.IN_PROGRESS,
                    )
                    .first()
                )
                if in_flight is not None:
                    # Bubble up to route layer for 409 mapping. Caught
                    # and re-raised by the inner ``except
                    # ArchiveBlockedError`` below — the broad
                    # ``except DB_EXCEPTIONS`` must not swallow it.
                    raise ArchiveBlockedError(  # noqa: TRY301 — re-raised by inner except ArchiveBlockedError
                        "Cannot archive: research in_progress. Stop it first."
                    )

                session.status = ChatSessionStatus.ARCHIVED.value
                db.commit()
                logger.info(f"Archived chat: {session_id[:8]}...")
                return True

        except ArchiveBlockedError:
            # Bubble up so the route layer can produce a 409 response.
            raise
        except DB_EXCEPTIONS:
            logger.exception("Error archiving chat session")
            return False

    def delete_session(self, session_id: str) -> bool:
        """
        Permanently delete a chat session.

        Cascades: ChatMessages deleted (CASCADE), ResearchHistory.chat_session_id set NULL.

        Args:
            session_id: ID of the session to delete

        Returns:
            True if deleted successfully
        """
        try:
            # Terminate any in-progress research tied to this session, so the
            # FK's ON DELETE SET NULL doesn't leave it alive with a null
            # chat_session_id — an orphan that keeps burning LLM cycles for a
            # conversation the user already discarded.
            #
            # Order matters: collect the in-flight ids inside the transaction,
            # but set the (in-memory, non-transactional) termination flags only
            # AFTER the delete commits. Flagging before the commit would, on a
            # commit failure, kill the research of a session that still exists.
            with get_user_db_session(self.username) as db:
                session = db.query(ChatSession).filter_by(id=session_id).first()
                if not session:
                    return False
                in_flight = (
                    db.query(ResearchHistory.id)
                    .filter(
                        ResearchHistory.chat_session_id == session_id,
                        ResearchHistory.status == ResearchStatus.IN_PROGRESS,
                    )
                    .all()
                )
                db.delete(session)
                db.commit()
                for (rid,) in in_flight:
                    set_termination_flag(rid)
                # Include the (truncated) username so a stolen-token bulk
                # delete leaves a forensic trail tying each deletion to an
                # account, not just an opaque session id (L_SEC1).
                logger.info(
                    f"Deleted chat: user={self.username[:8]}... "
                    f"session={session_id[:8]}..."
                )
                return True

        except DB_EXCEPTIONS:
            logger.exception("Error deleting chat session")
            return False

    def delete_attempt(self, session_id: str, research_id: str) -> bool:
        """Permanently delete one chat attempt (research + its messages + steps).

        An "attempt" is the user message that triggered a research, the
        research_history row itself, any assistant response message(s)
        tagged with that research, and all chat_progress_steps. The
        assistant message(s) and progress steps carry ``research_id``
        directly; the user message is inserted with ``research_id=NULL``
        (see ``_spawn_chat_research``) and is reachable only via
        ``research_meta.submission.message_id`` — both linkages are
        resolved below so the user bubble is removed too (issue #4659).

        Refuses to delete while the target research is IN_PROGRESS and
        its worker thread is alive (raises ``AttemptInProgress`` → 409).
        A stale IN_PROGRESS row whose thread is dead is reclaimed: the
        status flips to FAILED inside the same transaction that deletes
        the rows (mirrors the stale-reclaim sweep in
        chat/routes.py:903-923).

        Unlike ``delete_session``, this MUST decrement
        ``ChatSession.message_count`` — the session still exists, so the
        counter would otherwise drift permanently upward. Mirrors the
        per-message decrement in ``_cleanup_chat_send_rows``
        (chat/routes.py:175-179).

        Args:
            session_id: ID of the parent chat session (scoped lookup).
            research_id: ID of the research attempt to delete.

        Returns:
            True if the attempt existed and was deleted.

        Raises:
            AttemptNotFound: research_id does not belong to session_id.
                Route layer maps to HTTP 404.
            AttemptInProgress: research is IN_PROGRESS and its worker
                thread is alive. Route layer maps to HTTP 409 with
                ``active_research_id``.
        """
        try:
            # Phase 1 — load + liveness check, OUTSIDE the delete tx so
            # the (rare) set_termination_flag call below doesn't have to
            # roll back if the worker is still mid-flight. Also lets us
            # return AttemptNotFound before touching any rows.
            with get_user_db_session(self.username) as db:
                research = (
                    db.query(ResearchHistory)
                    .filter(
                        ResearchHistory.id == research_id,
                        ResearchHistory.chat_session_id == session_id,
                    )
                    .first()
                )
                if research is None:
                    raise AttemptNotFound(research_id)  # noqa: TRY301 — re-raised by outer except
                if (
                    research.status == ResearchStatus.IN_PROGRESS
                    and is_research_thread_alive(research_id)
                ):
                    # Signal the worker to drain; the route layer tells
                    # the client to Stop+retry. Don't hard-delete while a
                    # live thread owns the row — the worker's finally
                    # block would otherwise write back to a deleted id.
                    set_termination_flag(research_id)
                    raise AttemptInProgress(research_id)  # noqa: TRY301 — re-raised by outer except

            # Phase 2 — atomic delete. The row may be FAILED, COMPLETED,
            # SUSPENDED, or stale-IN_PROGRESS (thread dead); all of those
            # are safe to delete. Count the ChatMessage rows first so the
            # message_count decrement matches the rows removed.
            with get_user_db_session(self.username) as db:
                # Re-load inside this tx (the row may have changed
                # status between Phase 1 and Phase 2 — e.g. the worker
                # finished). If the session itself was deleted by a
                # concurrent request, the rows we're about to delete
                # CASCADE away anyway, so a missing row here is a clean
                # 404.
                research = (
                    db.query(ResearchHistory)
                    .filter(
                        ResearchHistory.id == research_id,
                        ResearchHistory.chat_session_id == session_id,
                    )
                    .first()
                )
                if research is None:
                    raise AttemptNotFound(research_id)  # noqa: TRY301 — re-raised by outer except

                # Belt-and-braces: re-check liveness. A thread that was
                # dead in Phase 1 can't come back, but a thread that was
                # alive (and tripped AttemptInProgress above) would have
                # returned already, so this branch only fires for the
                # narrow race where the worker went from dead→alive
                # between the two phases — not actually possible, kept
                # as a defensive guard.
                if (
                    research.status == ResearchStatus.IN_PROGRESS
                    and is_research_thread_alive(research_id)
                ):
                    set_termination_flag(research_id)
                    raise AttemptInProgress(research_id)  # noqa: TRY301 — re-raised by outer except

                # Resolve the user (query) message id. In current
                # production the user message is inserted by
                # _spawn_chat_research with research_id=NULL and linked to
                # the attempt only via research_meta.submission.message_id
                # (mirrors get_original_attempt_query). Assistant
                # response(s) — and legacy pre-research_meta user rows —
                # carry research_id directly. We must delete BOTH, otherwise
                # the user bubble lingers orphaned after the attempt is
                # removed (issue #4659).
                user_message_id = None
                meta = research.research_meta or {}
                submission = meta.get("submission") or {}
                if isinstance(submission, dict):
                    candidate = submission.get("message_id")
                    if isinstance(candidate, str) and candidate:
                        user_message_id = candidate

                # research_id matches assistant rows (+ legacy user rows);
                # the id branch matches the NULL-research_id user message.
                # session_id scopes the id branch so a forged/corrupt
                # message_id can't reach another session's row.
                msg_filter = ChatMessage.research_id == research_id
                if user_message_id:
                    msg_filter = or_(
                        msg_filter,
                        and_(
                            ChatMessage.id == user_message_id,
                            ChatMessage.session_id == session_id,
                        ),
                    )

                # Count messages BEFORE deleting them so the
                # message_count decrement is exact.
                removed_messages = (
                    db.query(ChatMessage).filter(msg_filter).count()
                )

                # ChatMessage.research_id FK is ON DELETE SET NULL, not
                # CASCADE — explicit delete is required to remove the rows
                # (otherwise the bubbles linger with a stale research_id).
                db.query(ChatMessage).filter(msg_filter).delete(
                    synchronize_session=False
                )

                # chat_progress_steps FK is ON DELETE CASCADE, so the
                # research_history.delete() below would clean them up.
                # Delete explicitly so the count is predictable and the
                # tx is self-contained if the CASCADE pragma ever flips.
                db.query(ChatProgressStep).filter(
                    ChatProgressStep.research_id == research_id
                ).delete(synchronize_session=False)

                # user_active_research row (per-user cap counter).
                # Filtered by research_id; the username filter is
                # belt-and-braces (this user's DB only contains their own
                # rows).
                db.query(UserActiveResearch).filter(
                    UserActiveResearch.research_id == research_id
                ).delete(synchronize_session=False)

                # ResearchHistory last so its CASCADE doesn't fire while
                # our explicit deletes are pending.
                db.query(ResearchHistory).filter(
                    ResearchHistory.id == research_id
                ).delete(synchronize_session=False)

                # Decrement message_count. Skipped when no messages were
                # removed (e.g. an attempt that crashed before any
                # assistant response landed) — avoids a pointless UPDATE.
                if removed_messages > 0:
                    db.query(ChatSession).filter(
                        ChatSession.id == session_id
                    ).update(
                        {
                            ChatSession.message_count: (
                                ChatSession.message_count - removed_messages
                            )
                        },
                        synchronize_session=False,
                    )

                db.commit()

            # Phase 3 — post-commit in-memory cleanup. The worker's own
            # finally block calls cleanup_research too, so this is a
            # no-op for the in-progress path; for the stale-IN_PROGRESS
            # path it frees the slot immediately.
            cleanup_research(research_id)

            logger.info(
                f"Deleted chat attempt: user={self.username[:8]}... "
                f"session={session_id[:8]}... research={research_id[:8]}... "
                f"({removed_messages} messages)"
            )
            return True

        except (AttemptNotFound, AttemptInProgress):
            raise
        except DB_EXCEPTIONS:
            logger.exception("Error deleting chat attempt")
            raise

    def get_original_attempt_query(
        self, session_id: str, research_id: str
    ) -> str:
        """Return the original user message content for a chat research.

        Used by the retry route to re-submit the same query without the
        client echoing it back. Looks up ``research_meta.submission.\
        message_id`` first (set at send time by chat/routes.py:1058);
        falls back to a query on ``ChatMessage.research_id == X AND
        role='user'`` for older rows that predate the meta field.

        Args:
            session_id: ID of the parent chat session (scoped lookup).
            research_id: ID of the research attempt.

        Returns:
            The original user message content as a string.

        Raises:
            AttemptNotFound: research_id does not belong to session_id,
                or no user message is reachable from it. Route layer
                maps to HTTP 404.
        """
        try:
            with get_user_db_session(self.username) as db:
                # Scope-by-session first: a research_id from another
                # session (e.g. user-supplied path param) should 404,
                # not silently return that other session's content.
                research = (
                    db.query(ResearchHistory)
                    .filter(
                        ResearchHistory.id == research_id,
                        ResearchHistory.chat_session_id == session_id,
                    )
                    .first()
                )
                if research is None:
                    raise AttemptNotFound(research_id)  # noqa: TRY301 — re-raised by outer except

                # Fast path: research_meta carries the original
                # message_id set at send time.
                user_message_id = None
                meta = research.research_meta or {}
                submission = meta.get("submission") or {}
                if isinstance(submission, dict):
                    candidate = submission.get("message_id")
                    if isinstance(candidate, str) and candidate:
                        user_message_id = candidate

                if user_message_id:
                    msg = (
                        db.query(ChatMessage)
                        .filter(
                            ChatMessage.id == user_message_id,
                            ChatMessage.session_id == session_id,
                            ChatMessage.role == ChatRole.USER.value,
                        )
                        .first()
                    )
                    if msg is not None and msg.content:
                        return str(msg.content)

                # Fallback: pre-research_meta rows. Look up the user
                # message by research_id + role.
                msg = (
                    db.query(ChatMessage)
                    .filter(
                        ChatMessage.research_id == research_id,
                        ChatMessage.session_id == session_id,
                        ChatMessage.role == ChatRole.USER.value,
                    )
                    .order_by(ChatMessage.created_at.asc())
                    .first()
                )
                if msg is None or not msg.content:
                    raise AttemptNotFound(research_id)  # noqa: TRY301 — re-raised by outer except
                return str(msg.content)

        except AttemptNotFound:
            raise
        except DB_EXCEPTIONS:
            logger.exception("Error fetching original attempt query")
            raise

    def update_accumulated_context(
        self,
        session_id: str,
        new_entities: Optional[List[str]] = None,
        new_topics: Optional[List[str]] = None,
        summary_addition: Optional[str] = None,
    ) -> bool:
        """
        Update the accumulated context for a session.

        Args:
            session_id: ID of the session
            new_entities: New entities to add
            new_topics: New topics to add
            summary_addition: Text to append to summary

        Returns:
            True if updated successfully
        """
        try:
            with get_user_db_session(self.username) as db:
                # with_for_update() is a no-op on SQLite but provides
                # row locking on PostgreSQL/MySQL if ever used
                session = (
                    db.query(ChatSession)
                    .filter_by(id=session_id)
                    .with_for_update()
                    .first()
                )
                if not session:
                    return False

                # Build a NEW dict and reassign so SQLAlchemy's plain JSON
                # column marks the row dirty. In-place mutation of the existing
                # dict (or reassigning the same object identity) is not
                # detected without MutableDict.as_mutable() — at flush time
                # the loaded snapshot equals the current value and no UPDATE
                # is emitted. Same convention as research_sources_service.py.
                existing_ctx = session.accumulated_context or {}
                ctx = dict(existing_ctx)

                # Merge entities (deduplicate)
                if new_entities:
                    existing = set(ctx.get("key_entities", []))
                    existing.update(new_entities)
                    ctx["key_entities"] = list(existing)[:50]

                # Merge topics
                if new_topics:
                    existing = set(ctx.get("topics", []))
                    existing.update(new_topics)
                    ctx["topics"] = list(existing)[:20]

                # Append to summary (with size limit)
                if summary_addition:
                    current = ctx.get("summary", "")
                    new_summary = (
                        f"{current}\n\n{summary_addition}"
                        if current
                        else summary_addition
                    )
                    ctx["summary"] = new_summary[-8000:]  # Keep last 8000 chars

                session.accumulated_context = ctx
                db.commit()
                return True

        except DB_EXCEPTIONS:
            logger.exception("Error updating accumulated context")
            return False

    def _fallback_title(self, query: Optional[str]) -> str:
        """Non-LLM title used at creation time (never blocks on I/O)."""
        if not query:
            return f"Chat {datetime.now(UTC).strftime('%Y-%m-%d %H:%M')}"
        if len(query) > 100:
            return query[:97].strip() + "..."
        return query.strip()

    def _generate_title(
        self,
        query: Optional[str],
        settings_snapshot: Optional[Dict[str, Any]] = None,
    ) -> str:
        """
        Generate a title from the initial query.

        When chat.llm_title_generation is enabled and settings_snapshot is
        provided, uses an LLM for concise titles. Otherwise returns the
        non-LLM fallback title.
        """
        if not query:
            return self._fallback_title(query)

        if settings_snapshot:
            from ..config.llm_config import get_llm
            from ..config.thread_settings import get_setting_from_snapshot

            if get_setting_from_snapshot(
                "chat.llm_title_generation",
                False,
                settings_snapshot=settings_snapshot,
            ):
                timeout = float(
                    get_setting_from_snapshot(
                        "chat.title_llm_timeout_seconds",
                        _DEFAULT_TITLE_LLM_TIMEOUT_SECONDS,
                        settings_snapshot=settings_snapshot,
                    )
                )
                # Run the blocking invoke in a worker thread so the request
                # thread isn't parked past `timeout` by an unresponsive LLM.
                # `with ThreadPoolExecutor(...) as pool:` would call
                # shutdown(wait=True) on __exit__, defeating the timeout —
                # use wait=False + cancel_futures so the timeout actually fires.
                pool = ThreadPoolExecutor(
                    max_workers=1,
                    thread_name_prefix="chat-title",
                )
                try:
                    llm = get_llm(settings_snapshot=settings_snapshot)
                    prompt = (
                        "Generate a concise 3-7 word title for this research "
                        "query. Return ONLY the title, no quotes or "
                        f"explanation.\n\nQuery: {query[:200]}"
                    )
                    future = pool.submit(llm.invoke, prompt)
                    try:
                        response = future.result(timeout=timeout)
                    except FuturesTimeoutError:
                        logger.warning(
                            "LLM title generation exceeded {}s timeout; "
                            "falling back to truncation",
                            timeout,
                        )
                        return self._fallback_title(query)
                    # Strip CR/LF before storing: the title is later
                    # interpolated into loguru f-strings (e.g. the
                    # "title already set" log line above) — an embedded
                    # newline forges what looks like a second log entry
                    # in aggregators. Also keeps document.title /
                    # chatTitle.textContent visually clean.
                    title = (
                        str(response.content)
                        .replace("\n", " ")
                        .replace("\r", " ")
                        .strip()
                        .strip("\"'")[:100]
                    )
                    if title:
                        return title
                except Exception:
                    # User opted into LLM title generation via the
                    # `chat.llm_title_generation` setting; a silent
                    # debug-level swallow would hide provider misconfig,
                    # auth failures, or response-shape regressions in
                    # production where stderr level is INFO. Log with
                    # traceback so operators can diagnose, then fall back
                    # to truncation for UX continuity.
                    logger.exception(
                        "LLM title generation failed, falling back to truncation"
                    )
                finally:
                    pool.shutdown(wait=False, cancel_futures=True)

        return self._fallback_title(query)
