"""
RAG Management API Routes

Provides endpoints for managing RAG indexing of library documents:
- Configure embedding models
- Index documents
- Get RAG statistics
- Bulk operations with progress tracking
"""

import os
import re

from flask import (
    Blueprint,
    jsonify,
    request,
    Response,
    render_template,
    session,
    stream_with_context,
)
from loguru import logger
import atexit
import glob
import json
import uuid
import time
import threading
import queue
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, UTC
from pathlib import Path
from typing import Optional

from sqlalchemy.orm import defer

from ...constants import (
    DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
    DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
    FILE_PATH_SENTINELS,
    FILE_PATH_TEXT_ONLY,
)
from ...security.decorators import require_json_body
from ...security.log_sanitizer import sanitize_error_message
from ...web.auth.decorators import login_required
from ...web.utils.request_helpers import parse_bool_arg
from ...utilities.db_utils import get_settings_manager
from ...utilities.resource_utils import safe_close
from ..services.library_rag_service import LibraryRAGService
from ...settings.manager import SettingsManager
from ...security.file_upload_validator import FileUploadValidator
from ...security.path_validator import PathValidator
from ...security.rate_limiter import (
    upload_rate_limit_ip,
    upload_rate_limit_user,
)
from ..utils import ensure_in_collection, handle_api_error
from ...database.models.library import (
    Document,
    Collection,
    DocumentCollection,
    RAGIndex,
    SourceType,
    EmbeddingProvider,
)
from ...database.models.queue import TaskMetadata
from ...database.thread_local_session import thread_cleanup
from ...security.rate_limiter import limiter
from ...config.paths import get_library_directory
from ...constants import DEFAULT_SEARCH_TOOL

rag_bp = Blueprint("rag", __name__, url_prefix="/library")

# Allowlist for user-supplied glob patterns. Each pattern is joined onto an
# already-validated base directory and globbed, so it must not be able to
# escape that directory. Only in-directory filename globs are allowed:
# any path separator, parent reference, or character outside this set is
# rejected. This also blocks a leading "/" — pathlib's "/" operator resets
# to an absolute path when the right operand is absolute (``base / "/abs"``
# yields ``/abs``), so a leading-slash absolute pattern would otherwise
# escape the base too.
# Anchored with ``\Z`` (not ``$``) so a trailing newline cannot slip through —
# ``$`` matches just before a final "\n", which would make the predicate
# depend on the caller stripping the pattern first.
_SAFE_GLOB_PATTERN = re.compile(r"^[A-Za-z0-9._*?\[\]-]+\Z")


def _is_safe_glob_pattern(pattern: str) -> bool:
    """Return True for in-directory filename globs that cannot traverse out."""
    return bool(_SAFE_GLOB_PATTERN.match(pattern)) and ".." not in pattern


# NOTE: Routes use session["username"] (not .get()) intentionally.
# @login_required guarantees the key exists; direct access fails fast
# if the decorator is ever removed.


def _agent_enabled_default_on(collection) -> bool:
    """Read a collection's ``agent_enabled`` flag with NULL → available.

    A missing attribute (pre-migration object) or a stored NULL both mean
    "available to the research agent" (default-on); only an explicit ``False``
    disables. Used by every serializer so the four read sites can't drift on
    null-handling. ``is not False`` is safe here because the source is a
    Boolean column (only ever None/True/False) — NOT raw JSON input, where a
    falsy ``0`` must still mean disabled (handled separately at the input edge).
    """
    return getattr(collection, "agent_enabled", True) is not False


# Global ThreadPoolExecutor for auto-indexing to prevent thread proliferation.
#
# ThreadPoolExecutor's internal _work_queue is unbounded. Without
# backpressure, a sustained burst of uploads (possible under the
# configurable upload rate cap from #3935) could queue thousands of indexing
# jobs in memory, exhausting RAM. We track pending submissions in a counter
# and reject new submissions once the queue is saturated.
_auto_index_executor: ThreadPoolExecutor | None = None
_auto_index_executor_lock = threading.Lock()

# Maximum in-flight + queued indexing jobs. Each job is one upload batch
# (an unbounded list of document IDs), so per-job duration varies with batch
# size; 100 is a buffer on the number of *queued batches*, not documents.
# The OOM bound still holds because a queued job holds only a small list of
# ID strings, not document contents. Tunable if real-world workloads need
# more headroom.
_MAX_PENDING_AUTO_INDEX_JOBS = 100
_pending_auto_index_jobs = 0
_pending_auto_index_lock = threading.Lock()


def _get_auto_index_executor() -> ThreadPoolExecutor:
    """Get or create the global auto-indexing executor (thread-safe)."""
    global _auto_index_executor
    with _auto_index_executor_lock:
        if _auto_index_executor is None:
            _auto_index_executor = ThreadPoolExecutor(
                max_workers=4,
                thread_name_prefix="auto_index_",
            )
    return _auto_index_executor


def _try_reserve_auto_index_slot() -> bool:
    """Reserve a queue slot if capacity allows. Returns True on success.

    Caller MUST call ``_release_auto_index_slot`` when the job completes
    (success OR failure), otherwise the counter leaks and eventually
    blocks all submissions.
    """
    global _pending_auto_index_jobs
    with _pending_auto_index_lock:
        if _pending_auto_index_jobs >= _MAX_PENDING_AUTO_INDEX_JOBS:
            return False
        _pending_auto_index_jobs += 1
        return True


def _release_auto_index_slot() -> None:
    """Release a previously reserved queue slot."""
    global _pending_auto_index_jobs
    with _pending_auto_index_lock:
        _pending_auto_index_jobs = max(0, _pending_auto_index_jobs - 1)


def _shutdown_auto_index_executor() -> None:
    """Shutdown the auto-index executor gracefully.

    Holds ``_auto_index_executor_lock`` for the read+nullify so two
    concurrent callers (e.g. atexit + a test teardown) can't race into an
    ``AttributeError`` on the ``None`` reference; the blocking ``shutdown``
    runs outside the lock. Resets the pending-jobs counter afterwards so a
    re-created executor (tests, WSGI reload) starts from a clean count
    instead of inheriting a stale, possibly-saturated value.
    """
    global _auto_index_executor, _pending_auto_index_jobs
    with _auto_index_executor_lock:
        executor = _auto_index_executor
        _auto_index_executor = None
    if executor is not None:
        executor.shutdown(wait=True)
    with _pending_auto_index_lock:
        _pending_auto_index_jobs = 0


atexit.register(_shutdown_auto_index_executor)


def get_rag_service(
    collection_id: Optional[str] = None,
    use_defaults: bool = False,
) -> LibraryRAGService:
    """
    Get RAG service instance with appropriate settings.

    Delegates to rag_service_factory.get_rag_service() with the current
    Flask session username. For non-Flask contexts, use the factory directly.

    Args:
        collection_id: Optional collection UUID to load stored settings from
        use_defaults: When True, ignore stored collection settings and use
            current defaults. Pass True on force-reindex so that the new
            default embedding model is picked up.
    """
    from ..services.rag_service_factory import (
        get_rag_service as _get_rag_service,
    )
    from ...database.session_passwords import session_password_store

    username = session["username"]
    session_id = session.get("session_id")
    db_password = None
    if session_id:
        db_password = session_password_store.get_session_password(
            username, session_id
        )
    return _get_rag_service(
        username,
        collection_id,
        use_defaults=use_defaults,
        db_password=db_password,
    )


def _get_text_separators(settings):
    """Return configured text separators, parsing string values if needed."""
    # Load the stored setting. It may already be a list or a JSON string.
    text_separators = settings.get_setting(
        "local_search_text_separators",
        DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
    )

    # If the setting is stored as a string, parse it into a list. A value that
    # is not valid JSON (e.g. a not-yet-migrated corrupt row) falls back to the
    # default separators — migration #4298 heals existing corrupt data.
    if isinstance(text_separators, str):
        try:
            text_separators = json.loads(text_separators)
        except json.JSONDecodeError:
            logger.warning(
                "Invalid JSON for local_search_text_separators setting: {!r}. Using default separators.",
                text_separators,
            )
            text_separators = DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS

    return text_separators


# Module prefix for exceptions DEFINED inside LDR itself. An exception whose
# class lives under ``local_deep_research`` is something only we can fix, so we
# steer the user to file a bug instead of second-guessing their model choice.
#
# We deliberately do NOT treat bare builtins (KeyError/TypeError/RuntimeError/
# ...) as internal: ``type(exc).__module__`` is where the class is *defined*,
# not where it was *raised*, so a builtin tells us nothing about whether LDR or
# a provider/langchain frame produced it. On this path an OpenAI-compatible
# server that returns a malformed 200 body can surface a bare TypeError from
# langchain's response parser; telling the user to "report it on GitHub" for
# that is exactly the misleading guidance #4208 set out to remove.
_INTERNAL_MODULE_PREFIX = "local_deep_research"

# Module prefixes whose exceptions are upstream-provider errors (network,
# auth, model not found, etc.).  Matched against ``type(exc).__module__``.
_UPSTREAM_MODULE_PREFIXES = (
    "openai",
    "httpx",
    "requests",
    "urllib3",
    "anthropic",
)


def _module_matches(type_module: str, prefix: str) -> bool:
    """True if ``type_module`` is ``prefix`` itself or one of its submodules."""
    return type_module == prefix or type_module.startswith(f"{prefix}.")


def _format_test_embedding_error(exc: Exception, model: str) -> str:
    """Build the user-facing error message for /api/rag/test-embedding.

    Categorizes the exception by the module its *class* is defined in so the
    UI distinguishes LDR-internal bugs (which the user can't fix and
    shouldn't be steered into changing their model over) from real
    upstream/provider errors. The previous implementation ran a keyword-match
    heuristic over every exception text — see #4208, where
    ``NoSettingsContextError`` surfaced as a misleading "try a dedicated
    embedding model" suggestion.

    The detail text is scrubbed with :func:`sanitize_error_message` first
    because an upstream error can echo an API key back in a URL or auth
    header, and this endpoint returns the detail to the browser.
    """
    raw_message = str(exc).strip() or type(exc).__name__
    detail = sanitize_error_message(raw_message)
    type_name = type(exc).__name__
    type_module = type(exc).__module__ or ""

    if _module_matches(type_module, _INTERNAL_MODULE_PREFIX):
        # Internal LDR exceptions can carry filesystem paths / SQL fragments
        # that sanitize_error_message() does not pattern-match, so we do not
        # echo their detail to the browser — the full trace is in the server
        # logs (logger.exception at the call site) which the user is asked to
        # attach.
        return (
            f"Embedding test failed for model '{model}' due to an "
            f"internal LDR error ({type_name}). This is a bug in LDR, "
            "not your configuration. Please report it on GitHub with "
            "the server logs."
        )

    if any(
        _module_matches(type_module, prefix)
        for prefix in _UPSTREAM_MODULE_PREFIXES
    ):
        return (
            f"Embedding test failed for model '{model}'. The provider "
            f"returned an error: {detail}"
        )

    return f"Embedding test failed for model '{model}': {detail}"


# Config API Routes


