import hashlib
import json
import re
import threading
import time
from datetime import datetime, UTC
from pathlib import Path

from loguru import logger

from ...exceptions import DuplicateResearchError, ResearchTerminatedException
from ...config.llm_config import get_llm
from ...settings.manager import SnapshotSettingsContext

# Output directory for research results
from ...config.paths import get_research_outputs_directory
from ...config.search_config import get_search
from ...constants import ResearchStatus
from ...database.models import ResearchHistory, ResearchStrategy
from ...database.session_context import get_user_db_session
from ...database.thread_local_session import thread_cleanup
from ...error_handling.openai_compat_errors import (
    friendly_openai_compatible_error,
    is_openai_compat_runtime_error,
)
from ...error_handling.report_generator import ErrorReportGenerator
from ...utilities.thread_context import set_search_context
from ...report_generator import IntegratedReportGenerator
from ...search_system import AdvancedSearchSystem
from ...text_optimization import CitationFormatter, CitationMode
from ...utilities.log_utils import log_for_research
from ...utilities.search_utilities import extract_links_from_search_results
from ...utilities.threading_utils import thread_context, thread_with_app_context
from ..models.database import calculate_duration
from ...settings.env_registry import get_env_setting
from .socket_service import SocketIOService

OUTPUT_DIR = get_research_outputs_directory()


# Global concurrent research limit (server-wide, across all users)
_MAX_GLOBAL_CONCURRENT = get_env_setting(
    "server.max_concurrent_research", default=10
)
_global_research_semaphore = threading.Semaphore(_MAX_GLOBAL_CONCURRENT)

# Progress allocation for detailed mode — report generation is the bulk of the work
_DETAILED_SEARCH_PROGRESS_CAP = 8  # Search/output phases capped here
_DETAILED_REPORT_PROGRESS_START = 10  # Report generation starts here
_DETAILED_REPORT_PROGRESS_END = 100  # Report generation ends here
# Phases that belong to the report-generation stage.  "report_generation" is
# emitted in this file; the other four are emitted by report_generator.py.  If
# you add or rename a phase, update both this set and the emitter.
_REPORT_PHASES = frozenset(
    {
        "report_generation",
        "report_section_research",
        "report_formatting",
        "report_structure",
        "report_complete",
    }
)

# Phases that produce user-visible step messages in chat mode.
# "complete" is excluded — it fires AFTER the response message is written,
# which would create a step with a higher sequence_number than the response.
_STEP_PHASES = frozenset(
    {
        "init",
        "setup",
        "search_planning",
        "search",
        "observation",
        "output_generation",
        "synthesis_error",
        "synthesis_fallback",
        "report_generation",
        "report_complete",
        "error",
    }
)

# Socket.IO emission throttling: minimum interval between progress emissions per research
_EMIT_THROTTLE_SECONDS = 0.2  # 200ms
_EMIT_TTL_SECONDS = 3600  # 1 hour — evict stale entries from orphaned research

# Cap on the partial-content buffer kept server-side so chat-mode termination
# can persist whatever was already streamed. Bounded to keep memory predictable
# under pathologically long answers (typical answers are a few KB).
_MAX_PARTIAL_BUFFER_BYTES = 256 * 1024  # 256 KB
_emit_cleanup_counter = 0
_last_emit_times: dict[str, float] = {}
_last_emit_lock = threading.Lock()


def _chat_step_decision(
    phase: str | None,
    last_step_phase: str | None,
    is_final: bool,
) -> tuple[bool, bool]:
    """Decide whether to persist + emit a chat-mode step event.

    Encodes the symmetry invariant the progress_callback in
    run_research_process enforces: for chat sessions, what the live UI
    surfaces over the socket must equal what `loadSession` reconstructs
    from chat_progress_steps on reload. The repeat-phase dedup must
    therefore block BOTH writes, not just the DB write.

    Returns:
        (persist, suppress_emit)
        - persist: True iff add_progress_step should be called for this event.
        - suppress_emit: True iff the caller should null the socket payload
          so the emit is dropped. Only suppressed when this is a non-final
          repeat — final phases (complete/error/report_complete) always
          emit so the client completion handler fires.

    Args:
        phase: the phase tag from this event (e.g. "search", "observation")
        last_step_phase: phase tag of the previously persisted chat step
            (None until the first persist of this research).
        is_final: True if this is a "final" phase event the client must
            see to fire its completion handler (complete | error |
            report_complete | progress==100).
    """
    if phase not in _STEP_PHASES:
        # Not a chat-step phase at all (e.g. "complete"). Persist no,
        # and let the emit through unchanged — completion / control
        # events still need to reach the client.
        return False, False
    dedup_ok = phase != last_step_phase or phase == "observation"
    if dedup_ok:
        return True, False
    return False, not is_final


def _parse_research_metadata(research_meta) -> dict:
    """Parse research_meta into a dict, handling both dict and JSON string types."""
    if isinstance(research_meta, dict):
        return dict(research_meta)
    if isinstance(research_meta, str):
        try:
            parsed = json.loads(research_meta)
            return dict(parsed) if isinstance(parsed, dict) else {}
        except json.JSONDecodeError:
            logger.exception("Failed to parse research_meta as JSON")
            return {}
    return {}


def _extract_synthesized_answer(results: dict) -> str:
    """Pull the LLM-synthesized answer out of a strategy result dict.

    ``report_content`` must store ONLY the synthesized answer (LLM
    prose with [N] inline citations). The strategy's
    ``formatted_findings`` is the full ``format_findings`` blob —
    answer + ``format_links_to_markdown`` source list +
    ``## SEARCH QUESTIONS BY ITERATION`` + ``## DETAILED FINDINGS``
    + ``## ALL SOURCES`` — and ``format_document_split`` only knows
    how to strip ``## Sources`` headers, not those other sections.
    Saving the blob would leak sources/findings into ``report_content``.

    Resolution order:
      1. ``Final synthesis`` finding (set by source_based, parallel,
         rapid, focused_iteration, iterdrag).
      2. ``current_knowledge`` (other strategies expose the answer
         there).
      3. Empty string — caller decides whether to fall back further.
    """
    for finding in results.get("findings") or []:
        if finding.get("phase") == "Final synthesis":
            content = finding.get("content") or ""
            if content:
                return content
    return results.get("current_knowledge") or ""


def get_citation_formatter():
    """Get citation formatter with settings from thread context."""
    # Import here to avoid circular imports
    from ...config.search_config import get_setting_from_snapshot

    citation_format = get_setting_from_snapshot(
        "report.citation_format", "number_hyperlinks"
    )
    mode_map = {
        "number_hyperlinks": CitationMode.NUMBER_HYPERLINKS,
        "domain_hyperlinks": CitationMode.DOMAIN_HYPERLINKS,
        "domain_id_hyperlinks": CitationMode.DOMAIN_ID_HYPERLINKS,
        "domain_id_always_hyperlinks": CitationMode.DOMAIN_ID_ALWAYS_HYPERLINKS,
        "source_tagged_hyperlinks": CitationMode.SOURCE_TAGGED_HYPERLINKS,
        "no_hyperlinks": CitationMode.NO_HYPERLINKS,
    }
    mode = mode_map.get(citation_format, CitationMode.NUMBER_HYPERLINKS)
    return CitationFormatter(mode=mode)


def export_report_to_memory(
    markdown_content: str, format: str, title: str | None = None
):
    """
    Export a markdown report to different formats in memory.

    Uses the modular exporter registry to support multiple formats.
    Available formats can be queried with ExporterRegistry.get_available_formats().

    Args:
        markdown_content: The markdown content to export
        format: Export format (e.g., 'pdf', 'odt', 'latex', 'quarto', 'ris')
        title: Optional title for the document

    Returns:
        Tuple of (content_bytes, filename, mimetype)
    """
    from ...exporters import ExporterRegistry, ExportOptions

    # Normalize format
    format_lower = format.lower()

    # Get exporter from registry
    exporter = ExporterRegistry.get_exporter(format_lower)

    if exporter is None:
        available = ExporterRegistry.get_available_formats()
        raise ValueError(
            f"Unsupported export format: {format}. "
            f"Available formats: {', '.join(available)}"
        )

    # Title prepending is now handled by each exporter via _prepend_title_if_needed()
    # PDF and ODT exporters prepend titles; RIS and other formats ignore them

    # Create options
    options = ExportOptions(title=title)

    # Export
    result = exporter.export(markdown_content, options)

    logger.info(
        f"Generated {format_lower} in memory, size: {len(result.content)} bytes"
    )

    return result.content, result.filename, result.mimetype


def save_research_strategy(research_id, strategy_name, *, username):
    """
    Save the strategy used for a research to the database.

    Args:
        research_id: The ID of the research
        strategy_name: The name of the strategy used
        username: The username whose encrypted DB to write. Required —
            without it get_user_db_session would silently fall back to
            the Flask session user (or fail off-request-context), so
            callers must state whose DB they mean.
    """
    try:
        logger.debug(
            f"save_research_strategy called with research_id={research_id}, strategy_name={strategy_name}"
        )
        with get_user_db_session(username) as session:
            # Check if a strategy already exists for this research
            existing_strategy = (
                session.query(ResearchStrategy)
                .filter_by(research_id=research_id)
                .first()
            )

            if existing_strategy:
                # Update existing strategy
                existing_strategy.strategy_name = strategy_name
                logger.debug(
                    f"Updating existing strategy for research {research_id}"
                )
            else:
                # Create new strategy record
                new_strategy = ResearchStrategy(
                    research_id=research_id, strategy_name=strategy_name
                )
                session.add(new_strategy)
                logger.debug(
                    f"Creating new strategy record for research {research_id}"
                )

            session.commit()
            logger.info(
                f"Saved strategy '{strategy_name}' for research {research_id}"
            )
    except Exception:
        logger.exception("Error saving research strategy")


def get_research_strategy(research_id, *, username):
    """
    Get the strategy used for a research.

    Args:
        research_id: The ID of the research
        username: The username whose encrypted DB to read. Required —
            without it get_user_db_session would silently fall back to
            the Flask session user (or fail off-request-context, which
            the except below would swallow into a None return), so
            callers must state whose DB they mean.

    Returns:
        str: The strategy name or None if not found
    """
    try:
        with get_user_db_session(username) as session:
            strategy = (
                session.query(ResearchStrategy)
                .filter_by(research_id=research_id)
                .first()
            )

            return strategy.strategy_name if strategy else None
    except Exception:
        logger.exception("Error getting research strategy")
        return None


