Coverage for src/local_deep_research/research_library/routes/rag_routes.py: 93%
1057 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2RAG Management API Routes
4Provides endpoints for managing RAG indexing of library documents:
5- Configure embedding models
6- Index documents
7- Get RAG statistics
8- Bulk operations with progress tracking
9"""
11import os
13from flask import (
14 Blueprint,
15 jsonify,
16 request,
17 Response,
18 render_template,
19 session,
20 stream_with_context,
21)
22from loguru import logger
23from sqlalchemy import case, func
24import atexit
25import json
26import uuid
27import time
28import threading
29import queue
30from concurrent.futures import ThreadPoolExecutor
31from datetime import datetime, UTC
32from pathlib import Path
33from typing import Optional
35from sqlalchemy.orm import defer
37from ...constants import (
38 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
39 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
40 FILE_PATH_SENTINELS,
41 FILE_PATH_TEXT_ONLY,
42)
43from ...security.decorators import require_json_body
44from ...security.log_sanitizer import sanitize_error_message
45from ...web.auth.decorators import login_required
46from ...web.utils.request_helpers import parse_bool_arg
47from ...utilities.db_utils import get_settings_manager
48from ...utilities.resource_utils import safe_close
49from ..services.library_rag_service import LibraryRAGService
50from ...settings.manager import SettingsManager
51from ...security.file_upload_validator import FileUploadValidator
52from ...security.rate_limiter import (
53 upload_rate_limit_ip,
54 upload_rate_limit_user,
55)
56from ..utils import ensure_in_collection, handle_api_error
57from ...database.models.library import (
58 Document,
59 Collection,
60 DocumentCollection,
61 RAGIndex,
62 SourceType,
63 EmbeddingProvider,
64)
65from ...database.models.queue import TaskMetadata
66from ...database.thread_local_session import thread_cleanup
67from ...security.rate_limiter import limiter
68from ...config.paths import get_library_directory
69from ...constants import DEFAULT_SEARCH_TOOL
71rag_bp = Blueprint("rag", __name__, url_prefix="/library")
73# NOTE: Routes use session["username"] (not .get()) intentionally.
74# @login_required guarantees the key exists; direct access fails fast
75# if the decorator is ever removed.
78def _agent_enabled_default_on(collection) -> bool:
79 """Read a collection's ``agent_enabled`` flag with NULL → available.
81 A missing attribute (pre-migration object) or a stored NULL both mean
82 "available to the research agent" (default-on); only an explicit ``False``
83 disables. Used by every serializer so the four read sites can't drift on
84 null-handling. ``is not False`` is safe here because the source is a
85 Boolean column (only ever None/True/False) — NOT raw JSON input, where a
86 falsy ``0`` must still mean disabled (handled separately at the input edge).
87 """
88 return getattr(collection, "agent_enabled", True) is not False
91def _is_protected_collection(collection) -> bool:
92 """True when the collection's type is deletion-protected.
94 Single source of truth is PROTECTED_COLLECTION_TYPES in the deletion
95 service (imported lazily, matching update_collection's usage); the
96 serializers expose the result as ``is_protected`` so UI surfaces can
97 hide destructive affordances instead of offering an action the server
98 categorically refuses with a 409.
99 """
100 from ..deletion.services.collection_deletion import (
101 PROTECTED_COLLECTION_TYPES,
102 )
104 return collection.collection_type in PROTECTED_COLLECTION_TYPES
107# Global ThreadPoolExecutor for auto-indexing to prevent thread proliferation.
108#
109# ThreadPoolExecutor's internal _work_queue is unbounded. Without
110# backpressure, a sustained burst of uploads (possible under the
111# configurable upload rate cap from #3935) could queue thousands of indexing
112# jobs in memory, exhausting RAM. We track pending submissions in a counter
113# and reject new submissions once the queue is saturated.
114_auto_index_executor: ThreadPoolExecutor | None = None
115_auto_index_executor_lock = threading.Lock()
117# Maximum in-flight + queued indexing jobs. Each job is one upload batch
118# (an unbounded list of document IDs), so per-job duration varies with batch
119# size; 100 is a buffer on the number of *queued batches*, not documents.
120# The OOM bound still holds because a queued job holds only a small list of
121# ID strings, not document contents. Tunable if real-world workloads need
122# more headroom.
123_MAX_PENDING_AUTO_INDEX_JOBS = 100
124_pending_auto_index_jobs = 0
125_pending_auto_index_lock = threading.Lock()
128def _get_auto_index_executor() -> ThreadPoolExecutor:
129 """Get or create the global auto-indexing executor (thread-safe)."""
130 global _auto_index_executor
131 with _auto_index_executor_lock:
132 if _auto_index_executor is None:
133 _auto_index_executor = ThreadPoolExecutor(
134 max_workers=4,
135 thread_name_prefix="auto_index_",
136 )
137 return _auto_index_executor
140def _try_reserve_auto_index_slot() -> bool:
141 """Reserve a queue slot if capacity allows. Returns True on success.
143 Caller MUST call ``_release_auto_index_slot`` when the job completes
144 (success OR failure), otherwise the counter leaks and eventually
145 blocks all submissions.
146 """
147 global _pending_auto_index_jobs
148 with _pending_auto_index_lock:
149 if _pending_auto_index_jobs >= _MAX_PENDING_AUTO_INDEX_JOBS:
150 return False
151 _pending_auto_index_jobs += 1
152 return True
155def _release_auto_index_slot() -> None:
156 """Release a previously reserved queue slot."""
157 global _pending_auto_index_jobs
158 with _pending_auto_index_lock:
159 _pending_auto_index_jobs = max(0, _pending_auto_index_jobs - 1)
162def _shutdown_auto_index_executor() -> None:
163 """Shutdown the auto-index executor gracefully.
165 Holds ``_auto_index_executor_lock`` for the read+nullify so two
166 concurrent callers (e.g. atexit + a test teardown) can't race into an
167 ``AttributeError`` on the ``None`` reference; the blocking ``shutdown``
168 runs outside the lock. Resets the pending-jobs counter afterwards so a
169 re-created executor (tests, WSGI reload) starts from a clean count
170 instead of inheriting a stale, possibly-saturated value.
171 """
172 global _auto_index_executor, _pending_auto_index_jobs
173 with _auto_index_executor_lock:
174 executor = _auto_index_executor
175 _auto_index_executor = None
176 if executor is not None:
177 executor.shutdown(wait=True)
178 with _pending_auto_index_lock:
179 _pending_auto_index_jobs = 0
182atexit.register(_shutdown_auto_index_executor)
185def get_rag_service(
186 collection_id: Optional[str] = None,
187 use_defaults: bool = False,
188) -> LibraryRAGService:
189 """
190 Get RAG service instance with appropriate settings.
192 Delegates to rag_service_factory.get_rag_service() with the current
193 Flask session username. For non-Flask contexts, use the factory directly.
195 Args:
196 collection_id: Optional collection UUID to load stored settings from
197 use_defaults: When True, ignore stored collection settings and use
198 current defaults. Pass True on force-reindex so that the new
199 default embedding model is picked up.
200 """
201 from ..services.rag_service_factory import (
202 get_rag_service as _get_rag_service,
203 )
204 from ...database.session_passwords import session_password_store
206 username = session["username"]
207 session_id = session.get("session_id")
208 db_password = None
209 if session_id:
210 db_password = session_password_store.get_session_password(
211 username, session_id
212 )
213 return _get_rag_service(
214 username,
215 collection_id,
216 use_defaults=use_defaults,
217 db_password=db_password,
218 )
221def _get_text_separators(settings):
222 """Return configured text separators, parsing string values if needed."""
223 # Load the stored setting. It may already be a list or a JSON string.
224 text_separators = settings.get_setting(
225 "local_search_text_separators",
226 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
227 )
229 # If the setting is stored as a string, parse it into a list. A value that
230 # is not valid JSON (e.g. a not-yet-migrated corrupt row) falls back to the
231 # default separators — migration #4298 heals existing corrupt data.
232 if isinstance(text_separators, str):
233 try:
234 text_separators = json.loads(text_separators)
235 except json.JSONDecodeError:
236 logger.warning(
237 "Invalid JSON for local_search_text_separators setting: {!r}. Using default separators.",
238 text_separators,
239 )
240 text_separators = DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
242 return text_separators
245# Module prefix for exceptions DEFINED inside LDR itself. An exception whose
246# class lives under ``local_deep_research`` is something only we can fix, so we
247# steer the user to file a bug instead of second-guessing their model choice.
248#
249# We deliberately do NOT treat bare builtins (KeyError/TypeError/RuntimeError/
250# ...) as internal: ``type(exc).__module__`` is where the class is *defined*,
251# not where it was *raised*, so a builtin tells us nothing about whether LDR or
252# a provider/langchain frame produced it. On this path an OpenAI-compatible
253# server that returns a malformed 200 body can surface a bare TypeError from
254# langchain's response parser; telling the user to "report it on GitHub" for
255# that is exactly the misleading guidance #4208 set out to remove.
256_INTERNAL_MODULE_PREFIX = "local_deep_research"
258# Module prefixes whose exceptions are upstream-provider errors (network,
259# auth, model not found, etc.). Matched against ``type(exc).__module__``.
260_UPSTREAM_MODULE_PREFIXES = (
261 "openai",
262 "httpx",
263 "requests",
264 "urllib3",
265 "anthropic",
266)
269def _module_matches(type_module: str, prefix: str) -> bool:
270 """True if ``type_module`` is ``prefix`` itself or one of its submodules."""
271 return type_module == prefix or type_module.startswith(f"{prefix}.")
274def _format_test_embedding_error(exc: Exception, model: str) -> str:
275 """Build the user-facing error message for /api/rag/test-embedding.
277 Categorizes the exception by the module its *class* is defined in so the
278 UI distinguishes LDR-internal bugs (which the user can't fix and
279 shouldn't be steered into changing their model over) from real
280 upstream/provider errors. The previous implementation ran a keyword-match
281 heuristic over every exception text — see #4208, where
282 ``NoSettingsContextError`` surfaced as a misleading "try a dedicated
283 embedding model" suggestion.
285 The detail text is scrubbed with :func:`sanitize_error_message` first
286 because an upstream error can echo an API key back in a URL or auth
287 header, and this endpoint returns the detail to the browser.
288 """
289 raw_message = str(exc).strip() or type(exc).__name__
290 detail = sanitize_error_message(raw_message)
291 type_name = type(exc).__name__
292 type_module = type(exc).__module__ or ""
294 if _module_matches(type_module, _INTERNAL_MODULE_PREFIX):
295 # Internal LDR exceptions can carry filesystem paths / SQL fragments
296 # that sanitize_error_message() does not pattern-match, so we do not
297 # echo their detail to the browser — the full trace is in the server
298 # logs (logger.exception at the call site) which the user is asked to
299 # attach.
300 return (
301 f"Embedding test failed for model '{model}' due to an "
302 f"internal LDR error ({type_name}). This is a bug in LDR, "
303 "not your configuration. Please report it on GitHub with "
304 "the server logs."
305 )
307 if any(
308 _module_matches(type_module, prefix)
309 for prefix in _UPSTREAM_MODULE_PREFIXES
310 ):
311 return (
312 f"Embedding test failed for model '{model}'. The provider "
313 f"returned an error: {detail}"
314 )
316 return f"Embedding test failed for model '{model}': {detail}"
319# Config API Routes
322@rag_bp.route("/api/config/supported-formats", methods=["GET"])
323@login_required
324def get_supported_formats():
325 """Return list of supported file formats for upload.
327 This endpoint provides the single source of truth for supported file
328 extensions, pulling from the document_loaders registry. The UI can
329 use this to dynamically update the file input accept attribute.
330 """
331 from ...document_loaders import get_supported_extensions
333 extensions = get_supported_extensions()
334 # Sort extensions for consistent display
335 extensions = sorted(extensions)
337 return jsonify(
338 {
339 "extensions": extensions,
340 "accept_string": ",".join(extensions),
341 "count": len(extensions),
342 }
343 )
346# Page Routes
349@rag_bp.route("/embedding-settings")
350@login_required
351def embedding_settings_page():
352 """Render the Embedding Settings page."""
353 return render_template(
354 "pages/embedding_settings.html", active_page="embedding-settings"
355 )
358@rag_bp.route("/document/<string:document_id>/chunks")
359@login_required
360def view_document_chunks(document_id):
361 """View all chunks for a document across all collections."""
362 from ...database.session_context import get_user_db_session
363 from ...database.models.library import DocumentChunk
365 username = session["username"]
367 with get_user_db_session(username) as db_session:
368 # Get document info
369 document = db_session.query(Document).filter_by(id=document_id).first()
371 if not document:
372 return "Document not found", 404
374 # Get all chunks for this document
375 chunks = (
376 db_session.query(DocumentChunk)
377 .filter(DocumentChunk.source_id == document_id)
378 .order_by(DocumentChunk.collection_name, DocumentChunk.chunk_index)
379 .all()
380 )
382 # Group chunks by collection
383 chunks_by_collection = {}
384 for chunk in chunks:
385 coll_name = chunk.collection_name
386 if coll_name not in chunks_by_collection: 386 ↛ 400line 386 didn't jump to line 400 because the condition on line 386 was always true
387 # Get collection display name
388 collection_id = coll_name.replace("collection_", "")
389 collection = (
390 db_session.query(Collection)
391 .filter_by(id=collection_id)
392 .first()
393 )
394 chunks_by_collection[coll_name] = {
395 "name": collection.name if collection else coll_name,
396 "id": collection_id,
397 "chunks": [],
398 }
400 chunks_by_collection[coll_name]["chunks"].append(
401 {
402 "id": chunk.id,
403 "index": chunk.chunk_index,
404 "text": chunk.chunk_text,
405 "word_count": chunk.word_count,
406 "start_char": chunk.start_char,
407 "end_char": chunk.end_char,
408 "embedding_model": chunk.embedding_model,
409 "embedding_model_type": chunk.embedding_model_type.value
410 if chunk.embedding_model_type
411 else None,
412 "embedding_dimension": chunk.embedding_dimension,
413 "created_at": chunk.created_at,
414 }
415 )
417 return render_template(
418 "pages/document_chunks.html",
419 document=document,
420 chunks_by_collection=chunks_by_collection,
421 total_chunks=len(chunks),
422 )
425@rag_bp.route("/collections")
426@login_required
427def collections_page():
428 """Render the Collections page."""
429 return render_template("pages/collections.html", active_page="collections")
432@rag_bp.route("/collections/<string:collection_id>")
433@login_required
434def collection_details_page(collection_id):
435 """Render the Collection Details page."""
436 return render_template(
437 "pages/collection_details.html",
438 active_page="collections",
439 collection_id=collection_id,
440 )
443@rag_bp.route("/collections/<string:collection_id>/upload")
444@login_required
445def collection_upload_page(collection_id):
446 """Render the Collection Upload page."""
447 # Get the upload PDF storage setting
448 settings = get_settings_manager()
449 upload_pdf_storage = settings.get_setting(
450 "research_library.upload_pdf_storage", "none"
451 )
452 # Only allow valid values for uploads (no filesystem)
453 if upload_pdf_storage not in ("database", "none"):
454 upload_pdf_storage = "none"
456 return render_template(
457 "pages/collection_upload.html",
458 active_page="collections",
459 collection_id=collection_id,
460 collection_name=None, # Could fetch from DB if needed
461 upload_pdf_storage=upload_pdf_storage,
462 )
465@rag_bp.route("/collections/create")
466@login_required
467def collection_create_page():
468 """Render the Create Collection page."""
469 return render_template(
470 "pages/collection_create.html", active_page="collections"
471 )
474# API Routes
477@rag_bp.route("/api/rag/settings", methods=["GET"])
478@login_required
479def get_current_settings():
480 """Get current RAG configuration from settings."""
482 try:
483 settings = get_settings_manager()
485 return jsonify(
486 {
487 "success": True,
488 "settings": {
489 "embedding_provider": settings.get_setting(
490 "local_search_embedding_provider",
491 "sentence_transformers",
492 ),
493 "embedding_model": settings.get_setting(
494 "local_search_embedding_model", "all-MiniLM-L6-v2"
495 ),
496 "chunk_size": settings.get_setting(
497 "local_search_chunk_size", 1000
498 ),
499 "chunk_overlap": settings.get_setting(
500 "local_search_chunk_overlap", 200
501 ),
502 "splitter_type": settings.get_setting(
503 "local_search_splitter_type", "recursive"
504 ),
505 "text_separators": _get_text_separators(settings),
506 "distance_metric": settings.get_setting(
507 "local_search_distance_metric", "cosine"
508 ),
509 "normalize_vectors": settings.get_setting(
510 "local_search_normalize_vectors", True
511 ),
512 "index_type": settings.get_setting(
513 "local_search_index_type", "flat"
514 ),
515 },
516 }
517 )
518 except Exception as e:
519 return handle_api_error("getting RAG settings", e)
522@rag_bp.route("/api/rag/test-embedding", methods=["POST"])
523@login_required
524@require_json_body(error_format="success")
525def test_embedding():
526 """Test an embedding configuration by generating a test embedding."""
528 try:
529 data = request.get_json()
530 provider = data.get("provider")
531 model = data.get("model")
532 test_text = data.get("test_text", "This is a test.")
534 if not provider or not model:
535 return jsonify(
536 {"success": False, "error": "Provider and model are required"}
537 ), 400
539 # Import embedding functions
540 from ...embeddings.embeddings_config import (
541 get_embedding_function,
542 )
544 logger.info(
545 f"Testing embedding with provider={provider}, model={model}"
546 )
548 # Get user's settings so provider URLs (e.g. Ollama) are resolved correctly
549 settings = get_settings_manager()
550 settings_snapshot = (
551 settings.get_all_settings()
552 if hasattr(settings, "get_all_settings")
553 else {}
554 )
556 # Get embedding function with the specified configuration
557 start_time = time.time()
558 embedding_func = get_embedding_function(
559 provider=provider,
560 model_name=model,
561 settings_snapshot=settings_snapshot,
562 )
564 # Generate test embedding
565 embedding = embedding_func([test_text])[0]
566 response_time_ms = int((time.time() - start_time) * 1000)
568 # Get embedding dimension
569 dimension = len(embedding) if hasattr(embedding, "__len__") else None
571 return jsonify(
572 {
573 "success": True,
574 "dimension": dimension,
575 "response_time_ms": response_time_ms,
576 "provider": provider,
577 "model": model,
578 }
579 )
581 except Exception as e:
582 # Egress policy denial → clean 4xx (the embedding never fired).
583 from ...security.egress.policy import PolicyDeniedError
585 if isinstance(e, PolicyDeniedError): 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true
586 return jsonify(
587 {
588 "success": False,
589 "error": (
590 f"Embedding provider '{provider}' refused by egress "
591 f"policy ({e.decision.reason}). Disable 'Require "
592 "local embeddings' or pick a local provider."
593 ),
594 }
595 ), 400
596 logger.exception("Error during testing embedding")
597 user_message = _format_test_embedding_error(e, model)
598 return jsonify({"success": False, "error": user_message}), 500
601@rag_bp.route("/api/rag/models", methods=["GET"])
602@login_required
603def get_available_models():
604 """Get list of available embedding providers and models."""
605 try:
606 from ...embeddings.embeddings_config import _get_provider_classes
608 # Get current settings for providers
609 settings = get_settings_manager()
610 settings_snapshot = (
611 settings.get_all_settings()
612 if hasattr(settings, "get_all_settings")
613 else {}
614 )
616 # Get provider classes
617 provider_classes = _get_provider_classes()
619 # Egress policy: fetching a cloud provider's model list opens a
620 # connection that carries the API key off-machine. Under an effective
621 # local-only embeddings posture (PRIVATE_ONLY / adaptive-private) we
622 # must NOT probe remote-classified providers — mirror the PEP in
623 # embeddings_config.get_embeddings(). Build the run context once and
624 # fail closed (skip remote model fetches) if the policy can't evaluate.
625 def _embeddings_model_fetch_allowed(provider_key: str) -> bool:
626 try:
627 from ...security.egress.policy import (
628 context_from_snapshot,
629 evaluate_embeddings,
630 )
631 from ...config.thread_settings import (
632 get_setting_from_snapshot,
633 )
635 primary = (
636 get_setting_from_snapshot(
637 "search.tool",
638 default=DEFAULT_SEARCH_TOOL,
639 settings_snapshot=settings_snapshot,
640 )
641 or DEFAULT_SEARCH_TOOL
642 )
643 ctx = context_from_snapshot(settings_snapshot, primary)
644 if not ctx.require_local_embeddings: 644 ↛ 646line 644 didn't jump to line 646 because the condition on line 644 was always true
645 return True
646 decision = evaluate_embeddings(
647 provider_key, ctx, settings_snapshot=settings_snapshot
648 )
649 if not decision.allowed:
650 logger.bind(policy_audit=True).info(
651 "skipping embeddings model-list probe "
652 "(local-only egress posture)",
653 provider=provider_key,
654 reason=decision.reason,
655 )
656 return decision.allowed
657 except Exception:
658 # Fail closed: a policy/snapshot error must not open a cloud
659 # probe under a local-only posture.
660 logger.bind(policy_audit=True).warning(
661 "embeddings model-list egress check failed; skipping probe",
662 provider=provider_key,
663 exc_info=True,
664 )
665 return False
667 # Provider display names
668 provider_labels = {
669 "sentence_transformers": "Sentence Transformers (Local)",
670 "ollama": "Ollama (Local)",
671 "openai": "OpenAI API",
672 }
674 # Get provider options and models by looping through providers
675 provider_options = []
676 providers = {}
678 for provider_key, provider_class in provider_classes.items():
679 available = provider_class.is_available(settings_snapshot)
681 # Always show the provider in the dropdown so users can
682 # configure its settings (e.g. fix a wrong Ollama URL).
683 provider_options.append(
684 {
685 "value": provider_key,
686 "label": provider_labels.get(provider_key, provider_key),
687 "available": available,
688 }
689 )
691 # Only fetch models when the provider is reachable AND egress
692 # policy permits probing it (cloud providers are skipped under a
693 # local-only posture).
694 if available and _embeddings_model_fetch_allowed(provider_key):
695 models = provider_class.get_available_models(settings_snapshot)
696 providers[provider_key] = [
697 {
698 "value": m["value"],
699 "label": m["label"],
700 "provider": provider_key,
701 **(
702 {"is_embedding": m["is_embedding"]}
703 if "is_embedding" in m
704 else {}
705 ),
706 }
707 for m in models
708 ]
709 else:
710 providers[provider_key] = []
712 return jsonify(
713 {
714 "success": True,
715 "provider_options": provider_options,
716 "providers": providers,
717 }
718 )
719 except Exception as e:
720 return handle_api_error("getting available models", e)
723@rag_bp.route("/api/rag/info", methods=["GET"])
724@login_required
725def get_index_info():
726 """Get information about the current RAG index."""
727 from ...database.library_init import get_default_library_id
729 try:
730 # Get collection_id from request or use default Library collection
731 collection_id = request.args.get("collection_id")
732 if not collection_id:
733 collection_id = get_default_library_id(session["username"])
735 logger.info(
736 f"Getting RAG index info for collection_id: {collection_id}"
737 )
739 with get_rag_service(collection_id) as rag_service:
740 info = rag_service.get_current_index_info(collection_id)
742 if info is None:
743 logger.info(
744 f"No RAG index found for collection_id: {collection_id}"
745 )
746 return jsonify(
747 {"success": True, "info": None, "message": "No index found"}
748 )
750 logger.info(f"Found RAG index for collection_id: {collection_id}")
751 return jsonify({"success": True, "info": info})
752 except Exception as e:
753 return handle_api_error("getting index info", e)
756@rag_bp.route("/api/rag/stats", methods=["GET"])
757@login_required
758def get_rag_stats():
759 """Get RAG statistics for a collection."""
760 from ...database.library_init import get_default_library_id
762 try:
763 # Get collection_id from request or use default Library collection
764 collection_id = request.args.get("collection_id")
765 if not collection_id:
766 collection_id = get_default_library_id(session["username"])
768 with get_rag_service(collection_id) as rag_service:
769 stats = rag_service.get_rag_stats(collection_id)
771 return jsonify({"success": True, "stats": stats})
772 except Exception as e:
773 return handle_api_error("getting RAG stats", e)
776@rag_bp.route("/api/rag/index-document", methods=["POST"])
777@login_required
778@require_json_body(error_format="success")
779def index_document():
780 """Index a single document in a collection."""
781 from ...database.library_init import get_default_library_id
783 try:
784 data = request.get_json()
785 text_doc_id = data.get("text_doc_id")
786 force_reindex = data.get("force_reindex", False)
787 collection_id = data.get("collection_id")
789 if not text_doc_id:
790 return jsonify(
791 {"success": False, "error": "text_doc_id is required"}
792 ), 400
794 # Get collection_id from request or use default Library collection
795 if not collection_id:
796 collection_id = get_default_library_id(session["username"])
798 with get_rag_service(collection_id) as rag_service:
799 result = rag_service.index_document(
800 text_doc_id, collection_id, force_reindex
801 )
803 if result["status"] == "error":
804 return jsonify(
805 {"success": False, "error": result.get("error")}
806 ), 400
808 return jsonify({"success": True, "result": result})
809 except Exception as e:
810 return handle_api_error(f"indexing document {text_doc_id}", e)
813@rag_bp.route("/api/rag/remove-document", methods=["POST"])
814@login_required
815@require_json_body(error_format="success")
816def remove_document():
817 """Remove a document from RAG in a collection."""
818 from ...database.library_init import get_default_library_id
820 try:
821 data = request.get_json()
822 text_doc_id = data.get("text_doc_id")
823 collection_id = data.get("collection_id")
825 if not text_doc_id:
826 return jsonify(
827 {"success": False, "error": "text_doc_id is required"}
828 ), 400
830 # Get collection_id from request or use default Library collection
831 if not collection_id: 831 ↛ 834line 831 didn't jump to line 834 because the condition on line 831 was always true
832 collection_id = get_default_library_id(session["username"])
834 with get_rag_service(collection_id) as rag_service:
835 result = rag_service.remove_document_from_rag(
836 text_doc_id, collection_id
837 )
839 if result["status"] == "error":
840 return jsonify(
841 {"success": False, "error": result.get("error")}
842 ), 400
844 return jsonify({"success": True, "result": result})
845 except Exception as e:
846 return handle_api_error(f"removing document {text_doc_id}", e)
849@rag_bp.route("/api/rag/index-all", methods=["GET"])
850@login_required
851def index_all():
852 """Index all documents in a collection with Server-Sent Events progress."""
853 from ...database.session_context import get_user_db_session
854 from ...database.library_init import get_default_library_id
856 force_reindex = parse_bool_arg("force_reindex")
857 username = session["username"]
859 # Get collection_id from request or use default Library collection
860 collection_id = request.args.get("collection_id")
861 if not collection_id: 861 ↛ 862line 861 didn't jump to line 862 because the condition on line 861 was never true
862 collection_id = get_default_library_id(username)
864 logger.info(
865 f"Starting index-all for collection_id: {collection_id}, force_reindex: {force_reindex}"
866 )
868 # Create RAG service in request context before generator runs.
869 # use_defaults=force_reindex mirrors index_collection / the background
870 # worker so a force-reindex resolves the same embedding config.
871 rag_service = get_rag_service(collection_id, use_defaults=force_reindex)
873 def generate():
874 """Generator function for SSE progress updates."""
875 try:
876 # Send initial status
877 yield f"data: {json.dumps({'type': 'start', 'message': 'Starting bulk indexing...'})}\n\n"
879 # Get document IDs to index from DocumentCollection
880 with get_user_db_session(username) as db_session:
881 # Persist embedding metadata + clean up on force-reindex via the
882 # shared helpers, so this bulk route behaves like the
883 # single-collection SSE route and the background worker
884 # (previously it stored no embedding_dimension and never reset
885 # stale chunks/indices on force-reindex).
886 collection = (
887 db_session.query(Collection)
888 .filter_by(id=collection_id)
889 .first()
890 )
892 if not collection:
893 yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
894 return
896 # Persist the new embedding config AND reset the old
897 # chunks/RAGIndex in ONE transaction. Two separate commits would
898 # leave a crash window where the Collection carries the NEW
899 # embedding config while the OLD RAGIndex row + FAISS file (built
900 # with the old config) still exist — a config/index mismatch.
901 changed = False
902 faiss_reset_paths = []
903 if collection is not None and (
904 collection.embedding_model is None or force_reindex
905 ):
906 _store_collection_embedding_metadata(
907 collection, rag_service
908 )
909 changed = True
910 if force_reindex:
911 faiss_reset_paths = _reset_collection_for_reindex(
912 db_session, collection_id
913 )
914 changed = True
915 if changed:
916 db_session.commit()
917 _unlink_reindex_faiss_files(faiss_reset_paths)
919 doc_info = [
920 (doc.id, doc.title)
921 for _dc, doc in _query_documents_to_index(
922 db_session, collection_id, force_reindex
923 )
924 ]
926 if not doc_info:
927 yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
928 return
930 results = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
931 total = len(doc_info)
933 # Process documents in batches to optimize performance
934 # Get batch size from settings
935 settings = get_settings_manager()
936 batch_size = int(
937 settings.get_setting("rag.indexing_batch_size", 15)
938 )
939 processed = 0
941 for i in range(0, len(doc_info), batch_size):
942 batch = doc_info[i : i + batch_size]
944 # Process batch with collection_id
945 batch_results = rag_service.index_documents_batch(
946 batch, collection_id, force_reindex
947 )
949 # Process results and send progress updates
950 for j, (doc_id, title) in enumerate(batch):
951 processed += 1
952 result = batch_results[doc_id]
954 # Send progress update
955 yield f"data: {json.dumps({'type': 'progress', 'current': processed, 'total': total, 'title': title, 'percent': int((processed / total) * 100)})}\n\n"
957 if result["status"] == "success": 957 ↛ 958line 957 didn't jump to line 958 because the condition on line 957 was never true
958 results["successful"] += 1
959 elif result["status"] in ("skipped", "cleared"): 959 ↛ 963line 959 didn't jump to line 963 because the condition on line 959 was always true
960 # "cleared" (empty-text purge) is a handled non-failure.
961 results["skipped"] += 1
962 else:
963 results["failed"] += 1
964 results["errors"].append(
965 {
966 "doc_id": doc_id,
967 "title": title,
968 "error": result.get("error"),
969 }
970 )
972 # Send completion status
973 yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
975 # Log final status for debugging
976 logger.info(
977 f"Bulk indexing complete: {results['successful']} successful, {results['skipped']} skipped, {results['failed']} failed"
978 )
980 except Exception:
981 logger.exception("Error in bulk indexing")
982 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
983 finally:
984 # Generator ``finally`` runs at stream completion (or client
985 # disconnect via ``GeneratorExit``) — the safe place to release
986 # the RAG service's embedding-manager httpx clients. Closing at
987 # the outer route scope would tear it down before the streamed
988 # generator runs. ``safe_close`` swallows close-time errors so
989 # a broken Ollama doesn't mask the original generator outcome.
990 safe_close(rag_service, "rag_service (index-all SSE)")
992 return Response(
993 stream_with_context(generate()), mimetype="text/event-stream"
994 )
997@rag_bp.route("/api/rag/configure", methods=["POST"])
998@login_required
999@require_json_body(error_format="success")
1000def configure_rag():
1001 """
1002 Change RAG configuration (embedding model, chunk size, etc.).
1003 This will create a new index with the new configuration.
1004 """
1005 import json as json_lib
1007 try:
1008 data = request.get_json()
1009 embedding_model = data.get("embedding_model")
1010 embedding_provider = data.get("embedding_provider")
1011 chunk_size = data.get("chunk_size")
1012 chunk_overlap = data.get("chunk_overlap")
1013 collection_id = data.get("collection_id")
1015 # Get new advanced settings (with defaults)
1016 splitter_type = data.get("splitter_type", "recursive")
1017 text_separators = data.get(
1018 "text_separators", ["\n\n", "\n", ". ", " ", ""]
1019 )
1020 distance_metric = data.get("distance_metric", "cosine")
1021 normalize_vectors = data.get("normalize_vectors", True)
1022 index_type = data.get("index_type", "flat")
1024 if not all(
1025 [
1026 embedding_model,
1027 embedding_provider,
1028 chunk_size,
1029 chunk_overlap,
1030 ]
1031 ):
1032 return jsonify(
1033 {
1034 "success": False,
1035 "error": "All configuration parameters are required (embedding_model, embedding_provider, chunk_size, chunk_overlap)",
1036 }
1037 ), 400
1039 # Save settings to database
1040 settings = get_settings_manager()
1041 settings.set_setting("local_search_embedding_model", embedding_model)
1042 settings.set_setting(
1043 "local_search_embedding_provider", embedding_provider
1044 )
1045 settings.set_setting("local_search_chunk_size", int(chunk_size))
1046 settings.set_setting("local_search_chunk_overlap", int(chunk_overlap))
1048 # Save new advanced settings
1049 settings.set_setting("local_search_splitter_type", splitter_type)
1051 # The setting is registered with ui_element "json", so store the
1052 # separators as a proper list. A string payload (e.g. a textarea
1053 # value) is parsed into a list first; if it is not valid JSON it is
1054 # stored as-is so the UI validation layer can flag it.
1055 if isinstance(text_separators, str):
1056 try:
1057 text_separators = json_lib.loads(text_separators)
1058 except json_lib.JSONDecodeError:
1059 logger.debug(
1060 "Invalid JSON for text_separators input; storing raw string for validation"
1061 )
1062 settings.set_setting("local_search_text_separators", text_separators)
1063 settings.set_setting("local_search_distance_metric", distance_metric)
1064 settings.set_setting(
1065 "local_search_normalize_vectors", bool(normalize_vectors)
1066 )
1067 settings.set_setting("local_search_index_type", index_type)
1069 # If collection_id is provided, update that collection's configuration
1070 if collection_id:
1071 # Create new RAG service with new configuration
1072 with LibraryRAGService(
1073 username=session["username"],
1074 embedding_model=embedding_model,
1075 embedding_provider=embedding_provider,
1076 chunk_size=int(chunk_size),
1077 chunk_overlap=int(chunk_overlap),
1078 splitter_type=splitter_type,
1079 text_separators=text_separators
1080 if isinstance(text_separators, list)
1081 else DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
1082 distance_metric=distance_metric,
1083 normalize_vectors=normalize_vectors,
1084 index_type=index_type,
1085 ) as new_rag_service:
1086 # Get or create new index with this configuration
1087 rag_index = new_rag_service._get_or_create_rag_index(
1088 collection_id
1089 )
1091 return jsonify(
1092 {
1093 "success": True,
1094 "message": "Configuration updated for collection. You can now index documents with the new settings.",
1095 "index_hash": rag_index.index_hash,
1096 }
1097 )
1098 else:
1099 # Just saving default settings without updating a specific collection
1100 return jsonify(
1101 {
1102 "success": True,
1103 "message": "Default embedding settings saved successfully. New collections will use these settings.",
1104 }
1105 )
1107 except Exception as e:
1108 return handle_api_error("configuring RAG", e)
1111@rag_bp.route("/api/rag/documents", methods=["GET"])
1112@login_required
1113def get_documents():
1114 """Get library documents with their RAG status for the default Library collection (paginated)."""
1115 from ...database.session_context import get_user_db_session
1116 from ...database.library_init import get_default_library_id
1118 try:
1119 # Get pagination parameters
1120 page = request.args.get("page", 1, type=int)
1121 per_page = request.args.get("per_page", 50, type=int)
1122 filter_type = request.args.get(
1123 "filter", "all"
1124 ) # all, indexed, unindexed
1126 # Validate pagination parameters
1127 page = max(1, page)
1128 per_page = min(max(10, per_page), 100) # Limit between 10-100
1130 # Close current thread's session to force fresh connection
1131 from ...database.thread_local_session import cleanup_current_thread
1133 cleanup_current_thread()
1135 username = session["username"]
1137 # Get collection_id from request or use default Library collection
1138 collection_id = request.args.get("collection_id")
1139 if not collection_id:
1140 collection_id = get_default_library_id(username)
1142 logger.info(
1143 f"Getting documents for collection_id: {collection_id}, filter: {filter_type}, page: {page}"
1144 )
1146 with get_user_db_session(username) as db_session:
1147 # Expire all cached objects to ensure we get fresh data from DB
1148 db_session.expire_all()
1150 # Import RagDocumentStatus model
1151 from ...database.models.library import RagDocumentStatus
1153 # Build base query - join Document with DocumentCollection for the collection
1154 # LEFT JOIN with rag_document_status to check indexed status
1155 query = (
1156 db_session.query(
1157 Document, DocumentCollection, RagDocumentStatus
1158 )
1159 .join(
1160 DocumentCollection,
1161 (DocumentCollection.document_id == Document.id)
1162 & (DocumentCollection.collection_id == collection_id),
1163 )
1164 .outerjoin(
1165 RagDocumentStatus,
1166 (RagDocumentStatus.document_id == Document.id)
1167 & (RagDocumentStatus.collection_id == collection_id),
1168 )
1169 )
1171 logger.debug(f"Base query for collection {collection_id}: {query}")
1173 # Apply filters based on rag_document_status existence
1174 if filter_type == "indexed":
1175 query = query.filter(RagDocumentStatus.document_id.isnot(None))
1176 elif filter_type == "unindexed":
1177 # Documents in collection but not indexed yet
1178 query = query.filter(RagDocumentStatus.document_id.is_(None))
1180 # Get total count before pagination
1181 total_count = query.count()
1182 logger.info(
1183 f"Found {total_count} total documents for collection {collection_id} with filter {filter_type}"
1184 )
1186 # Apply pagination
1187 results = (
1188 query.order_by(Document.created_at.desc())
1189 .limit(per_page)
1190 .offset((page - 1) * per_page)
1191 .all()
1192 )
1194 documents = [
1195 {
1196 "id": doc.id,
1197 "title": doc.title,
1198 "original_url": doc.original_url,
1199 "rag_indexed": rag_status is not None,
1200 "chunk_count": rag_status.chunk_count if rag_status else 0,
1201 "created_at": doc.created_at.isoformat()
1202 if doc.created_at
1203 else None,
1204 }
1205 for doc, doc_collection, rag_status in results
1206 ]
1208 # Debug logging to help diagnose indexing status issues
1209 indexed_count = sum(1 for d in documents if d["rag_indexed"])
1211 # Additional debug: check rag_document_status for this collection
1212 all_indexed_statuses = (
1213 db_session.query(RagDocumentStatus)
1214 .filter_by(collection_id=collection_id)
1215 .all()
1216 )
1217 logger.info(
1218 f"rag_document_status table shows: {len(all_indexed_statuses)} documents indexed for collection {collection_id}"
1219 )
1221 logger.info(
1222 f"Returning {len(documents)} documents on page {page}: "
1223 f"{indexed_count} indexed, {len(documents) - indexed_count} not indexed"
1224 )
1226 return jsonify(
1227 {
1228 "success": True,
1229 "documents": documents,
1230 "pagination": {
1231 "page": page,
1232 "per_page": per_page,
1233 "total": total_count,
1234 "pages": (total_count + per_page - 1) // per_page,
1235 },
1236 }
1237 )
1238 except Exception as e:
1239 return handle_api_error("getting documents", e)
1242# Collection Management Routes
1245@rag_bp.route("/api/collections", methods=["GET"])
1246@login_required
1247def get_collections():
1248 """Get all document collections for the current user."""
1249 from ...database.session_context import get_user_db_session
1251 try:
1252 username = session["username"]
1253 with get_user_db_session(username) as db_session:
1254 # No need to filter by username - each user has their own database
1255 collections = db_session.query(Collection).all()
1257 # How many documents in each collection are already indexed for
1258 # RAG search. Computed once as a grouped aggregate (mirrors
1259 # LibraryService.get_all_collections) so the per-collection
1260 # "X of Y indexed" / pending status doesn't trigger an N+1.
1261 indexed_counts = dict(
1262 db_session.query(
1263 DocumentCollection.collection_id,
1264 func.count(
1265 case(
1266 (
1267 DocumentCollection.indexed == True, # noqa: E712
1268 DocumentCollection.document_id,
1269 ),
1270 else_=None,
1271 )
1272 ),
1273 )
1274 .group_by(DocumentCollection.collection_id)
1275 .all()
1276 )
1278 result = []
1279 for coll in collections:
1280 document_count = (
1281 len(coll.document_links)
1282 if hasattr(coll, "document_links")
1283 else 0
1284 )
1285 indexed_document_count = int(
1286 indexed_counts.get(coll.id, 0) or 0
1287 )
1288 collection_data = {
1289 "id": coll.id,
1290 "name": coll.name,
1291 "description": coll.description,
1292 "created_at": coll.created_at.isoformat()
1293 if coll.created_at
1294 else None,
1295 "collection_type": coll.collection_type,
1296 "is_default": coll.is_default
1297 if hasattr(coll, "is_default")
1298 else False,
1299 "is_public": bool(getattr(coll, "is_public", False)),
1300 "agent_enabled": _agent_enabled_default_on(coll),
1301 "document_count": document_count,
1302 "indexed_document_count": indexed_document_count,
1303 "folder_count": len(coll.linked_folders)
1304 if hasattr(coll, "linked_folders")
1305 else 0,
1306 }
1308 # Include embedding metadata if available
1309 if coll.embedding_model:
1310 collection_data["embedding"] = {
1311 "model": coll.embedding_model,
1312 "provider": coll.embedding_model_type.value
1313 if coll.embedding_model_type
1314 else None,
1315 "dimension": coll.embedding_dimension,
1316 "chunk_size": coll.chunk_size,
1317 "chunk_overlap": coll.chunk_overlap,
1318 }
1319 else:
1320 collection_data["embedding"] = None
1322 result.append(collection_data)
1324 return jsonify({"success": True, "collections": result})
1325 except Exception as e:
1326 return handle_api_error("getting collections", e)
1329@rag_bp.route("/api/collections", methods=["POST"])
1330@login_required
1331@require_json_body(error_format="success")
1332def create_collection():
1333 """Create a new document collection."""
1334 from ...database.session_context import get_user_db_session
1336 try:
1337 data = request.get_json()
1338 name = data.get("name", "").strip()
1339 description = data.get("description", "").strip()
1340 collection_type = data.get("type", "user_uploads")
1341 # Allowlist user-creatable types. System types (notes,
1342 # default_library, research_history) are lazy-created by their
1343 # owning subsystems; a user-crafted impostor (e.g. type="notes")
1344 # would be undeletable under PROTECTED_COLLECTION_TYPES and could
1345 # nondeterministically win _get_or_create_notes_collection's
1346 # .first() lookup, splitting the notes corpus.
1347 allowed_types = {"user_uploads", "user_collection"}
1348 if collection_type not in allowed_types:
1349 return jsonify(
1350 {
1351 "success": False,
1352 "error": (
1353 f"Invalid collection type '{collection_type}'. "
1354 f"Allowed: {sorted(allowed_types)}"
1355 ),
1356 }
1357 ), 400
1358 # Egress classification, default private (the safe choice). A public
1359 # collection counts as a public engine and may use cloud inference.
1360 is_public = bool(data.get("is_public", False))
1361 # Usability switch (NOT egress): offer this collection to the research
1362 # agent? Default True (available) — behaviour-preserving. An explicit
1363 # JSON null normalizes to True so it matches how NULL is read back
1364 # everywhere else (NULL → available); only an explicit false disables.
1365 agent_enabled_raw = data.get("agent_enabled", True)
1366 agent_enabled = (
1367 True if agent_enabled_raw is None else bool(agent_enabled_raw)
1368 )
1370 if not name:
1371 return jsonify({"success": False, "error": "Name is required"}), 400
1373 username = session["username"]
1374 with get_user_db_session(username) as db_session:
1375 # Check if collection with this name already exists in this user's database
1376 existing = db_session.query(Collection).filter_by(name=name).first()
1378 if existing:
1379 return jsonify(
1380 {
1381 "success": False,
1382 "error": f"Collection '{name}' already exists",
1383 }
1384 ), 400
1386 # Create new collection (no username needed - each user has their own DB)
1387 # Note: created_at uses default=utcnow() in the model, so we don't need to set it manually
1388 collection = Collection(
1389 id=str(uuid.uuid4()), # Generate UUID for collection
1390 name=name,
1391 description=description,
1392 collection_type=collection_type,
1393 is_public=is_public,
1394 agent_enabled=agent_enabled,
1395 )
1397 db_session.add(collection)
1398 db_session.commit()
1400 return jsonify(
1401 {
1402 "success": True,
1403 "collection": {
1404 "id": collection.id,
1405 "name": collection.name,
1406 "description": collection.description,
1407 "created_at": collection.created_at.isoformat(),
1408 "collection_type": collection.collection_type,
1409 "is_public": bool(collection.is_public),
1410 "agent_enabled": _agent_enabled_default_on(collection),
1411 },
1412 }
1413 )
1414 except Exception as e:
1415 return handle_api_error("creating collection", e)
1418@rag_bp.route("/api/collections/<string:collection_id>", methods=["PUT"])
1419@login_required
1420@require_json_body(error_format="success")
1421def update_collection(collection_id):
1422 """Update a collection's details."""
1423 from ...database.session_context import get_user_db_session
1425 try:
1426 data = request.get_json()
1427 name = data.get("name", "").strip()
1428 description = data.get("description", "").strip()
1430 username = session["username"]
1431 with get_user_db_session(username) as db_session:
1432 # No need to filter by username - each user has their own database
1433 collection = (
1434 db_session.query(Collection).filter_by(id=collection_id).first()
1435 )
1437 if not collection:
1438 return jsonify(
1439 {"success": False, "error": "Collection not found"}
1440 ), 404
1442 # System collections (Notes, Library, Research History) own
1443 # first-class user data whose lifecycle is managed by other
1444 # subsystems. Renaming them via this generic endpoint
1445 # bypasses their intended invariants and confuses every UI
1446 # surface that lists collections. Mirrors PROTECTED_COLLECTION_TYPES
1447 # in the deletion service. Only identity fields (name,
1448 # description) are locked — the is_public / agent_enabled
1449 # toggles below are deliberate per-collection settings that
1450 # must keep working for system collections too.
1451 from ..deletion.services.collection_deletion import (
1452 PROTECTED_COLLECTION_TYPES,
1453 )
1455 if collection.collection_type in PROTECTED_COLLECTION_TYPES and (
1456 name or "description" in data
1457 ):
1458 return jsonify(
1459 {
1460 "success": False,
1461 "error": (
1462 f"Cannot rename or redescribe system "
1463 f"collection '{collection.name}' "
1464 f"(type={collection.collection_type})."
1465 ),
1466 }
1467 ), 409
1469 if name:
1470 # Check if new name conflicts with existing collection
1471 existing = (
1472 db_session.query(Collection)
1473 .filter(
1474 Collection.name == name,
1475 Collection.id != collection_id,
1476 )
1477 .first()
1478 )
1480 if existing:
1481 return jsonify(
1482 {
1483 "success": False,
1484 "error": f"Collection '{name}' already exists",
1485 }
1486 ), 400
1488 collection.name = name
1490 # Only write when the caller sends the key (sending "" still
1491 # clears) — otherwise toggle-only PUTs wipe the description.
1492 if "description" in data:
1493 collection.description = description
1495 # Egress classification toggle (only when the caller sends it).
1496 if "is_public" in data:
1497 collection.is_public = bool(data.get("is_public"))
1499 # Research-agent availability toggle (only when the caller sends it).
1500 # Explicit null normalizes to True (available) for parity with how
1501 # NULL is read back elsewhere; only an explicit false disables.
1502 if "agent_enabled" in data:
1503 raw = data.get("agent_enabled")
1504 collection.agent_enabled = True if raw is None else bool(raw)
1506 db_session.commit()
1508 return jsonify(
1509 {
1510 "success": True,
1511 "collection": {
1512 "id": collection.id,
1513 "name": collection.name,
1514 "description": collection.description,
1515 "created_at": collection.created_at.isoformat()
1516 if collection.created_at
1517 else None,
1518 "collection_type": collection.collection_type,
1519 "is_public": bool(collection.is_public),
1520 "agent_enabled": _agent_enabled_default_on(collection),
1521 },
1522 }
1523 )
1524 except Exception as e:
1525 return handle_api_error("updating collection", e)
1528@rag_bp.route(
1529 "/api/collections/<string:collection_id>/upload", methods=["POST"]
1530)
1531@login_required
1532@upload_rate_limit_user
1533@upload_rate_limit_ip
1534def upload_to_collection(collection_id):
1535 """Upload files to a collection."""
1536 from ...database.session_context import (
1537 get_user_db_session,
1538 safe_rollback,
1539 )
1540 from ...security import sanitize_filename, UnsafeFilenameError
1541 import hashlib
1542 import uuid
1543 from ..services.pdf_storage_manager import PDFStorageManager
1545 try:
1546 if "files" not in request.files:
1547 return jsonify(
1548 {"success": False, "error": "No files provided"}
1549 ), 400
1551 files = request.files.getlist("files")
1552 if not files: 1552 ↛ 1553line 1552 didn't jump to line 1553 because the condition on line 1552 was never true
1553 return jsonify(
1554 {"success": False, "error": "No files selected"}
1555 ), 400
1557 # Bound the per-request file count BEFORE doing any work. The
1558 # request-level MAX_CONTENT_LENGTH gate covers total bytes, but
1559 # not file *count*; a request with 10000 zero-byte files would
1560 # otherwise reach the loop below.
1561 is_valid, error_msg = FileUploadValidator.validate_file_count(
1562 len(files)
1563 )
1564 if not is_valid:
1565 return jsonify({"success": False, "error": error_msg}), 400
1567 username = session["username"]
1568 with get_user_db_session(username) as db_session:
1569 # Verify collection exists in this user's database
1570 collection = (
1571 db_session.query(Collection).filter_by(id=collection_id).first()
1572 )
1574 if not collection:
1575 return jsonify(
1576 {"success": False, "error": "Collection not found"}
1577 ), 404
1579 # Get PDF storage mode from form data, falling back to user's setting
1580 settings = get_settings_manager()
1581 default_pdf_storage = settings.get_setting(
1582 "research_library.upload_pdf_storage", "none"
1583 )
1584 pdf_storage = request.form.get("pdf_storage", default_pdf_storage)
1585 if pdf_storage not in ("database", "none"): 1585 ↛ 1588line 1585 didn't jump to line 1588 because the condition on line 1585 was never true
1586 # Security: user uploads can only use database (encrypted) or none (text-only)
1587 # Filesystem storage is not allowed for user uploads
1588 pdf_storage = "none"
1590 # Initialize PDF storage manager if storing PDFs in database
1591 pdf_storage_manager = None
1592 if pdf_storage == "database":
1593 library_root = settings.get_setting(
1594 "research_library.storage_path",
1595 str(get_library_directory()),
1596 )
1597 library_root = str(
1598 Path(os.path.expandvars(library_root))
1599 .expanduser()
1600 .resolve()
1601 )
1602 pdf_storage_manager = PDFStorageManager(
1603 library_root=Path(library_root), storage_mode="database"
1604 )
1605 logger.info("PDF storage mode: database (encrypted)")
1606 else:
1607 logger.info("PDF storage mode: none (text-only)")
1609 uploaded_files = []
1610 errors = []
1612 for file in files:
1613 if not file.filename:
1614 continue
1616 try:
1617 filename = sanitize_filename(file.filename)
1618 except UnsafeFilenameError:
1619 errors.append(
1620 {
1621 "filename": "rejected",
1622 "error": "Invalid or unsafe filename",
1623 }
1624 )
1625 continue
1627 try:
1628 # Pre-flight size check on Content-Length BEFORE reading
1629 # bytes into memory. Cheap rejection for oversized files;
1630 # avoids loading 50MB+ into memory just to discard it.
1631 is_valid, error_msg = (
1632 FileUploadValidator.validate_file_size(
1633 content_length=file.content_length,
1634 file_content=None,
1635 )
1636 )
1637 if not is_valid: 1637 ↛ 1638line 1637 didn't jump to line 1638 because the condition on line 1637 was never true
1638 errors.append(
1639 {"filename": filename, "error": error_msg}
1640 )
1641 continue
1643 # Read file content
1644 file_content = file.read()
1645 file.seek(0) # Reset for potential re-reading
1647 # Post-read size check (Content-Length can be missing or
1648 # spoofed; the actual byte count is authoritative).
1649 is_valid, error_msg = (
1650 FileUploadValidator.validate_file_size(
1651 content_length=None,
1652 file_content=file_content,
1653 )
1654 )
1655 if not is_valid:
1656 errors.append(
1657 {"filename": filename, "error": error_msg}
1658 )
1659 continue
1661 # Calculate file hash for deduplication
1662 file_hash = hashlib.sha256(file_content).hexdigest()
1664 # Check if document already exists
1665 existing_doc = (
1666 db_session.query(Document)
1667 .filter_by(document_hash=file_hash)
1668 .first()
1669 )
1671 if existing_doc:
1672 # Document exists, check if we can upgrade to include PDF
1673 pdf_upgraded = False
1674 if (
1675 pdf_storage == "database"
1676 and pdf_storage_manager is not None
1677 ):
1678 # NOTE: Only the PDF magic-byte check is needed here.
1679 # File count validation is already handled by Flask's MAX_CONTENT_LENGTH.
1680 # Filename sanitization already happens via sanitize_filename() above.
1681 # See PR #3145 review for details.
1682 if file_content[:4] != b"%PDF": 1682 ↛ 1683line 1682 didn't jump to line 1683 because the condition on line 1682 was never true
1683 logger.debug(
1684 "Skipping PDF upgrade for {}: not a PDF file",
1685 filename,
1686 )
1687 else:
1688 pdf_upgraded = (
1689 pdf_storage_manager.upgrade_to_pdf(
1690 document=existing_doc,
1691 pdf_content=file_content,
1692 session=db_session,
1693 )
1694 )
1696 # Check if already in collection
1697 existing_link = (
1698 db_session.query(DocumentCollection)
1699 .filter_by(
1700 document_id=existing_doc.id,
1701 collection_id=collection_id,
1702 )
1703 .first()
1704 )
1706 if not existing_link:
1707 ensure_in_collection(
1708 db_session, existing_doc.id, collection_id
1709 )
1710 status = "added_to_collection"
1711 if pdf_upgraded:
1712 status = "added_to_collection_pdf_upgraded"
1713 uploaded_files.append(
1714 {
1715 "filename": existing_doc.filename,
1716 "status": status,
1717 "id": existing_doc.id,
1718 "pdf_upgraded": pdf_upgraded,
1719 }
1720 )
1721 else:
1722 status = "already_in_collection"
1723 if pdf_upgraded:
1724 status = "pdf_upgraded"
1725 uploaded_files.append(
1726 {
1727 "filename": existing_doc.filename,
1728 "status": status,
1729 "id": existing_doc.id,
1730 "pdf_upgraded": pdf_upgraded,
1731 }
1732 )
1733 else:
1734 # Create new document
1735 from ...document_loaders import (
1736 extract_text_from_bytes,
1737 is_extension_supported,
1738 )
1740 file_extension = Path(filename).suffix.lower()
1742 # Validate extension is supported before extraction
1743 if not is_extension_supported(file_extension):
1744 errors.append(
1745 {
1746 "filename": filename,
1747 "error": f"Unsupported format: {file_extension}",
1748 }
1749 )
1750 continue
1752 # Use file_type without leading dot for storage
1753 file_type = (
1754 file_extension[1:]
1755 if file_extension.startswith(".")
1756 else file_extension
1757 )
1759 # Extract text using document_loaders module
1760 extracted_text = extract_text_from_bytes(
1761 file_content, file_extension, filename
1762 )
1764 # Clean the extracted text to remove surrogate characters
1765 if extracted_text:
1766 from ...text_processing import remove_surrogates
1768 extracted_text = remove_surrogates(extracted_text)
1770 if not extracted_text:
1771 errors.append(
1772 {
1773 "filename": filename,
1774 "error": f"Could not extract text from {file_type} file",
1775 }
1776 )
1777 logger.warning(
1778 f"Skipping file {filename} - no text could be extracted"
1779 )
1780 continue
1782 # Get or create the user_upload source type
1783 logger.info(
1784 f"Getting or creating user_upload source type for {filename}"
1785 )
1786 source_type = (
1787 db_session.query(SourceType)
1788 .filter_by(name="user_upload")
1789 .first()
1790 )
1791 if not source_type: 1791 ↛ 1792line 1791 didn't jump to line 1792 because the condition on line 1791 was never true
1792 logger.info("Creating new user_upload source type")
1793 source_type = SourceType(
1794 id=str(uuid.uuid4()),
1795 name="user_upload",
1796 display_name="User Upload",
1797 description="Documents uploaded by users",
1798 icon="fas fa-upload",
1799 )
1800 db_session.add(source_type)
1801 db_session.flush()
1802 logger.info(
1803 f"Created source type with ID: {source_type.id}"
1804 )
1805 else:
1806 logger.info(
1807 f"Found existing source type with ID: {source_type.id}"
1808 )
1810 # Create document with extracted text (no username needed - in user's own database)
1811 # Note: uploaded_at uses default=utcnow() in the model, so we don't need to set it manually
1812 doc_id = str(uuid.uuid4())
1813 logger.info(
1814 f"Creating document {doc_id} for {filename}"
1815 )
1817 # Determine storage mode and file_path
1818 store_pdf_in_db = (
1819 pdf_storage == "database"
1820 and file_type == "pdf"
1821 and pdf_storage_manager is not None
1822 )
1824 new_doc = Document(
1825 id=doc_id,
1826 source_type_id=source_type.id,
1827 filename=filename,
1828 document_hash=file_hash,
1829 file_size=len(file_content),
1830 file_type=file_type,
1831 text_content=extracted_text, # Always store extracted text
1832 file_path=None
1833 if store_pdf_in_db
1834 else FILE_PATH_TEXT_ONLY,
1835 storage_mode="database"
1836 if store_pdf_in_db
1837 else "none",
1838 )
1839 db_session.add(new_doc)
1840 db_session.flush() # Get the ID
1841 logger.info(
1842 f"Document {new_doc.id} created successfully"
1843 )
1845 # Store PDF in encrypted database if requested
1846 pdf_stored = False
1847 if store_pdf_in_db:
1848 try:
1849 pdf_storage_manager.save_pdf(
1850 pdf_content=file_content,
1851 document=new_doc,
1852 session=db_session,
1853 filename=filename,
1854 )
1855 pdf_stored = True
1856 logger.info(
1857 f"PDF stored in encrypted database for {filename}"
1858 )
1859 except Exception:
1860 logger.exception(
1861 f"Failed to store PDF in database for {filename}"
1862 )
1863 # Continue without PDF storage - text is still saved
1865 # Add to collection
1866 ensure_in_collection(
1867 db_session, new_doc.id, collection_id
1868 )
1870 uploaded_files.append(
1871 {
1872 "filename": filename,
1873 "status": "uploaded",
1874 "id": new_doc.id,
1875 "text_length": len(extracted_text),
1876 "pdf_stored": pdf_stored,
1877 }
1878 )
1880 except Exception:
1881 # A failed flush / save_pdf / ensure_in_collection above
1882 # poisons the shared request session. Without a rollback the
1883 # NEXT file in the loop — and the post-loop commit — cascade
1884 # into PendingRollbackError, failing the whole batch (and
1885 # 500ing files that already succeeded). Recover so one bad
1886 # file doesn't take the rest down.
1887 safe_rollback(db_session, "upload_to_collection")
1888 errors.append(
1889 {
1890 "filename": filename,
1891 "error": "Failed to upload file",
1892 }
1893 )
1894 logger.exception(f"Error uploading file {filename}")
1896 db_session.commit()
1898 # Trigger auto-indexing for successfully uploaded documents
1899 document_ids = [
1900 f["id"]
1901 for f in uploaded_files
1902 if f.get("status") in ("uploaded", "added_to_collection")
1903 ]
1904 if document_ids:
1905 from ...database.session_passwords import session_password_store
1907 session_id = session.get("session_id")
1908 db_password = session_password_store.get_session_password(
1909 username, session_id
1910 )
1911 if db_password:
1912 trigger_auto_index(
1913 document_ids, collection_id, username, db_password
1914 )
1916 return jsonify(
1917 {
1918 "success": True,
1919 "uploaded": uploaded_files,
1920 "errors": errors,
1921 "summary": {
1922 "total": len(files),
1923 "successful": len(uploaded_files),
1924 "failed": len(errors),
1925 },
1926 }
1927 )
1929 except Exception as e:
1930 return handle_api_error("uploading files", e)
1933@rag_bp.route(
1934 "/api/collections/<string:collection_id>/documents", methods=["GET"]
1935)
1936@login_required
1937def get_collection_documents(collection_id):
1938 """Get all documents in a collection."""
1939 from ...database.session_context import get_user_db_session
1941 try:
1942 username = session["username"]
1943 with get_user_db_session(username) as db_session:
1944 # Verify collection exists in this user's database
1945 collection = (
1946 db_session.query(Collection).filter_by(id=collection_id).first()
1947 )
1949 if not collection:
1950 return jsonify(
1951 {"success": False, "error": "Collection not found"}
1952 ), 404
1954 # Look up note source type to filter notes into separate array
1955 note_source_type = (
1956 db_session.query(SourceType).filter_by(name="note").first()
1957 )
1958 note_source_type_id = (
1959 note_source_type.id if note_source_type else None
1960 )
1962 # Get documents through junction table. Compute "has text in DB"
1963 # at the SQL level and defer the text_content column, so listing a
1964 # collection doesn't pull every document's full text body into
1965 # memory just to test it for truthiness (#4560).
1966 has_text_col = (
1967 Document.text_content.isnot(None)
1968 & (Document.text_content != "")
1969 ).label("has_text_db")
1970 doc_links = (
1971 db_session.query(DocumentCollection, Document, has_text_col)
1972 .join(Document)
1973 .options(defer(Document.text_content))
1974 .filter(DocumentCollection.collection_id == collection_id)
1975 .all()
1976 )
1978 documents = []
1979 for link, doc, has_text_db in doc_links:
1980 # Skip notes — they go in the separate notes array
1981 if (
1982 note_source_type_id
1983 and doc.source_type_id == note_source_type_id
1984 ):
1985 continue
1986 # Check if PDF file is stored
1987 has_pdf = bool(
1988 doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
1989 )
1990 has_text_db = bool(has_text_db)
1992 # Use title if available, otherwise filename
1993 display_title = doc.title or doc.filename or "Untitled"
1995 # Get source type name
1996 source_type_name = (
1997 doc.source_type.name if doc.source_type else "unknown"
1998 )
2000 # Check if document is in other collections
2001 other_collections_count = (
2002 db_session.query(DocumentCollection)
2003 .filter(
2004 DocumentCollection.document_id == doc.id,
2005 DocumentCollection.collection_id != collection_id,
2006 )
2007 .count()
2008 )
2010 documents.append(
2011 {
2012 "id": doc.id,
2013 "filename": display_title,
2014 "title": display_title,
2015 "file_type": doc.file_type,
2016 "file_size": doc.file_size,
2017 "uploaded_at": doc.created_at.isoformat()
2018 if doc.created_at
2019 else None,
2020 "indexed": link.indexed,
2021 "chunk_count": link.chunk_count,
2022 "last_indexed_at": link.last_indexed_at.isoformat()
2023 if link.last_indexed_at
2024 else None,
2025 "has_pdf": has_pdf,
2026 "has_text_db": has_text_db,
2027 "source_type": source_type_name,
2028 "in_other_collections": other_collections_count > 0,
2029 "other_collections_count": other_collections_count,
2030 }
2031 )
2033 # Get notes in this collection (note_source_type resolved above)
2034 notes = []
2035 if note_source_type:
2036 note_links = (
2037 db_session.query(DocumentCollection, Document)
2038 .join(Document)
2039 .filter(
2040 DocumentCollection.collection_id == collection_id,
2041 Document.source_type_id == note_source_type.id,
2042 )
2043 .all()
2044 )
2045 for link, note in note_links:
2046 content = note.text_content or ""
2047 notes.append(
2048 {
2049 "id": note.id,
2050 "title": note.title or "Untitled",
2051 "content_preview": content[:200] + "..."
2052 if len(content) > 200
2053 else content,
2054 "tags": note.tags or [],
2055 "pinned": note.favorite,
2056 "created_at": note.created_at.isoformat()
2057 if note.created_at
2058 else None,
2059 "updated_at": note.updated_at.isoformat()
2060 if note.updated_at
2061 else None,
2062 "indexed": link.indexed,
2063 "chunk_count": link.chunk_count,
2064 "source_type": "note",
2065 }
2066 )
2068 # Get index file size if available
2069 index_file_size = None
2070 index_file_size_bytes = None
2071 collection_name = f"collection_{collection_id}"
2072 rag_index = (
2073 db_session.query(RAGIndex)
2074 .filter_by(collection_name=collection_name)
2075 .first()
2076 )
2077 if rag_index and rag_index.index_path:
2078 from pathlib import Path
2080 index_path = Path(rag_index.index_path)
2081 if index_path.exists(): 2081 ↛ 2092line 2081 didn't jump to line 2092 because the condition on line 2081 was always true
2082 size_bytes = index_path.stat().st_size
2083 index_file_size_bytes = size_bytes
2084 # Format as human-readable
2085 if size_bytes < 1024:
2086 index_file_size = f"{size_bytes} B"
2087 elif size_bytes < 1024 * 1024:
2088 index_file_size = f"{size_bytes / 1024:.1f} KB"
2089 else:
2090 index_file_size = f"{size_bytes / (1024 * 1024):.1f} MB"
2092 return jsonify(
2093 {
2094 "success": True,
2095 "collection": {
2096 "id": collection.id,
2097 "name": collection.name,
2098 "description": collection.description,
2099 "is_public": bool(
2100 getattr(collection, "is_public", False)
2101 ),
2102 "agent_enabled": _agent_enabled_default_on(collection),
2103 "embedding_model": collection.embedding_model,
2104 "embedding_model_type": collection.embedding_model_type.value
2105 if collection.embedding_model_type
2106 else None,
2107 "embedding_dimension": collection.embedding_dimension,
2108 "chunk_size": collection.chunk_size,
2109 "chunk_overlap": collection.chunk_overlap,
2110 # Advanced settings
2111 "splitter_type": collection.splitter_type,
2112 "distance_metric": collection.distance_metric,
2113 "index_type": collection.index_type,
2114 "normalize_vectors": collection.normalize_vectors,
2115 # Index file info
2116 "index_file_size": index_file_size,
2117 "index_file_size_bytes": index_file_size_bytes,
2118 "collection_type": collection.collection_type,
2119 # Deletion policy comes from the server so the UI
2120 # can't drift from PROTECTED_COLLECTION_TYPES: the
2121 # details page hides its Delete button for system
2122 # collections, whose deletion the service refuses
2123 # with a 409 anyway.
2124 "is_protected": _is_protected_collection(collection),
2125 },
2126 "documents": documents,
2127 "notes": notes,
2128 }
2129 )
2131 except Exception as e:
2132 return handle_api_error("getting collection documents", e)
2135# =============================================================================
2136# Shared collection-indexing helpers
2137#
2138# Three routes index a collection: the single-collection SSE route
2139# (index_collection), the bulk SSE route (index_all), and the background worker
2140# (_background_index_worker). These helpers hold the logic that must stay
2141# identical across all three so it cannot drift again — the embedding metadata
2142# to persist, the force-reindex cleanup, and the query for documents to index.
2143# The callers differ only in how they report progress (SSE stream vs
2144# TaskMetadata) and so keep their own loops.
2145# =============================================================================
2148def _store_collection_embedding_metadata(collection, rag_service):
2149 """Persist the embedding/index configuration used to index a collection.
2151 Run on first index or force-reindex. Includes the embedding dimension,
2152 probed by embedding a test string (the same way LibraryRAGService derives
2153 it for the RAGIndex row). Previously this read
2154 ``embedding_manager.provider.embedding_dimension``, an attribute the real
2155 LocalEmbeddingManager does not have, so the probe always failed and
2156 ``embedding_dimension`` was persisted as NULL via both indexing paths. Does
2157 not commit — the caller owns the transaction.
2158 """
2159 embedding_dim = None
2160 try:
2161 # Derive the dimension by embedding a test string — mirrors
2162 # LibraryRAGService.* which computes len(embed_query("test")) for the
2163 # RAGIndex row. The embedding manager has no usable dimension attribute.
2164 test_embedding = rag_service.embedding_manager.embeddings.embed_query(
2165 "test"
2166 )
2167 embedding_dim = len(test_embedding)
2168 except Exception:
2169 logger.debug("Could not determine embedding dimension", exc_info=True)
2171 collection.embedding_model = rag_service.embedding_model
2172 collection.embedding_model_type = EmbeddingProvider(
2173 rag_service.embedding_provider
2174 )
2175 collection.embedding_dimension = embedding_dim
2176 collection.chunk_size = rag_service.chunk_size
2177 collection.chunk_overlap = rag_service.chunk_overlap
2178 # Advanced settings
2179 collection.splitter_type = rag_service.splitter_type
2180 collection.text_separators = rag_service.text_separators
2181 collection.distance_metric = rag_service.distance_metric
2182 # Ensure normalize_vectors is a proper boolean for the database
2183 collection.normalize_vectors = bool(rag_service.normalize_vectors)
2184 collection.index_type = rag_service.index_type
2185 logger.info(
2186 f"Stored embedding metadata for collection: provider={rag_service.embedding_provider}"
2187 )
2190def _reset_collection_for_reindex(db_session, collection_id):
2191 """Clear old chunks, FAISS indices, and indexed-state before a force rebuild.
2193 Prevents mixed-model vectors / stale chunks when re-indexing (e.g. if a
2194 previous run was cancelled midway). Shared so a force-reindex behaves the
2195 same whether started via the SSE route or the background worker. Does not
2196 commit — the caller owns the transaction.
2198 Returns the list of on-disk FAISS index paths to unlink AFTER the caller
2199 commits. The RAGIndex ROWS are deleted in this transaction, but the files
2200 must NOT be removed before the commit that could still roll back — a
2201 rollback would restore the rows while the files were already gone, leaving
2202 the collection pointing at missing indices (mirrors collection deletion).
2203 """
2204 # Import stays function-local: hoisting it would pull in the whole
2205 # deletion package (deletion/__init__ imports its services) at
2206 # route-module import time for a path only exercised on force-reindex.
2207 from ..deletion.utils.cascade_helper import CascadeHelper
2209 collection_name = f"collection_{collection_id}"
2211 # Delete all old document chunks from DB
2212 deleted_chunks = CascadeHelper.delete_collection_chunks(
2213 db_session, collection_name
2214 )
2215 logger.info(
2216 f"Cleared {deleted_chunks} old chunks for collection {collection_id}"
2217 )
2219 # Delete old RAGIndex records (DB only). Files unlinked by the caller after
2220 # commit (see docstring). RagDocumentStatus cascade-deletes via FK.
2221 rag_result = CascadeHelper.delete_rag_indices_for_collection(
2222 db_session, collection_name, unlink_files=False
2223 )
2224 logger.info(
2225 f"Cleared old RAG indices for collection {collection_id}: {rag_result}"
2226 )
2228 # Mark all documents as unindexed
2229 db_session.query(DocumentCollection).filter_by(
2230 collection_id=collection_id
2231 ).update(
2232 {
2233 DocumentCollection.indexed: False,
2234 DocumentCollection.chunk_count: 0,
2235 }
2236 )
2237 logger.info(f"Reset indexing state for collection {collection_id}")
2238 return rag_result["index_paths"]
2241def _unlink_reindex_faiss_files(index_paths):
2242 """Unlink FAISS index files returned by _reset_collection_for_reindex, AFTER
2243 the caller's DB commit. Best-effort: the DB delete is already durable."""
2244 if not index_paths:
2245 return
2246 from ..deletion.utils.cascade_helper import CascadeHelper
2248 for path in index_paths: 2248 ↛ 2249line 2248 didn't jump to line 2249 because the loop on line 2248 never started
2249 CascadeHelper.delete_faiss_index_files(path)
2252def _query_documents_to_index(db_session, collection_id, force_reindex):
2253 """Return the ``(DocumentCollection, Document)`` pairs to index.
2255 When not force-reindexing, only un-indexed documents are returned.
2256 """
2257 query = (
2258 db_session.query(DocumentCollection, Document)
2259 .join(Document)
2260 # Don't load the large text_content body: the indexing loop only
2261 # reads doc.id/filename/title (index_document re-fetches by id in a
2262 # worker thread), so loading every document's full text here just to
2263 # iterate is what exhausts memory on big collections (#4560).
2264 .options(defer(Document.text_content))
2265 .filter(DocumentCollection.collection_id == collection_id)
2266 )
2268 if not force_reindex:
2269 # IMPORTANT: DO NOT CHANGE THIS LINE! DO NOT REMOVE THIS COMMENT!
2270 # We MUST use `== False` instead of `not DocumentCollection.indexed`
2271 # The Python `not` operator does NOT work correctly in SQLAlchemy filters.
2272 # Using `not` will cause the query to return NO results (zero documents).
2273 # SQLAlchemy requires explicit comparison: `== False` or `== True`
2274 # This has been fixed multiple times - DO NOT change it back to `not`!
2275 query = query.filter(DocumentCollection.indexed == False) # noqa: E712
2277 return query.all()
2280@rag_bp.route("/api/collections/<string:collection_id>/index", methods=["GET"])
2281@login_required
2282def index_collection(collection_id):
2283 """Index all documents in a collection with Server-Sent Events progress."""
2284 from ...database.session_context import get_user_db_session
2285 from ...database.session_passwords import session_password_store
2287 force_reindex = parse_bool_arg("force_reindex")
2288 username = session["username"]
2289 session_id = session.get("session_id")
2291 logger.info(f"Starting index_collection, force_reindex={force_reindex}")
2293 # Get password for thread access to encrypted database
2294 db_password = None
2295 if session_id: 2295 ↛ 2301line 2295 didn't jump to line 2301 because the condition on line 2295 was always true
2296 db_password = session_password_store.get_session_password(
2297 username, session_id
2298 )
2300 # Create RAG service — on force reindex use current default model
2301 rag_service = get_rag_service(collection_id, use_defaults=force_reindex)
2302 logger.info(
2303 f"RAG service created: provider={rag_service.embedding_provider}"
2304 )
2306 def generate():
2307 """Generator for SSE progress updates."""
2308 logger.info("SSE generator started")
2309 try:
2310 with get_user_db_session(username, db_password) as db_session:
2311 # Verify collection exists in this user's database
2312 collection = (
2313 db_session.query(Collection)
2314 .filter_by(id=collection_id)
2315 .first()
2316 )
2318 if not collection:
2319 yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
2320 return
2322 # Store embedding metadata AND reset the old index in ONE commit
2323 # (prevents stale / mixed-model vectors). Committing the config
2324 # write separately from the reset leaves a window where a crash
2325 # keeps the new config but the old vectors survive.
2326 changed = False
2327 faiss_reset_paths = []
2328 if collection.embedding_model is None or force_reindex:
2329 _store_collection_embedding_metadata(
2330 collection, rag_service
2331 )
2332 changed = True
2333 if force_reindex:
2334 faiss_reset_paths = _reset_collection_for_reindex(
2335 db_session, collection_id
2336 )
2337 changed = True
2338 if changed:
2339 db_session.commit()
2340 _unlink_reindex_faiss_files(faiss_reset_paths)
2342 # Get documents to index
2343 doc_links = _query_documents_to_index(
2344 db_session, collection_id, force_reindex
2345 )
2347 if not doc_links:
2348 logger.info("No documents to index in collection")
2349 yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
2350 return
2352 total = len(doc_links)
2353 logger.info(f"Found {total} documents to index")
2354 results = {
2355 "successful": 0,
2356 "skipped": 0,
2357 "failed": 0,
2358 "errors": [],
2359 }
2361 yield f"data: {json.dumps({'type': 'start', 'message': f'Indexing {total} documents in collection: {collection.name}'})}\n\n"
2363 for idx, (link, doc) in enumerate(doc_links, 1):
2364 filename = doc.filename or doc.title or "Unknown"
2365 yield f"data: {json.dumps({'type': 'progress', 'current': idx, 'total': total, 'filename': filename, 'percent': int((idx / total) * 100)})}\n\n"
2367 try:
2368 logger.debug(
2369 f"Indexing document {idx}/{total}: {filename}"
2370 )
2372 # Run index_document in a separate thread to allow sending SSE heartbeats.
2373 # This keeps the HTTP connection alive during long indexing operations,
2374 # preventing timeouts from proxy servers (nginx) and browsers.
2375 # The main thread periodically yields heartbeat comments while waiting.
2376 result_queue = queue.Queue()
2377 error_queue = queue.Queue()
2379 def index_in_thread():
2380 try:
2381 r = rag_service.index_document(
2382 document_id=doc.id,
2383 collection_id=collection_id,
2384 force_reindex=force_reindex,
2385 )
2386 result_queue.put(r)
2387 except Exception as ex:
2388 error_queue.put(ex)
2389 finally:
2390 try:
2391 from ...database.thread_local_session import (
2392 cleanup_current_thread,
2393 )
2395 cleanup_current_thread()
2396 except Exception:
2397 logger.debug(
2398 "best-effort thread-local DB session cleanup",
2399 exc_info=True,
2400 )
2402 thread = threading.Thread(target=index_in_thread)
2403 thread.start()
2405 # Send heartbeats while waiting for the thread to complete
2406 heartbeat_interval = 5 # seconds
2407 while thread.is_alive(): 2407 ↛ 2408line 2407 didn't jump to line 2408 because the condition on line 2407 was never true
2408 thread.join(timeout=heartbeat_interval)
2409 if thread.is_alive():
2410 # Send SSE comment as heartbeat (keeps connection alive)
2411 yield f": heartbeat {idx}/{total}\n\n"
2413 # Check for errors from thread
2414 if not error_queue.empty():
2415 raise error_queue.get() # noqa: TRY301 — re-raises thread exception for per-document error handling
2417 result = result_queue.get()
2418 logger.info(
2419 f"Indexed document {idx}/{total}: {filename} - status={result.get('status')}"
2420 )
2422 if result.get("status") == "success":
2423 results["successful"] += 1
2424 # DocumentCollection status is already updated in index_document
2425 # No need to update link here
2426 elif result.get("status") in ("skipped", "cleared"): 2426 ↛ 2430line 2426 didn't jump to line 2430 because the condition on line 2426 was always true
2427 # "cleared" (empty-text purge) is a handled non-failure.
2428 results["skipped"] += 1
2429 else:
2430 results["failed"] += 1
2431 error_msg = result.get("error", "Unknown error")
2432 results["errors"].append(
2433 {
2434 "filename": filename,
2435 "error": error_msg,
2436 }
2437 )
2438 logger.warning(
2439 f"Failed to index {filename} ({idx}/{total}): {error_msg}"
2440 )
2441 except Exception as e:
2442 results["failed"] += 1
2443 # Scrub creds/keys before surfacing to the browser
2444 # (SSE yield below); full trace stays in server logs.
2445 error_msg = (
2446 sanitize_error_message(str(e))
2447 or "Failed to index document"
2448 )
2449 results["errors"].append(
2450 {
2451 "filename": filename,
2452 "error": error_msg,
2453 }
2454 )
2455 logger.exception(
2456 f"Exception indexing document {filename} ({idx}/{total})"
2457 )
2458 # Send error update to client so they know indexing is continuing
2459 yield f"data: {json.dumps({'type': 'doc_error', 'filename': filename, 'error': error_msg})}\n\n"
2461 db_session.commit()
2462 # Ensure all changes are written to disk
2463 db_session.flush()
2465 logger.info(
2466 f"Indexing complete: {results['successful']} successful, {results['failed']} failed, {results['skipped']} skipped"
2467 )
2468 yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
2469 logger.info("SSE generator finished successfully")
2471 except Exception:
2472 logger.exception("Error in collection indexing")
2473 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
2474 finally:
2475 # See parallel comment in index-all generator above — generator
2476 # ``finally`` is the safe close site for streamed RAG services.
2477 safe_close(rag_service, "rag_service (index-collection SSE)")
2479 response = Response(
2480 stream_with_context(generate()), mimetype="text/event-stream"
2481 )
2482 # Prevent buffering for proper SSE streaming
2483 response.headers["Cache-Control"] = "no-cache, no-transform"
2484 response.headers["Connection"] = "keep-alive"
2485 response.headers["X-Accel-Buffering"] = "no"
2486 return response
2489# =============================================================================
2490# Background Indexing Endpoints
2491# =============================================================================
2494def _get_rag_service_for_thread(
2495 collection_id: str,
2496 username: str,
2497 db_password: str,
2498 use_defaults: bool = False,
2499) -> LibraryRAGService:
2500 """
2501 Create RAG service for use in background threads (no Flask context).
2503 Delegates settings resolution to the shared rag_service_factory, then
2504 propagates db_password to the embedding manager for thread-safe DB access.
2505 """
2506 from ..services.rag_service_factory import (
2507 get_rag_service as _get_rag_service,
2508 )
2510 service = _get_rag_service(
2511 username,
2512 collection_id,
2513 use_defaults=use_defaults,
2514 db_password=db_password,
2515 )
2516 # The factory passes db_password to LibraryRAGService, but __init__ stores
2517 # it in the backing field (_db_password) without propagating to sub-managers.
2518 # Re-assign via the property setter to propagate to embedding_manager and
2519 # integrity_manager, which need it for thread-safe session access.
2520 service.db_password = db_password
2521 return service
2524def trigger_auto_index(
2525 document_ids: list[str],
2526 collection_id: str,
2527 username: str,
2528 db_password: str,
2529) -> None:
2530 """
2531 Trigger automatic RAG indexing for documents if auto-indexing is enabled.
2533 This function checks the auto_index_enabled setting and spawns a background
2534 thread to index the specified documents. It does not block the caller.
2536 Args:
2537 document_ids: List of document IDs to index
2538 collection_id: The collection to index into
2539 username: The username for database access
2540 db_password: The user's database password for thread-safe access
2541 """
2542 from ...database.session_context import get_user_db_session
2544 if not document_ids:
2545 logger.debug("No documents to auto-index")
2546 return
2548 # Check if auto-indexing is enabled
2549 try:
2550 with get_user_db_session(username, db_password) as db_session:
2551 settings = SettingsManager(db_session)
2552 auto_index_enabled = settings.get_bool_setting(
2553 "research_library.auto_index_enabled", True
2554 )
2556 if not auto_index_enabled:
2557 logger.debug("Auto-indexing is disabled, skipping")
2558 return
2559 except Exception:
2560 logger.exception(
2561 "Failed to check auto-index setting, skipping auto-index"
2562 )
2563 return
2565 # Reserve a queue slot before submission. If the indexing queue is
2566 # saturated (too many uploads in flight), drop this auto-index job
2567 # rather than letting the executor's unbounded internal queue grow
2568 # without bound. The user can reindex manually later via the UI.
2569 if not _try_reserve_auto_index_slot():
2570 logger.warning(
2571 "Auto-index queue saturated ({}+ jobs pending); dropping "
2572 "auto-index for {} document(s) in collection {}. "
2573 "Documents are uploaded; trigger a manual reindex if needed.",
2574 _MAX_PENDING_AUTO_INDEX_JOBS,
2575 len(document_ids),
2576 collection_id,
2577 )
2578 return
2580 logger.info(
2581 f"Auto-indexing {len(document_ids)} documents in collection {collection_id}"
2582 )
2584 # Release the reserved slot EXACTLY ONCE per submission. CPython's
2585 # ThreadPoolExecutor.submit() enqueues the work item BEFORE it tries to
2586 # spin up a worker thread; if thread-start raises (e.g.
2587 # RuntimeError("can't start new thread")), the item is already queued, so
2588 # a live worker can run _wrapped_worker (releasing in its finally) AND the
2589 # except block below also fires. A per-call lock + flag makes the release
2590 # idempotent so that double-fire only releases once — otherwise the
2591 # counter under-counts in-flight jobs and erodes the OOM bound (the
2592 # max(0, ...) floor would silently hide the drift).
2593 _release_lock = threading.Lock()
2594 _slot_released = {"done": False}
2596 def _release_slot_once():
2597 with _release_lock:
2598 if _slot_released["done"]:
2599 return
2600 _slot_released["done"] = True
2601 _release_auto_index_slot()
2603 def _wrapped_worker(*args, **kwargs):
2604 try:
2605 _auto_index_documents_worker(*args, **kwargs)
2606 finally:
2607 _release_slot_once()
2609 # Submit to thread pool (bounded concurrency, prevents thread proliferation).
2610 # Build the executor inside the try too: if _get_auto_index_executor()
2611 # itself fails (OS thread/mutex exhaustion), the slot must still be
2612 # released — otherwise a reserved slot leaks permanently.
2613 try:
2614 executor = _get_auto_index_executor()
2615 executor.submit(
2616 _wrapped_worker,
2617 document_ids,
2618 collection_id,
2619 username,
2620 db_password,
2621 )
2622 except Exception:
2623 # If submit itself fails (executor shutting down, OOM, etc.), the
2624 # wrapped worker may never run, so release the slot here. The release
2625 # is idempotent: if the work item was already enqueued and a worker
2626 # ran (and released) before thread-start failed, this is a no-op. Do
2627 # NOT re-raise: the upload has already been committed by the caller,
2628 # so propagating would turn a successful upload into a 500 (and prompt
2629 # the client to retry, creating duplicates). Auto-indexing is simply
2630 # skipped; the user can trigger a manual reindex later via the UI.
2631 _release_slot_once()
2632 logger.exception(
2633 "Failed to submit auto-index job for {} document(s) in "
2634 "collection {}; upload succeeded, auto-indexing skipped. "
2635 "Trigger a manual reindex if needed.",
2636 len(document_ids),
2637 collection_id,
2638 )
2641@thread_cleanup
2642def _auto_index_documents_worker(
2643 document_ids: list[str],
2644 collection_id: str,
2645 username: str,
2646 db_password: str,
2647) -> None:
2648 """
2649 Background worker to index documents automatically.
2651 This is a simpler worker than _background_index_worker - it doesn't track
2652 progress via TaskMetadata since it's meant to be a lightweight auto-indexing
2653 operation.
2654 """
2656 try:
2657 # Create RAG service (thread-safe, no Flask context needed)
2658 with _get_rag_service_for_thread(
2659 collection_id, username, db_password
2660 ) as rag_service:
2661 indexed_count = 0
2662 for doc_id in document_ids:
2663 try:
2664 result = rag_service.index_document(
2665 doc_id, collection_id, force_reindex=False
2666 )
2667 if result.get("status") == "success":
2668 indexed_count += 1
2669 logger.debug(f"Auto-indexed document {doc_id}")
2670 elif result.get("status") == "skipped": 2670 ↛ 2662line 2670 didn't jump to line 2662 because the condition on line 2670 was always true
2671 logger.debug(
2672 f"Document {doc_id} already indexed, skipped"
2673 )
2674 except Exception:
2675 logger.exception(f"Failed to auto-index document {doc_id}")
2677 logger.info(
2678 f"Auto-indexing complete: {indexed_count}/{len(document_ids)} documents indexed"
2679 )
2681 except Exception:
2682 logger.exception("Auto-indexing worker failed")
2685@thread_cleanup
2686def _background_index_worker(
2687 task_id: str,
2688 collection_id: str,
2689 username: str,
2690 db_password: str,
2691 force_reindex: bool,
2692):
2693 """
2694 Background worker thread for indexing documents.
2695 Updates TaskMetadata with progress and checks for cancellation.
2696 """
2697 from ...database.session_context import get_user_db_session
2699 try:
2700 # Create RAG service (thread-safe, no Flask context needed)
2701 with _get_rag_service_for_thread(
2702 collection_id, username, db_password, use_defaults=force_reindex
2703 ) as rag_service:
2704 with get_user_db_session(username, db_password) as db_session:
2705 # Get collection
2706 collection = (
2707 db_session.query(Collection)
2708 .filter_by(id=collection_id)
2709 .first()
2710 )
2712 if not collection:
2713 _update_task_status(
2714 username,
2715 db_password,
2716 task_id,
2717 status="failed",
2718 error_message="Collection not found",
2719 )
2720 return
2722 # Store embedding metadata AND reset the old index in ONE commit,
2723 # so a crash between them can't leave the new config committed
2724 # while the old (stale/mixed-model) vectors survive.
2725 changed = False
2726 faiss_reset_paths = []
2727 if collection.embedding_model is None or force_reindex:
2728 _store_collection_embedding_metadata(
2729 collection, rag_service
2730 )
2731 changed = True
2732 if force_reindex:
2733 faiss_reset_paths = _reset_collection_for_reindex(
2734 db_session, collection_id
2735 )
2736 changed = True
2737 if changed:
2738 db_session.commit()
2739 _unlink_reindex_faiss_files(faiss_reset_paths)
2741 # Get documents to index
2742 doc_links = _query_documents_to_index(
2743 db_session, collection_id, force_reindex
2744 )
2746 if not doc_links:
2747 _update_task_status(
2748 username,
2749 db_password,
2750 task_id,
2751 status="completed",
2752 progress_message="No documents to index",
2753 )
2754 return
2756 total = len(doc_links)
2757 results = {"successful": 0, "skipped": 0, "failed": 0}
2759 # Update task with total count
2760 _update_task_status(
2761 username,
2762 db_password,
2763 task_id,
2764 progress_total=total,
2765 progress_message=f"Indexing {total} documents",
2766 )
2768 for idx, (link, doc) in enumerate(doc_links, 1):
2769 # Check if cancelled
2770 if _is_task_cancelled(username, db_password, task_id):
2771 _update_task_status(
2772 username,
2773 db_password,
2774 task_id,
2775 status="cancelled",
2776 progress_message=f"Cancelled after {idx - 1}/{total} documents",
2777 )
2778 logger.info(f"Indexing task {task_id} was cancelled")
2779 return
2781 filename = doc.filename or doc.title or "Unknown"
2783 # Update progress with filename
2784 _update_task_status(
2785 username,
2786 db_password,
2787 task_id,
2788 progress_current=idx,
2789 progress_message=f"Indexing {idx}/{total}: {filename}",
2790 )
2792 try:
2793 result = rag_service.index_document(
2794 document_id=doc.id,
2795 collection_id=collection_id,
2796 force_reindex=force_reindex,
2797 )
2799 if result.get("status") == "success":
2800 results["successful"] += 1
2801 elif result.get("status") in ("skipped", "cleared"):
2802 # "cleared" (empty-text purge) is a handled non-failure.
2803 results["skipped"] += 1
2804 else:
2805 results["failed"] += 1
2807 except Exception:
2808 results["failed"] += 1
2809 logger.exception(
2810 f"Error indexing document {idx}/{total}"
2811 )
2813 db_session.commit()
2815 # Mark as completed
2816 _update_task_status(
2817 username,
2818 db_password,
2819 task_id,
2820 status="completed",
2821 progress_current=total,
2822 progress_message=f"Completed: {results['successful']} indexed, {results['failed']} failed, {results['skipped']} skipped",
2823 )
2824 logger.info(
2825 f"Background indexing task {task_id} completed: {results}"
2826 )
2828 except Exception as e:
2829 logger.exception(f"Background indexing task {task_id} failed")
2830 _update_task_status(
2831 username,
2832 db_password,
2833 task_id,
2834 status="failed",
2835 # Scrub creds/keys: this is later returned to the client by the
2836 # index-status endpoint; full trace stays in server logs above.
2837 error_message=sanitize_error_message(str(e)),
2838 )
2841def _update_task_status(
2842 username: str,
2843 db_password: str,
2844 task_id: str,
2845 status: str = None,
2846 progress_current: int = None,
2847 progress_total: int = None,
2848 progress_message: str = None,
2849 error_message: str = None,
2850):
2851 """Update task metadata in the database."""
2852 from ...database.session_context import get_user_db_session
2854 try:
2855 with get_user_db_session(username, db_password) as db_session:
2856 task = (
2857 db_session.query(TaskMetadata)
2858 .filter_by(task_id=task_id)
2859 .first()
2860 )
2861 if task:
2862 if status is not None:
2863 task.status = status
2864 if status == "completed":
2865 task.completed_at = datetime.now(UTC)
2866 if progress_current is not None:
2867 task.progress_current = progress_current
2868 if progress_total is not None:
2869 task.progress_total = progress_total
2870 if progress_message is not None:
2871 task.progress_message = progress_message
2872 if error_message is not None:
2873 task.error_message = error_message
2874 db_session.commit()
2875 except Exception:
2876 logger.exception(f"Failed to update task status for {task_id}")
2879def _is_task_cancelled(username: str, db_password: str, task_id: str) -> bool:
2880 """Check if a task has been cancelled."""
2881 from ...database.session_context import get_user_db_session
2883 try:
2884 with get_user_db_session(username, db_password) as db_session:
2885 task = (
2886 db_session.query(TaskMetadata)
2887 .filter_by(task_id=task_id)
2888 .first()
2889 )
2890 return task and task.status == "cancelled"
2891 except Exception:
2892 logger.warning(
2893 "Could not check cancellation status for task {}", task_id
2894 )
2895 return False
2898@rag_bp.route(
2899 "/api/collections/<string:collection_id>/index/start", methods=["POST"]
2900)
2901@login_required
2902def start_background_index(collection_id):
2903 """Start background indexing for a collection."""
2904 from ...database.session_context import get_user_db_session
2905 from ...database.session_passwords import session_password_store
2907 username = session["username"]
2908 session_id = session.get("session_id")
2910 # Get password for thread access
2911 db_password = None
2912 if session_id: 2912 ↛ 2918line 2912 didn't jump to line 2918 because the condition on line 2912 was always true
2913 db_password = session_password_store.get_session_password(
2914 username, session_id
2915 )
2917 # Parse request body
2918 data = request.get_json() or {}
2919 force_reindex = data.get("force_reindex", False)
2921 try:
2922 with get_user_db_session(username, db_password) as db_session:
2923 # Check if there's already an active indexing task for this collection
2924 existing_task = (
2925 db_session.query(TaskMetadata)
2926 .filter(
2927 TaskMetadata.task_type == "indexing",
2928 TaskMetadata.status == "processing",
2929 )
2930 .first()
2931 )
2933 if existing_task:
2934 # Check if it's for this collection
2935 metadata = existing_task.metadata_json or {}
2936 if metadata.get("collection_id") == collection_id:
2937 return jsonify(
2938 {
2939 "success": False,
2940 "error": "Indexing is already in progress for this collection",
2941 "task_id": existing_task.task_id,
2942 }
2943 ), 409
2945 # Create new task
2946 task_id = str(uuid.uuid4())
2947 task = TaskMetadata(
2948 task_id=task_id,
2949 status="processing",
2950 task_type="indexing",
2951 created_at=datetime.now(UTC),
2952 started_at=datetime.now(UTC),
2953 progress_current=0,
2954 progress_total=0,
2955 progress_message="Starting indexing...",
2956 metadata_json={
2957 "collection_id": collection_id,
2958 "force_reindex": force_reindex,
2959 },
2960 )
2961 db_session.add(task)
2962 db_session.commit()
2964 # Start background thread
2965 thread = threading.Thread(
2966 target=_background_index_worker,
2967 args=(task_id, collection_id, username, db_password, force_reindex),
2968 daemon=True,
2969 )
2970 thread.start()
2972 logger.info(
2973 f"Started background indexing task {task_id} for collection {collection_id}"
2974 )
2976 return jsonify(
2977 {
2978 "success": True,
2979 "task_id": task_id,
2980 "message": "Indexing started in background",
2981 }
2982 )
2984 except Exception:
2985 logger.exception("Failed to start background indexing")
2986 return jsonify(
2987 {
2988 "success": False,
2989 "error": "Failed to start indexing. Please try again.",
2990 }
2991 ), 500
2994@rag_bp.route(
2995 "/api/collections/<string:collection_id>/index/status", methods=["GET"]
2996)
2997@limiter.exempt
2998@login_required
2999def get_index_status(collection_id):
3000 """Get the current indexing status for a collection."""
3001 from ...database.session_context import get_user_db_session
3002 from ...database.session_passwords import session_password_store
3004 username = session["username"]
3005 session_id = session.get("session_id")
3007 db_password = None
3008 if session_id: 3008 ↛ 3013line 3008 didn't jump to line 3013 because the condition on line 3008 was always true
3009 db_password = session_password_store.get_session_password(
3010 username, session_id
3011 )
3013 try:
3014 with get_user_db_session(username, db_password) as db_session:
3015 # Find the most recent indexing task FOR THIS COLLECTION.
3016 #
3017 # collection_id is stored inside the metadata_json JSON column, so
3018 # we can't portably filter on it in SQL (SQLite/SQLCipher JSON
3019 # support varies). Instead, scan recent indexing tasks newest-first
3020 # and return the first whose metadata.collection_id matches. This
3021 # is scoped per collection: a newer indexing task for a DIFFERENT
3022 # collection no longer makes this one falsely report "idle".
3023 #
3024 # This endpoint is polled every ~2s during a reindex and the
3025 # indexing-task table is never pruned, so an unbounded .all() would
3026 # materialize the entire history on every poll (and task_type /
3027 # created_at are unindexed). Bound the scan to the newest N tasks.
3028 # The task we're looking for was just created by the reindex that
3029 # started this poll, so it sits near the top; N=200 is large enough
3030 # that concurrent reindexes of other collections can't push it out
3031 # of the window before it terminates. Trade-off: if more than ~200
3032 # newer indexing tasks for OTHER collections appear while this one
3033 # is still in-flight, the scoped lookup falls back to "idle".
3034 recent_task_scan_limit = 200
3035 recent_tasks = (
3036 db_session.query(TaskMetadata)
3037 .filter(TaskMetadata.task_type == "indexing")
3038 .order_by(TaskMetadata.created_at.desc())
3039 .limit(recent_task_scan_limit)
3040 .all()
3041 )
3043 task = None
3044 for candidate in recent_tasks:
3045 metadata = candidate.metadata_json or {}
3046 if metadata.get("collection_id") == collection_id:
3047 task = candidate
3048 break
3050 if not task:
3051 return jsonify(
3052 {
3053 "status": "idle",
3054 "collection_id": collection_id,
3055 "message": "No indexing task for this collection",
3056 }
3057 )
3059 return jsonify(
3060 {
3061 "task_id": task.task_id,
3062 "collection_id": collection_id,
3063 "status": task.status,
3064 "progress_current": task.progress_current or 0,
3065 "progress_total": task.progress_total or 0,
3066 "progress_message": task.progress_message,
3067 "error_message": task.error_message,
3068 "created_at": task.created_at.isoformat()
3069 if task.created_at
3070 else None,
3071 "completed_at": task.completed_at.isoformat()
3072 if task.completed_at
3073 else None,
3074 }
3075 )
3077 except Exception:
3078 logger.exception("Failed to get index status")
3079 return jsonify(
3080 {
3081 "status": "error",
3082 "error": "Failed to get indexing status. Please try again.",
3083 }
3084 ), 500
3087@rag_bp.route(
3088 "/api/collections/<string:collection_id>/index/cancel", methods=["POST"]
3089)
3090@login_required
3091def cancel_indexing(collection_id):
3092 """Cancel an active indexing task for a collection."""
3093 from ...database.session_context import get_user_db_session
3094 from ...database.session_passwords import session_password_store
3096 username = session["username"]
3097 session_id = session.get("session_id")
3099 db_password = None
3100 if session_id: 3100 ↛ 3105line 3100 didn't jump to line 3105 because the condition on line 3100 was always true
3101 db_password = session_password_store.get_session_password(
3102 username, session_id
3103 )
3105 try:
3106 with get_user_db_session(username, db_password) as db_session:
3107 # Find active indexing task for this collection
3108 task = (
3109 db_session.query(TaskMetadata)
3110 .filter(
3111 TaskMetadata.task_type == "indexing",
3112 TaskMetadata.status == "processing",
3113 )
3114 .first()
3115 )
3117 if not task:
3118 return jsonify(
3119 {
3120 "success": False,
3121 "error": "No active indexing task found",
3122 }
3123 ), 404
3125 # Check if it's for this collection
3126 metadata = task.metadata_json or {}
3127 if metadata.get("collection_id") != collection_id:
3128 return jsonify(
3129 {
3130 "success": False,
3131 "error": "No active indexing task for this collection",
3132 }
3133 ), 404
3135 # Mark as cancelled - the worker thread will check this
3136 task.status = "cancelled"
3137 task.progress_message = "Cancellation requested..."
3138 db_session.commit()
3140 logger.info(
3141 f"Cancelled indexing task {task.task_id} for collection {collection_id}"
3142 )
3144 return jsonify(
3145 {
3146 "success": True,
3147 "message": "Cancellation requested",
3148 "task_id": task.task_id,
3149 }
3150 )
3152 except Exception:
3153 logger.exception("Failed to cancel indexing")
3154 return jsonify(
3155 {
3156 "success": False,
3157 "error": "Failed to cancel indexing. Please try again.",
3158 }
3159 ), 500
3162# Research History Semantic Search Routes have been moved to
3163# research_library.search.routes.search_routes