@rag_bp.route("/api/config/supported-formats", methods=["GET"])
@login_required
def get_supported_formats():
    """Return list of supported file formats for upload.

    This endpoint provides the single source of truth for supported file
    extensions, pulling from the document_loaders registry. The UI can
    use this to dynamically update the file input accept attribute.
    """
    from ...document_loaders import get_supported_extensions

    extensions = get_supported_extensions()
    # Sort extensions for consistent display
    extensions = sorted(extensions)

    return jsonify(
        {
            "extensions": extensions,
            "accept_string": ",".join(extensions),
            "count": len(extensions),
        }
    )


# Page Routes


@rag_bp.route("/embedding-settings")
@login_required
def embedding_settings_page():
    """Render the Embedding Settings page."""
    return render_template(
        "pages/embedding_settings.html", active_page="embedding-settings"
    )


@rag_bp.route("/document/<string:document_id>/chunks")
@login_required
def view_document_chunks(document_id):
    """View all chunks for a document across all collections."""
    from ...database.session_context import get_user_db_session
    from ...database.models.library import DocumentChunk

    username = session["username"]

    with get_user_db_session(username) as db_session:
        # Get document info
        document = db_session.query(Document).filter_by(id=document_id).first()

        if not document:
            return "Document not found", 404

        # Get all chunks for this document
        chunks = (
            db_session.query(DocumentChunk)
            .filter(DocumentChunk.source_id == document_id)
            .order_by(DocumentChunk.collection_name, DocumentChunk.chunk_index)
            .all()
        )

        # Group chunks by collection
        chunks_by_collection = {}
        for chunk in chunks:
            coll_name = chunk.collection_name
            if coll_name not in chunks_by_collection:
                # Get collection display name
                collection_id = coll_name.replace("collection_", "")
                collection = (
                    db_session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )
                chunks_by_collection[coll_name] = {
                    "name": collection.name if collection else coll_name,
                    "id": collection_id,
                    "chunks": [],
                }

            chunks_by_collection[coll_name]["chunks"].append(
                {
                    "id": chunk.id,
                    "index": chunk.chunk_index,
                    "text": chunk.chunk_text,
                    "word_count": chunk.word_count,
                    "start_char": chunk.start_char,
                    "end_char": chunk.end_char,
                    "embedding_model": chunk.embedding_model,
                    "embedding_model_type": chunk.embedding_model_type.value
                    if chunk.embedding_model_type
                    else None,
                    "embedding_dimension": chunk.embedding_dimension,
                    "created_at": chunk.created_at,
                }
            )

        return render_template(
            "pages/document_chunks.html",
            document=document,
            chunks_by_collection=chunks_by_collection,
            total_chunks=len(chunks),
        )


@rag_bp.route("/collections")
@login_required
def collections_page():
    """Render the Collections page."""
    return render_template("pages/collections.html", active_page="collections")


@rag_bp.route("/collections/<string:collection_id>")
@login_required
def collection_details_page(collection_id):
    """Render the Collection Details page."""
    return render_template(
        "pages/collection_details.html",
        active_page="collections",
        collection_id=collection_id,
    )


@rag_bp.route("/collections/<string:collection_id>/upload")
@login_required
def collection_upload_page(collection_id):
    """Render the Collection Upload page."""
    # Get the upload PDF storage setting
    settings = get_settings_manager()
    upload_pdf_storage = settings.get_setting(
        "research_library.upload_pdf_storage", "none"
    )
    # Only allow valid values for uploads (no filesystem)
    if upload_pdf_storage not in ("database", "none"):
        upload_pdf_storage = "none"

    return render_template(
        "pages/collection_upload.html",
        active_page="collections",
        collection_id=collection_id,
        collection_name=None,  # Could fetch from DB if needed
        upload_pdf_storage=upload_pdf_storage,
    )


@rag_bp.route("/collections/create")
@login_required
def collection_create_page():
    """Render the Create Collection page."""
    return render_template(
        "pages/collection_create.html", active_page="collections"
    )


# API Routes


@rag_bp.route("/api/rag/settings", methods=["GET"])
@login_required
def get_current_settings():
    """Get current RAG configuration from settings."""

    try:
        settings = get_settings_manager()

        return jsonify(
            {
                "success": True,
                "settings": {
                    "embedding_provider": settings.get_setting(
                        "local_search_embedding_provider",
                        "sentence_transformers",
                    ),
                    "embedding_model": settings.get_setting(
                        "local_search_embedding_model", "all-MiniLM-L6-v2"
                    ),
                    "chunk_size": settings.get_setting(
                        "local_search_chunk_size", 1000
                    ),
                    "chunk_overlap": settings.get_setting(
                        "local_search_chunk_overlap", 200
                    ),
                    "splitter_type": settings.get_setting(
                        "local_search_splitter_type", "recursive"
                    ),
                    "text_separators": _get_text_separators(settings),
                    "distance_metric": settings.get_setting(
                        "local_search_distance_metric", "cosine"
                    ),
                    "normalize_vectors": settings.get_setting(
                        "local_search_normalize_vectors", True
                    ),
                    "index_type": settings.get_setting(
                        "local_search_index_type", "flat"
                    ),
                },
            }
        )
    except Exception as e:
        return handle_api_error("getting RAG settings", e)


@rag_bp.route("/api/rag/test-embedding", methods=["POST"])
@login_required
@require_json_body(error_format="success")
def test_embedding():
    """Test an embedding configuration by generating a test embedding."""

    try:
        data = request.get_json()
        provider = data.get("provider")
        model = data.get("model")
        test_text = data.get("test_text", "This is a test.")

        if not provider or not model:
            return jsonify(
                {"success": False, "error": "Provider and model are required"}
            ), 400

        # Import embedding functions
        from ...embeddings.embeddings_config import (
            get_embedding_function,
        )

        logger.info(
            f"Testing embedding with provider={provider}, model={model}"
        )

        # Get user's settings so provider URLs (e.g. Ollama) are resolved correctly
        settings = get_settings_manager()
        settings_snapshot = (
            settings.get_all_settings()
            if hasattr(settings, "get_all_settings")
            else {}
        )

        # Get embedding function with the specified configuration
        start_time = time.time()
        embedding_func = get_embedding_function(
            provider=provider,
            model_name=model,
            settings_snapshot=settings_snapshot,
        )

        # Generate test embedding
        embedding = embedding_func([test_text])[0]
        response_time_ms = int((time.time() - start_time) * 1000)

        # Get embedding dimension
        dimension = len(embedding) if hasattr(embedding, "__len__") else None

        return jsonify(
            {
                "success": True,
                "dimension": dimension,
                "response_time_ms": response_time_ms,
                "provider": provider,
                "model": model,
            }
        )

    except Exception as e:
        # Egress policy denial → clean 4xx (the embedding never fired).
        from ...security.egress.policy import PolicyDeniedError

        if isinstance(e, PolicyDeniedError):
            return jsonify(
                {
                    "success": False,
                    "error": (
                        f"Embedding provider '{provider}' refused by egress "
                        f"policy ({e.decision.reason}). Disable 'Require "
                        "local embeddings' or pick a local provider."
                    ),
                }
            ), 400
        logger.exception("Error during testing embedding")
        user_message = _format_test_embedding_error(e, model)
        return jsonify({"success": False, "error": user_message}), 500


@rag_bp.route("/api/rag/models", methods=["GET"])
@login_required
def get_available_models():
    """Get list of available embedding providers and models."""
    try:
        from ...embeddings.embeddings_config import _get_provider_classes

        # Get current settings for providers
        settings = get_settings_manager()
        settings_snapshot = (
            settings.get_all_settings()
            if hasattr(settings, "get_all_settings")
            else {}
        )

        # Get provider classes
        provider_classes = _get_provider_classes()

        # Egress policy: fetching a cloud provider's model list opens a
        # connection that carries the API key off-machine. Under an effective
        # local-only embeddings posture (PRIVATE_ONLY / adaptive-private) we
        # must NOT probe remote-classified providers — mirror the PEP in
        # embeddings_config.get_embeddings(). Build the run context once and
        # fail closed (skip remote model fetches) if the policy can't evaluate.
        def _embeddings_model_fetch_allowed(provider_key: str) -> bool:
            try:
                from ...security.egress.policy import (
                    context_from_snapshot,
                    evaluate_embeddings,
                )
                from ...config.thread_settings import (
                    get_setting_from_snapshot,
                )

                primary = (
                    get_setting_from_snapshot(
                        "search.tool",
                        default=DEFAULT_SEARCH_TOOL,
                        settings_snapshot=settings_snapshot,
                    )
                    or DEFAULT_SEARCH_TOOL
                )
                ctx = context_from_snapshot(settings_snapshot, primary)
                if not ctx.require_local_embeddings:
                    return True
                decision = evaluate_embeddings(
                    provider_key, ctx, settings_snapshot=settings_snapshot
                )
                if not decision.allowed:
                    logger.bind(policy_audit=True).info(
                        "skipping embeddings model-list probe "
                        "(local-only egress posture)",
                        provider=provider_key,
                        reason=decision.reason,
                    )
                return decision.allowed
            except Exception:
                # Fail closed: a policy/snapshot error must not open a cloud
                # probe under a local-only posture.
                logger.bind(policy_audit=True).warning(
                    "embeddings model-list egress check failed; skipping probe",
                    provider=provider_key,
                    exc_info=True,
                )
                return False

        # Provider display names
        provider_labels = {
            "sentence_transformers": "Sentence Transformers (Local)",
            "ollama": "Ollama (Local)",
            "openai": "OpenAI API",
        }

        # Get provider options and models by looping through providers
        provider_options = []
        providers = {}

        for provider_key, provider_class in provider_classes.items():
            available = provider_class.is_available(settings_snapshot)

            # Always show the provider in the dropdown so users can
            # configure its settings (e.g. fix a wrong Ollama URL).
            provider_options.append(
                {
                    "value": provider_key,
                    "label": provider_labels.get(provider_key, provider_key),
                    "available": available,
                }
            )

            # Only fetch models when the provider is reachable AND egress
            # policy permits probing it (cloud providers are skipped under a
            # local-only posture).
            if available and _embeddings_model_fetch_allowed(provider_key):
                models = provider_class.get_available_models(settings_snapshot)
                providers[provider_key] = [
                    {
                        "value": m["value"],
                        "label": m["label"],
                        "provider": provider_key,
                        **(
                            {"is_embedding": m["is_embedding"]}
                            if "is_embedding" in m
                            else {}
                        ),
                    }
                    for m in models
                ]
            else:
                providers[provider_key] = []

        return jsonify(
            {
                "success": True,
                "provider_options": provider_options,
                "providers": providers,
            }
        )
    except Exception as e:
        return handle_api_error("getting available models", e)


@rag_bp.route("/api/rag/info", methods=["GET"])
@login_required
def get_index_info():
    """Get information about the current RAG index."""
    from ...database.library_init import get_default_library_id

    try:
        # Get collection_id from request or use default Library collection
        collection_id = request.args.get("collection_id")
        if not collection_id:
            collection_id = get_default_library_id(session["username"])

        logger.info(
            f"Getting RAG index info for collection_id: {collection_id}"
        )

        with get_rag_service(collection_id) as rag_service:
            info = rag_service.get_current_index_info(collection_id)

        if info is None:
            logger.info(
                f"No RAG index found for collection_id: {collection_id}"
            )
            return jsonify(
                {"success": True, "info": None, "message": "No index found"}
            )

        logger.info(f"Found RAG index for collection_id: {collection_id}")
        return jsonify({"success": True, "info": info})
    except Exception as e:
        return handle_api_error("getting index info", e)