def start_research_process(
    research_id,
    query,
    mode,
    run_research_callback,
    **kwargs,
):
    """
    Start a research process in a background thread.

    Args:
        research_id: The ID of the research
        query: The research query
        mode: The research mode (quick/detailed)
        run_research_callback: The callback function to run the research
        **kwargs: Additional parameters to pass to the research process (model, search_engine, etc.)

    Returns:
        threading.Thread: The thread running the research
    """
    from ..routes.globals import check_and_start_research
    from ...exceptions import SystemAtCapacityError

    # Acquire the global concurrency semaphore SYNCHRONOUSLY in the caller's
    # thread. Previously this happened inside the worker after the HTTP route
    # had already returned 200 — at capacity, the worker parked and the user
    # saw an infinite thinking spinner with the partial unique in-progress
    # index blocking retries. Surfacing capacity as an exception lets the
    # route return HTTP 429 before committing any DB state.
    if not _global_research_semaphore.acquire(blocking=False):
        raise SystemAtCapacityError(
            f"At research capacity (max {_MAX_GLOBAL_CONCURRENT} concurrent)"
        )

    # Pass the app context to the thread.
    run_research_callback = thread_with_app_context(run_research_callback)

    # Wrap callback so the worker releases the already-held semaphore on exit.
    original_callback = run_research_callback

    def _release_semaphore_on_exit(*args, **kw):
        try:
            return original_callback(*args, **kw)
        finally:
            _global_research_semaphore.release()

    # Prepare (but do not start) the background thread.
    thread = threading.Thread(
        target=_release_semaphore_on_exit,
        args=(
            thread_context(),
            research_id,
            query,
            mode,
        ),
        kwargs=kwargs,
    )
    thread.daemon = True

    # Atomic check-and-start: refuses to spawn a second live thread
    # for the same research_id. Guards against the double-spawn window
    # where a post-spawn commit failure in the queue processor could
    # otherwise cause the retry loop to dispatch the same research twice.
    try:
        started = check_and_start_research(
            research_id,
            {
                "thread": thread,
                "progress": 0,
                "status": ResearchStatus.IN_PROGRESS,
                "log": [],
                "settings": kwargs,
            },
        )
    except Exception:
        # check_and_start_research raised before the thread ran — the
        # semaphore won't be released by the worker, so release it here.
        _global_research_semaphore.release()
        raise
    if not started:
        # No thread will run → no _release_semaphore_on_exit → release here.
        _global_research_semaphore.release()
        raise DuplicateResearchError(
            f"Research {research_id} already has a live thread"
        )

    return thread


def _generate_report_path(query: str) -> Path:
    """
    Generates a path for a new report file based on the query.

    Args:
        query: The query used for the report.

    Returns:
        The path that it generated.

    """
    # Generate a unique filename that does not contain
    # non-alphanumeric characters.
    query_hash = hashlib.md5(  # DevSkim: ignore DS126858
        query.encode("utf-8"), usedforsecurity=False
    ).hexdigest()[:10]
    return OUTPUT_DIR / (
        f"research_report_{query_hash}_{int(datetime.now(UTC).timestamp())}.md"
    )


def _save_chat_message_and_context(
    chat_session_id,
    research_id,
    username,
    report_content,
    streaming_enabled,
    streaming_state,
    socket_service,
    settings_snapshot=None,
):
    """Save assistant message to chat and update accumulated context."""
    from ...chat.service import ChatService
    from ...chat.context import ChatContextManager

    chat_service = ChatService(username)
    # chat_messages.content is NOT NULL. Write report_content
    # inline (snapshot pattern). Falls back to a placeholder marker only
    # if report_content is itself empty — the same pattern used by the
    # terminate path's _STOPPED_BEFORE_OUTPUT_MARKER.
    snapshot_content = report_content or _NO_OUTPUT_MARKER
    # allow_archived=True: a multi-tab race can flip the session to
    # archived between research.status=COMPLETED commit and this write.
    # Losing the final assistant answer is worse than violating the
    # "no writes to archived sessions" rule for a system-generated row;
    # user-message writes from chat/routes.py still keep the default.
    chat_service.add_message(
        session_id=chat_session_id,
        role="assistant",
        content=snapshot_content,
        message_type="response",
        research_id=research_id,
        allow_archived=True,
    )
    # Mark the response row as persisted so the trailing
    # progress_callback("Research completed successfully", 100) — which
    # runs the termination check and could fire if the user clicks Stop
    # in the small window between this write and the final emit — does
    # NOT write a duplicate row via _save_partial_chat_message_on_terminate.
    if streaming_state is not None:
        streaming_state["_persisted"] = True
    logger.info(f"Added research result to chat {chat_session_id[:8]}...")

    try:
        # report_content is the answer-only string; no extraction needed.
        chat_content = report_content or ""
        ctx_manager = ChatContextManager(
            session_id=chat_session_id,
            messages=[],
            settings_snapshot=settings_snapshot,
        )
        updates = ctx_manager.extract_context_updates(new_content=chat_content)
        chat_service.update_accumulated_context(
            session_id=chat_session_id, **updates
        )
        logger.info(
            f"Updated accumulated context for chat {chat_session_id[:8]}..."
        )
    except Exception:
        # Bumped from debug to warning: a failed accumulated-context
        # update silently degrades multi-turn context for the next
        # follow-up in this chat (entities/topics/source counts are
        # missing). Ops need visibility to catch widespread breakage
        # before users notice "the AI forgot what we were talking
        # about" — debug-level was invisible in production.
        logger.opt(exception=True).warning(
            "Could not update accumulated context"
        )

    if streaming_enabled and streaming_state.get("chunks_sent", 0) > 0:
        # Flush any partial-bracket fragment held in the citation carry
        # buffer before sending the final empty sentinel. Without this,
        # a stream that ends mid-token (LLM emits "[12" as its last
        # bytes before closing) silently drops the leading "[" from
        # what the client renders — the carry would be discarded when
        # the callback's closure goes out of scope.
        flush = streaming_state.get("_flush_carry")
        if flush:
            leftover = flush()
            if leftover:
                try:
                    socket_service.emit_to_subscribers(
                        "response_chunk",
                        research_id,
                        {
                            "chunk": leftover,
                            "is_streaming": True,
                            "is_final": False,
                        },
                    )
                except Exception:
                    logger.debug(
                        "Carry-buffer flush emit failed (non-critical)"
                    )
        socket_service.emit_to_subscribers(
            "response_chunk",
            research_id,
            {"chunk": "", "is_streaming": True, "is_final": True},
        )


_STOPPED_BEFORE_OUTPUT_MARKER = "[Stopped before any output was generated.]"
_STOPPED_FOOTER = "\n\n_— Stopped by user._"
_NO_OUTPUT_MARKER = "_(Research completed without producing output.)_"


# Match a trailing incomplete citation opener at the chunk boundary.
# Covers both ASCII "[" and the lenticular "【" (U+3010) — some LLMs
# emit Chinese-style brackets and the citation formatter accepts them,
# so we have to hold those back the same way to avoid breaking a token
# across the next chunk.
_PARTIAL_BRACKET_RE = re.compile(r"[\[【]\d*$")

# Upper bound on the inline-citation carry buffer. A real citation token
# is a handful of bytes ("[123"); anything longer means the "[" wasn't a
# citation opener or the stream is pathological. Flush raw past this so a
# never-closing "[" + endless digits can't grow the buffer without limit.
_MAX_CARRY_BYTES = 64


def _make_chat_stream_callback(
    research_id,
    streaming_state,
    socket_service,
    source_resolver=None,
    formatter=None,
):
    """Build the chat-mode streaming callback.

    The callback:
      * Counts chunks (``chunks_sent``).
      * Buffers RAW chunks in ``streaming_state['chunks']`` so partial
        content survives termination — the citation handler's local list
        is discarded on raise. Capped at ``_MAX_PARTIAL_BUFFER_BYTES``;
        once capped, ``streaming_state['_truncated']`` flips to True and
        further chunks aren't accumulated server-side.
      * Raises ``ResearchTerminatedException`` if the user clicked Stop
        mid-stream — fails fast instead of letting the LLM finish.
        ``ResearchTerminatedException`` inherits from ``BaseException``,
        so the citation handler's ``except Exception`` blocks naturally
        propagate it.
      * Emits ``response_chunk`` over Socket.IO for live display, with
        inline citation hyperlinks applied per-chunk when both
        ``source_resolver`` and ``formatter`` are provided — so the
        client sees ``[[arxiv.org-1]](url)`` appearing live rather than
        plain ``[1]`` brackets that only get hyperlinked after the full
        response saves. Bracket tokens split across chunk boundaries are
        held in a small carry buffer until the closing ``]`` arrives.

    ``source_resolver`` — optional ``() -> list[dict]`` returning the
        current ``all_links_of_system`` (so we read it lazily; the list
        grows as the agent collects sources).
    ``formatter`` — optional :class:`CitationFormatter` instance. Its
        ``mode`` controls the inline format (``[[arxiv.org-1]](url)``
        etc.), matching what the final-save formatter produces so the
        live display doesn't mode-shift when ``handleResearchComplete``
        swaps in the DB-saved version.

    Extracted to module level so it can be unit-tested without spinning
    up the full ``run_research_process``.
    """

    # Per-call closure state for the streaming-substitution carry
    # buffer (holds a trailing incomplete bracket like "[12" so the
    # closing "]3]" on the next chunk completes the citation token).
    carry = [""]

    def _flush_carry() -> str:
        """Release and clear any held partial-bracket fragment.

        The completion finalizer (``_save_chat_message_and_context``)
        lives in a different scope and can't reach ``carry`` directly,
        so it calls this through ``streaming_state['_flush_carry']`` to
        avoid silently dropping the tail of a stream that ends mid-token
        like ``"[12"``.
        """
        released, carry[0] = carry[0], ""
        return released

    # Expose to the completion path via the shared state dict.
    streaming_state["_flush_carry"] = _flush_carry

    def _hyperlink_chunk(chunk: str) -> str:
        """Apply inline citation hyperlinks to a single chunk.

        Maintains the closure-level ``carry`` buffer for incomplete
        ``[N`` tokens straddling chunk boundaries. The carry contains
        the trailing partial-bracket fragment from the previous chunk
        that we haven't been able to substitute yet — it's prepended
        to the next chunk so the regex can see the full token. The
        DELTA returned is what the client should append next: it
        includes the just-completed carry (now hyperlinked) plus the
        new chunk's safe portion, MINUS the new chunk's own trailing
        partial bracket (which becomes the new carry).
        """
        if source_resolver is None or formatter is None:
            return chunk
        try:
            sources = source_resolver() or []
            if not sources:
                # Reset carry so the leading "[" we held onto doesn't
                # disappear from the client's accumulated text.
                released, carry[0] = carry[0], ""
                return released + chunk
            text = carry[0] + chunk
            pending = _PARTIAL_BRACKET_RE.search(text)
            if pending:
                safe = text[: pending.start()]
                new_carry = text[pending.start() :]
                # Bound the carry. A well-formed citation token is a few
                # bytes (`[12`); if the held fragment grows past this, the
                # "[" was not actually opening a citation (or a hostile /
                # misbehaving LLM is streaming `[` + endless digits with no
                # closing `]`). Flush it raw rather than buffering without
                # limit — preserves the text, just doesn't hyperlink it.
                if len(new_carry) > _MAX_CARRY_BYTES:
                    safe = text
                    carry[0] = ""
                else:
                    carry[0] = new_carry
            else:
                safe = text
                carry[0] = ""
            if not safe:
                return ""
            return formatter.apply_inline_hyperlinks(safe, sources)
        except Exception:
            # Hyperlinking is quality-of-life. On any failure fall back
            # to emitting the raw chunk so the user still sees the text.
            logger.debug(
                "Inline citation hyperlinking failed; emitting raw chunk",
                exc_info=True,
            )
            released, carry[0] = carry[0], ""
            return released + chunk

    def stream_callback(chunk: str):
        # Resolve through the module namespace each call so tests can
        # ``patch("local_deep_research.web.routes.globals.is_termination_requested")``.
        # Cached in sys.modules — negligible cost.
        from ..routes.globals import is_termination_requested

        if not chunk:
            return
        streaming_state["chunks_sent"] += 1
        if not streaming_state["_truncated"]:
            chunk_bytes = len(chunk.encode("utf-8"))
            if (
                streaming_state["_bytes"] + chunk_bytes
                <= _MAX_PARTIAL_BUFFER_BYTES
            ):
                # IMPORTANT: store the RAW chunk in the partial-content
                # buffer (terminate handler joins these and saves them as
                # the partial assistant message). If we stored the
                # hyperlinked version, the saved-on-terminate text would
                # be double-formatted when the user resumes.
                streaming_state["chunks"].append(chunk)
                streaming_state["_bytes"] += chunk_bytes
            else:
                streaming_state["_truncated"] = True
                logger.warning(
                    f"Partial-content buffer hit {_MAX_PARTIAL_BUFFER_BYTES} bytes "
                    f"for research {research_id}; further chunks won't be persisted on terminate"
                )
        # Mid-stream interrupt — fail fast if the user clicked Stop while
        # the LLM is still streaming.
        if is_termination_requested(research_id):
            raise ResearchTerminatedException(  # noqa: TRY301 — propagated through citation handler
                "Research was terminated by user during streaming"
            )
        # Apply citation hyperlinks for the client emit only — the
        # client accumulates substituted text into the streaming bubble
        # so the user sees [[arxiv.org-1]](url) appearing live as the
        # model writes "According to [1]…".
        display_chunk = _hyperlink_chunk(chunk)
        if not display_chunk:
            return  # nothing safe to emit yet (all held in carry buffer)
        try:
            socket_service.emit_to_subscribers(
                "response_chunk",
                research_id,
                {
                    "chunk": display_chunk,
                    "is_streaming": True,
                    "is_final": False,
                },
            )
        except Exception:
            logger.debug("Stream chunk emit failed (non-critical)")

    return stream_callback