@rag_bp.route("/api/rag/stats", methods=["GET"])
@login_required
def get_rag_stats():
    """Get RAG statistics for a collection."""
    from ...database.library_init import get_default_library_id

    try:
        # Get collection_id from request or use default Library collection
        collection_id = request.args.get("collection_id")
        if not collection_id:
            collection_id = get_default_library_id(session["username"])

        with get_rag_service(collection_id) as rag_service:
            stats = rag_service.get_rag_stats(collection_id)

        return jsonify({"success": True, "stats": stats})
    except Exception as e:
        return handle_api_error("getting RAG stats", e)


@rag_bp.route("/api/rag/index-document", methods=["POST"])
@login_required
@require_json_body(error_format="success")
def index_document():
    """Index a single document in a collection."""
    from ...database.library_init import get_default_library_id

    try:
        data = request.get_json()
        text_doc_id = data.get("text_doc_id")
        force_reindex = data.get("force_reindex", False)
        collection_id = data.get("collection_id")

        if not text_doc_id:
            return jsonify(
                {"success": False, "error": "text_doc_id is required"}
            ), 400

        # Get collection_id from request or use default Library collection
        if not collection_id:
            collection_id = get_default_library_id(session["username"])

        with get_rag_service(collection_id) as rag_service:
            result = rag_service.index_document(
                text_doc_id, collection_id, force_reindex
            )

        if result["status"] == "error":
            return jsonify(
                {"success": False, "error": result.get("error")}
            ), 400

        return jsonify({"success": True, "result": result})
    except Exception as e:
        return handle_api_error(f"indexing document {text_doc_id}", e)


@rag_bp.route("/api/rag/remove-document", methods=["POST"])
@login_required
@require_json_body(error_format="success")
def remove_document():
    """Remove a document from RAG in a collection."""
    from ...database.library_init import get_default_library_id

    try:
        data = request.get_json()
        text_doc_id = data.get("text_doc_id")
        collection_id = data.get("collection_id")

        if not text_doc_id:
            return jsonify(
                {"success": False, "error": "text_doc_id is required"}
            ), 400

        # Get collection_id from request or use default Library collection
        if not collection_id:
            collection_id = get_default_library_id(session["username"])

        with get_rag_service(collection_id) as rag_service:
            result = rag_service.remove_document_from_rag(
                text_doc_id, collection_id
            )

        if result["status"] == "error":
            return jsonify(
                {"success": False, "error": result.get("error")}
            ), 400

        return jsonify({"success": True, "result": result})
    except Exception as e:
        return handle_api_error(f"removing document {text_doc_id}", e)


@rag_bp.route("/api/rag/index-research", methods=["POST"])
@login_required
@require_json_body(error_format="success")
def index_research():
    """Index all documents from a research."""
    try:
        data = request.get_json()
        research_id = data.get("research_id")
        force_reindex = data.get("force_reindex", False)

        if not research_id:
            return jsonify(
                {"success": False, "error": "research_id is required"}
            ), 400

        with get_rag_service() as rag_service:
            results = rag_service.index_research_documents(
                research_id, force_reindex
            )

        return jsonify({"success": True, "results": results})
    except Exception as e:
        return handle_api_error(f"indexing research {research_id}", e)


@rag_bp.route("/api/rag/index-all", methods=["GET"])
@login_required
def index_all():
    """Index all documents in a collection with Server-Sent Events progress."""
    from ...database.session_context import get_user_db_session
    from ...database.library_init import get_default_library_id

    force_reindex = parse_bool_arg("force_reindex")
    username = session["username"]

    # Get collection_id from request or use default Library collection
    collection_id = request.args.get("collection_id")
    if not collection_id:
        collection_id = get_default_library_id(username)

    logger.info(
        f"Starting index-all for collection_id: {collection_id}, force_reindex: {force_reindex}"
    )

    # Create RAG service in request context before generator runs.
    # use_defaults=force_reindex mirrors index_collection / the background
    # worker so a force-reindex resolves the same embedding config.
    rag_service = get_rag_service(collection_id, use_defaults=force_reindex)

    def generate():
        """Generator function for SSE progress updates."""
        try:
            # Send initial status
            yield f"data: {json.dumps({'type': 'start', 'message': 'Starting bulk indexing...'})}\n\n"

            # Get document IDs to index from DocumentCollection
            with get_user_db_session(username) as db_session:
                # Persist embedding metadata + clean up on force-reindex via the
                # shared helpers, so this bulk route behaves like the
                # single-collection SSE route and the background worker
                # (previously it stored no embedding_dimension and never reset
                # stale chunks/indices on force-reindex).
                collection = (
                    db_session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )
                if collection is not None and (
                    collection.embedding_model is None or force_reindex
                ):
                    _store_collection_embedding_metadata(
                        collection, rag_service
                    )
                    db_session.commit()
                if force_reindex:
                    _reset_collection_for_reindex(db_session, collection_id)
                    db_session.commit()

                doc_info = [
                    (doc.id, doc.title)
                    for _dc, doc in _query_documents_to_index(
                        db_session, collection_id, force_reindex
                    )
                ]

            if not doc_info:
                yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
                return

            results = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
            total = len(doc_info)

            # Process documents in batches to optimize performance
            # Get batch size from settings
            settings = get_settings_manager()
            batch_size = int(
                settings.get_setting("rag.indexing_batch_size", 15)
            )
            processed = 0

            for i in range(0, len(doc_info), batch_size):
                batch = doc_info[i : i + batch_size]

                # Process batch with collection_id
                batch_results = rag_service.index_documents_batch(
                    batch, collection_id, force_reindex
                )

                # Process results and send progress updates
                for j, (doc_id, title) in enumerate(batch):
                    processed += 1
                    result = batch_results[doc_id]

                    # Send progress update
                    yield f"data: {json.dumps({'type': 'progress', 'current': processed, 'total': total, 'title': title, 'percent': int((processed / total) * 100)})}\n\n"

                    if result["status"] == "success":
                        results["successful"] += 1
                    elif result["status"] == "skipped":
                        results["skipped"] += 1
                    else:
                        results["failed"] += 1
                        results["errors"].append(
                            {
                                "doc_id": doc_id,
                                "title": title,
                                "error": result.get("error"),
                            }
                        )

            # Send completion status
            yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"

            # Log final status for debugging
            logger.info(
                f"Bulk indexing complete: {results['successful']} successful, {results['skipped']} skipped, {results['failed']} failed"
            )

        except Exception:
            logger.exception("Error in bulk indexing")
            yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
        finally:
            # Generator ``finally`` runs at stream completion (or client
            # disconnect via ``GeneratorExit``) — the safe place to release
            # the RAG service's embedding-manager httpx clients. Closing at
            # the outer route scope would tear it down before the streamed
            # generator runs. ``safe_close`` swallows close-time errors so
            # a broken Ollama doesn't mask the original generator outcome.
            safe_close(rag_service, "rag_service (index-all SSE)")

    return Response(
        stream_with_context(generate()), mimetype="text/event-stream"
    )


@rag_bp.route("/api/rag/configure", methods=["POST"])
@login_required
@require_json_body(error_format="success")
def configure_rag():
    """
    Change RAG configuration (embedding model, chunk size, etc.).
    This will create a new index with the new configuration.
    """
    import json as json_lib

    try:
        data = request.get_json()
        embedding_model = data.get("embedding_model")
        embedding_provider = data.get("embedding_provider")
        chunk_size = data.get("chunk_size")
        chunk_overlap = data.get("chunk_overlap")
        collection_id = data.get("collection_id")

        # Get new advanced settings (with defaults)
        splitter_type = data.get("splitter_type", "recursive")
        text_separators = data.get(
            "text_separators", ["\n\n", "\n", ". ", " ", ""]
        )
        distance_metric = data.get("distance_metric", "cosine")
        normalize_vectors = data.get("normalize_vectors", True)
        index_type = data.get("index_type", "flat")

        if not all(
            [
                embedding_model,
                embedding_provider,
                chunk_size,
                chunk_overlap,
            ]
        ):
            return jsonify(
                {
                    "success": False,
                    "error": "All configuration parameters are required (embedding_model, embedding_provider, chunk_size, chunk_overlap)",
                }
            ), 400

        # Save settings to database
        settings = get_settings_manager()
        settings.set_setting("local_search_embedding_model", embedding_model)
        settings.set_setting(
            "local_search_embedding_provider", embedding_provider
        )
        settings.set_setting("local_search_chunk_size", int(chunk_size))
        settings.set_setting("local_search_chunk_overlap", int(chunk_overlap))

        # Save new advanced settings
        settings.set_setting("local_search_splitter_type", splitter_type)

        # The setting is registered with ui_element "json", so store the
        # separators as a proper list. A string payload (e.g. a textarea
        # value) is parsed into a list first; if it is not valid JSON it is
        # stored as-is so the UI validation layer can flag it.
        if isinstance(text_separators, str):
            try:
                text_separators = json_lib.loads(text_separators)
            except json_lib.JSONDecodeError:
                logger.debug(
                    "Invalid JSON for text_separators input; storing raw string for validation"
                )
        settings.set_setting("local_search_text_separators", text_separators)
        settings.set_setting("local_search_distance_metric", distance_metric)
        settings.set_setting(
            "local_search_normalize_vectors", bool(normalize_vectors)
        )
        settings.set_setting("local_search_index_type", index_type)

        # If collection_id is provided, update that collection's configuration
        if collection_id:
            # Create new RAG service with new configuration
            with LibraryRAGService(
                username=session["username"],
                embedding_model=embedding_model,
                embedding_provider=embedding_provider,
                chunk_size=int(chunk_size),
                chunk_overlap=int(chunk_overlap),
                splitter_type=splitter_type,
                text_separators=text_separators
                if isinstance(text_separators, list)
                else DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
                distance_metric=distance_metric,
                normalize_vectors=normalize_vectors,
                index_type=index_type,
            ) as new_rag_service:
                # Get or create new index with this configuration
                rag_index = new_rag_service._get_or_create_rag_index(
                    collection_id
                )

                return jsonify(
                    {
                        "success": True,
                        "message": "Configuration updated for collection. You can now index documents with the new settings.",
                        "index_hash": rag_index.index_hash,
                    }
                )
        else:
            # Just saving default settings without updating a specific collection
            return jsonify(
                {
                    "success": True,
                    "message": "Default embedding settings saved successfully. New collections will use these settings.",
                }
            )

    except Exception as e:
        return handle_api_error("configuring RAG", e)


@rag_bp.route("/api/rag/documents", methods=["GET"])
@login_required
def get_documents():
    """Get library documents with their RAG status for the default Library collection (paginated)."""
    from ...database.session_context import get_user_db_session
    from ...database.library_init import get_default_library_id

    try:
        # Get pagination parameters
        page = request.args.get("page", 1, type=int)
        per_page = request.args.get("per_page", 50, type=int)
        filter_type = request.args.get(
            "filter", "all"
        )  # all, indexed, unindexed

        # Validate pagination parameters
        page = max(1, page)
        per_page = min(max(10, per_page), 100)  # Limit between 10-100

        # Close current thread's session to force fresh connection
        from ...database.thread_local_session import cleanup_current_thread

        cleanup_current_thread()

        username = session["username"]

        # Get collection_id from request or use default Library collection
        collection_id = request.args.get("collection_id")
        if not collection_id:
            collection_id = get_default_library_id(username)

        logger.info(
            f"Getting documents for collection_id: {collection_id}, filter: {filter_type}, page: {page}"
        )

        with get_user_db_session(username) as db_session:
            # Expire all cached objects to ensure we get fresh data from DB
            db_session.expire_all()

            # Import RagDocumentStatus model
            from ...database.models.library import RagDocumentStatus

            # Build base query - join Document with DocumentCollection for the collection
            # LEFT JOIN with rag_document_status to check indexed status
            query = (
                db_session.query(
                    Document, DocumentCollection, RagDocumentStatus
                )
                .join(
                    DocumentCollection,
                    (DocumentCollection.document_id == Document.id)
                    & (DocumentCollection.collection_id == collection_id),
                )
                .outerjoin(
                    RagDocumentStatus,
                    (RagDocumentStatus.document_id == Document.id)
                    & (RagDocumentStatus.collection_id == collection_id),
                )
            )

            logger.debug(f"Base query for collection {collection_id}: {query}")

            # Apply filters based on rag_document_status existence
            if filter_type == "indexed":
                query = query.filter(RagDocumentStatus.document_id.isnot(None))
            elif filter_type == "unindexed":
                # Documents in collection but not indexed yet
                query = query.filter(RagDocumentStatus.document_id.is_(None))

            # Get total count before pagination
            total_count = query.count()
            logger.info(
                f"Found {total_count} total documents for collection {collection_id} with filter {filter_type}"
            )

            # Apply pagination
            results = (
                query.order_by(Document.created_at.desc())
                .limit(per_page)
                .offset((page - 1) * per_page)
                .all()
            )

            documents = [
                {
                    "id": doc.id,
                    "title": doc.title,
                    "original_url": doc.original_url,
                    "rag_indexed": rag_status is not None,
                    "chunk_count": rag_status.chunk_count if rag_status else 0,
                    "created_at": doc.created_at.isoformat()
                    if doc.created_at
                    else None,
                }
                for doc, doc_collection, rag_status in results
            ]

            # Debug logging to help diagnose indexing status issues
            indexed_count = sum(1 for d in documents if d["rag_indexed"])

            # Additional debug: check rag_document_status for this collection
            all_indexed_statuses = (
                db_session.query(RagDocumentStatus)
                .filter_by(collection_id=collection_id)
                .all()
            )
            logger.info(
                f"rag_document_status table shows: {len(all_indexed_statuses)} documents indexed for collection {collection_id}"
            )

            logger.info(
                f"Returning {len(documents)} documents on page {page}: "
                f"{indexed_count} indexed, {len(documents) - indexed_count} not indexed"
            )

        return jsonify(
            {
                "success": True,
                "documents": documents,
                "pagination": {
                    "page": page,
                    "per_page": per_page,
                    "total": total_count,
                    "pages": (total_count + per_page - 1) // per_page,
                },
            }
        )
    except Exception as e:
        return handle_api_error("getting documents", e)


@rag_bp.route("/api/rag/index-local", methods=["GET"])
@login_required
def index_local_library():
    """Index documents from a local folder with Server-Sent Events progress."""
    folder_path = request.args.get("path")
    raw_patterns = request.args.get(
        "patterns", "*.pdf,*.txt,*.md,*.html"
    ).split(",")
    file_patterns = [p.strip() for p in raw_patterns if p.strip()]
    recursive = parse_bool_arg("recursive", default=True)

    if not folder_path:
        return jsonify({"success": False, "error": "Path is required"}), 400

    # Reject glob patterns that could escape the validated base directory.
    # The folder is validated below, but each pattern is joined onto it and
    # globbed; a pattern with ".." or a leading-slash absolute pattern would
    # otherwise read files outside the folder. Only in-directory filename
    # globs are allowed.
    if not file_patterns or not all(
        _is_safe_glob_pattern(p) for p in file_patterns
    ):
        return jsonify({"success": False, "error": "Invalid file pattern"}), 400

    # Validate and sanitize the path to prevent traversal attacks
    try:
        validated_path = PathValidator.validate_local_filesystem_path(
            folder_path
        )
        # Re-sanitize for static analyzer recognition (CodeQL)
        path = PathValidator.sanitize_for_filesystem_ops(validated_path)
    except ValueError:
        # Don't log the raw user-submitted path — it can contain a username.
        # A static message keeps operator visibility without leaking the input
        # (project policy disallows exception variables in logger.warning).
        logger.warning("Path validation failed")
        return jsonify({"success": False, "error": "Invalid path"}), 400

    # Check path exists and is a directory
    if not path.exists():
        return jsonify({"success": False, "error": "Path does not exist"}), 400
    if not path.is_dir():
        return jsonify(
            {"success": False, "error": "Path is not a directory"}
        ), 400

    # Create RAG service in request context
    rag_service = get_rag_service()

    def generate():
        """Generator function for SSE progress updates."""
        try:
            # Send initial status
            yield f"data: {json.dumps({'type': 'start', 'message': f'Scanning folder: {path}'})}\n\n"

            # Find all matching files
            files_to_index = []
            for pattern in file_patterns:
                pattern = pattern.strip()
                if recursive:
                    search_pattern = str(path / "**" / pattern)
                else:
                    search_pattern = str(path / pattern)

                matching_files = glob.glob(search_pattern, recursive=recursive)
                files_to_index.extend(matching_files)

            # Remove duplicates and sort
            files_to_index = sorted(set(files_to_index))

            if not files_to_index:
                yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No matching files found'}})}\n\n"
                return

            results = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
            total = len(files_to_index)

            # Index each file
            for idx, file_path in enumerate(files_to_index, 1):
                file_name = Path(file_path).name

                # Send progress update
                yield f"data: {json.dumps({'type': 'progress', 'current': idx, 'total': total, 'filename': file_name, 'percent': int((idx / total) * 100)})}\n\n"

                try:
                    # Index the file directly using RAG service
                    result = rag_service.index_local_file(file_path)

                    if result.get("status") == "success":
                        results["successful"] += 1
                    elif result.get("status") == "skipped":
                        results["skipped"] += 1
                    else:
                        results["failed"] += 1
                        results["errors"].append(
                            {
                                "file": file_name,
                                "error": result.get("error", "Unknown error"),
                            }
                        )
                except Exception:
                    results["failed"] += 1
                    results["errors"].append(
                        {"file": file_name, "error": "Failed to index file"}
                    )
                    # Log the file name only — the full path can contain a
                    # username.
                    logger.exception(f"Error indexing file {file_name}")

            # Send completion status
            yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"

            # Log the folder name only — the full path can contain a username.
            logger.info(
                f"Local library indexing complete for {path.name}: "
                f"{results['successful']} successful, "
                f"{results['skipped']} skipped, "
                f"{results['failed']} failed"
            )

        except Exception:
            logger.exception("Error in local library indexing")
            yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
        finally:
            # See parallel comment in index-all generator above — generator
            # ``finally`` is the safe close site for streamed RAG services.
            safe_close(rag_service, "rag_service (index-local SSE)")

    return Response(
        stream_with_context(generate()), mimetype="text/event-stream"
    )


# Collection Management Routes


@rag_bp.route("/api/collections", methods=["GET"])
@login_required
def get_collections():
    """Get all document collections for the current user."""
    from ...database.session_context import get_user_db_session

    try:
        username = session["username"]
        with get_user_db_session(username) as db_session:
            # No need to filter by username - each user has their own database
            collections = db_session.query(Collection).all()

            result = []
            for coll in collections:
                collection_data = {
                    "id": coll.id,
                    "name": coll.name,
                    "description": coll.description,
                    "created_at": coll.created_at.isoformat()
                    if coll.created_at
                    else None,
                    "collection_type": coll.collection_type,
                    "is_default": coll.is_default
                    if hasattr(coll, "is_default")
                    else False,
                    "is_public": bool(getattr(coll, "is_public", False)),
                    "agent_enabled": _agent_enabled_default_on(coll),
                    "document_count": len(coll.document_links)
                    if hasattr(coll, "document_links")
                    else 0,
                    "folder_count": len(coll.linked_folders)
                    if hasattr(coll, "linked_folders")
                    else 0,
                }

                # Include embedding metadata if available
                if coll.embedding_model:
                    collection_data["embedding"] = {
                        "model": coll.embedding_model,
                        "provider": coll.embedding_model_type.value
                        if coll.embedding_model_type
                        else None,
                        "dimension": coll.embedding_dimension,
                        "chunk_size": coll.chunk_size,
                        "chunk_overlap": coll.chunk_overlap,
                    }
                else:
                    collection_data["embedding"] = None

                result.append(collection_data)

        return jsonify({"success": True, "collections": result})
    except Exception as e:
        return handle_api_error("getting collections", e)


@rag_bp.route("/api/collections", methods=["POST"])
@login_required
@require_json_body(error_format="success")
def create_collection():
    """Create a new document collection."""
    from ...database.session_context import get_user_db_session

    try:
        data = request.get_json()
        name = data.get("name", "").strip()
        description = data.get("description", "").strip()
        collection_type = data.get("type", "user_uploads")
        # Egress classification, default private (the safe choice). A public
        # collection counts as a public engine and may use cloud inference.
        is_public = bool(data.get("is_public", False))
        # Usability switch (NOT egress): offer this collection to the research
        # agent? Default True (available) — behaviour-preserving. An explicit
        # JSON null normalizes to True so it matches how NULL is read back
        # everywhere else (NULL → available); only an explicit false disables.
        agent_enabled_raw = data.get("agent_enabled", True)
        agent_enabled = (
            True if agent_enabled_raw is None else bool(agent_enabled_raw)
        )

        if not name:
            return jsonify({"success": False, "error": "Name is required"}), 400

        username = session["username"]
        with get_user_db_session(username) as db_session:
            # Check if collection with this name already exists in this user's database
            existing = db_session.query(Collection).filter_by(name=name).first()

            if existing:
                return jsonify(
                    {
                        "success": False,
                        "error": f"Collection '{name}' already exists",
                    }
                ), 400

            # Create new collection (no username needed - each user has their own DB)
            # Note: created_at uses default=utcnow() in the model, so we don't need to set it manually
            collection = Collection(
                id=str(uuid.uuid4()),  # Generate UUID for collection
                name=name,
                description=description,
                collection_type=collection_type,
                is_public=is_public,
                agent_enabled=agent_enabled,
            )

            db_session.add(collection)
            db_session.commit()

            return jsonify(
                {
                    "success": True,
                    "collection": {
                        "id": collection.id,
                        "name": collection.name,
                        "description": collection.description,
                        "created_at": collection.created_at.isoformat(),
                        "collection_type": collection.collection_type,
                        "is_public": bool(collection.is_public),
                        "agent_enabled": _agent_enabled_default_on(collection),
                    },
                }
            )
    except Exception as e:
        return handle_api_error("creating collection", e)