def _save_partial_chat_message_on_terminate(
    chat_session_id,
    research_id,
    username,
    partial_content,
    truncated=False,
    streaming_state=None,
):
    """Persist a chat 'response' row capturing whatever was streamed before
    termination, and emit a final ``response_chunk`` so the client strips
    the streaming class from the bubble.

    Must run BEFORE ``handle_termination()`` because that path runs
    ``cleanup_research_resources()`` which removes the Socket.IO room
    subscriptions — anything emitted afterwards goes nowhere.

    Idempotent: when ``streaming_state`` is supplied, sets a ``_persisted``
    flag so duplicate calls (one from the progress callback, one from the
    outer except handler when a stream_callback raises mid-stream) only
    write a single row.

    Skips silently when ``chat_session_id`` is falsy (single-turn case).
    All failures are swallowed — termination cleanup must never crash the
    worker.
    """
    if not chat_session_id:
        return
    if streaming_state is not None and streaming_state.get("_persisted"):
        return
    try:
        from ...chat.service import ChatService

        if partial_content:
            content = partial_content + _STOPPED_FOOTER
            if truncated:
                content += " _(output was very long; truncated.)_"
        else:
            content = _STOPPED_BEFORE_OUTPUT_MARKER

        # allow_archived=True: same rationale as the completion path —
        # the partial response on Stop is system-generated and must
        # survive a concurrent archive (see _save_chat_message_and_context).
        ChatService(username).add_message(
            session_id=chat_session_id,
            role="assistant",
            content=content,
            message_type="response",
            research_id=research_id,
            allow_archived=True,
        )
        # Set the idempotency flag ONLY after the write succeeds. If
        # `add_message` raises (DB lock, encryption error, archived
        # session), the outer ResearchTerminatedException handler retries
        # this helper — flipping the flag pre-write would short-circuit
        # the retry and silently lose the partial response.
        if streaming_state is not None:
            streaming_state["_persisted"] = True
        logger.info(
            f"Persisted partial chat response for terminated research "
            f"{research_id} ({len(content)} chars)"
        )
    except Exception:
        logger.opt(exception=True).warning(
            "Failed to persist partial chat message on terminate"
        )

    try:
        SocketIOService().emit_to_subscribers(
            "response_chunk",
            research_id,
            {"chunk": "", "is_streaming": True, "is_final": True},
        )
    except Exception:
        logger.debug("Final-chunk emit on terminate failed (non-critical)")