@rag_bp.route("/api/collections/<string:collection_id>", methods=["PUT"])
@login_required
@require_json_body(error_format="success")
def update_collection(collection_id):
    """Update a collection's details."""
    from ...database.session_context import get_user_db_session

    try:
        data = request.get_json()
        name = data.get("name", "").strip()
        description = data.get("description", "").strip()

        username = session["username"]
        with get_user_db_session(username) as db_session:
            # No need to filter by username - each user has their own database
            collection = (
                db_session.query(Collection).filter_by(id=collection_id).first()
            )

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

            if name:
                # Check if new name conflicts with existing collection
                existing = (
                    db_session.query(Collection)
                    .filter(
                        Collection.name == name,
                        Collection.id != collection_id,
                    )
                    .first()
                )

                if existing:
                    return jsonify(
                        {
                            "success": False,
                            "error": f"Collection '{name}' already exists",
                        }
                    ), 400

                collection.name = name

            if description is not None:  # Allow empty description
                collection.description = description

            # Egress classification toggle (only when the caller sends it).
            if "is_public" in data:
                collection.is_public = bool(data.get("is_public"))

            # Research-agent availability toggle (only when the caller sends it).
            # Explicit null normalizes to True (available) for parity with how
            # NULL is read back elsewhere; only an explicit false disables.
            if "agent_enabled" in data:
                raw = data.get("agent_enabled")
                collection.agent_enabled = True if raw is None else bool(raw)

            db_session.commit()

            return jsonify(
                {
                    "success": True,
                    "collection": {
                        "id": collection.id,
                        "name": collection.name,
                        "description": collection.description,
                        "created_at": collection.created_at.isoformat()
                        if collection.created_at
                        else None,
                        "collection_type": collection.collection_type,
                        "is_public": bool(collection.is_public),
                        "agent_enabled": _agent_enabled_default_on(collection),
                    },
                }
            )
    except Exception as e:
        return handle_api_error("updating collection", e)


@rag_bp.route(
    "/api/collections/<string:collection_id>/upload", methods=["POST"]
)
@login_required
@upload_rate_limit_user
@upload_rate_limit_ip
def upload_to_collection(collection_id):
    """Upload files to a collection."""
    from ...database.session_context import get_user_db_session
    from ...security import sanitize_filename, UnsafeFilenameError
    from pathlib import Path
    import hashlib
    import uuid
    from ..services.pdf_storage_manager import PDFStorageManager

    try:
        if "files" not in request.files:
            return jsonify(
                {"success": False, "error": "No files provided"}
            ), 400

        files = request.files.getlist("files")
        if not files:
            return jsonify(
                {"success": False, "error": "No files selected"}
            ), 400

        # Bound the per-request file count BEFORE doing any work. The
        # request-level MAX_CONTENT_LENGTH gate covers total bytes, but
        # not file *count*; a request with 10000 zero-byte files would
        # otherwise reach the loop below.
        is_valid, error_msg = FileUploadValidator.validate_file_count(
            len(files)
        )
        if not is_valid:
            return jsonify({"success": False, "error": error_msg}), 400

        username = session["username"]
        with get_user_db_session(username) as db_session:
            # Verify collection exists in this user's database
            collection = (
                db_session.query(Collection).filter_by(id=collection_id).first()
            )

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

            # Get PDF storage mode from form data, falling back to user's setting
            settings = get_settings_manager()
            default_pdf_storage = settings.get_setting(
                "research_library.upload_pdf_storage", "none"
            )
            pdf_storage = request.form.get("pdf_storage", default_pdf_storage)
            if pdf_storage not in ("database", "none"):
                # Security: user uploads can only use database (encrypted) or none (text-only)
                # Filesystem storage is not allowed for user uploads
                pdf_storage = "none"

            # Initialize PDF storage manager if storing PDFs in database
            pdf_storage_manager = None
            if pdf_storage == "database":
                library_root = settings.get_setting(
                    "research_library.storage_path",
                    str(get_library_directory()),
                )
                library_root = str(
                    Path(os.path.expandvars(library_root))
                    .expanduser()
                    .resolve()
                )
                pdf_storage_manager = PDFStorageManager(
                    library_root=Path(library_root), storage_mode="database"
                )
                logger.info("PDF storage mode: database (encrypted)")
            else:
                logger.info("PDF storage mode: none (text-only)")

            uploaded_files = []
            errors = []

            for file in files:
                if not file.filename:
                    continue

                try:
                    filename = sanitize_filename(file.filename)
                except UnsafeFilenameError:
                    errors.append(
                        {
                            "filename": "rejected",
                            "error": "Invalid or unsafe filename",
                        }
                    )
                    continue

                try:
                    # Pre-flight size check on Content-Length BEFORE reading
                    # bytes into memory. Cheap rejection for oversized files;
                    # avoids loading 50MB+ into memory just to discard it.
                    is_valid, error_msg = (
                        FileUploadValidator.validate_file_size(
                            content_length=file.content_length,
                            file_content=None,
                        )
                    )
                    if not is_valid:
                        errors.append(
                            {"filename": filename, "error": error_msg}
                        )
                        continue

                    # Read file content
                    file_content = file.read()
                    file.seek(0)  # Reset for potential re-reading

                    # Post-read size check (Content-Length can be missing or
                    # spoofed; the actual byte count is authoritative).
                    is_valid, error_msg = (
                        FileUploadValidator.validate_file_size(
                            content_length=None,
                            file_content=file_content,
                        )
                    )
                    if not is_valid:
                        errors.append(
                            {"filename": filename, "error": error_msg}
                        )
                        continue

                    # Calculate file hash for deduplication
                    file_hash = hashlib.sha256(file_content).hexdigest()

                    # Check if document already exists
                    existing_doc = (
                        db_session.query(Document)
                        .filter_by(document_hash=file_hash)
                        .first()
                    )

                    if existing_doc:
                        # Document exists, check if we can upgrade to include PDF
                        pdf_upgraded = False
                        if (
                            pdf_storage == "database"
                            and pdf_storage_manager is not None
                        ):
                            # NOTE: Only the PDF magic-byte check is needed here.
                            # File count validation is already handled by Flask's MAX_CONTENT_LENGTH.
                            # Filename sanitization already happens via sanitize_filename() above.
                            # See PR #3145 review for details.
                            if file_content[:4] != b"%PDF":
                                logger.debug(
                                    "Skipping PDF upgrade for {}: not a PDF file",
                                    filename,
                                )
                            else:
                                pdf_upgraded = (
                                    pdf_storage_manager.upgrade_to_pdf(
                                        document=existing_doc,
                                        pdf_content=file_content,
                                        session=db_session,
                                    )
                                )

                        # Check if already in collection
                        existing_link = (
                            db_session.query(DocumentCollection)
                            .filter_by(
                                document_id=existing_doc.id,
                                collection_id=collection_id,
                            )
                            .first()
                        )

                        if not existing_link:
                            ensure_in_collection(
                                db_session, existing_doc.id, collection_id
                            )
                            status = "added_to_collection"
                            if pdf_upgraded:
                                status = "added_to_collection_pdf_upgraded"
                            uploaded_files.append(
                                {
                                    "filename": existing_doc.filename,
                                    "status": status,
                                    "id": existing_doc.id,
                                    "pdf_upgraded": pdf_upgraded,
                                }
                            )
                        else:
                            status = "already_in_collection"
                            if pdf_upgraded:
                                status = "pdf_upgraded"
                            uploaded_files.append(
                                {
                                    "filename": existing_doc.filename,
                                    "status": status,
                                    "id": existing_doc.id,
                                    "pdf_upgraded": pdf_upgraded,
                                }
                            )
                    else:
                        # Create new document
                        from ...document_loaders import (
                            extract_text_from_bytes,
                            is_extension_supported,
                        )

                        file_extension = Path(filename).suffix.lower()

                        # Validate extension is supported before extraction
                        if not is_extension_supported(file_extension):
                            errors.append(
                                {
                                    "filename": filename,
                                    "error": f"Unsupported format: {file_extension}",
                                }
                            )
                            continue

                        # Use file_type without leading dot for storage
                        file_type = (
                            file_extension[1:]
                            if file_extension.startswith(".")
                            else file_extension
                        )

                        # Extract text using document_loaders module
                        extracted_text = extract_text_from_bytes(
                            file_content, file_extension, filename
                        )

                        # Clean the extracted text to remove surrogate characters
                        if extracted_text:
                            from ...text_processing import remove_surrogates

                            extracted_text = remove_surrogates(extracted_text)

                        if not extracted_text:
                            errors.append(
                                {
                                    "filename": filename,
                                    "error": f"Could not extract text from {file_type} file",
                                }
                            )
                            logger.warning(
                                f"Skipping file {filename} - no text could be extracted"
                            )
                            continue

                        # Get or create the user_upload source type
                        logger.info(
                            f"Getting or creating user_upload source type for {filename}"
                        )
                        source_type = (
                            db_session.query(SourceType)
                            .filter_by(name="user_upload")
                            .first()
                        )
                        if not source_type:
                            logger.info("Creating new user_upload source type")
                            source_type = SourceType(
                                id=str(uuid.uuid4()),
                                name="user_upload",
                                display_name="User Upload",
                                description="Documents uploaded by users",
                                icon="fas fa-upload",
                            )
                            db_session.add(source_type)
                            db_session.flush()
                            logger.info(
                                f"Created source type with ID: {source_type.id}"
                            )
                        else:
                            logger.info(
                                f"Found existing source type with ID: {source_type.id}"
                            )

                        # Create document with extracted text (no username needed - in user's own database)
                        # Note: uploaded_at uses default=utcnow() in the model, so we don't need to set it manually
                        doc_id = str(uuid.uuid4())
                        logger.info(
                            f"Creating document {doc_id} for {filename}"
                        )

                        # Determine storage mode and file_path
                        store_pdf_in_db = (
                            pdf_storage == "database"
                            and file_type == "pdf"
                            and pdf_storage_manager is not None
                        )

                        new_doc = Document(
                            id=doc_id,
                            source_type_id=source_type.id,
                            filename=filename,
                            document_hash=file_hash,
                            file_size=len(file_content),
                            file_type=file_type,
                            text_content=extracted_text,  # Always store extracted text
                            file_path=None
                            if store_pdf_in_db
                            else FILE_PATH_TEXT_ONLY,
                            storage_mode="database"
                            if store_pdf_in_db
                            else "none",
                        )
                        db_session.add(new_doc)
                        db_session.flush()  # Get the ID
                        logger.info(
                            f"Document {new_doc.id} created successfully"
                        )

                        # Store PDF in encrypted database if requested
                        pdf_stored = False
                        if store_pdf_in_db:
                            try:
                                pdf_storage_manager.save_pdf(
                                    pdf_content=file_content,
                                    document=new_doc,
                                    session=db_session,
                                    filename=filename,
                                )
                                pdf_stored = True
                                logger.info(
                                    f"PDF stored in encrypted database for {filename}"
                                )
                            except Exception:
                                logger.exception(
                                    f"Failed to store PDF in database for {filename}"
                                )
                                # Continue without PDF storage - text is still saved

                        # Add to collection
                        ensure_in_collection(
                            db_session, new_doc.id, collection_id
                        )

                        uploaded_files.append(
                            {
                                "filename": filename,
                                "status": "uploaded",
                                "id": new_doc.id,
                                "text_length": len(extracted_text),
                                "pdf_stored": pdf_stored,
                            }
                        )

                except Exception:
                    errors.append(
                        {
                            "filename": filename,
                            "error": "Failed to upload file",
                        }
                    )
                    logger.exception(f"Error uploading file {filename}")

            db_session.commit()

            # Trigger auto-indexing for successfully uploaded documents
            document_ids = [
                f["id"]
                for f in uploaded_files
                if f.get("status") in ("uploaded", "added_to_collection")
            ]
            if document_ids:
                from ...database.session_passwords import session_password_store

                session_id = session.get("session_id")
                db_password = session_password_store.get_session_password(
                    username, session_id
                )
                if db_password:
                    trigger_auto_index(
                        document_ids, collection_id, username, db_password
                    )

            return jsonify(
                {
                    "success": True,
                    "uploaded": uploaded_files,
                    "errors": errors,
                    "summary": {
                        "total": len(files),
                        "successful": len(uploaded_files),
                        "failed": len(errors),
                    },
                }
            )

    except Exception as e:
        return handle_api_error("uploading files", e)


@rag_bp.route(
    "/api/collections/<string:collection_id>/documents", methods=["GET"]
)
@login_required
def get_collection_documents(collection_id):
    """Get all documents in a collection."""
    from ...database.session_context import get_user_db_session

    try:
        username = session["username"]
        with get_user_db_session(username) as db_session:
            # Verify collection exists in this user's database
            collection = (
                db_session.query(Collection).filter_by(id=collection_id).first()
            )

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

            # Get documents through junction table. Compute "has text in DB"
            # at the SQL level and defer the text_content column, so listing a
            # collection doesn't pull every document's full text body into
            # memory just to test it for truthiness (#4560).
            has_text_col = (
                Document.text_content.isnot(None)
                & (Document.text_content != "")
            ).label("has_text_db")
            doc_links = (
                db_session.query(DocumentCollection, Document, has_text_col)
                .join(Document)
                .options(defer(Document.text_content))
                .filter(DocumentCollection.collection_id == collection_id)
                .all()
            )

            documents = []
            for link, doc, has_text_db in doc_links:
                # Check if PDF file is stored
                has_pdf = bool(
                    doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
                )
                has_text_db = bool(has_text_db)

                # Use title if available, otherwise filename
                display_title = doc.title or doc.filename or "Untitled"

                # Get source type name
                source_type_name = (
                    doc.source_type.name if doc.source_type else "unknown"
                )

                # Check if document is in other collections
                other_collections_count = (
                    db_session.query(DocumentCollection)
                    .filter(
                        DocumentCollection.document_id == doc.id,
                        DocumentCollection.collection_id != collection_id,
                    )
                    .count()
                )

                documents.append(
                    {
                        "id": doc.id,
                        "filename": display_title,
                        "title": display_title,
                        "file_type": doc.file_type,
                        "file_size": doc.file_size,
                        "uploaded_at": doc.created_at.isoformat()
                        if doc.created_at
                        else None,
                        "indexed": link.indexed,
                        "chunk_count": link.chunk_count,
                        "last_indexed_at": link.last_indexed_at.isoformat()
                        if link.last_indexed_at
                        else None,
                        "has_pdf": has_pdf,
                        "has_text_db": has_text_db,
                        "source_type": source_type_name,
                        "in_other_collections": other_collections_count > 0,
                        "other_collections_count": other_collections_count,
                    }
                )

            # Get index file size if available
            index_file_size = None
            index_file_size_bytes = None
            collection_name = f"collection_{collection_id}"
            rag_index = (
                db_session.query(RAGIndex)
                .filter_by(collection_name=collection_name)
                .first()
            )
            if rag_index and rag_index.index_path:
                from pathlib import Path

                index_path = Path(rag_index.index_path)
                if index_path.exists():
                    size_bytes = index_path.stat().st_size
                    index_file_size_bytes = size_bytes
                    # Format as human-readable
                    if size_bytes < 1024:
                        index_file_size = f"{size_bytes} B"
                    elif size_bytes < 1024 * 1024:
                        index_file_size = f"{size_bytes / 1024:.1f} KB"
                    else:
                        index_file_size = f"{size_bytes / (1024 * 1024):.1f} MB"

            return jsonify(
                {
                    "success": True,
                    "collection": {
                        "id": collection.id,
                        "name": collection.name,
                        "description": collection.description,
                        "is_public": bool(
                            getattr(collection, "is_public", False)
                        ),
                        "agent_enabled": _agent_enabled_default_on(collection),
                        "embedding_model": collection.embedding_model,
                        "embedding_model_type": collection.embedding_model_type.value
                        if collection.embedding_model_type
                        else None,
                        "embedding_dimension": collection.embedding_dimension,
                        "chunk_size": collection.chunk_size,
                        "chunk_overlap": collection.chunk_overlap,
                        # Advanced settings
                        "splitter_type": collection.splitter_type,
                        "distance_metric": collection.distance_metric,
                        "index_type": collection.index_type,
                        "normalize_vectors": collection.normalize_vectors,
                        # Index file info
                        "index_file_size": index_file_size,
                        "index_file_size_bytes": index_file_size_bytes,
                        "collection_type": collection.collection_type,
                    },
                    "documents": documents,
                }
            )

    except Exception as e:
        return handle_api_error("getting collection documents", e)


# =============================================================================
# Shared collection-indexing helpers
#
# Three routes index a collection: the single-collection SSE route
# (index_collection), the bulk SSE route (index_all), and the background worker
# (_background_index_worker). These helpers hold the logic that must stay
# identical across all three so it cannot drift again — the embedding metadata
# to persist, the force-reindex cleanup, and the query for documents to index.
# The callers differ only in how they report progress (SSE stream vs
# TaskMetadata) and so keep their own loops.
# =============================================================================


def _store_collection_embedding_metadata(collection, rag_service):
    """Persist the embedding/index configuration used to index a collection.

    Run on first index or force-reindex. Includes the embedding dimension,
    probed by embedding a test string (the same way LibraryRAGService derives
    it for the RAGIndex row). Previously this read
    ``embedding_manager.provider.embedding_dimension``, an attribute the real
    LocalEmbeddingManager does not have, so the probe always failed and
    ``embedding_dimension`` was persisted as NULL via both indexing paths. Does
    not commit — the caller owns the transaction.
    """
    embedding_dim = None
    try:
        # Derive the dimension by embedding a test string — mirrors
        # LibraryRAGService.* which computes len(embed_query("test")) for the
        # RAGIndex row. The embedding manager has no usable dimension attribute.
        test_embedding = rag_service.embedding_manager.embeddings.embed_query(
            "test"
        )
        embedding_dim = len(test_embedding)
    except Exception:
        logger.debug("Could not determine embedding dimension", exc_info=True)

    collection.embedding_model = rag_service.embedding_model
    collection.embedding_model_type = EmbeddingProvider(
        rag_service.embedding_provider
    )
    collection.embedding_dimension = embedding_dim
    collection.chunk_size = rag_service.chunk_size
    collection.chunk_overlap = rag_service.chunk_overlap
    # Advanced settings
    collection.splitter_type = rag_service.splitter_type
    collection.text_separators = rag_service.text_separators
    collection.distance_metric = rag_service.distance_metric
    # Ensure normalize_vectors is a proper boolean for the database
    collection.normalize_vectors = bool(rag_service.normalize_vectors)
    collection.index_type = rag_service.index_type
    logger.info(
        f"Stored embedding metadata for collection: provider={rag_service.embedding_provider}"
    )


def _reset_collection_for_reindex(db_session, collection_id):
    """Clear old chunks, FAISS indices, and indexed-state before a force rebuild.

    Prevents mixed-model vectors / stale chunks when re-indexing (e.g. if a
    previous run was cancelled midway). Shared so a force-reindex behaves the
    same whether started via the SSE route or the background worker. Does not
    commit — the caller owns the transaction.
    """
    # Import stays function-local: hoisting it would pull in the whole
    # deletion package (deletion/__init__ imports its services) at
    # route-module import time for a path only exercised on force-reindex.
    from ..deletion.utils.cascade_helper import CascadeHelper

    collection_name = f"collection_{collection_id}"

    # Delete all old document chunks from DB
    deleted_chunks = CascadeHelper.delete_collection_chunks(
        db_session, collection_name
    )
    logger.info(
        f"Cleared {deleted_chunks} old chunks for collection {collection_id}"
    )

    # Delete old FAISS index files (.faiss + .pkl) and RAGIndex records.
    # RagDocumentStatus rows cascade-delete via FK ondelete="CASCADE".
    rag_result = CascadeHelper.delete_rag_indices_for_collection(
        db_session, collection_name
    )
    logger.info(
        f"Cleared old RAG indices for collection {collection_id}: {rag_result}"
    )

    # Mark all documents as unindexed
    db_session.query(DocumentCollection).filter_by(
        collection_id=collection_id
    ).update(
        {
            DocumentCollection.indexed: False,
            DocumentCollection.chunk_count: 0,
        }
    )
    logger.info(f"Reset indexing state for collection {collection_id}")


def _query_documents_to_index(db_session, collection_id, force_reindex):
    """Return the ``(DocumentCollection, Document)`` pairs to index.

    When not force-reindexing, only un-indexed documents are returned.
    """
    query = (
        db_session.query(DocumentCollection, Document)
        .join(Document)
        # Don't load the large text_content body: the indexing loop only
        # reads doc.id/filename/title (index_document re-fetches by id in a
        # worker thread), so loading every document's full text here just to
        # iterate is what exhausts memory on big collections (#4560).
        .options(defer(Document.text_content))
        .filter(DocumentCollection.collection_id == collection_id)
    )

    if not force_reindex:
        # IMPORTANT: DO NOT CHANGE THIS LINE! DO NOT REMOVE THIS COMMENT!
        # We MUST use `== False` instead of `not DocumentCollection.indexed`
        # The Python `not` operator does NOT work correctly in SQLAlchemy filters.
        # Using `not` will cause the query to return NO results (zero documents).
        # SQLAlchemy requires explicit comparison: `== False` or `== True`
        # This has been fixed multiple times - DO NOT change it back to `not`!
        query = query.filter(DocumentCollection.indexed == False)  # noqa: E712

    return query.all()