@log_for_research
@thread_cleanup
def run_research_process(research_id, query, mode, **kwargs):
    """
    Run the research process in the background for a given research ID.

    Args:
        research_id: The ID of the research
        query: The research query
        mode: The research mode (quick/detailed)
        **kwargs: Additional parameters for the research (model_provider, model, search_engine, etc.)
                 MUST include 'username' for database access
    """
    from ..routes.globals import (
        is_research_active,
        is_termination_requested,
        update_progress_and_check_active,
    )

    # Extract username - required for database access
    username = kwargs.get("username")
    if not username:
        logger.error("No username provided to research thread")
        raise ValueError("Username is required for research process")
    # Extract user_password early so it's available for all cleanup paths
    user_password = kwargs.get("user_password")

    # Establish thread context FIRST so every subsequent log line in this
    # thread can be attributed to the correct user/research and persisted
    # to the user's encrypted ResearchLog. Otherwise the early INFO logs
    # below ("Research thread started", "Research strategy", "Research
    # parameters") fire before start_research_process gets to its own
    # set_search_context call (~line 417) and the daemon can't open the
    # encrypted DB to write them — silently dropped via the bare-except.
    set_search_context(
        {
            "research_id": research_id,
            "username": username,
            "user_password": user_password,
            # Settings snapshot may be absent here (early init); the
            # shared context built later overwrites this with the
            # populated snapshot. Defaults to {} so cache reads in any
            # intermediate code path see a non-None dict.
            "settings_snapshot": kwargs.get("settings_snapshot") or {},
        }
    )

    logger.info(f"Research thread started with username: {username}")

    try:
        # Check if this research has been terminated before we even start
        if is_termination_requested(research_id):
            logger.info(
                f"Research {research_id} was terminated before starting"
            )
            cleanup_research_resources(
                research_id,
                username,
                user_password=user_password,
                final_status=ResearchStatus.SUSPENDED,
            )
            return

        logger.info(
            f"Starting research process for ID {research_id}, query: {query}"
        )

        # Extract key parameters
        model_provider = kwargs.get("model_provider")
        model = kwargs.get("model")
        custom_endpoint = kwargs.get("custom_endpoint")
        search_engine = kwargs.get("search_engine")
        max_results = kwargs.get("max_results")
        time_period = kwargs.get("time_period")
        iterations = kwargs.get("iterations")
        questions_per_iteration = kwargs.get("questions_per_iteration")
        strategy = kwargs.get(
            "strategy", "source-based"
        )  # Default to source-based
        settings_snapshot = kwargs.get(
            "settings_snapshot", {}
        )  # Complete settings snapshot
        # NOTE: the "_username" injection that lets the agent register the
        # user's document collections is done downstream in
        # AdvancedSearchSystem.__init__ (the narrowest common consumer of the
        # strategy-running paths — web run and the programmatic API), not here.
        # Run-start egress (below) doesn't need it: it passes username
        # explicitly. See _ensure_snapshot_username in search_system.py.

        # Log settings snapshot to debug
        from ...settings.logger import log_settings

        log_settings(settings_snapshot, "Settings snapshot received in thread")

        # Strategy should already be saved in the database before thread starts
        logger.info(f"Research strategy: {strategy}")

        # Log all parameters for debugging
        logger.info(
            f"Research parameters: provider={model_provider}, model={model}, "
            f"search_engine={search_engine}, max_results={max_results}, "
            f"time_period={time_period}, iterations={iterations}, "
            f"questions_per_iteration={questions_per_iteration}, "
            f"custom_endpoint={custom_endpoint}, strategy={strategy}"
        )

        # Set up the AI Context Manager
        output_dir = OUTPUT_DIR / f"research_{research_id}"
        output_dir.mkdir(parents=True, exist_ok=True)

        # Create a settings context that uses snapshot - no database access in threads
        settings_context = SnapshotSettingsContext(
            settings_snapshot, username=username
        )

        # Only log settings if explicitly enabled via LDR_LOG_SETTINGS env var
        from ...settings.logger import log_settings

        log_settings(
            settings_context.values, "SettingsContext values extracted"
        )

        # Set the settings context for this thread
        from ...config.thread_settings import (
            set_settings_context,
        )

        set_settings_context(settings_context)

        # Defense-in-depth: register an EgressContext on the audit-hook
        # thread-local so socket.connect calls that bypass our explicit
        # PEPs (third-party libraries, future contributors reaching for
        # raw requests.get, prompt-injection-steered tools) are gated
        # against the same scope. Cleared by the @thread_cleanup exit
        # handler in database/thread_local_session.py when the worker
        # winds down. If snapshot construction errors here, we log and
        # continue — the explicit PEPs still gate the known callers, so
        # the run is not less secure than before this hook existed.
        try:
            from ...security import set_active_context as _set_egress_ctx
            from ...security.egress.policy import (
                PolicyDeniedError,
                context_from_snapshot,
                resolve_run_primary_engine,
            )

            try:
                _egress_primary = resolve_run_primary_engine(settings_snapshot)
                _egress_ctx = context_from_snapshot(
                    settings_snapshot,
                    _egress_primary,
                    username=username,
                )
                _set_egress_ctx(_egress_ctx)
            except (PolicyDeniedError, ValueError) as _ctx_err:
                # Corrupted scope or invalid policy config. The run-start
                # precheck already rejects this at the API boundary;
                # if we get here it is because the run was launched
                # without going through the precheck (CLI, scheduler).
                # Re-raise so the worker fails fast with a clear reason
                # rather than running unprotected.
                logger.bind(policy_audit=True).warning(
                    "research worker refused: egress policy unevaluable",
                    research_id=research_id,
                    reason=str(_ctx_err),
                )
                raise
        except ImportError:
            # security module unavailable — preserve legacy behaviour.
            logger.debug("egress audit hook unavailable for research worker")

        # user_password already extracted above (before termination check)

        # Create shared research context that can be updated during research
        shared_research_context = {
            "research_id": research_id,
            "research_query": query,
            "research_mode": mode,
            "research_phase": "init",
            "search_iteration": 0,
            "search_engines_planned": None,
            "search_engine_selected": search_engine,
            "username": username,  # Add username for queue operations
            "user_password": user_password,  # Add password for metrics access
            # Thread-safe settings snapshot propagated to background search
            # threads (engine config, per-user resolution, egress scope).
            "settings_snapshot": settings_snapshot,
            "chat_session_id": kwargs.get("chat_session_id"),
        }

        # If this is a follow-up research, include the parent context
        if "research_context" in kwargs and kwargs["research_context"]:
            logger.info(
                f"Adding parent research context with {len(kwargs['research_context'].get('past_findings', ''))} chars of findings"
            )
            shared_research_context.update(kwargs["research_context"])

        # Do not log context keys as they may contain sensitive information
        logger.info(f"Created shared_research_context for user: {username}")

        # Set search context for search tracking
        set_search_context(shared_research_context)

        # Per-research dedup state for step message persistence
        last_step_phase = None

        # Pre-bind streaming_state so progress_callback's closure cell has a
        # value even if an exception fires between this point and the full
        # initialization later in this function. Without this, the except handler's
        # call to progress_callback during a concurrent termination raises
        # UnboundLocalError, silently skipping the DB FAILED update and the
        # error socket emit. The full streaming_state is reassigned below
        # before any real streaming starts; keys must match the canonical shape.
        streaming_state: dict = {
            "chunks_sent": 0,
            "chunks": [],
            "_bytes": 0,
            "_truncated": False,
        }
        streaming_enabled = False

        # Set up progress callback
        def progress_callback(message, progress_percent, metadata):
            nonlocal last_step_phase
            # Frequent termination check
            if is_termination_requested(research_id):
                # Persist the partial chat row + emit final chunk BEFORE
                # handle_termination — afterwards the Socket.IO room is gone.
                _save_partial_chat_message_on_terminate(
                    shared_research_context.get("chat_session_id"),
                    research_id,
                    username,
                    "".join(streaming_state.get("chunks", [])),
                    truncated=streaming_state.get("_truncated", False),
                    streaming_state=streaming_state,
                )
                handle_termination(research_id, username)
                streaming_state["_termination_handled"] = True
                raise ResearchTerminatedException(  # noqa: TRY301 — inside nested callback, not caught by enclosing try
                    "Research was terminated by user"
                )

            # Silent phase — no UI logging or socket emission needed
            if metadata.get("phase") == "termination_check":
                return

            # Bind research_id AND username so the database_sink + queue
            # daemon can resolve the per-user encrypted DB. Without username
            # the daemon's _write_log_to_database hits "No authenticated
            # user", silently swallows the error, and ResearchLog ends up
            # with zero milestone rows — leaving /api/research/<id>/status
            # without a log_entry to render and the frontend stuck on the
            # "Performing research..." fallback.
            bound_logger = logger.bind(
                research_id=research_id, username=username
            )
            bound_logger.log("MILESTONE", message)

            if "SEARCH_PLAN:" in message:
                engines = message.split("SEARCH_PLAN:")[1].strip()
                metadata["planned_engines"] = engines
                metadata["phase"] = "search_planning"  # Use existing phase
                # Update shared context for token tracking
                shared_research_context["search_engines_planned"] = engines
                shared_research_context["research_phase"] = "search_planning"

            if "ENGINE_SELECTED:" in message:
                engine = message.split("ENGINE_SELECTED:")[1].strip()
                metadata["selected_engine"] = engine
                metadata["phase"] = "search"  # Use existing 'search' phase
                # Update shared context for token tracking
                shared_research_context["search_engine_selected"] = engine
                shared_research_context["research_phase"] = "search"

            # Capture other research phases for better context tracking
            if metadata.get("phase"):
                shared_research_context["research_phase"] = metadata["phase"]

            # Update search iteration if available
            if "iteration" in metadata:
                shared_research_context["search_iteration"] = metadata[
                    "iteration"
                ]

            # Adjust progress based on research mode
            adjusted_progress = progress_percent
            phase = metadata.get("phase", "")

            if mode == "detailed":
                # Report phases pass through (already mapped by wrapper).
                # All other phases — including "complete" emitted by each
                # strategy when its analyze_topic finishes (report
                # generation runs analyze_topic per subsection, so a
                # strategy "complete" fires mid-report and must NOT be
                # treated as the end of the whole run) — are capped.
                # update_progress_and_check_active enforces global
                # monotonicity, so backwards jumps from non-monotonic
                # strategy updates are absorbed there.
                # None values (error path, sub-search relays) pass through
                # unchanged.
                if phase not in _REPORT_PHASES and progress_percent is not None:
                    adjusted_progress = min(
                        _DETAILED_SEARCH_PROGRESS_CAP, progress_percent
                    )
            elif (
                mode == "quick"
                and phase == "output_generation"
                and progress_percent is not None
            ):
                # For quick mode, scale output_generation to 85-95% range
                if progress_percent > 0:
                    adjusted_progress = 85 + (progress_percent / 100) * 10
                else:
                    adjusted_progress = 85

            # Atomically update progress and check if research is still active
            if adjusted_progress is not None:
                adjusted_progress, still_active = (
                    update_progress_and_check_active(
                        research_id, adjusted_progress
                    )
                )
            else:
                still_active = is_research_active(research_id)

            if still_active:
                # Queue the progress update to be processed in main thread
                if adjusted_progress is not None:
                    from ..queue.processor_v2 import queue_processor

                    if username:
                        queue_processor.queue_progress_update(
                            username, research_id, adjusted_progress
                        )
                    else:
                        logger.warning(
                            f"Cannot queue progress update for research {research_id} - no username available"
                        )

                # Determine socket emit throttling
                phase = metadata.get("phase", "")
                is_final = (
                    phase
                    in (
                        "complete",
                        "error",
                        "report_complete",
                    )
                    or adjusted_progress == 100
                )

                should_emit = is_final
                if not is_final:
                    now = time.monotonic()
                    with _last_emit_lock:
                        last = _last_emit_times.get(research_id, 0)
                        if now - last >= _EMIT_THROTTLE_SECONDS:
                            _last_emit_times[research_id] = now
                            should_emit = True
                        # Periodic TTL cleanup for orphaned entries
                        global _emit_cleanup_counter  # noqa: PLW0603
                        _emit_cleanup_counter += 1
                        if _emit_cleanup_counter % 100 == 0:
                            stale = [
                                rid
                                for rid, t in _last_emit_times.items()
                                if now - t > _EMIT_TTL_SECONDS
                            ]
                            for rid in stale:
                                del _last_emit_times[rid]

                # Build event data (before emit, shared scope)
                event_data = None
                if should_emit:
                    event_data = {
                        "progress": adjusted_progress,
                        "message": message,
                        "phase": phase,
                    }
                    # Include additional metadata for MCP/ReAct strategy display
                    if metadata.get("thought"):
                        event_data["thought"] = metadata["thought"]
                    if metadata.get("tool"):
                        event_data["tool"] = metadata["tool"]
                    if metadata.get("arguments"):
                        event_data["arguments"] = metadata["arguments"]
                    if metadata.get("iteration"):
                        event_data["iteration"] = metadata["iteration"]
                    if metadata.get("error"):
                        event_data["error"] = metadata["error"]
                    if metadata.get("content"):
                        event_data["content"] = metadata["content"]

                # Persist step message to chat session BEFORE emitting socket
                # (ensures DB has the step when clients receive the event).
                #
                # Symmetry invariant: for chat sessions, what the user sees
                # live must equal what `loadSession` reconstructs on reload.
                # If dedup blocks persistence of a repeat step phase, drop
                # the socket emit too (unless this is a final-phase event
                # the completion handler depends on) so live UI doesn't
                # surface events that vanish on reload.
                _chat_session_id = shared_research_context.get(
                    "chat_session_id"
                )
                if _chat_session_id:
                    _persist, _suppress_emit = _chat_step_decision(
                        phase, last_step_phase, is_final
                    )
                    if _persist:
                        try:
                            from ...chat.service import ChatService

                            ChatService(username).add_progress_step(
                                session_id=_chat_session_id,
                                research_id=research_id,
                                content=message,
                                phase=phase,
                            )
                            last_step_phase = phase
                        except Exception:
                            logger.opt(exception=True).warning(
                                "Failed to persist progress step"
                            )
                            # Symmetry invariant: if persistence failed
                            # (e.g. OperationalError under DB contention),
                            # suppress the live emit too — otherwise the
                            # client sees a step that vanishes on reload.
                            event_data = None
                    if _suppress_emit:
                        event_data = None

                # Emit socket event AFTER DB persistence
                if event_data is not None:
                    try:
                        SocketIOService().emit_to_subscribers(
                            "progress", research_id, event_data
                        )
                    except Exception:
                        logger.exception("Socket emit error (non-critical)")

        # Function to check termination during long-running operations
        def check_termination():
            if is_termination_requested(research_id):
                _save_partial_chat_message_on_terminate(
                    shared_research_context.get("chat_session_id"),
                    research_id,
                    username,
                    "".join(streaming_state.get("chunks", [])),
                    truncated=streaming_state.get("_truncated", False),
                    streaming_state=streaming_state,
                )
                handle_termination(research_id, username)
                streaming_state["_termination_handled"] = True
                raise ResearchTerminatedException(  # noqa: TRY301 — inside nested callback, not caught by enclosing try
                    "Research was terminated by user during long-running operation"
                )
            return False  # Not terminated

        # Configure the system with the specified parameters
        use_llm = None
        if model or search_engine or model_provider:
            # Log that we're overriding system settings
            logger.info(
                f"Overriding system settings with: provider={model_provider}, model={model}, search_engine={search_engine}"
            )

        # Override LLM if model or model_provider specified
        if model or model_provider:
            try:
                # Get LLM with the overridden settings.
                # Pass settings_snapshot explicitly — without it the LLM
                # PEP (llm_config.py:295) skips evaluate_llm_endpoint
                # because the gate is "if settings_snapshot is not None
                # and provider:". The snapshot is already stored on
                # shared_research_context as "settings_snapshot".
                use_llm = get_llm(
                    model_name=model,
                    provider=model_provider,
                    openai_endpoint_url=custom_endpoint,
                    research_id=research_id,
                    research_context=shared_research_context,
                    settings_snapshot=shared_research_context.get(
                        "settings_snapshot"
                    ),
                )

                logger.info(
                    f"Successfully set LLM to: provider={model_provider}, model={model}"
                )
            except Exception as e:
                logger.exception(
                    f"Error setting LLM provider={model_provider}, model={model}"
                )
                error_msg = str(e)
                # Surface configuration errors to user instead of silently continuing
                config_error_keywords = [
                    "model path",
                    "llamacpp",
                    "cannot connect",
                    "server",
                    "not configured",
                    "not responding",
                    "directory",
                    ".gguf",
                ]
                if any(
                    keyword in error_msg.lower()
                    for keyword in config_error_keywords
                ):
                    # This is a configuration error the user can fix
                    raise ValueError(
                        f"LLM Configuration Error: {error_msg}"
                    ) from e
                # For other errors, re-raise to avoid silent failures
                raise

        # Create search engine first if specified, to avoid default creation without username
        use_search = None
        if search_engine:
            try:
                # Create a new search object with these settings
                use_search = get_search(
                    search_tool=search_engine,
                    llm_instance=use_llm,
                    username=username,
                    settings_snapshot=settings_snapshot,
                )
                logger.info(
                    f"Successfully created search engine: {search_engine}"
                )
            except Exception as e:
                logger.exception(
                    f"Error creating search engine {search_engine}"
                )
                error_msg = str(e)
                # Surface configuration errors to user instead of silently continuing
                config_error_keywords = [
                    "searxng",
                    "instance_url",
                    "api_key",
                    "cannot connect",
                    "connection",
                    "timeout",
                    "not configured",
                ]
                if any(
                    keyword in error_msg.lower()
                    for keyword in config_error_keywords
                ):
                    # This is a configuration error the user can fix
                    raise ValueError(
                        f"Search Engine Configuration Error ({search_engine}): {error_msg}"
                    ) from e
                # For other errors, re-raise to avoid silent failures
                raise

        # Set the progress callback in the system
        system = AdvancedSearchSystem(
            llm=use_llm,  # type: ignore[arg-type]
            search=use_search,  # type: ignore[arg-type]
            strategy_name=strategy,
            max_iterations=iterations,
            questions_per_iteration=questions_per_iteration,
            username=username,
            settings_snapshot=settings_snapshot,
            research_id=research_id,
            research_context=shared_research_context,
        )
        system.set_progress_callback(progress_callback)

        # Chat mode: set up LLM streaming callback for real-time response chunks
        streaming_enabled = False
        # chunks: server-side buffer so partial content survives termination
        # (the citation handler's local list is discarded on raise).
        # _bytes / _truncated cap the buffer to bound memory on pathologically
        # long answers; once capped we still forward to the frontend but stop
        # accumulating server-side.
        streaming_state = {
            "chunks_sent": 0,
            "chunks": [],
            "_bytes": 0,
            "_truncated": False,
        }
        chat_session_id = shared_research_context.get("chat_session_id")

        if chat_session_id:
            try:
                socket_service = SocketIOService()

                # Source resolver returns the strategy's currently-collected
                # source list. Late-bound so the streaming callback can
                # apply inline hyperlinks as the agent finishes adding to
                # all_links_of_system (sources may still be growing when
                # synthesis starts; reading via the closure picks up the
                # final list at chunk-emit time, not callback-build time).
                def _resolve_sources():
                    if not hasattr(system, "all_links_of_system"):
                        return []
                    return list(system.all_links_of_system or [])

                # Build a formatter matching the user's report.citation_format
                # so live-display brackets ([[arxiv.org-1]] / [[arxiv-1]] /
                # [[1]] etc.) match what the final-save formatter will emit
                # — avoids a visible format-flip when handleResearchComplete
                # swaps in the DB-saved version.
                live_formatter = get_citation_formatter()

                stream_callback = _make_chat_stream_callback(
                    research_id,
                    streaming_state,
                    socket_service,
                    source_resolver=_resolve_sources,
                    formatter=live_formatter,
                )

                # Hook into the citation handler's streaming
                if hasattr(system, "strategy") and hasattr(
                    system.strategy, "citation_handler"
                ):
                    handler = system.strategy.citation_handler
                    if hasattr(handler, "set_stream_callback"):
                        handler.set_stream_callback(stream_callback)
                        streaming_enabled = True
                        logger.info(
                            f"Streaming enabled for chat {chat_session_id[:8]}..."
                        )
            except Exception:
                # exception=True so the traceback is visible: streaming is
                # non-critical (research still completes without it), but a
                # silent warning would hide real bugs in the setup above.
                logger.opt(exception=True).warning(
                    "Could not set up streaming (non-critical)"
                )

        # Helper to save chat message (closes over outer scope vars)
        def _maybe_save_chat_message(content):
            _chat_sid = shared_research_context.get("chat_session_id")
            if not _chat_sid:
                return
            try:
                _save_chat_message_and_context(
                    _chat_sid,
                    research_id,
                    username,
                    content,
                    streaming_enabled,
                    streaming_state,
                    SocketIOService(),
                    settings_snapshot=settings_snapshot,
                )
            except Exception:
                # Promoted from debug→warning: at debug level this is invisible
                # in production and the user sees "Research completed but no
                # report available" with no operator signal. Same rationale as
                # the accumulated_context handler in _save_chat_message_and_context.
                logger.opt(exception=True).warning(
                    "Could not add message to chat session — assistant "
                    "response NOT persisted; user will see 'no report available'"
                )
                # The DB write failed, so _save_chat_message_and_context never
                # emitted its is_final response_chunk. Emit one here so the
                # streaming UI clears its 'thinking' state instead of stalling.
                try:
                    SocketIOService().emit_to_subscribers(
                        "response_chunk",
                        research_id,
                        {"chunk": "", "is_streaming": True, "is_final": True},
                    )
                except Exception:
                    logger.opt(exception=True).debug(
                        "Failed to emit final chunk after chat-persist error"
                    )

        # Run the search
        progress_callback("Starting research process", 5, {"phase": "init"})

        try:
            results = system.analyze_topic(query)
            if mode == "quick":
                progress_callback(
                    "Search complete, preparing to generate summary...",
                    85,
                    {"phase": "output_generation"},
                )
            else:
                progress_callback(
                    "Search complete, generating output",
                    80,
                    {"phase": "output_generation"},
                )
        except Exception as search_error:
            # Better handling of specific search errors
            error_message = str(search_error)
            error_type = "unknown"

            # OpenAI-compatible runtime failures (LM Studio / vLLM / llama.cpp
            # server / OpenRouter / custom endpoint) -- rewrite to a message
            # that names the provider, base URL, and model (#3878).
            if model_provider in {
                "openai_endpoint",
                "lmstudio",
                "llamacpp",
                "openai",
                "openrouter",
                "google",
                "ionos",
                "xai",
            } and is_openai_compat_runtime_error(search_error):
                rewritten = friendly_openai_compatible_error(
                    search_error,
                    provider=model_provider,
                    base_url=custom_endpoint,
                    model=model,
                )
                raise RuntimeError(rewritten) from search_error

            # Extract error details for common issues
            if "status code: 503" in error_message:
                error_message = "Ollama AI service is unavailable (HTTP 503). Please check that Ollama is running properly on your system."
                error_type = "ollama_unavailable"
            elif "status code: 404" in error_message:
                error_message = "Ollama model not found (HTTP 404). Please check that you have pulled the required model."
                error_type = "model_not_found"
            elif "status code:" in error_message:
                # Extract the status code for other HTTP errors
                status_code = error_message.split("status code:")[1].strip()
                error_message = f"API request failed with status code {status_code}. Please check your configuration."
                error_type = "api_error"
            elif "connection" in error_message.lower():
                error_message = "Connection error. Please check that your LLM service (Ollama/API) is running and accessible."
                error_type = "connection_error"

            # Raise with improved error message
            raise RuntimeError(
                f"{error_message} (Error type: {error_type})"
            ) from search_error

        # Generate output based on mode
        if mode == "quick":
            # Quick Summary
            if results.get("findings") or results.get("formatted_findings"):
                raw_formatted_findings = results["formatted_findings"]

                # Check if formatted_findings contains an error message
                if isinstance(
                    raw_formatted_findings, str
                ) and raw_formatted_findings.startswith("Error:"):
                    logger.error(
                        f"Detected error in formatted findings: {raw_formatted_findings[:100]}..."
                    )

                    # Determine error type for better user feedback
                    error_type = "unknown"
                    error_message = raw_formatted_findings.lower()

                    if (
                        "token limit" in error_message
                        or "context length" in error_message
                    ):
                        error_type = "token_limit"
                        # Log specific error type
                        logger.warning(
                            "Detected token limit error in synthesis"
                        )

                        # Update progress with specific error type
                        progress_callback(
                            "Synthesis hit token limits. Attempting fallback...",
                            87,
                            {
                                "phase": "synthesis_error",
                                "error_type": error_type,
                            },
                        )
                    elif (
                        "timeout" in error_message
                        or "timed out" in error_message
                    ):
                        error_type = "timeout"
                        logger.warning("Detected timeout error in synthesis")
                        progress_callback(
                            "Synthesis timed out. Attempting fallback...",
                            87,
                            {
                                "phase": "synthesis_error",
                                "error_type": error_type,
                            },
                        )
                    elif "rate limit" in error_message:
                        error_type = "rate_limit"
                        logger.warning("Detected rate limit error in synthesis")
                        progress_callback(
                            "LLM rate limit reached. Attempting fallback...",
                            87,
                            {
                                "phase": "synthesis_error",
                                "error_type": error_type,
                            },
                        )
                    elif (
                        "connection" in error_message
                        or "network" in error_message
                    ):
                        error_type = "connection"
                        logger.warning("Detected connection error in synthesis")
                        progress_callback(
                            "Connection issue with LLM. Attempting fallback...",
                            87,
                            {
                                "phase": "synthesis_error",
                                "error_type": error_type,
                            },
                        )
                    elif (
                        "llm error" in error_message
                        or "final answer synthesis fail" in error_message
                    ):
                        error_type = "llm_error"
                        logger.warning(
                            "Detected general LLM error in synthesis"
                        )
                        progress_callback(
                            "LLM error during synthesis. Attempting fallback...",
                            87,
                            {
                                "phase": "synthesis_error",
                                "error_type": error_type,
                            },
                        )
                    else:
                        # Generic error
                        logger.warning("Detected unknown error in synthesis")
                        progress_callback(
                            "Error during synthesis. Attempting fallback...",
                            87,
                            {
                                "phase": "synthesis_error",
                                "error_type": "unknown",
                            },
                        )

                    # Extract synthesized content from findings if available
                    synthesized_content = ""
                    for finding in results.get("findings", []):
                        if finding.get("phase") == "Final synthesis":
                            synthesized_content = finding.get("content", "")
                            break

                    # Use synthesized content as fallback
                    if (
                        synthesized_content
                        and not synthesized_content.startswith("Error:")
                    ):
                        logger.info(
                            "Using existing synthesized content as fallback"
                        )
                        raw_formatted_findings = synthesized_content

                    # Or use current_knowledge as another fallback
                    elif results.get("current_knowledge"):
                        logger.info("Using current_knowledge as fallback")
                        raw_formatted_findings = results["current_knowledge"]

                    # Or combine all finding contents as last resort
                    elif results.get("findings"):
                        logger.info("Combining all findings as fallback")
                        # First try to use any findings that are not errors
                        valid_findings = [
                            f"## {finding.get('phase', 'Finding')}\n\n{finding.get('content', '')}"
                            for finding in results.get("findings", [])
                            if finding.get("content")
                            and not finding.get("content", "").startswith(
                                "Error:"
                            )
                        ]

                        synthesis_error = raw_formatted_findings
                        if valid_findings:
                            raw_formatted_findings = (
                                "# Research Results (Fallback Mode)\n\n"
                            )
                            raw_formatted_findings += "\n\n".join(
                                valid_findings
                            )
                            raw_formatted_findings += (
                                f"\n\n## Error Information\n{synthesis_error}"
                            )
                        else:
                            # Last resort: use everything including errors
                            raw_formatted_findings = (
                                "# Research Results (Emergency Fallback)\n\n"
                            )
                            raw_formatted_findings += "The system encountered errors during final synthesis.\n\n"
                            raw_formatted_findings += "\n\n".join(
                                f"## {finding.get('phase', 'Finding')}\n\n{finding.get('content', '')}"
                                for finding in results.get("findings", [])
                                if finding.get("content")
                            )

                    progress_callback(
                        f"Using fallback synthesis due to {error_type} error",
                        88,
                        {
                            "phase": "synthesis_fallback",
                            "error_type": error_type,
                        },
                    )

                logger.info(
                    "Found formatted_findings of length: {}",
                    len(str(raw_formatted_findings)),
                )

                try:
                    # Check if we have an error in the findings and use enhanced error handling
                    if isinstance(
                        raw_formatted_findings, str
                    ) and raw_formatted_findings.startswith("Error:"):
                        logger.info(
                            "Generating enhanced error report using ErrorReportGenerator"
                        )

                        # Generate comprehensive error report
                        # ErrorReportGenerator does not use LLM (kept for compat)
                        error_generator = ErrorReportGenerator()
                        clean_markdown = error_generator.generate_error_report(
                            error_message=raw_formatted_findings,
                            query=query,
                            partial_results=results,
                            search_iterations=results.get("iterations", 0),
                            research_id=research_id,
                        )

                        logger.info(
                            "Generated enhanced error report with {} characters",
                            len(clean_markdown),
                        )
                    else:
                        # report_content stores the synthesized answer
                        # only (see _extract_synthesized_answer for the
                        # full rationale). Fall back to the formatted
                        # blob only when neither Final synthesis nor
                        # current_knowledge is populated — leaks
                        # sources, but at least we save *something*.
                        clean_markdown = (
                            _extract_synthesized_answer(results)
                            or raw_formatted_findings
                        )

                    # Pull sources from the search-system's accumulated link
                    # buffer first — same source the detailed-report path
                    # uses at the equivalent point below. Wrapper
                    # strategies (e.g. EnhancedContextualFollowUpStrategy,
                    # IterativeRefinementStrategy) delegate the actual
                    # search to an inner strategy that populates
                    # `self.all_links_of_system`, but they don't bubble
                    # that list back into the result dict's `findings`. So
                    # the legacy `findings[*].search_results` extraction
                    # below stays empty for chat follow-ups, leaving the
                    # citation formatter with no urls to hyperlink. Prefer
                    # the system-level accumulator; fall back to the
                    # legacy extraction so direct strategies that bypass
                    # the system buffer still work.
                    all_links = list(
                        getattr(system, "all_links_of_system", None) or []
                    )
                    if not all_links:
                        for finding in results.get("findings", []):
                            search_results = finding.get("search_results", [])
                            if search_results:
                                try:
                                    links = extract_links_from_search_results(
                                        search_results
                                    )
                                    all_links.extend(links)
                                except Exception:
                                    logger.exception(
                                        "Error processing search results/links"
                                    )

                    logger.info(
                        "Successfully converted to clean markdown of length: {}",
                        len(clean_markdown),
                    )

                    # First send a progress update for generating the summary
                    progress_callback(
                        "Generating clean summary from research data...",
                        90,
                        {"phase": "output_generation"},
                    )

                    # Send progress update for saving report
                    progress_callback(
                        "Saving research report to database...",
                        95,
                        {"phase": "report_complete"},
                    )

                    # Format citations in the markdown content. The
                    # split returns the answer-with-hyperlinks half
                    # separately from the trailing sources section the
                    # LLM may have emitted (which is discarded — sources
                    # live in research_resources, the canonical store).
                    # When no Sources section is found, fall back to
                    # structured-source hyperlinking — never re-parse
                    # concatenated formatter output downstream.
                    formatter = get_citation_formatter()
                    try:
                        answer_with_links, llm_sources = (
                            formatter.format_document_split(clean_markdown)
                        )
                        if not llm_sources:
                            answer_with_links = (
                                formatter.apply_inline_hyperlinks(
                                    clean_markdown, all_links
                                )
                            )
                        # Safety check: a >50% strip on a long input
                        # likely means the regex over-stripped on a
                        # "Sources:" header inside the answer body.
                        # Fall back to structured-source hyperlinking
                        # on the full text. Min-length floor prevents
                        # false-fires on legitimately short answers.
                        SAFETY_MIN_LEN = 800
                        if (
                            llm_sources
                            and len(clean_markdown) > SAFETY_MIN_LEN
                            and len(answer_with_links)
                            < len(clean_markdown) * 0.5
                        ):
                            logger.warning(
                                "format_document_split appears to have "
                                "over-stripped (answer={} chars, "
                                "original={} chars) for research {}. "
                                "Falling back to structured-source "
                                "hyperlinking on full input.",
                                len(answer_with_links),
                                len(clean_markdown),
                                research_id,
                            )
                            answer_with_links = (
                                formatter.apply_inline_hyperlinks(
                                    clean_markdown, all_links
                                )
                            )
                    except Exception:
                        # Hyperlinking is quality-of-life, not a hard
                        # requirement. If anything blows up, save the
                        # raw LLM text rather than fail the research.
                        logger.exception(
                            "Citation formatter failed; saving raw answer"
                        )
                        answer_with_links = clean_markdown

                    # report_content stores ONLY the synthesized answer.
                    # The legacy "answer + ## Sources + ## Research
                    # Metrics" view is reconstructed at render time by
                    # report_assembly_service.assemble_full_report.
                    full_report_content = answer_with_links

                    # Save report FIRST, then sources:
                    # a chat read between commits sees a report with no
                    # sources (assembler renders just the answer) — better
                    # failure mode than partial assembly with sources but
                    # no answer body.
                    from ...storage import get_report_storage

                    with get_user_db_session(username) as db_session:
                        storage = get_report_storage(session=db_session)

                        # Prepare metadata
                        metadata = {
                            "iterations": results["iterations"],
                            "generated_at": datetime.now(UTC).isoformat(),
                        }

                        # Save report using storage abstraction
                        success = storage.save_report(
                            research_id=research_id,
                            content=full_report_content,
                            metadata=metadata,
                            username=username,
                        )

                        if not success:
                            raise RuntimeError("Failed to save research report")  # noqa: TRY301 — triggers research failure handling in outer except

                    # Save sources to database (non-fatal - report
                    # already saved; sources missing is recoverable
                    # because the assembler omits empty Sources blocks)
                    try:
                        from .research_sources_service import (
                            ResearchSourcesService,
                        )

                        sources_service = ResearchSourcesService()
                        if all_links:
                            logger.info(
                                f"Quick summary: Saving {len(all_links)} sources to database"
                            )
                            sources_saved = (
                                sources_service.save_research_sources(
                                    research_id=research_id,
                                    sources=all_links,
                                    username=username,
                                )
                            )
                            logger.info(
                                f"Quick summary: Saved {sources_saved} sources for research {research_id}"
                            )
                    except Exception:
                        logger.exception(
                            f"Failed to save sources for research {research_id} (continuing with report save)"
                        )

                    logger.info(f"Report saved for research_id: {research_id}")

                    # Skip export to additional formats - we're storing in database only

                    # Update research status in database
                    completed_at = datetime.now(UTC).isoformat()

                    with get_user_db_session(username) as db_session:
                        research = (
                            db_session.query(ResearchHistory)
                            .filter_by(id=research_id)
                            .first()
                        )

                        # Preserve existing metadata and update with new values
                        metadata = _parse_research_metadata(
                            research.research_meta
                        )

                        metadata.update(
                            {
                                "iterations": results["iterations"],
                                "generated_at": datetime.now(UTC).isoformat(),
                            }
                        )

                        # Use the helper function for consistent duration calculation
                        duration_seconds = calculate_duration(
                            research.created_at, completed_at
                        )

                        research.status = ResearchStatus.COMPLETED
                        research.completed_at = completed_at
                        research.duration_seconds = duration_seconds
                        # report_content was already saved above via the report
                        # storage abstraction; this block only updates
                        # status/metadata. report_path is not used in the
                        # encrypted-database version.

                        # Generate headline and topics only for news searches
                        if (
                            metadata.get("is_news_search")
                            or metadata.get("search_type") == "news_analysis"
                        ):
                            try:
                                from ...news.utils.headline_generator import (
                                    generate_headline,
                                )
                                from ...news.utils.topic_generator import (
                                    generate_topics,
                                )

                                # Get the report content from database for better headline/topic generation
                                report_content = ""
                                try:
                                    research = (
                                        db_session.query(ResearchHistory)
                                        .filter_by(id=research_id)
                                        .first()
                                    )
                                    if research and research.report_content:
                                        report_content = research.report_content
                                        logger.info(
                                            f"Retrieved {len(report_content)} chars from database for headline generation"
                                        )
                                    else:
                                        logger.warning(
                                            f"No report content found in database for research_id: {research_id}"
                                        )
                                except Exception:
                                    logger.warning(
                                        "Could not retrieve report content from database"
                                    )

                                # Generate headline
                                logger.info(
                                    f"Generating headline for query: {query[:100]}"
                                )
                                headline = generate_headline(
                                    query, report_content
                                )
                                metadata["generated_headline"] = headline

                                # Generate topics
                                logger.info(
                                    f"Generating topics with category: {metadata.get('category', 'News')}"
                                )
                                topics = generate_topics(
                                    query=query,
                                    findings=report_content,
                                    category=metadata.get("category", "News"),
                                    max_topics=6,
                                )
                                metadata["generated_topics"] = topics

                                logger.info(f"Generated headline: {headline}")
                                logger.info(f"Generated topics: {topics}")

                            except Exception:
                                logger.warning(
                                    "Could not generate headline/topics"
                                )

                        research.research_meta = metadata

                        db_session.commit()
                        logger.info(
                            f"Database commit completed for research_id: {research_id}"
                        )

                        # Update subscription if this was triggered by a subscription
                        if metadata.get("subscription_id"):
                            try:
                                from ...news.subscription_runner import (
                                    advance_refresh_schedule_by_id,
                                )

                                subscription_id = metadata["subscription_id"]
                                if advance_refresh_schedule_by_id(
                                    db_session, subscription_id
                                ):
                                    db_session.commit()
                                    logger.info(
                                        f"Updated subscription {subscription_id} refresh times"
                                    )
                            except Exception:
                                logger.warning(
                                    "Could not update subscription refresh time"
                                )

                    logger.info(
                        f"Database updated successfully for research_id: {research_id}"
                    )

                    _maybe_save_chat_message(full_report_content)

                    # Send the final completion message
                    progress_callback(
                        "Research completed successfully",
                        100,
                        {"phase": "complete"},
                    )

                    # Clean up resources
                    logger.info(
                        "Cleaning up resources for research_id: {}", research_id
                    )
                    cleanup_research_resources(
                        research_id, username, user_password=user_password
                    )
                    logger.info(
                        "Resources cleaned up for research_id: {}", research_id
                    )

                except Exception as inner_e:
                    logger.exception("Error during quick summary generation")
                    raise RuntimeError(
                        f"Error generating quick summary: {inner_e!s}"
                    )
            else:
                raise RuntimeError(  # noqa: TRY301 — triggers research failure handling in outer except
                    "No research findings were generated. Please try again."
                )
        else:
            # Full Report
            progress_callback(
                "Generating detailed report...",
                _DETAILED_REPORT_PROGRESS_START,
                {"phase": "report_generation"},
            )

            # Extract the search system from the results if available
            search_system = results.get("search_system", None)

            # Wrapper that maps report generator's 0-100% to the configured
            # detailed-mode range and relays cancellation checks through the
            # outer progress_callback
            _report_range = (
                _DETAILED_REPORT_PROGRESS_END - _DETAILED_REPORT_PROGRESS_START
            )

            def report_progress_callback(message, progress_percent, metadata):
                if progress_percent is not None:
                    adjusted = (
                        _DETAILED_REPORT_PROGRESS_START
                        + (progress_percent / 100) * _report_range
                    )
                else:
                    adjusted = progress_percent
                progress_callback(message, adjusted, metadata)

            # Pass the existing search system to maintain citation indices
            report_generator = IntegratedReportGenerator(
                search_system=search_system,
                settings_snapshot=settings_snapshot,
            )
            final_report = report_generator.generate_report(
                results, query, progress_callback=report_progress_callback
            )

            progress_callback(
                "Report generation complete",
                _DETAILED_REPORT_PROGRESS_END,
                {"phase": "report_complete"},
            )

            # Format citations and split off the trailing Sources
            # section. Save only the answer half — sources are
            # persisted structurally to research_resources.
            all_links = (
                getattr(search_system, "all_links_of_system", None) or []
            )
            formatter = get_citation_formatter()
            try:
                answer_with_links, llm_sources = (
                    formatter.format_document_split(final_report["content"])
                )
                if not llm_sources:
                    answer_with_links = formatter.apply_inline_hyperlinks(
                        final_report["content"], all_links
                    )
                SAFETY_MIN_LEN = 800
                if (
                    llm_sources
                    and len(final_report["content"]) > SAFETY_MIN_LEN
                    and len(answer_with_links)
                    < len(final_report["content"]) * 0.5
                ):
                    logger.warning(
                        "format_document_split appears to have over-stripped "
                        "(answer={} chars, original={} chars) for research {}.",
                        len(answer_with_links),
                        len(final_report["content"]),
                        research_id,
                    )
                    answer_with_links = formatter.apply_inline_hyperlinks(
                        final_report["content"], all_links
                    )
            except Exception:
                logger.exception("Citation formatter failed; saving raw answer")
                answer_with_links = final_report["content"]
            formatted_content = answer_with_links

            # Save report FIRST, sources after.
            # See quick-summary path for rationale.
            from ...storage import get_report_storage

            with get_user_db_session(username) as db_session:
                storage = get_report_storage(session=db_session)

                # Update metadata. Include generated_at like the quick path so
                # the detailed file backup's _metadata.json has parity with
                # quick-mode (the detailed path previously omitted it).
                metadata = final_report["metadata"]
                metadata["iterations"] = results["iterations"]
                metadata["generated_at"] = datetime.now(UTC).isoformat()

                # Save the report through the storage abstraction, exactly as
                # the quick-summary branch does, so detailed reports also honor
                # the report.enable_file_backup setting. This previously did a
                # raw ORM write of report_content that silently skipped the
                # file backup, so a user who enabled file backup got it for
                # quick research but never for detailed research.
                success = storage.save_report(
                    research_id=research_id,
                    content=formatted_content,
                    metadata=metadata,
                    username=username,
                )

                if not success:
                    raise RuntimeError("Failed to save research report")  # noqa: TRY301 — triggers research failure handling in outer except

                logger.info(
                    f"Report saved to database for research_id: {research_id}"
                )

            # Save sources AFTER report (non-fatal; assembler omits
            # empty Sources blocks if this fails).
            try:
                from .research_sources_service import ResearchSourcesService

                sources_service = ResearchSourcesService()
                if all_links:
                    logger.info(f"Saving {len(all_links)} sources to database")
                    sources_saved = sources_service.save_research_sources(
                        research_id=research_id,
                        sources=all_links,
                        username=username,
                    )
                    logger.info(
                        f"Saved {sources_saved} sources for research {research_id}"
                    )
            except Exception:
                logger.exception(
                    f"Failed to save sources for research {research_id} (continuing)"
                )

            # Update research status in database
            completed_at = datetime.now(UTC).isoformat()

            with get_user_db_session(username) as db_session:
                research = (
                    db_session.query(ResearchHistory)
                    .filter_by(id=research_id)
                    .first()
                )

                # Preserve existing metadata and merge with report metadata
                metadata = _parse_research_metadata(research.research_meta)

                metadata.update(final_report["metadata"])
                metadata["iterations"] = results["iterations"]

                # Use the helper function for consistent duration calculation
                duration_seconds = calculate_duration(
                    research.created_at, completed_at
                )

                research.status = ResearchStatus.COMPLETED
                research.completed_at = completed_at
                research.duration_seconds = duration_seconds
                # report_content was already saved above via the report storage
                # abstraction; this block only updates status/metadata.
                # report_path is not used in the encrypted-database version.

                # Generate headline and topics only for news searches
                if (
                    metadata.get("is_news_search")
                    or metadata.get("search_type") == "news_analysis"
                ):
                    try:
                        from ...news.utils.headline_generator import (
                            generate_headline,  # type: ignore[no-redef]
                        )
                        from ...news.utils.topic_generator import (
                            generate_topics,  # type: ignore[no-redef]
                        )

                        # Get the report content from database for better headline/topic generation
                        report_content = ""
                        try:
                            research = (
                                db_session.query(ResearchHistory)
                                .filter_by(id=research_id)
                                .first()
                            )
                            if research and research.report_content:
                                report_content = research.report_content
                            else:
                                logger.warning(
                                    f"No report content found in database for research_id: {research_id}"
                                )
                        except Exception:
                            logger.warning(
                                "Could not retrieve report content from database"
                            )

                        # Generate headline
                        headline = generate_headline(query, report_content)
                        metadata["generated_headline"] = headline

                        # Generate topics
                        topics = generate_topics(
                            query=query,
                            findings=report_content,
                            category=metadata.get("category", "News"),
                            max_topics=6,
                        )
                        metadata["generated_topics"] = topics

                        logger.info(f"Generated headline: {headline}")
                        logger.info(f"Generated topics: {topics}")

                    except Exception:
                        logger.warning("Could not generate headline/topics")

                research.research_meta = metadata

                db_session.commit()

                # Update subscription if this was triggered by a subscription
                if metadata.get("subscription_id"):
                    try:
                        from ...news.subscription_runner import (
                            advance_refresh_schedule_by_id,
                        )

                        subscription_id = metadata["subscription_id"]
                        if advance_refresh_schedule_by_id(
                            db_session, subscription_id
                        ):
                            db_session.commit()
                            logger.info(
                                f"Updated subscription {subscription_id} refresh times"
                            )
                    except Exception:
                        logger.warning(
                            "Could not update subscription refresh time"
                        )

            _maybe_save_chat_message(formatted_content)

            progress_callback(
                "Research completed successfully",
                100,
                {"phase": "complete"},
            )

            # Clean up resources
            cleanup_research_resources(
                research_id, username, user_password=user_password
            )

    except ResearchTerminatedException:
        logger.info(f"Research {research_id} terminated by user")
        # Fallback path: when termination was raised from the streaming
        # callback (mid-stream interrupt), progress_callback hasn't run
        # since the flag was set, so handle_termination() was NOT called
        # and the partial row hasn't been persisted yet. The helper is
        # idempotent via streaming_state["_persisted"]; if the in-callback
        # path already ran, both calls are no-ops.
        _save_partial_chat_message_on_terminate(
            shared_research_context.get("chat_session_id"),
            research_id,
            username,
            "".join(streaming_state.get("chunks", [])),
            truncated=streaming_state.get("_truncated", False),
            streaming_state=streaming_state,
        )
        # Ensure the SUSPENDED status update + cleanup runs even when the
        # exception was raised mid-stream. The in-callback termination paths
        # set "_termination_handled"; only run here when they did NOT, so a
        # single termination doesn't queue two SUSPENDED updates, emit two
        # final socket messages, and (in test mode) sleep twice.
        if not streaming_state.get("_termination_handled"):
            try:
                handle_termination(research_id, username)
            except Exception:
                logger.opt(exception=True).debug(
                    "handle_termination in except block failed"
                )

    except Exception as e:
        # Handle error
        error_message = f"Research failed: {e!s}"
        logger.exception(error_message)

        try:
            # Check for common Ollama error patterns in the exception and provide more user-friendly errors
            user_friendly_error = str(e)
            error_context = {}

            if "Error type: ollama_unavailable" in user_friendly_error:
                user_friendly_error = "Ollama AI service is unavailable. Please check that Ollama is running properly on your system."
                error_context = {
                    "solution": "Start Ollama with 'ollama serve' or check if it's installed correctly."
                }
            elif "Error type: model_not_found" in user_friendly_error:
                user_friendly_error = "Required Ollama model not found. Please pull the model first."
                error_context = {
                    "solution": "Run 'ollama pull mistral' to download the required model."
                }
            elif "Error type: connection_error" in user_friendly_error:
                user_friendly_error = "Connection error with LLM service. Please check that your AI service is running."
                error_context = {
                    "solution": "Ensure Ollama or your API service is running and accessible."
                }
            elif "Error type: api_error" in user_friendly_error:
                # Keep the original error message as it's already improved
                error_context = {
                    "solution": "Check API configuration and credentials."
                }
            # OpenAI-compatible runtime tokens (#3878). The friendly message
            # built by friendly_openai_compatible_error() already names the
            # provider, base URL, and model -- keep it as-is.
            elif "Error type: openai_connection_refused" in user_friendly_error:
                error_context = {
                    "solution": "Start your LLM server (LM Studio / vLLM / llama.cpp server) and verify the base URL in Settings -> LLM Providers."
                }
            elif "Error type: openai_timeout" in user_friendly_error:
                error_context = {
                    "solution": "The server is reachable but slow -- it may be loading a model. Retry, or increase the request timeout."
                }
            elif "Error type: openai_auth" in user_friendly_error:
                error_context = {
                    "solution": "Set or correct the API key for this provider in Settings -> LLM Providers. Local servers usually accept any non-empty key."
                }
            elif "Error type: openai_permission_denied" in user_friendly_error:
                error_context = {
                    "solution": "Your API key is valid but lacks access to this model. Pick a model your account/server is permitted to use."
                }
            elif "Error type: openai_model_not_found" in user_friendly_error:
                error_context = {
                    "solution": "The model id is not loaded on this server. Pick a currently-loaded model in the provider's UI/config."
                }
            elif "Error type: openai_bad_request" in user_friendly_error:
                error_context = {
                    "solution": "The server rejected the request. Check the model id and any provider-specific parameters."
                }
            elif "Error type: openai_unknown" in user_friendly_error:
                error_context = {
                    "solution": "Check the provider's logs for the full error and verify the base URL / model id."
                }
            elif "Error type: openai_rate_limit" in user_friendly_error:
                error_context = {
                    "solution": "The provider rate-limited the request. Wait a moment and retry, or enable LLM Rate Limiting in Settings."
                }
            elif "LLM Configuration Error:" in user_friendly_error:
                # Surface the raw config detail to the operator (which model
                # path / .gguf / endpoint is misconfigured) rather than hiding
                # it behind a generic message: this is a self-hosted tool, so
                # the operator needs the specifics to fix their own setup.
                # Keep ``user_friendly_error`` as-is and just attach a hint.
                error_context = {
                    "solution": "Review your LLM model settings and ensure they are correct."
                }

            # Generate enhanced error report for failed research
            enhanced_report_content = None
            try:
                # Get partial results if they exist
                partial_results = results if "results" in locals() else None
                search_iterations = (
                    results.get("iterations", 0) if partial_results else 0
                )

                # Generate comprehensive error report
                # ErrorReportGenerator does not use LLM (kept for compat)
                error_generator = ErrorReportGenerator()
                enhanced_report_content = error_generator.generate_error_report(
                    error_message=f"Research failed: {e!s}",
                    query=query,
                    partial_results=partial_results,
                    search_iterations=search_iterations,
                    research_id=research_id,
                )

                logger.info(
                    "Generated enhanced error report for failed research (length: {})",
                    len(enhanced_report_content),
                )

                # Save enhanced error report to encrypted database
                try:
                    # username already available from function scope (line 281)
                    if username:
                        from ...storage import get_report_storage

                        with get_user_db_session(username) as db_session:
                            storage = get_report_storage(session=db_session)
                            success = storage.save_report(
                                research_id=research_id,
                                content=enhanced_report_content,
                                metadata={"error_report": True},
                                username=username,
                            )
                            if success:
                                logger.info(
                                    "Saved enhanced error report to encrypted database for research {}",
                                    research_id,
                                )
                            else:
                                logger.warning(
                                    "Failed to save enhanced error report to database for research {}",
                                    research_id,
                                )
                    else:
                        logger.warning(
                            "Cannot save error report: username not available"
                        )

                except Exception as report_error:
                    logger.exception(
                        "Failed to save enhanced error report: {}", report_error
                    )

            except Exception as error_gen_error:
                logger.exception(
                    "Failed to generate enhanced error report: {}",
                    error_gen_error,
                )
                enhanced_report_content = None

            # Get existing metadata from database first
            existing_metadata = {}
            try:
                # username already available from function scope (line 281)
                if username:
                    with get_user_db_session(username) as db_session:
                        research = (
                            db_session.query(ResearchHistory)
                            .filter_by(id=research_id)
                            .first()
                        )
                        if research and research.research_meta:
                            existing_metadata = dict(research.research_meta)
            except Exception:
                logger.exception("Failed to get existing metadata")

            # Update metadata with more context about the error while preserving existing values
            metadata = existing_metadata
            metadata.update({"phase": "error", "error": user_friendly_error})
            if error_context:
                metadata.update(error_context)
            if enhanced_report_content:
                metadata["has_enhanced_report"] = True

            # If we still have an active research record, update its log
            if is_research_active(research_id):
                progress_callback(user_friendly_error, None, metadata)

            # If termination was requested, mark as suspended instead of failed
            status = (
                ResearchStatus.SUSPENDED
                if is_termination_requested(research_id)
                else ResearchStatus.FAILED
            )
            message = (
                "Research was terminated by user"
                if status == ResearchStatus.SUSPENDED
                else user_friendly_error
            )

            # A subscription-triggered run that FAILED must be made due again.
            # run_subscription_now / the overdue sweep advance next_refresh at
            # spawn time (to avoid the scheduler double-running an in-flight
            # subscription), but the completion-time advance only runs on
            # success. Without this reset a failed run would leave next_refresh
            # pushed a full interval out, silently hiding the subscription from
            # the scheduler. Skipped for SUSPENDED (user-terminated) runs.
            if (
                status == ResearchStatus.FAILED
                and username
                and metadata.get("subscription_id")
            ):
                try:
                    from ...news.subscription_runner import (
                        mark_subscription_due_by_id,
                    )

                    with get_user_db_session(username) as sub_db:
                        if mark_subscription_due_by_id(
                            sub_db, metadata["subscription_id"]
                        ):
                            sub_db.commit()
                            logger.info(
                                f"Reset subscription {metadata['subscription_id']} "
                                "to due after failed run"
                            )
                except Exception:
                    logger.warning(
                        "Could not reset subscription refresh time after "
                        "failed run"
                    )

            # Calculate duration up to termination point - using UTC consistently
            now = datetime.now(UTC)
            completed_at = now.isoformat()

            # NOTE: Database updates from threads are handled by queue processor
            # The queue_processor.queue_error_update() method is already being used below
            # to safely update the database from the main thread

            # Queue the error update to be processed in main thread
            # Using the queue processor v2 system
            from ..queue.processor_v2 import queue_processor

            if username:
                queue_processor.queue_error_update(
                    username=username,
                    research_id=research_id,
                    status=status,
                    error_message=message,
                    metadata=metadata,
                    completed_at=completed_at,
                    report_path=None,
                )
                logger.info(
                    f"Queued error update for research {research_id} with status '{status}'"
                )
            else:
                logger.error(
                    f"Cannot queue error update for research {research_id} - no username provided. "
                    f"Status: '{status}', Message: {message}"
                )

            try:
                SocketIOService().emit_to_subscribers(
                    "progress",
                    research_id,
                    {"status": status, "error": message},
                )
            except Exception:
                logger.exception("Failed to emit error via socket")

            # Add error message to chat session if applicable.
            # Read chat_session_id from shared_research_context (the canonical
            # source — kept consistent with the six other reads in this file).
            chat_session_id = shared_research_context.get("chat_session_id")
            if chat_session_id and username:
                try:
                    from ...chat.service import ChatService

                    chat_service = ChatService(username)
                    # allow_archived=True: same multi-tab race rationale as
                    # the completion / stop-and-partial paths — if the user
                    # archived (or deleted) the session between research
                    # start and failure, the error message would otherwise
                    # be silently dropped by the active-only insert guard
                    # and the user sees nothing in the chat.
                    chat_service.add_message(
                        session_id=chat_session_id,
                        role="assistant",
                        content=f"Sorry, the research failed: {message}",
                        message_type="response",
                        allow_archived=True,
                    )
                except Exception:
                    # Promoted from debug → warning to match the success-path
                    # rationale: if this write fails the user never sees the
                    # error and operators get no signal at debug-off level.
                    logger.opt(exception=True).warning(
                        "Could not add error message to chat session"
                    )

        except Exception:
            logger.exception("Error in error handler")

        # Clean up resources. This is the error path, so report FAILED on
        # the final socket message rather than a spurious "completed".
        cleanup_research_resources(
            research_id,
            username,
            user_password=user_password,
            final_status=ResearchStatus.FAILED,
        )

    finally:
        # RESOURCE CLEANUP: Close search engine HTTP sessions.
        #
        # Search engines (created via get_search()) may hold HTTP connection
        # pools. Currently only SemanticScholarSearchEngine creates a
        # persistent SafeSession; other engines use stateless safe_get()/
        # safe_post() utility functions. However, BaseSearchEngine.close()
        # is safe to call on any engine — it checks for a 'session'
        # attribute and is fully idempotent (SemanticScholar sets
        # self.session = None after close).
        #
        # Neither @thread_cleanup nor cleanup_research_resources() close
        # the search engine — @thread_cleanup only handles database sessions
        # and context cleanup, and cleanup_research_resources() only handles
        # status updates, notifications, and tracking dict removal.
        #
        # Without this explicit close, search engine sessions rely on
        # Python's non-deterministic garbage collection (__del__) for
        # cleanup, which can cause file descriptor exhaustion under
        # sustained load.
        from ...utilities.resource_utils import safe_close

        if "use_search" in locals():
            safe_close(use_search, "research search engine")
        # Close search system (cascades to strategy thread pools).
        # See AdvancedSearchSystem.close() for details.
        if "system" in locals():
            safe_close(system, "research system")
        # Close the LLM instance created for model/provider overrides.
        # system.close() does NOT close the LLM passed to it via system.model,
        # so we must close it explicitly here.
        if "use_llm" in locals():
            safe_close(use_llm, "research LLM")