@rag_bp.route("/api/collections/<string:collection_id>/index", methods=["GET"])
@login_required
def index_collection(collection_id):
    """Index all documents in a collection with Server-Sent Events progress."""
    from ...database.session_context import get_user_db_session
    from ...database.session_passwords import session_password_store

    force_reindex = parse_bool_arg("force_reindex")
    username = session["username"]
    session_id = session.get("session_id")

    logger.info(f"Starting index_collection, force_reindex={force_reindex}")

    # Get password for thread access to encrypted database
    db_password = None
    if session_id:
        db_password = session_password_store.get_session_password(
            username, session_id
        )

    # Create RAG service — on force reindex use current default model
    rag_service = get_rag_service(collection_id, use_defaults=force_reindex)
    logger.info(
        f"RAG service created: provider={rag_service.embedding_provider}"
    )

    def generate():
        """Generator for SSE progress updates."""
        logger.info("SSE generator started")
        try:
            with get_user_db_session(username, db_password) as db_session:
                # Verify collection exists in this user's database
                collection = (
                    db_session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )

                if not collection:
                    yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
                    return

                # Store embedding metadata on first index or force reindex
                if collection.embedding_model is None or force_reindex:
                    _store_collection_embedding_metadata(
                        collection, rag_service
                    )
                    db_session.commit()

                # Clean up old index data for a fresh rebuild (prevents stale /
                # mixed-model vectors). Shared with the background worker.
                if force_reindex:
                    _reset_collection_for_reindex(db_session, collection_id)
                    db_session.commit()

                # Get documents to index
                doc_links = _query_documents_to_index(
                    db_session, collection_id, force_reindex
                )

                if not doc_links:
                    logger.info("No documents to index in collection")
                    yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
                    return

                total = len(doc_links)
                logger.info(f"Found {total} documents to index")
                results = {
                    "successful": 0,
                    "skipped": 0,
                    "failed": 0,
                    "errors": [],
                }

                yield f"data: {json.dumps({'type': 'start', 'message': f'Indexing {total} documents in collection: {collection.name}'})}\n\n"

                for idx, (link, doc) in enumerate(doc_links, 1):
                    filename = doc.filename or doc.title or "Unknown"
                    yield f"data: {json.dumps({'type': 'progress', 'current': idx, 'total': total, 'filename': filename, 'percent': int((idx / total) * 100)})}\n\n"

                    try:
                        logger.debug(
                            f"Indexing document {idx}/{total}: {filename}"
                        )

                        # Run index_document in a separate thread to allow sending SSE heartbeats.
                        # This keeps the HTTP connection alive during long indexing operations,
                        # preventing timeouts from proxy servers (nginx) and browsers.
                        # The main thread periodically yields heartbeat comments while waiting.
                        result_queue = queue.Queue()
                        error_queue = queue.Queue()

                        def index_in_thread():
                            try:
                                r = rag_service.index_document(
                                    document_id=doc.id,
                                    collection_id=collection_id,
                                    force_reindex=force_reindex,
                                )
                                result_queue.put(r)
                            except Exception as ex:
                                error_queue.put(ex)
                            finally:
                                try:
                                    from ...database.thread_local_session import (
                                        cleanup_current_thread,
                                    )

                                    cleanup_current_thread()
                                except Exception:
                                    logger.debug(
                                        "best-effort thread-local DB session cleanup",
                                        exc_info=True,
                                    )

                        thread = threading.Thread(target=index_in_thread)
                        thread.start()

                        # Send heartbeats while waiting for the thread to complete
                        heartbeat_interval = 5  # seconds
                        while thread.is_alive():
                            thread.join(timeout=heartbeat_interval)
                            if thread.is_alive():
                                # Send SSE comment as heartbeat (keeps connection alive)
                                yield f": heartbeat {idx}/{total}\n\n"

                        # Check for errors from thread
                        if not error_queue.empty():
                            raise error_queue.get()  # noqa: TRY301 — re-raises thread exception for per-document error handling

                        result = result_queue.get()
                        logger.info(
                            f"Indexed document {idx}/{total}: {filename} - status={result.get('status')}"
                        )

                        if result.get("status") == "success":
                            results["successful"] += 1
                            # DocumentCollection status is already updated in index_document
                            # No need to update link here
                        elif result.get("status") == "skipped":
                            results["skipped"] += 1
                        else:
                            results["failed"] += 1
                            error_msg = result.get("error", "Unknown error")
                            results["errors"].append(
                                {
                                    "filename": filename,
                                    "error": error_msg,
                                }
                            )
                            logger.warning(
                                f"Failed to index {filename} ({idx}/{total}): {error_msg}"
                            )
                    except Exception as e:
                        results["failed"] += 1
                        # Scrub creds/keys before surfacing to the browser
                        # (SSE yield below); full trace stays in server logs.
                        error_msg = (
                            sanitize_error_message(str(e))
                            or "Failed to index document"
                        )
                        results["errors"].append(
                            {
                                "filename": filename,
                                "error": error_msg,
                            }
                        )
                        logger.exception(
                            f"Exception indexing document {filename} ({idx}/{total})"
                        )
                        # Send error update to client so they know indexing is continuing
                        yield f"data: {json.dumps({'type': 'doc_error', 'filename': filename, 'error': error_msg})}\n\n"

                db_session.commit()
                # Ensure all changes are written to disk
                db_session.flush()

            logger.info(
                f"Indexing complete: {results['successful']} successful, {results['failed']} failed, {results['skipped']} skipped"
            )
            yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
            logger.info("SSE generator finished successfully")

        except Exception:
            logger.exception("Error in collection indexing")
            yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
        finally:
            # See parallel comment in index-all generator above — generator
            # ``finally`` is the safe close site for streamed RAG services.
            safe_close(rag_service, "rag_service (index-collection SSE)")

    response = Response(
        stream_with_context(generate()), mimetype="text/event-stream"
    )
    # Prevent buffering for proper SSE streaming
    response.headers["Cache-Control"] = "no-cache, no-transform"
    response.headers["Connection"] = "keep-alive"
    response.headers["X-Accel-Buffering"] = "no"
    return response


# =============================================================================
# Background Indexing Endpoints
# =============================================================================


def _get_rag_service_for_thread(
    collection_id: str,
    username: str,
    db_password: str,
    use_defaults: bool = False,
) -> LibraryRAGService:
    """
    Create RAG service for use in background threads (no Flask context).

    Delegates settings resolution to the shared rag_service_factory, then
    propagates db_password to the embedding manager for thread-safe DB access.
    """
    from ..services.rag_service_factory import (
        get_rag_service as _get_rag_service,
    )

    service = _get_rag_service(
        username,
        collection_id,
        use_defaults=use_defaults,
        db_password=db_password,
    )
    # The factory passes db_password to LibraryRAGService, but __init__ stores
    # it in the backing field (_db_password) without propagating to sub-managers.
    # Re-assign via the property setter to propagate to embedding_manager and
    # integrity_manager, which need it for thread-safe session access.
    service.db_password = db_password
    return service


def trigger_auto_index(
    document_ids: list[str],
    collection_id: str,
    username: str,
    db_password: str,
) -> None:
    """
    Trigger automatic RAG indexing for documents if auto-indexing is enabled.

    This function checks the auto_index_enabled setting and spawns a background
    thread to index the specified documents. It does not block the caller.

    Args:
        document_ids: List of document IDs to index
        collection_id: The collection to index into
        username: The username for database access
        db_password: The user's database password for thread-safe access
    """
    from ...database.session_context import get_user_db_session

    if not document_ids:
        logger.debug("No documents to auto-index")
        return

    # Check if auto-indexing is enabled
    try:
        with get_user_db_session(username, db_password) as db_session:
            settings = SettingsManager(db_session)
            auto_index_enabled = settings.get_bool_setting(
                "research_library.auto_index_enabled", True
            )

            if not auto_index_enabled:
                logger.debug("Auto-indexing is disabled, skipping")
                return
    except Exception:
        logger.exception(
            "Failed to check auto-index setting, skipping auto-index"
        )
        return

    # Reserve a queue slot before submission. If the indexing queue is
    # saturated (too many uploads in flight), drop this auto-index job
    # rather than letting the executor's unbounded internal queue grow
    # without bound. The user can reindex manually later via the UI.
    if not _try_reserve_auto_index_slot():
        logger.warning(
            "Auto-index queue saturated ({}+ jobs pending); dropping "
            "auto-index for {} document(s) in collection {}. "
            "Documents are uploaded; trigger a manual reindex if needed.",
            _MAX_PENDING_AUTO_INDEX_JOBS,
            len(document_ids),
            collection_id,
        )
        return

    logger.info(
        f"Auto-indexing {len(document_ids)} documents in collection {collection_id}"
    )

    # Release the reserved slot EXACTLY ONCE per submission. CPython's
    # ThreadPoolExecutor.submit() enqueues the work item BEFORE it tries to
    # spin up a worker thread; if thread-start raises (e.g.
    # RuntimeError("can't start new thread")), the item is already queued, so
    # a live worker can run _wrapped_worker (releasing in its finally) AND the
    # except block below also fires. A per-call lock + flag makes the release
    # idempotent so that double-fire only releases once — otherwise the
    # counter under-counts in-flight jobs and erodes the OOM bound (the
    # max(0, ...) floor would silently hide the drift).
    _release_lock = threading.Lock()
    _slot_released = {"done": False}

    def _release_slot_once():
        with _release_lock:
            if _slot_released["done"]:
                return
            _slot_released["done"] = True
        _release_auto_index_slot()

    def _wrapped_worker(*args, **kwargs):
        try:
            _auto_index_documents_worker(*args, **kwargs)
        finally:
            _release_slot_once()

    # Submit to thread pool (bounded concurrency, prevents thread proliferation).
    # Build the executor inside the try too: if _get_auto_index_executor()
    # itself fails (OS thread/mutex exhaustion), the slot must still be
    # released — otherwise a reserved slot leaks permanently.
    try:
        executor = _get_auto_index_executor()
        executor.submit(
            _wrapped_worker,
            document_ids,
            collection_id,
            username,
            db_password,
        )
    except Exception:
        # If submit itself fails (executor shutting down, OOM, etc.), the
        # wrapped worker may never run, so release the slot here. The release
        # is idempotent: if the work item was already enqueued and a worker
        # ran (and released) before thread-start failed, this is a no-op. Do
        # NOT re-raise: the upload has already been committed by the caller,
        # so propagating would turn a successful upload into a 500 (and prompt
        # the client to retry, creating duplicates). Auto-indexing is simply
        # skipped; the user can trigger a manual reindex later via the UI.
        _release_slot_once()
        logger.exception(
            "Failed to submit auto-index job for {} document(s) in "
            "collection {}; upload succeeded, auto-indexing skipped. "
            "Trigger a manual reindex if needed.",
            len(document_ids),
            collection_id,
        )


@thread_cleanup
def _auto_index_documents_worker(
    document_ids: list[str],
    collection_id: str,
    username: str,
    db_password: str,
) -> None:
    """
    Background worker to index documents automatically.

    This is a simpler worker than _background_index_worker - it doesn't track
    progress via TaskMetadata since it's meant to be a lightweight auto-indexing
    operation.
    """

    try:
        # Create RAG service (thread-safe, no Flask context needed)
        with _get_rag_service_for_thread(
            collection_id, username, db_password
        ) as rag_service:
            indexed_count = 0
            for doc_id in document_ids:
                try:
                    result = rag_service.index_document(
                        doc_id, collection_id, force_reindex=False
                    )
                    if result.get("status") == "success":
                        indexed_count += 1
                        logger.debug(f"Auto-indexed document {doc_id}")
                    elif result.get("status") == "skipped":
                        logger.debug(
                            f"Document {doc_id} already indexed, skipped"
                        )
                except Exception:
                    logger.exception(f"Failed to auto-index document {doc_id}")

            logger.info(
                f"Auto-indexing complete: {indexed_count}/{len(document_ids)} documents indexed"
            )

    except Exception:
        logger.exception("Auto-indexing worker failed")