def cleanup_research_resources(
    research_id,
    username=None,
    user_password=None,
    final_status=ResearchStatus.COMPLETED,
):
    """
    Clean up resources for a completed research.

    Args:
        research_id: The ID of the research
        username: The username for database access (required for thread context)
        final_status: The terminal status to report on the final socket
            message. Callers that end a research for a reason other than
            normal completion MUST pass the real status (e.g. SUSPENDED on
            user termination, FAILED on error) so the final ``progress``
            event matches reality. Defaulting this to COMPLETED — and
            previously hard-coding it — caused the stop/error paths to emit
            a spurious "completed" signal to subscribers.
    """
    from ..routes.globals import cleanup_research

    logger.info("Cleaning up resources for research {}", research_id)

    # For testing: Add a small delay to simulate research taking time
    # This helps test concurrent research limits
    from ...settings.env_registry import is_test_mode

    if is_test_mode():
        import time

        logger.info(
            f"Test mode: Adding 5 second delay before cleanup for {research_id}"
        )
        time.sleep(5)

    # The terminal status to report on the final socket message. This comes
    # from the caller (which knows why the research ended) rather than a
    # hard-coded COMPLETED, so termination (SUSPENDED) and error (FAILED)
    # paths no longer emit a false "completed" signal to subscribers.
    current_status = final_status

    # NOTE: Queue processor already handles database updates from the main thread
    # The notify_research_completed() method is called at the end of this function
    # which safely updates the database status

    # Notify queue processor that research completed
    # This uses processor_v2 which handles database updates in the main thread
    # avoiding the Flask request context issues that occur in background threads
    from ..queue.processor_v2 import queue_processor

    if username:
        queue_processor.notify_research_completed(
            username, research_id, user_password=user_password
        )
        logger.info(
            f"Notified queue processor of completion for research {research_id} (user: {username})"
        )
    else:
        logger.warning(
            f"Cannot notify completion for research {research_id} - no username provided"
        )

    # Remove from active research and termination flags atomically
    cleanup_research(research_id)

    # Clean up throttle state for this research
    with _last_emit_lock:
        _last_emit_times.pop(research_id, None)

    # Send a final message to subscribers
    try:
        # Send a final message to any remaining subscribers with explicit status
        # Use the proper status message based on database status
        if current_status in (
            ResearchStatus.SUSPENDED,
            ResearchStatus.FAILED,
        ):
            final_message = {
                "status": current_status,
                "message": f"Research was {current_status}",
                "progress": 0,  # For suspended research, show 0% not 100%
            }
        else:
            final_message = {
                "status": ResearchStatus.COMPLETED,
                "message": "Research process has ended and resources have been cleaned up",
                "progress": 100,
            }

        logger.info(
            "Sending final {} socket message for research {}",
            current_status,
            research_id,
        )

        SocketIOService().emit_to_subscribers(
            "progress", research_id, final_message
        )

        # Clean up socket subscriptions for this research
        SocketIOService().remove_subscriptions_for_research(research_id)

    except Exception:
        logger.exception("Error sending final cleanup message")