@thread_cleanup
def _background_index_worker(
    task_id: str,
    collection_id: str,
    username: str,
    db_password: str,
    force_reindex: bool,
):
    """
    Background worker thread for indexing documents.
    Updates TaskMetadata with progress and checks for cancellation.
    """
    from ...database.session_context import get_user_db_session

    try:
        # Create RAG service (thread-safe, no Flask context needed)
        with _get_rag_service_for_thread(
            collection_id, username, db_password, use_defaults=force_reindex
        ) as rag_service:
            with get_user_db_session(username, db_password) as db_session:
                # Get collection
                collection = (
                    db_session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )

                if not collection:
                    _update_task_status(
                        username,
                        db_password,
                        task_id,
                        status="failed",
                        error_message="Collection not found",
                    )
                    return

                # Store embedding metadata on first index or force reindex
                if collection.embedding_model is None or force_reindex:
                    _store_collection_embedding_metadata(
                        collection, rag_service
                    )
                    db_session.commit()

                # Clean up old index data for a fresh rebuild.
                # This prevents mixed-model vectors if cancelled midway
                # and ensures accurate stats during partial reindex.
                if force_reindex:
                    _reset_collection_for_reindex(db_session, collection_id)
                    db_session.commit()

                # Get documents to index
                doc_links = _query_documents_to_index(
                    db_session, collection_id, force_reindex
                )

                if not doc_links:
                    _update_task_status(
                        username,
                        db_password,
                        task_id,
                        status="completed",
                        progress_message="No documents to index",
                    )
                    return

                total = len(doc_links)
                results = {"successful": 0, "skipped": 0, "failed": 0}

                # Update task with total count
                _update_task_status(
                    username,
                    db_password,
                    task_id,
                    progress_total=total,
                    progress_message=f"Indexing {total} documents",
                )

                for idx, (link, doc) in enumerate(doc_links, 1):
                    # Check if cancelled
                    if _is_task_cancelled(username, db_password, task_id):
                        _update_task_status(
                            username,
                            db_password,
                            task_id,
                            status="cancelled",
                            progress_message=f"Cancelled after {idx - 1}/{total} documents",
                        )
                        logger.info(f"Indexing task {task_id} was cancelled")
                        return

                    filename = doc.filename or doc.title or "Unknown"

                    # Update progress with filename
                    _update_task_status(
                        username,
                        db_password,
                        task_id,
                        progress_current=idx,
                        progress_message=f"Indexing {idx}/{total}: {filename}",
                    )

                    try:
                        result = rag_service.index_document(
                            document_id=doc.id,
                            collection_id=collection_id,
                            force_reindex=force_reindex,
                        )

                        if result.get("status") == "success":
                            results["successful"] += 1
                        elif result.get("status") == "skipped":
                            results["skipped"] += 1
                        else:
                            results["failed"] += 1

                    except Exception:
                        results["failed"] += 1
                        logger.exception(
                            f"Error indexing document {idx}/{total}"
                        )

                db_session.commit()

            # Mark as completed
            _update_task_status(
                username,
                db_password,
                task_id,
                status="completed",
                progress_current=total,
                progress_message=f"Completed: {results['successful']} indexed, {results['failed']} failed, {results['skipped']} skipped",
            )
            logger.info(
                f"Background indexing task {task_id} completed: {results}"
            )

    except Exception as e:
        logger.exception(f"Background indexing task {task_id} failed")
        _update_task_status(
            username,
            db_password,
            task_id,
            status="failed",
            # Scrub creds/keys: this is later returned to the client by the
            # index-status endpoint; full trace stays in server logs above.
            error_message=sanitize_error_message(str(e)),
        )


def _update_task_status(
    username: str,
    db_password: str,
    task_id: str,
    status: str = None,
    progress_current: int = None,
    progress_total: int = None,
    progress_message: str = None,
    error_message: str = None,
):
    """Update task metadata in the database."""
    from ...database.session_context import get_user_db_session

    try:
        with get_user_db_session(username, db_password) as db_session:
            task = (
                db_session.query(TaskMetadata)
                .filter_by(task_id=task_id)
                .first()
            )
            if task:
                if status is not None:
                    task.status = status
                    if status == "completed":
                        task.completed_at = datetime.now(UTC)
                if progress_current is not None:
                    task.progress_current = progress_current
                if progress_total is not None:
                    task.progress_total = progress_total
                if progress_message is not None:
                    task.progress_message = progress_message
                if error_message is not None:
                    task.error_message = error_message
                db_session.commit()
    except Exception:
        logger.exception(f"Failed to update task status for {task_id}")


def _is_task_cancelled(username: str, db_password: str, task_id: str) -> bool:
    """Check if a task has been cancelled."""
    from ...database.session_context import get_user_db_session

    try:
        with get_user_db_session(username, db_password) as db_session:
            task = (
                db_session.query(TaskMetadata)
                .filter_by(task_id=task_id)
                .first()
            )
            return task and task.status == "cancelled"
    except Exception:
        logger.warning(
            "Could not check cancellation status for task {}", task_id
        )
        return False


@rag_bp.route(
    "/api/collections/<string:collection_id>/index/start", methods=["POST"]
)
@login_required
def start_background_index(collection_id):
    """Start background indexing for a collection."""
    from ...database.session_context import get_user_db_session
    from ...database.session_passwords import session_password_store

    username = session["username"]
    session_id = session.get("session_id")

    # Get password for thread access
    db_password = None
    if session_id:
        db_password = session_password_store.get_session_password(
            username, session_id
        )

    # Parse request body
    data = request.get_json() or {}
    force_reindex = data.get("force_reindex", False)

    try:
        with get_user_db_session(username, db_password) as db_session:
            # Check if there's already an active indexing task for this collection
            existing_task = (
                db_session.query(TaskMetadata)
                .filter(
                    TaskMetadata.task_type == "indexing",
                    TaskMetadata.status == "processing",
                )
                .first()
            )

            if existing_task:
                # Check if it's for this collection
                metadata = existing_task.metadata_json or {}
                if metadata.get("collection_id") == collection_id:
                    return jsonify(
                        {
                            "success": False,
                            "error": "Indexing is already in progress for this collection",
                            "task_id": existing_task.task_id,
                        }
                    ), 409

            # Create new task
            task_id = str(uuid.uuid4())
            task = TaskMetadata(
                task_id=task_id,
                status="processing",
                task_type="indexing",
                created_at=datetime.now(UTC),
                started_at=datetime.now(UTC),
                progress_current=0,
                progress_total=0,
                progress_message="Starting indexing...",
                metadata_json={
                    "collection_id": collection_id,
                    "force_reindex": force_reindex,
                },
            )
            db_session.add(task)
            db_session.commit()

        # Start background thread
        thread = threading.Thread(
            target=_background_index_worker,
            args=(task_id, collection_id, username, db_password, force_reindex),
            daemon=True,
        )
        thread.start()

        logger.info(
            f"Started background indexing task {task_id} for collection {collection_id}"
        )

        return jsonify(
            {
                "success": True,
                "task_id": task_id,
                "message": "Indexing started in background",
            }
        )

    except Exception:
        logger.exception("Failed to start background indexing")
        return jsonify(
            {
                "success": False,
                "error": "Failed to start indexing. Please try again.",
            }
        ), 500


@rag_bp.route(
    "/api/collections/<string:collection_id>/index/status", methods=["GET"]
)
@limiter.exempt
@login_required
def get_index_status(collection_id):
    """Get the current indexing status for a collection."""
    from ...database.session_context import get_user_db_session
    from ...database.session_passwords import session_password_store

    username = session["username"]
    session_id = session.get("session_id")

    db_password = None
    if session_id:
        db_password = session_password_store.get_session_password(
            username, session_id
        )

    try:
        with get_user_db_session(username, db_password) as db_session:
            # Find the most recent indexing task for this collection
            task = (
                db_session.query(TaskMetadata)
                .filter(TaskMetadata.task_type == "indexing")
                .order_by(TaskMetadata.created_at.desc())
                .first()
            )

            if not task:
                return jsonify(
                    {
                        "status": "idle",
                        "message": "No indexing task found",
                    }
                )

            # Check if it's for this collection
            metadata = task.metadata_json or {}
            if metadata.get("collection_id") != collection_id:
                return jsonify(
                    {
                        "status": "idle",
                        "message": "No indexing task for this collection",
                    }
                )

            return jsonify(
                {
                    "task_id": task.task_id,
                    "status": task.status,
                    "progress_current": task.progress_current or 0,
                    "progress_total": task.progress_total or 0,
                    "progress_message": task.progress_message,
                    "error_message": task.error_message,
                    "created_at": task.created_at.isoformat()
                    if task.created_at
                    else None,
                    "completed_at": task.completed_at.isoformat()
                    if task.completed_at
                    else None,
                }
            )

    except Exception:
        logger.exception("Failed to get index status")
        return jsonify(
            {
                "status": "error",
                "error": "Failed to get indexing status. Please try again.",
            }
        ), 500


@rag_bp.route(
    "/api/collections/<string:collection_id>/index/cancel", methods=["POST"]
)
@login_required
def cancel_indexing(collection_id):
    """Cancel an active indexing task for a collection."""
    from ...database.session_context import get_user_db_session
    from ...database.session_passwords import session_password_store

    username = session["username"]
    session_id = session.get("session_id")

    db_password = None
    if session_id:
        db_password = session_password_store.get_session_password(
            username, session_id
        )

    try:
        with get_user_db_session(username, db_password) as db_session:
            # Find active indexing task for this collection
            task = (
                db_session.query(TaskMetadata)
                .filter(
                    TaskMetadata.task_type == "indexing",
                    TaskMetadata.status == "processing",
                )
                .first()
            )

            if not task:
                return jsonify(
                    {
                        "success": False,
                        "error": "No active indexing task found",
                    }
                ), 404

            # Check if it's for this collection
            metadata = task.metadata_json or {}
            if metadata.get("collection_id") != collection_id:
                return jsonify(
                    {
                        "success": False,
                        "error": "No active indexing task for this collection",
                    }
                ), 404

            # Mark as cancelled - the worker thread will check this
            task.status = "cancelled"
            task.progress_message = "Cancellation requested..."
            db_session.commit()

            logger.info(
                f"Cancelled indexing task {task.task_id} for collection {collection_id}"
            )

            return jsonify(
                {
                    "success": True,
                    "message": "Cancellation requested",
                    "task_id": task.task_id,
                }
            )

    except Exception:
        logger.exception("Failed to cancel indexing")
        return jsonify(
            {
                "success": False,
                "error": "Failed to cancel indexing. Please try again.",
            }
        ), 500


# Research History Semantic Search Routes have been moved to
# research_library.search.routes.search_routes