def handle_termination(research_id, username=None):
    """
    Handle the termination of a research process.

    Args:
        research_id: The ID of the research
        username: The username for database access (required for thread context)
    """
    logger.info(f"Handling termination for research {research_id}")

    # Queue the status update to be processed in the main thread
    # This avoids Flask request context errors in background threads
    try:
        from ..queue.processor_v2 import queue_processor

        now = datetime.now(UTC)
        completed_at = now.isoformat()

        # Queue the suspension update
        queue_processor.queue_error_update(
            username=username,
            research_id=research_id,
            status=ResearchStatus.SUSPENDED,
            error_message="Research was terminated by user",
            metadata={"terminated_at": completed_at},
            completed_at=completed_at,
            report_path=None,
        )

        logger.info(f"Queued suspension update for research {research_id}")
    except Exception:
        logger.exception(
            f"Error queueing termination update for research {research_id}"
        )

    # Clean up resources (this already handles things properly).
    # Pass SUSPENDED so the final socket message reports the real terminal
    # status — not a spurious "completed" — to chat/progress subscribers.
    cleanup_research_resources(
        research_id, username, final_status=ResearchStatus.SUSPENDED
    )


def cancel_research(research_id, username):
    """
    Cancel/terminate a research process using ORM.

    Args:
        research_id: The ID of the research to cancel
        username: The username of the user cancelling the research

    Returns:
        bool: True if the research was found and cancelled, False otherwise
    """
    try:
        from ..routes.globals import is_research_active, set_termination_flag

        # Set termination flag
        set_termination_flag(research_id)

        # Check if the research is active
        if is_research_active(research_id):
            # Call handle_termination to update database
            handle_termination(research_id, username)
            return True
        try:
            with get_user_db_session(username) as db_session:
                research = (
                    db_session.query(ResearchHistory)
                    .filter_by(id=research_id)
                    .first()
                )
                if not research:
                    logger.info(f"Research {research_id} not found in database")
                    return False

                # Check if already in a terminal state
                if research.status in (
                    ResearchStatus.COMPLETED,
                    ResearchStatus.SUSPENDED,
                    ResearchStatus.FAILED,
                    ResearchStatus.ERROR,
                ):
                    logger.info(
                        f"Research {research_id} already in terminal state: {research.status}"
                    )
                    return True  # Consider this a success since it's already stopped

                # If it exists but isn't in active_research, still update status
                research.status = ResearchStatus.SUSPENDED
                db_session.commit()
                logger.info(f"Successfully suspended research {research_id}")
        except Exception:
            logger.exception(
                f"Error accessing database for research {research_id}"
            )
            return False

        return True
    except Exception:
        logger.exception(
            f"Unexpected error in cancel_research for {research_id}"
        )
        return False
