Coverage for src/local_deep_research/web/routers/rag.py: 95%
1317 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 asyncio
13from fastapi import APIRouter, Depends, HTTPException, Request
14from fastapi.responses import HTMLResponse, JSONResponse
15from ..dependencies.auth import require_auth
16from ..dependencies.rate_limit import (
17 limiter,
18 upload_rate_limit_ip,
19 upload_rate_limit_user,
20)
21from ..dependencies.threadpool import (
22 WorkerCleanupStreamingResponse,
23 run_db_sync,
24)
25from ..template_config import templates
27import os
29from loguru import logger
30from sqlalchemy import case, func
31import atexit
32import json
33import uuid
34import time
35import contextvars
36import threading
37import queue
38from concurrent.futures import ThreadPoolExecutor
39from datetime import datetime, UTC
40from pathlib import Path
41from typing import Any, Dict, Optional, Set, Tuple, Annotated
42from tenacity import (
43 retry,
44 retry_if_exception,
45 stop_after_attempt,
46 wait_exponential,
47)
49from sqlalchemy.orm import defer
51from ...constants import (
52 DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
53 DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
54 DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
55 DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
56 DEFAULT_LOCAL_SEARCH_MODEL,
57 DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
58 DEFAULT_LOCAL_SEARCH_PROVIDER,
59 DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
60 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
61 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
62 FILE_PATH_SENTINELS,
63 FILE_PATH_TEXT_ONLY,
64)
65from ...security.log_sanitizer import (
66 sanitize_error_for_client,
67 sanitize_error_message,
68)
69from ...utilities.db_utils import get_settings_manager
70from ...utilities.resource_utils import safe_close
71from ...research_library.services.library_rag_service import LibraryRAGService
72from ...settings.manager import SettingsManager, check_env_setting
73from ...research_library.utils import (
74 apply_user_subdir,
75 ensure_in_collection,
76 handle_api_error,
77)
78from ...database.models.library import (
79 Document,
80 Collection,
81 DocumentCollection,
82 RAGIndex,
83 SourceType,
84 EmbeddingProvider,
85)
86from ...database.models.queue import TaskMetadata
87from ...database.thread_local_session import thread_cleanup
89from ...config.paths import get_library_directory
90from ...constants import DEFAULT_SEARCH_TOOL
91from ..dependencies.json_body import json_body_error
93#: Upper bound on ``page`` -- see context_overflow_api for the rationale.
94_MAX_PAGE = 10_000
96# Keep the configure endpoint's boundary aligned with the registered setting
97# schema in defaults/settings_local_search.json. Text splitters require integer
98# sizes; accepting strings, floats, or booleans here would silently coerce or
99# truncate them only after the request reached the settings/index write path.
100_RAG_CHUNK_SIZE_MIN = 100
101_RAG_CHUNK_SIZE_MAX = 5_000
102_RAG_CHUNK_OVERLAP_MIN = 0
103_RAG_CHUNK_OVERLAP_MAX = 1_000
105router = APIRouter(prefix="/library", tags=["rag"])
107# Process-local registry tracking active SSE indexing streams for cancellation.
108# Keyed by (username, collection_id) -> Set[threading.Event]
109# NOTE: Process-local registry assumes single-process deployment (or sticky sessions).
110# Multi-worker deployments (e.g. uvicorn with multiple workers without sticky sessions)
111# would require cross-process signaling (e.g. DB-backed cancel flag or Redis pub/sub).
112_active_sse_indexers: Dict[Tuple[str, str], Set[threading.Event]] = {}
113_active_sse_indexers_lock = threading.Lock()
115# NOTE: Routes use username (not .get()) intentionally.
116# Depends(require_auth) guarantees the key exists; direct access fails
117# fast if the dependency is ever removed.
120def _agent_enabled_default_on(collection) -> bool:
121 """Read a collection's ``agent_enabled`` flag with NULL → available.
123 A missing attribute (pre-migration object) or a stored NULL both mean
124 "available to the research agent" (default-on); only an explicit ``False``
125 disables. Used by every serializer so the four read sites can't drift on
126 null-handling. ``is not False`` is safe here because the source is a
127 Boolean column (only ever None/True/False) — NOT raw JSON input, where a
128 falsy ``0`` must still mean disabled (handled separately at the input edge).
129 """
130 return getattr(collection, "agent_enabled", True) is not False
133def _is_protected_collection(collection) -> bool:
134 """True when the collection's type is deletion-protected.
136 Single source of truth is PROTECTED_COLLECTION_TYPES in the deletion
137 service (imported lazily, matching update_collection's usage); the
138 serializers expose the result as ``is_protected`` so UI surfaces can
139 hide destructive affordances instead of offering an action the server
140 categorically refuses with a 409.
141 """
142 from ...research_library.deletion.services.collection_deletion import (
143 PROTECTED_COLLECTION_TYPES,
144 )
146 return collection.collection_type in PROTECTED_COLLECTION_TYPES
149# Global ThreadPoolExecutor for auto-indexing to prevent thread proliferation.
150#
151# ThreadPoolExecutor's internal _work_queue is unbounded. Without
152# backpressure, a sustained burst of uploads (possible under the
153# configurable upload rate cap from #3935) could queue thousands of indexing
154# jobs in memory, exhausting RAM. We track pending submissions in a counter
155# and reject new submissions once the queue is saturated.
156_auto_index_executor: ThreadPoolExecutor | None = None
157_auto_index_executor_lock = threading.Lock()
159# Per-(user, collection) locks guarding the start-background-index
160# check-and-create path. Keyed by tuple so two users indexing different
161# collections don't serialize against each other.
162_start_bg_index_locks: dict[tuple[str, str], threading.Lock] = {}
164# Maximum in-flight + queued indexing jobs. Each job is one upload batch
165# (an unbounded list of document IDs), so per-job duration varies with batch
166# size; 100 is a buffer on the number of *queued batches*, not documents.
167# The OOM bound still holds because a queued job holds only a small list of
168# ID strings, not document contents. Tunable if real-world workloads need
169# more headroom.
170_MAX_PENDING_AUTO_INDEX_JOBS = 100
171_pending_auto_index_jobs = 0
172_pending_auto_index_lock = threading.Lock()
175def _get_auto_index_executor() -> ThreadPoolExecutor:
176 """Get or create the global auto-indexing executor (thread-safe)."""
177 global _auto_index_executor
178 with _auto_index_executor_lock:
179 if _auto_index_executor is None:
180 _auto_index_executor = ThreadPoolExecutor(
181 max_workers=4,
182 thread_name_prefix="auto_index_",
183 )
184 return _auto_index_executor
187def _try_reserve_auto_index_slot() -> bool:
188 """Reserve a queue slot if capacity allows. Returns True on success.
190 Caller MUST call ``_release_auto_index_slot`` when the job completes
191 (success OR failure), otherwise the counter leaks and eventually
192 blocks all submissions.
193 """
194 global _pending_auto_index_jobs
195 with _pending_auto_index_lock:
196 if _pending_auto_index_jobs >= _MAX_PENDING_AUTO_INDEX_JOBS:
197 return False
198 _pending_auto_index_jobs += 1
199 return True
202def _release_auto_index_slot() -> None:
203 """Release a previously reserved queue slot."""
204 global _pending_auto_index_jobs
205 with _pending_auto_index_lock:
206 _pending_auto_index_jobs = max(0, _pending_auto_index_jobs - 1)
209def _shutdown_auto_index_executor() -> None:
210 """Shutdown the auto-index executor gracefully.
212 Holds ``_auto_index_executor_lock`` for the read+nullify so two
213 concurrent callers (e.g. atexit + a test teardown) can't race into an
214 ``AttributeError`` on the ``None`` reference; the blocking ``shutdown``
215 runs outside the lock. Resets the pending-jobs counter afterwards so a
216 re-created executor (tests, a dev-server reload) starts from a clean count
217 instead of inheriting a stale, possibly-saturated value.
218 """
219 global _auto_index_executor, _pending_auto_index_jobs
220 with _auto_index_executor_lock:
221 executor = _auto_index_executor
222 _auto_index_executor = None
223 if executor is not None:
224 executor.shutdown(wait=True)
225 with _pending_auto_index_lock:
226 _pending_auto_index_jobs = 0
229atexit.register(_shutdown_auto_index_executor)
232def get_rag_service(
233 request: Request,
234 username: str,
235 collection_id: Optional[str] = None,
236 use_defaults: bool = False,
237) -> LibraryRAGService:
238 """
239 Get RAG service instance with appropriate settings.
241 Args:
242 request: The FastAPI request (for session lookup).
243 username: The authenticated username.
244 collection_id: Optional collection UUID to load stored settings from
245 use_defaults: When True, ignore stored collection settings and use
246 current defaults. Pass True on force-reindex so that the new
247 default embedding model is picked up.
248 """
249 from ...research_library.services.rag_service_factory import (
250 get_rag_service as _get_rag_service,
251 )
252 from ...database.session_passwords import session_password_store
254 session_id = request.session.get("session_id")
255 db_password = None
256 if session_id:
257 db_password = session_password_store.get_session_password(
258 username, session_id
259 )
260 if not db_password:
261 db_password = session_password_store.get_any_session_password(username)
262 return _get_rag_service(
263 username,
264 collection_id,
265 use_defaults=use_defaults,
266 db_password=db_password,
267 )
270def _get_text_separators(settings):
271 """Return configured text separators, parsing string values if needed."""
272 # Load the stored setting. It may already be a list or a JSON string.
273 text_separators = settings.get_setting(
274 "local_search_text_separators",
275 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
276 )
278 # If the setting is stored as a string, parse it into a list. A value that
279 # is not valid JSON (e.g. a not-yet-migrated corrupt row) falls back to the
280 # default separators — migration #4298 heals existing corrupt data.
281 if isinstance(text_separators, str):
282 try:
283 text_separators = json.loads(text_separators)
284 except json.JSONDecodeError:
285 logger.warning(
286 "Invalid JSON for local_search_text_separators setting: {!r}. Using default separators.",
287 text_separators,
288 )
289 text_separators = DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
291 return text_separators
294def _parse_configured_text_separators(value):
295 """Coerce a text-separator setting to a list of strings, or None.
297 Called via ``asyncio.to_thread`` from ``configure_rag``: ``value``
298 comes straight from the request body, so a caller can hand it a
299 string up to that route's 16 MB body cap. ``json.loads`` runs at
300 ~110 ms/MB, which is ~1.7 s at the cap -- fine on a worker thread
301 (Flask charged the same work to a request thread), a whole-instance
302 freeze on the event loop under single-worker uvicorn.
303 """
304 separators = value
305 if isinstance(value, str):
306 try:
307 separators = json.loads(value)
308 except json.JSONDecodeError:
309 return None
310 if not isinstance(separators, list) or not all(
311 isinstance(separator, str) for separator in separators
312 ):
313 return None
314 return separators
317# Module prefix for exceptions DEFINED inside LDR itself. An exception whose
318# class lives under ``local_deep_research`` is something only we can fix, so we
319# steer the user to file a bug instead of second-guessing their model choice.
320#
321# We deliberately do NOT treat bare builtins (KeyError/TypeError/RuntimeError/
322# ...) as internal: ``type(exc).__module__`` is where the class is *defined*,
323# not where it was *raised*, so a builtin tells us nothing about whether LDR or
324# a provider/langchain frame produced it. On this path an OpenAI-compatible
325# server that returns a malformed 200 body can surface a bare TypeError from
326# langchain's response parser; telling the user to "report it on GitHub" for
327# that is exactly the misleading guidance #4208 set out to remove.
328_INTERNAL_MODULE_PREFIX = "local_deep_research"
330# Module prefixes whose exceptions are upstream-provider errors (network,
331# auth, model not found, etc.). Matched against ``type(exc).__module__``.
332_UPSTREAM_MODULE_PREFIXES = (
333 "openai",
334 "httpx",
335 "requests",
336 "urllib3",
337 "anthropic",
338)
341def _module_matches(type_module: str, prefix: str) -> bool:
342 """True if ``type_module`` is ``prefix`` itself or one of its submodules."""
343 return type_module == prefix or type_module.startswith(f"{prefix}.")
346def _format_test_embedding_error(exc: Exception, model: str) -> str:
347 """Build the user-facing error message for /api/rag/test-embedding.
349 Categorizes the exception by the module its *class* is defined in so the
350 UI distinguishes LDR-internal bugs (which the user can't fix and
351 shouldn't be steered into changing their model over) from real
352 upstream/provider errors. The previous implementation ran a keyword-match
353 heuristic over every exception text — see #4208, where
354 ``NoSettingsContextError`` surfaced as a misleading "try a dedicated
355 embedding model" suggestion.
357 Echoing the exception text at all is *opt-in* (CWE-209 / CodeQL alert
358 8001). Only the allowlisted upstream-provider modules get their detail
359 surfaced; every other module is treated the way LDR-internal exceptions
360 already were — class name only. ``sanitize_error_message`` removes
361 credential *shapes*, not server filesystem paths, SQL text or dependency
362 internals, and those are exactly what a ``sqlalchemy.exc.*`` or
363 ``builtins.OSError`` raised under ``get_embedding_function`` carries.
365 The detail that *is* surfaced goes through
366 :func:`sanitize_error_for_client` (credential redaction, control-char
367 strip and a length cap) rather than the bare credential scrubber, so a
368 provider cannot replay a multi-kilobyte body into the response.
369 """
370 raw_message = str(exc).strip() or type(exc).__name__
371 type_name = type(exc).__name__
372 type_module = type(exc).__module__ or ""
374 if _module_matches(type_module, _INTERNAL_MODULE_PREFIX):
375 # Internal LDR exceptions can carry filesystem paths / SQL fragments
376 # that sanitize_error_message() does not pattern-match, so we do not
377 # echo their detail to the browser — the full trace is in the server
378 # logs (logger.exception at the call site) which the user is asked to
379 # attach.
380 return (
381 f"Embedding test failed for model '{model}' due to an "
382 f"internal LDR error ({type_name}). This is a bug in LDR, "
383 "not your configuration. Please report it on GitHub with "
384 "the server logs."
385 )
387 if any(
388 _module_matches(type_module, prefix)
389 for prefix in _UPSTREAM_MODULE_PREFIXES
390 ):
391 return (
392 f"Embedding test failed for model '{model}'. The provider "
393 f"returned an error: {sanitize_error_for_client(raw_message)}"
394 )
396 # Default-deny. Everything outside the upstream allowlist reaches here:
397 # sqlalchemy.exc.* (the try block opens the user's SQLCipher database, and
398 # a DBAPIError's str() renders the driver message plus [SQL: ...] and
399 # [parameters: ...]), builtins.OSError/PermissionError/FileNotFoundError
400 # (absolute server paths and the OS account name), chromadb,
401 # sentence_transformers, langchain_core, botocore... None of that is
402 # touched by the credential-shape scrubber, so the detail is withheld and
403 # only the class name — a fixed identifier, not reflected input — is
404 # returned. The full text is in the server logs (logger.exception at the
405 # call site).
406 #
407 # Worded deliberately unlike the internal-LDR branch above: a bare
408 # TypeError from langchain's response parser is not an LDR bug, and
409 # steering the user to "report it on GitHub" for it is the misleading
410 # guidance #4208 removed.
411 return (
412 f"Embedding test failed for model '{model}' ({type_name}). "
413 "Check the provider URL and model name; the full error is in the "
414 "server logs."
415 )
418# Config API Routes
421@router.get("/api/config/supported-formats")
422def get_supported_formats(
423 request: Request, username: Annotated[str, Depends(require_auth)]
424):
425 """Return list of supported file formats for upload.
427 This endpoint provides the single source of truth for supported file
428 extensions, pulling from the document_loaders registry. The UI can
429 use this to dynamically update the file input accept attribute.
430 """
431 from ...document_loaders import get_supported_extensions
433 extensions = get_supported_extensions()
434 # Sort extensions for consistent display
435 extensions = sorted(extensions)
437 return {
438 "extensions": extensions,
439 "accept_string": ",".join(extensions),
440 "count": len(extensions),
441 }
444# Page Routes
447@router.get("/embedding-settings")
448def embedding_settings_page(
449 request: Request, username: Annotated[str, Depends(require_auth)]
450):
451 """Render the Embedding Settings page."""
452 return templates.TemplateResponse(
453 request=request,
454 name="pages/embedding_settings.html",
455 context={"request": request, "active_page": "embedding-settings"},
456 )
459@router.get("/document/{document_id}/chunks")
460def view_document_chunks(
461 request: Request,
462 document_id,
463 username: Annotated[str, Depends(require_auth)],
464):
465 """View all chunks for a document across all collections."""
466 from ...database.session_context import get_user_db_session
467 from ...database.models.library import DocumentChunk
469 with get_user_db_session(username) as db_session:
470 # Get document info
471 document = db_session.query(Document).filter_by(id=document_id).first()
473 if not document:
474 # Browser navigation: return text/html, not JSON. These four page
475 # routes are reached from library.html as ordinary <a href> links, so a
476 # stale link showed the user a raw {"error": ...} body in the browser's
477 # JSON viewer. main returned `"Document not found", 404` (text/html)
478 # here; the sibling /api/ routes keep JSON.
479 return HTMLResponse("Document not found", status_code=404)
481 # Get all chunks for this document
482 chunks = (
483 db_session.query(DocumentChunk)
484 .filter(DocumentChunk.source_id == document_id)
485 .order_by(DocumentChunk.collection_name, DocumentChunk.chunk_index)
486 .all()
487 )
489 # Group chunks by collection
490 chunks_by_collection = {}
491 for chunk in chunks:
492 coll_name = chunk.collection_name
493 if coll_name not in chunks_by_collection: 493 ↛ 507line 493 didn't jump to line 507 because the condition on line 493 was always true
494 # Get collection display name
495 collection_id = coll_name.replace("collection_", "")
496 collection = (
497 db_session.query(Collection)
498 .filter_by(id=collection_id)
499 .first()
500 )
501 chunks_by_collection[coll_name] = {
502 "name": collection.name if collection else coll_name,
503 "id": collection_id,
504 "chunks": [],
505 }
507 chunks_by_collection[coll_name]["chunks"].append(
508 {
509 "id": chunk.id,
510 "index": chunk.chunk_index,
511 "text": chunk.chunk_text,
512 "word_count": chunk.word_count,
513 "start_char": chunk.start_char,
514 "end_char": chunk.end_char,
515 "embedding_model": chunk.embedding_model,
516 "embedding_model_type": chunk.embedding_model_type.value
517 if chunk.embedding_model_type
518 else None,
519 "embedding_dimension": chunk.embedding_dimension,
520 "created_at": chunk.created_at,
521 }
522 )
524 return templates.TemplateResponse(
525 request=request,
526 name="pages/document_chunks.html",
527 context={
528 "request": request,
529 "document": document,
530 "chunks_by_collection": chunks_by_collection,
531 "total_chunks": len(chunks),
532 },
533 )
536@router.get("/collections")
537def collections_page(
538 request: Request, username: Annotated[str, Depends(require_auth)]
539):
540 """Render the Collections page."""
541 return templates.TemplateResponse(
542 request=request,
543 name="pages/collections.html",
544 context={"request": request, "active_page": "collections"},
545 )
548# NOTE: this static route MUST be registered before the parameterized
549# /collections/{collection_id} below — FastAPI matches in registration
550# order, so the param route would otherwise swallow /collections/create
551# as collection_id="create" and render the details page instead of the
552# create form (fenced by tests/web/routers/test_route_ordering.py).
553@router.get("/collections/create")
554def collection_create_page(
555 request: Request, username: Annotated[str, Depends(require_auth)]
556):
557 """Render the Create Collection page."""
558 return templates.TemplateResponse(
559 request=request,
560 name="pages/collection_create.html",
561 context={"request": request, "active_page": "collections"},
562 )
565def _validated_collection_id(collection_id: str) -> str:
566 """Return `collection_id` if it is a well-formed collection id, else 404.
568 Collection ids are server-generated UUID4 strings (`Collection.id` is
569 String(36), assigned from `str(uuid.uuid4())`), so anything else is either
570 a typo or an injection attempt and cannot match a real row.
572 This is the fix for a stored-reflection XSS: the page templates interpolate
573 this value into an inline `onclick` handler, and HTML-escaping is NOT
574 sufficient there -- the browser entity-decodes attribute content before
575 treating it as JS, so an escaped quote reopens the string. Validating the
576 shape at the boundary means no template can receive a value capable of
577 breaking out, regardless of the context it is interpolated into. That is
578 strictly more robust than fixing the one template, because it also covers
579 collection_details.html and anything added later.
581 (`/collections/create` is a separate static route registered ahead of the
582 parameterised ones, so it never reaches this.)
583 """
584 try:
585 uuid.UUID(str(collection_id))
586 except (ValueError, AttributeError, TypeError):
587 logger.warning(f"Rejected malformed collection_id: {collection_id!r}")
588 raise HTTPException(status_code=404, detail="Collection not found")
589 return str(collection_id)
592@router.get("/collections/{collection_id}")
593def collection_details_page(
594 request: Request,
595 collection_id,
596 username: Annotated[str, Depends(require_auth)],
597):
598 """Render the Collection Details page."""
599 collection_id = _validated_collection_id(collection_id)
600 return templates.TemplateResponse(
601 request=request,
602 name="pages/collection_details.html",
603 context={
604 "active_page": "collections",
605 "collection_id": collection_id,
606 },
607 )
610@router.get("/collections/{collection_id}/upload")
611def collection_upload_page(
612 request: Request,
613 collection_id,
614 username: Annotated[str, Depends(require_auth)],
615):
616 """Render the Collection Upload page."""
617 collection_id = _validated_collection_id(collection_id)
619 from ...database.session_context import get_user_db_session
620 from ...utilities.db_utils import get_settings_manager
622 # Get the upload PDF storage setting
623 with get_user_db_session(username) as db_session:
624 settings = get_settings_manager(db_session, username)
625 upload_pdf_storage = settings.get_setting(
626 "research_library.upload_pdf_storage", "none"
627 )
628 # Only allow valid values for uploads (no filesystem)
629 if upload_pdf_storage not in ("database", "none"):
630 upload_pdf_storage = "none"
632 return templates.TemplateResponse(
633 request=request,
634 name="pages/collection_upload.html",
635 context={
636 "active_page": "collections",
637 "collection_id": collection_id,
638 "collection_name": None,
639 "upload_pdf_storage": upload_pdf_storage,
640 },
641 )
644# API Routes
647@router.get("/api/rag/settings")
648def get_current_settings(
649 request: Request,
650 username: Annotated[str, Depends(require_auth)],
651):
652 """Get current RAG configuration from settings."""
653 from ...database.session_context import get_user_db_session
655 try:
656 with get_user_db_session(username) as db_session:
657 settings = get_settings_manager(db_session, username)
659 return {
660 "success": True,
661 "settings": {
662 "embedding_provider": settings.get_setting(
663 "local_search_embedding_provider",
664 DEFAULT_LOCAL_SEARCH_PROVIDER,
665 ),
666 "embedding_model": settings.get_setting(
667 "local_search_embedding_model",
668 DEFAULT_LOCAL_SEARCH_MODEL,
669 ),
670 "chunk_size": settings.get_setting(
671 "local_search_chunk_size",
672 DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
673 ),
674 "chunk_overlap": settings.get_setting(
675 "local_search_chunk_overlap",
676 DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
677 ),
678 "splitter_type": settings.get_setting(
679 "local_search_splitter_type",
680 DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
681 ),
682 "text_separators": _get_text_separators(settings),
683 "distance_metric": settings.get_setting(
684 "local_search_distance_metric",
685 DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
686 ),
687 "normalize_vectors": settings.get_setting(
688 "local_search_normalize_vectors",
689 DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
690 ),
691 "index_type": settings.get_setting(
692 "local_search_index_type",
693 DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
694 ),
695 },
696 }
697 except Exception as e:
698 return handle_api_error("getting RAG settings", e)
701@router.post("/api/rag/test-embedding")
702async def test_embedding(
703 request: Request,
704 username: Annotated[str, Depends(require_auth)],
705):
706 """Test an embedding configuration by generating a test embedding."""
707 from ...database.session_context import get_user_db_session
708 from ...utilities.db_utils import get_settings_manager
710 # Pre-bound: the except block below formats an error using these, so a
711 # malformed/empty body would otherwise raise UnboundLocalError from
712 # inside the handler's own error path.
713 provider = None
714 model = None
715 try:
716 data = await request.json()
717 if not isinstance(data, dict):
718 return json_body_error("success", "Request body must be valid JSON")
719 provider = data.get("provider")
720 model = data.get("model")
721 test_text = data.get("test_text", "This is a test.")
723 if not provider or not model:
724 return JSONResponse(
725 {"success": False, "error": "Provider and model are required"},
726 status_code=400,
727 )
729 # Import embedding functions
730 from ...embeddings.embeddings_config import (
731 get_embedding_function,
732 )
734 logger.info(
735 f"Testing embedding with provider={provider}, model={model}"
736 )
738 # Get user's settings so provider URLs (e.g. Ollama) are resolved correctly.
739 # Sync SQLAlchemy + a network-bound embedding call — both must run in a
740 # thread, never on the uvicorn event loop.
741 def _run_embedding_test() -> tuple[Any, int]:
742 with get_user_db_session(username) as db_session:
743 settings = get_settings_manager(db_session, username)
744 settings_snapshot = (
745 settings.get_all_settings()
746 if hasattr(settings, "get_all_settings")
747 else {}
748 )
749 start = time.time()
750 ef = get_embedding_function(
751 provider=provider,
752 model_name=model,
753 settings_snapshot=settings_snapshot,
754 )
755 emb = ef([test_text])[0]
756 return emb, int((time.time() - start) * 1000)
758 # run_db_sync (not raw to_thread): the thunk opens a
759 # get_user_db_session block, and to_thread's reused workers would
760 # keep the user's thread-local session attached after the task.
761 embedding, response_time_ms = await run_db_sync(_run_embedding_test)
763 # Get embedding dimension
764 dimension = len(embedding) if hasattr(embedding, "__len__") else None
766 return {
767 "success": True,
768 "dimension": dimension,
769 "response_time_ms": response_time_ms,
770 "provider": provider,
771 "model": model,
772 }
774 except json.JSONDecodeError:
775 # Caught here (ahead of the broad `except Exception` below) so a
776 # malformed body gets the route's own "success"-shaped 400 instead
777 # of falling into `_format_test_embedding_error`, which would echo
778 # the raw decoder message (e.g. "Expecting value: line 1 column 1")
779 # back to the client as part of a hardcoded 500.
780 return json_body_error("success", "Request body must be valid JSON")
781 except Exception as e:
782 # Egress policy denial → clean 4xx (the embedding never fired).
783 from ...security.egress.policy import PolicyDeniedError
785 if isinstance(e, PolicyDeniedError): 785 ↛ 786line 785 didn't jump to line 786 because the condition on line 785 was never true
786 return JSONResponse(
787 {
788 "success": False,
789 "error": (
790 f"Embedding provider '{provider}' refused by egress "
791 f"policy ({e.decision.reason}). Disable 'Require "
792 "local embeddings' or pick a local provider."
793 ),
794 },
795 status_code=400,
796 )
797 logger.exception("Error during testing embedding")
798 user_message = _format_test_embedding_error(e, model)
799 # CWE-209 (CodeQL "Information exposure through an exception"):
800 # user_message is exception-derived, but _format_test_embedding_error()
801 # echoes exception text only for the allowlisted upstream-provider
802 # modules, and only after sanitize_error_for_client(). Every other
803 # exception — LDR-internal, sqlalchemy.exc.*, OSError, chromadb... —
804 # is reduced to its class name, so no server path, SQL fragment or
805 # dependency internal reaches this response. See that function's
806 # docstring for the full rationale.
807 return JSONResponse(
808 {"success": False, "error": user_message}, status_code=500
809 )
812@router.get("/api/rag/models")
813def get_available_models(
814 request: Request,
815 username: Annotated[str, Depends(require_auth)],
816):
817 """Get list of available embedding providers and models."""
818 from ...database.session_context import get_user_db_session
819 from ...utilities.db_utils import get_settings_manager
821 try:
822 from ...embeddings.embeddings_config import _get_provider_classes
823 from ...security.egress.policy import (
824 Decision,
825 PolicyDeniedError,
826 )
828 # Get current settings for providers. STRICT snapshot is mandatory
829 # at this policy-sensitive seam: a non-strict read silently falls
830 # back to JSON defaults when the underlying settings query fails
831 # (SQLAlchemyError / stale enum row), and a defaults-only snapshot
832 # can lack the operator-selected scope/provider inputs that the
833 # local-only posture is supposed to enforce. We refuse before any
834 # provider discovery, credential read, or network probe instead of
835 # admitting a cloud embedder under a permissive fallback.
836 with get_user_db_session(username) as db_session:
837 settings = get_settings_manager(db_session, username)
838 try:
839 settings_snapshot = (
840 settings.get_all_settings(strict=True)
841 if hasattr(settings, "get_all_settings")
842 else {}
843 )
844 except PolicyDeniedError:
845 raise
846 except Exception as exc:
847 raise PolicyDeniedError(
848 Decision(False, "settings_unavailable"),
849 target="available_rag_models",
850 ) from exc
852 # Get provider classes
853 provider_classes = _get_provider_classes()
855 # Egress policy: fetching a cloud provider's model list opens a
856 # connection that carries the API key off-machine. Under an effective
857 # local-only embeddings posture (PRIVATE_ONLY / adaptive-private) we
858 # must NOT probe remote-classified providers — mirror the PEP in
859 # embeddings_config.get_embeddings(). Build the run context once and
860 # fail closed (skip remote model fetches) if the policy can't evaluate.
861 def _embeddings_model_fetch_allowed(provider_key: str) -> bool:
862 try:
863 from ...security.egress.policy import (
864 context_from_snapshot,
865 evaluate_embeddings,
866 )
867 from ...config.thread_settings import (
868 get_setting_from_snapshot,
869 )
870 from ...search_system import username_from_snapshot
872 primary = (
873 get_setting_from_snapshot(
874 "search.tool",
875 default=DEFAULT_SEARCH_TOOL,
876 settings_snapshot=settings_snapshot,
877 )
878 or DEFAULT_SEARCH_TOOL
879 )
880 # Thread username so a per-user private retriever primary
881 # resolves PRIVATE_ONLY (forcing local embeddings) and this
882 # route can't probe a remote embedder for that user.
883 ctx = context_from_snapshot(
884 settings_snapshot,
885 primary,
886 # `session.get(...)` here was a leftover Flask global --
887 # undefined in this module, so this line raised NameError
888 # on EVERY call (the snapshot never carries `_username`:
889 # only ensure_snapshot_username injects it, and this route
890 # does not call it). The enclosing `except Exception` failed
891 # closed and returned False, so the model-list probe was
892 # skipped for every provider and the embeddings dropdown
893 # was permanently empty for every user. `username` is the
894 # route's authenticated user, closed over from
895 # get_available_models.
896 username=username_from_snapshot(settings_snapshot)
897 or username,
898 )
899 if not ctx.require_local_embeddings: 899 ↛ 901line 899 didn't jump to line 901 because the condition on line 899 was always true
900 return True
901 decision = evaluate_embeddings(
902 provider_key, ctx, settings_snapshot=settings_snapshot
903 )
904 if not decision.allowed:
905 logger.bind(policy_audit=True).info(
906 "skipping embeddings model-list probe "
907 "(local-only egress posture)",
908 provider=provider_key,
909 reason=decision.reason,
910 )
911 return decision.allowed
912 except Exception:
913 # Fail closed: a policy/snapshot error must not open a cloud
914 # probe under a local-only posture.
915 logger.bind(policy_audit=True).warning(
916 "embeddings model-list egress check failed; skipping probe",
917 provider=provider_key,
918 exc_info=True,
919 )
920 return False
922 # Provider display names
923 provider_labels = {
924 "sentence_transformers": "Sentence Transformers (Local)",
925 "ollama": "Ollama (Local)",
926 "openai": "OpenAI API",
927 }
929 # Get provider options and models by looping through providers
930 provider_options = []
931 providers = {}
933 for provider_key, provider_class in provider_classes.items():
934 available = provider_class.is_available(settings_snapshot)
936 # Always show the provider in the dropdown so users can
937 # configure its settings (e.g. fix a wrong Ollama URL).
938 provider_options.append(
939 {
940 "value": provider_key,
941 "label": provider_labels.get(provider_key, provider_key),
942 "available": available,
943 }
944 )
946 # Only fetch models when the provider is reachable AND egress
947 # policy permits probing it (cloud providers are skipped under a
948 # local-only posture).
949 if available and _embeddings_model_fetch_allowed(provider_key):
950 models = provider_class.get_available_models(settings_snapshot)
951 providers[provider_key] = [
952 {
953 "value": m["value"],
954 "label": m["label"],
955 "provider": provider_key,
956 **(
957 {"is_embedding": m["is_embedding"]}
958 if "is_embedding" in m
959 else {}
960 ),
961 }
962 for m in models
963 ]
964 else:
965 providers[provider_key] = []
967 return {
968 "success": True,
969 "provider_options": provider_options,
970 "providers": providers,
971 }
973 except Exception as e:
974 # Egress policy denial (incl. settings_unavailable) → clean 4xx/5xx
975 # so the caller distinguishes a refused probe from a server crash.
976 from ...security.egress.policy import PolicyDeniedError as _PDE
978 if isinstance(e, _PDE):
979 reason = e.decision.reason
980 status_code = 503 if reason == "settings_unavailable" else 400
981 logger.bind(policy_audit=True).info(
982 "available-models refused by egress policy",
983 reason=reason,
984 target=e.target,
985 )
986 return JSONResponse(
987 {
988 "success": False,
989 "error": (
990 "Settings are currently unavailable; cannot safely "
991 "enumerate embedding models."
992 )
993 if reason == "settings_unavailable"
994 else f"Available-models request refused ({reason}).",
995 },
996 status_code=status_code,
997 )
998 return handle_api_error("getting available models", e)
1001@router.get("/api/rag/info")
1002def get_index_info(
1003 request: Request, username: Annotated[str, Depends(require_auth)]
1004):
1005 """Get information about the current RAG index."""
1006 from ...database.library_init import get_default_library_id
1008 try:
1009 # Get collection_id from request or use default Library collection
1010 collection_id = request.query_params.get("collection_id")
1011 if not collection_id:
1012 collection_id = get_default_library_id(username)
1014 logger.info(
1015 f"Getting RAG index info for collection_id: {collection_id}"
1016 )
1018 # `with`: LocalEmbeddingManager eagerly opens httpx client pairs for
1019 # Ollama embeddings, so an unclosed service leaks file descriptors on
1020 # every call (#4407). This is a hot endpoint — the collection page
1021 # hits it on each load.
1022 with get_rag_service(request, username, collection_id) as rag_service:
1023 info = rag_service.get_current_index_info(collection_id)
1025 if info is None:
1026 logger.info(
1027 f"No RAG index found for collection_id: {collection_id}"
1028 )
1029 return {"success": True, "info": None, "message": "No index found"}
1031 logger.info(f"Found RAG index for collection_id: {collection_id}")
1032 return {"success": True, "info": info}
1033 except Exception as e:
1034 return handle_api_error("getting index info", e)
1037@router.get("/api/rag/stats")
1038def get_rag_stats(
1039 request: Request, username: Annotated[str, Depends(require_auth)]
1040):
1041 """Get RAG statistics for a collection."""
1042 from ...database.library_init import get_default_library_id
1044 try:
1045 # Get collection_id from request or use default Library collection
1046 collection_id = request.query_params.get("collection_id")
1047 if not collection_id:
1048 collection_id = get_default_library_id(username)
1050 with get_rag_service(request, username, collection_id) as rag_service:
1051 stats = rag_service.get_rag_stats(collection_id)
1053 return {"success": True, "stats": stats}
1054 except Exception as e:
1055 return handle_api_error("getting RAG stats", e)
1058@router.post("/api/rag/index-document")
1059async def index_document(
1060 request: Request, username: Annotated[str, Depends(require_auth)]
1061):
1062 """Index a single document in a collection."""
1063 from ...database.library_init import get_default_library_id
1065 text_doc_id = None
1066 try:
1067 data = await request.json()
1068 if not isinstance(data, dict):
1069 return json_body_error("success", "Request body must be valid JSON")
1070 text_doc_id = data.get("text_doc_id")
1071 force_reindex = data.get("force_reindex", False)
1072 collection_id = data.get("collection_id")
1074 if not isinstance(force_reindex, bool):
1075 return JSONResponse(
1076 {
1077 "success": False,
1078 "error": "force_reindex must be a boolean",
1079 },
1080 status_code=400,
1081 )
1083 if not text_doc_id:
1084 return JSONResponse(
1085 {"success": False, "error": "text_doc_id is required"},
1086 status_code=400,
1087 )
1089 if not collection_id:
1090 # Opens the user's SQLCipher session (sync) — threadpool it.
1091 collection_id = await run_db_sync(get_default_library_id, username)
1093 def _index_sync():
1094 with get_rag_service(
1095 request, username, collection_id
1096 ) as rag_service:
1097 return rag_service.index_document(
1098 text_doc_id, collection_id, force_reindex
1099 )
1101 result = await run_db_sync(_index_sync)
1103 if result["status"] == "error":
1104 return JSONResponse(
1105 {"success": False, "error": result.get("error")},
1106 status_code=400,
1107 )
1109 return {"success": True, "result": result}
1110 except json.JSONDecodeError:
1111 return json_body_error("success", "Request body must be valid JSON")
1112 except Exception as e:
1113 return handle_api_error(f"indexing document {text_doc_id}", e)
1116@router.post("/api/rag/remove-document")
1117async def remove_document(
1118 request: Request, username: Annotated[str, Depends(require_auth)]
1119):
1120 """Remove a document from RAG in a collection."""
1121 from ...database.library_init import get_default_library_id
1123 # Pre-bound for the same reason as test_embedding above: the except
1124 # block interpolates text_doc_id into its error message.
1125 text_doc_id = None
1126 try:
1127 data = await request.json()
1128 if not isinstance(data, dict):
1129 return json_body_error("success", "Request body must be valid JSON")
1130 text_doc_id = data.get("text_doc_id")
1131 collection_id = data.get("collection_id")
1133 if not text_doc_id:
1134 return JSONResponse(
1135 {"success": False, "error": "text_doc_id is required"},
1136 status_code=400,
1137 )
1139 if not collection_id: 1139 ↛ 1143line 1139 didn't jump to line 1143 because the condition on line 1139 was always true
1140 # Opens the user's SQLCipher session (sync) — threadpool it.
1141 collection_id = await run_db_sync(get_default_library_id, username)
1143 def _remove_sync():
1144 with get_rag_service(
1145 request, username, collection_id
1146 ) as rag_service:
1147 return rag_service.remove_document_from_rag(
1148 text_doc_id, collection_id
1149 )
1151 result = await run_db_sync(_remove_sync)
1153 if result["status"] == "error":
1154 return JSONResponse(
1155 {"success": False, "error": result.get("error")},
1156 status_code=400,
1157 )
1159 return {"success": True, "result": result}
1160 except json.JSONDecodeError:
1161 return json_body_error("success", "Request body must be valid JSON")
1162 except Exception as e:
1163 return handle_api_error(f"removing document {text_doc_id}", e)
1166@router.get("/api/rag/index-all")
1167def index_all(
1168 request: Request,
1169 username: Annotated[str, Depends(require_auth)],
1170):
1171 """Index all documents in a collection with Server-Sent Events progress."""
1172 from ...database.session_context import get_user_db_session
1173 from ...database.library_init import get_default_library_id
1174 from ...utilities.db_utils import get_settings_manager
1176 # Parse as bool — a raw string "false" is truthy, which would silently
1177 # force a full reindex for every caller that asked to skip already-indexed
1178 # documents.
1179 force_reindex = (
1180 request.query_params.get("force_reindex", "false").lower() == "true"
1181 )
1183 # Get collection_id from request or use default Library collection
1184 collection_id = request.query_params.get("collection_id")
1185 if not collection_id:
1186 collection_id = get_default_library_id(username)
1188 logger.info(
1189 f"Starting index-all for collection_id: {collection_id}, force_reindex: {force_reindex}"
1190 )
1192 # Create RAG service in request context before generator runs.
1193 # use_defaults=force_reindex mirrors index_collection / the background
1194 # worker so a force-reindex resolves the same embedding config.
1195 rag_service = get_rag_service(
1196 request, username, collection_id, use_defaults=force_reindex
1197 )
1199 # Fetch batch size from settings in this thread before generator runs
1200 with get_user_db_session(username) as db_session:
1201 settings = get_settings_manager(db_session, username)
1202 batch_size = int(settings.get_setting("rag.indexing_batch_size", 15))
1203 try:
1204 _max_workers = int(
1205 settings.get_setting("rag.indexing_max_parallel_docs", 4)
1206 )
1207 except Exception:
1208 _max_workers = 4
1209 _max_workers = max(1, min(_max_workers, 16))
1211 def generate():
1212 """Generator function for SSE progress updates."""
1213 _sse_cancel = threading.Event()
1214 try:
1215 # Send initial status
1216 yield f"data: {json.dumps({'type': 'start', 'message': 'Starting bulk indexing...'})}\n\n"
1218 # Get document IDs to index from DocumentCollection. All DB work
1219 # happens inside this SHORT-LIVED session; NO ``yield`` runs while
1220 # it is held (the "collection not found" error is yielded AFTER the
1221 # block) — keeping a get_user_db_session scope open across a yield
1222 # risks the Starlette/anyio thread-affinity corruption fixed in
1223 # download_bulk (__enter__/__exit__ can land on different threads).
1224 collection_found = True
1225 doc_info = []
1226 with get_user_db_session(username) as db_session:
1227 # Persist embedding metadata + clean up on force-reindex via the
1228 # shared helpers, so this bulk route behaves like the
1229 # single-collection SSE route and the background worker
1230 # (previously it stored no embedding_dimension and never reset
1231 # stale chunks/indices on force-reindex).
1232 collection = (
1233 db_session.query(Collection)
1234 .filter_by(id=collection_id)
1235 .first()
1236 )
1238 if not collection:
1239 collection_found = False
1240 else:
1241 # Persist the new embedding config AND reset the old
1242 # chunks/RAGIndex in ONE transaction. Two separate commits
1243 # would leave a crash window where the Collection carries
1244 # the NEW embedding config while the OLD RAGIndex row +
1245 # FAISS file (built with the old config) still exist — a
1246 # config/index mismatch.
1247 changed = False
1248 faiss_reset_paths = []
1249 if collection.embedding_model is None or force_reindex:
1250 _store_collection_embedding_metadata(
1251 collection, rag_service
1252 )
1253 changed = True
1254 if force_reindex:
1255 faiss_reset_paths = _reset_collection_for_reindex(
1256 db_session, collection_id
1257 )
1258 changed = True
1259 if changed:
1260 db_session.commit()
1261 _unlink_reindex_faiss_files(faiss_reset_paths)
1263 doc_info = [
1264 (doc.id, doc.title)
1265 for _dc, doc in _query_documents_to_index(
1266 db_session, collection_id, force_reindex
1267 )
1268 ]
1270 if not collection_found:
1271 yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
1272 return
1274 if not doc_info:
1275 yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
1276 return
1278 results = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
1279 total = len(doc_info)
1281 # Process documents in batches to pace SSE progress events.
1282 # The batch boundary is purely cosmetic now — documents within
1283 # a batch run in parallel via the bounded worker pool below.
1284 # batch_size and _max_workers are read from settings before the
1285 # generator starts (closure capture) — no DB work in-stream.
1286 processed = 0
1288 for i in range(0, len(doc_info), batch_size):
1289 batch = doc_info[i : i + batch_size]
1291 # Run the per-batch docs through the bounded parallel
1292 # helper. ``as_completed`` order means progress events
1293 # fire in completion order, not submission order — the
1294 # `processed` counter below keeps the SSE payload
1295 # monotonically increasing so the UI percent stays sane.
1296 # ``is_cancelled`` polls the SSE-disconnect event so an
1297 # early-disconnect watcher (or this generator's own
1298 # ``finally``) can short-circuit the batch before all
1299 # queued futures run. The helper still drains in-flight
1300 # workers via ``pool.shutdown(wait=True, ...)`` so no
1301 # worker outlives the helper return — see
1302 # ``index_documents_parallel`` for the rationale.
1303 parallel_result = rag_service.index_documents_parallel(
1304 batch,
1305 collection_id,
1306 force_reindex=force_reindex,
1307 max_workers=_max_workers,
1308 is_cancelled=_sse_cancel.is_set,
1309 )
1310 batch_results = parallel_result["results"]
1312 for doc_id, title in batch:
1313 processed += 1
1314 result = batch_results[doc_id]
1316 # Send progress update
1317 yield f"data: {json.dumps({'type': 'progress', 'current': processed, 'total': total, 'title': title, 'percent': int((processed / total) * 100)})}\n\n"
1319 if result["status"] == "success": 1319 ↛ 1320line 1319 didn't jump to line 1320 because the condition on line 1319 was never true
1320 results["successful"] += 1
1321 elif result["status"] in ("skipped", "cleared"): 1321 ↛ 1325line 1321 didn't jump to line 1325 because the condition on line 1321 was always true
1322 # "cleared" (empty-text purge) is a handled non-failure.
1323 results["skipped"] += 1
1324 else:
1325 results["failed"] += 1
1326 results["errors"].append(
1327 {
1328 "doc_id": doc_id,
1329 "title": title,
1330 "error": result.get("error"),
1331 }
1332 )
1334 # Send completion status
1335 yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
1337 # Log final status for debugging
1338 logger.info(
1339 f"Bulk indexing complete: {results['successful']} successful, {results['skipped']} skipped, {results['failed']} failed"
1340 )
1342 except Exception:
1343 logger.exception("Error in bulk indexing")
1344 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
1345 finally:
1346 # Generator ``finally`` runs at stream completion (or client
1347 # disconnect via ``GeneratorExit``) — the safe place to release
1348 # the RAG service's embedding-manager httpx clients. Closing at
1349 # the outer route scope would tear it down before the streamed
1350 # generator runs. ``safe_close`` swallows close-time errors so
1351 # a broken Ollama doesn't mask the original generator outcome.
1352 #
1353 # Signal cancellation BEFORE closing: the helper has already
1354 # drained in-flight workers (its ``shutdown(wait=True, ...)``
1355 # guarantees no worker outlives the helper return), so setting
1356 # the event here is for any in-progress batch loop iteration
1357 # and any future code path that bypasses the helper.
1358 _sse_cancel.set()
1359 safe_close(rag_service, "rag_service (index-all SSE)")
1361 return WorkerCleanupStreamingResponse(
1362 generate(),
1363 media_type="text/event-stream",
1364 headers={
1365 "Cache-Control": "no-cache, no-transform",
1366 "X-Accel-Buffering": "no",
1367 },
1368 )
1371@router.post("/api/rag/configure")
1372async def configure_rag(
1373 request: Request,
1374 username: Annotated[str, Depends(require_auth)],
1375):
1376 """
1377 Change RAG configuration (embedding model, chunk size, etc.).
1378 This will create a new index with the new configuration.
1379 """
1380 from ...database.session_context import get_user_db_session
1381 from ...utilities.db_utils import get_settings_manager
1383 try:
1384 data = await request.json()
1385 if not isinstance(data, dict):
1386 return json_body_error("success", "Request body must be valid JSON")
1387 embedding_model = data.get("embedding_model")
1388 embedding_provider = data.get("embedding_provider")
1389 chunk_size = data.get("chunk_size")
1390 chunk_overlap = data.get("chunk_overlap")
1391 collection_id = data.get("collection_id")
1393 # Get new advanced settings (with defaults)
1394 splitter_type = data.get(
1395 "splitter_type", DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE
1396 )
1397 text_separators = data.get(
1398 "text_separators", DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
1399 )
1400 distance_metric = data.get(
1401 "distance_metric", DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC
1402 )
1403 normalize_vectors = data.get(
1404 "normalize_vectors", DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS
1405 )
1406 index_type = data.get("index_type", DEFAULT_LOCAL_SEARCH_INDEX_TYPE)
1408 if (
1409 not embedding_model
1410 or not embedding_provider
1411 or chunk_size is None
1412 or chunk_overlap is None
1413 ):
1414 return JSONResponse(
1415 {
1416 "success": False,
1417 "error": "All configuration parameters are required (embedding_model, embedding_provider, chunk_size, chunk_overlap",
1418 },
1419 status_code=400,
1420 )
1422 if type(chunk_size) is not int or not (
1423 _RAG_CHUNK_SIZE_MIN <= chunk_size <= _RAG_CHUNK_SIZE_MAX
1424 ):
1425 return JSONResponse(
1426 {
1427 "success": False,
1428 "error": (
1429 "chunk_size must be an integer between "
1430 f"{_RAG_CHUNK_SIZE_MIN} and {_RAG_CHUNK_SIZE_MAX}"
1431 ),
1432 },
1433 status_code=400,
1434 )
1436 if type(chunk_overlap) is not int or not (
1437 _RAG_CHUNK_OVERLAP_MIN <= chunk_overlap <= _RAG_CHUNK_OVERLAP_MAX
1438 ):
1439 return JSONResponse(
1440 {
1441 "success": False,
1442 "error": (
1443 "chunk_overlap must be an integer between "
1444 f"{_RAG_CHUNK_OVERLAP_MIN} and "
1445 f"{_RAG_CHUNK_OVERLAP_MAX}"
1446 ),
1447 },
1448 status_code=400,
1449 )
1451 if chunk_overlap > chunk_size:
1452 return JSONResponse(
1453 {
1454 "success": False,
1455 "error": (
1456 "chunk_overlap must be less than or equal to chunk_size"
1457 ),
1458 },
1459 status_code=400,
1460 )
1462 if not isinstance(normalize_vectors, bool):
1463 return JSONResponse(
1464 {
1465 "success": False,
1466 "error": "normalize_vectors must be a boolean",
1467 },
1468 status_code=400,
1469 )
1471 # The local_search_text_separators setting is registered with
1472 # ui_element "json", so store the separators as a proper list. A
1473 # string payload (e.g. a textarea value) is parsed into a list
1474 # first; anything that isn't a JSON array of strings is rejected
1475 # outright (400) rather than silently coerced to defaults, so a
1476 # malformed payload can't get written into settings or fed into
1477 # the chunker.
1478 text_separators = await asyncio.to_thread(
1479 _parse_configured_text_separators, text_separators
1480 )
1481 if text_separators is None:
1482 return JSONResponse(
1483 {
1484 "success": False,
1485 "error": "text_separators must be a JSON array of strings",
1486 },
1487 status_code=400,
1488 )
1490 # All settings writes + the collection reconfiguration (FAISS index
1491 # creation) run sync I/O. Offload to a worker thread so the uvicorn
1492 # event loop is free to serve other requests during the potentially
1493 # slow index-create path.
1494 def _persist_configuration():
1495 """Persist the settings and (optionally) create/promote the
1496 collection's index in ONE database transaction.
1498 Settings and the selected index must land atomically: a
1499 mid-way failure (locked/environment-locked setting, DB error)
1500 must leave BOTH unchanged, never commit the settings while the
1501 old index stays "current" (or vice versa) — that mismatch is
1502 exactly what would let a subsequent search silently run
1503 against a stale/inconsistent configuration.
1505 Returns a ``JSONResponse`` for an error that should short-
1506 circuit the request, or the new/reused index hash (``None``
1507 when no ``collection_id`` was supplied) on success.
1508 """
1509 with get_user_db_session(username) as db_session:
1510 settings = get_settings_manager(db_session, username)
1512 requested_settings = (
1513 ("local_search_embedding_model", embedding_model),
1514 ("local_search_embedding_provider", embedding_provider),
1515 ("local_search_chunk_size", int(chunk_size)),
1516 ("local_search_chunk_overlap", int(chunk_overlap)),
1517 ("local_search_splitter_type", splitter_type),
1518 ("local_search_text_separators", text_separators),
1519 ("local_search_distance_metric", distance_metric),
1520 (
1521 "local_search_normalize_vectors",
1522 normalize_vectors,
1523 ),
1524 ("local_search_index_type", index_type),
1525 )
1526 requested_setting_keys = [
1527 key for key, _value in requested_settings
1528 ]
1530 if settings.settings_locked:
1531 return JSONResponse(
1532 {
1533 "success": False,
1534 "error": "RAG configuration is locked by app.lock_settings",
1535 },
1536 status_code=403,
1537 )
1539 # Environment-locked settings (LDR_* env vars) must never be
1540 # overwritten through a settings-mutation endpoint. Refuse
1541 # the WHOLE request up front, before any write, rather than
1542 # silently skipping just the locked keys — that would both
1543 # mislead the caller with a 200 and partially apply an
1544 # inconsistent configuration.
1545 environment_locked_keys = [
1546 key
1547 for key, _value in requested_settings
1548 if check_env_setting(key) is not None
1549 ]
1550 if environment_locked_keys:
1551 return JSONResponse(
1552 {
1553 "success": False,
1554 "error": (
1555 "Settings "
1556 f"{', '.join(environment_locked_keys)} "
1557 "are environment-locked"
1558 ),
1559 },
1560 status_code=403,
1561 )
1563 for key, value in requested_settings:
1564 if not settings.set_setting(key, value, commit=False):
1565 db_session.rollback()
1566 return JSONResponse(
1567 {
1568 "success": False,
1569 "error": "Unable to save RAG configuration. No changes were applied.",
1570 },
1571 status_code=500,
1572 )
1574 index_hash = None
1575 if collection_id:
1576 with LibraryRAGService(
1577 username=username,
1578 embedding_model=embedding_model,
1579 embedding_provider=embedding_provider,
1580 chunk_size=int(chunk_size),
1581 chunk_overlap=int(chunk_overlap),
1582 splitter_type=splitter_type,
1583 text_separators=text_separators,
1584 distance_metric=distance_metric,
1585 normalize_vectors=normalize_vectors,
1586 index_type=index_type,
1587 ) as new_rag_service:
1588 rag_index = new_rag_service._get_or_create_rag_index(
1589 collection_id, db_session=db_session, commit=False
1590 )
1591 index_hash = rag_index.index_hash
1593 db_session.commit()
1594 settings.emit_settings_changed_after_commit(
1595 requested_setting_keys
1596 )
1597 return index_hash
1599 # run_db_sync (not raw to_thread): get_user_db_session/LibraryRAGService
1600 # open the user's DB session; reused to_thread workers must not retain it.
1601 result = await run_db_sync(_persist_configuration)
1602 if isinstance(result, JSONResponse):
1603 return result
1604 index_hash = result
1606 if collection_id:
1607 return {
1608 "success": True,
1609 "message": "Configuration updated for collection. You can now index documents with the new settings.",
1610 "index_hash": index_hash,
1611 }
1613 # Just saving default settings without updating a specific collection
1614 return {
1615 "success": True,
1616 "message": "Default embedding settings saved successfully. New collections will use these settings.",
1617 }
1619 except json.JSONDecodeError:
1620 return json_body_error("success", "Request body must be valid JSON")
1621 except Exception as e:
1622 return handle_api_error("configuring RAG", e)
1625@router.get("/api/rag/documents")
1626def get_documents(
1627 request: Request, username: Annotated[str, Depends(require_auth)]
1628):
1629 """Get library documents with their RAG status for the default Library collection (paginated)."""
1630 from ...database.session_context import get_user_db_session
1631 from ...database.library_init import get_default_library_id
1633 try:
1634 # Get pagination parameters
1635 try:
1636 page = int(request.query_params.get("page", "1"))
1637 except (TypeError, ValueError):
1638 page = 1
1639 try:
1640 per_page = int(request.query_params.get("per_page", "50"))
1641 except (TypeError, ValueError):
1642 per_page = 50
1643 page = max(1, page)
1644 per_page = max(1, min(per_page, 500))
1645 filter_type = request.query_params.get(
1646 "filter", "all"
1647 ) # all, indexed, unindexed
1649 # Validate pagination parameters
1650 # Upper-bound the page as well. A numerically valid but astronomical
1651 # ?page=10**40 reaches .offset() and raises OverflowError from
1652 # SQLite's 64-bit integer conversion, surfacing as a 500. The
1653 # try/except above only rescues NON-numeric input. Mirrors
1654 # metrics.py's _MAX_PAGE, which already handles this correctly.
1655 page = max(1, min(page, _MAX_PAGE))
1656 per_page = min(max(10, per_page), 100) # Limit between 10-100
1658 # Close current thread's session to force fresh connection
1659 from ...database.thread_local_session import cleanup_current_thread
1661 cleanup_current_thread()
1663 # Get collection_id from request or use default Library collection
1664 collection_id = request.query_params.get("collection_id")
1665 if not collection_id:
1666 collection_id = get_default_library_id(username)
1668 logger.info(
1669 f"Getting documents for collection_id: {collection_id}, filter: {filter_type}, page: {page}"
1670 )
1672 with get_user_db_session(username) as db_session:
1673 # Expire all cached objects to ensure we get fresh data from DB
1674 db_session.expire_all()
1676 # Import RagDocumentStatus model
1677 from ...database.models.library import RagDocumentStatus
1679 # Build base query - join Document with DocumentCollection for the collection
1680 # LEFT JOIN with rag_document_status to check indexed status
1681 query = (
1682 db_session.query(
1683 Document, DocumentCollection, RagDocumentStatus
1684 )
1685 .join(
1686 DocumentCollection,
1687 (DocumentCollection.document_id == Document.id)
1688 & (DocumentCollection.collection_id == collection_id),
1689 )
1690 .outerjoin(
1691 RagDocumentStatus,
1692 (RagDocumentStatus.document_id == Document.id)
1693 & (RagDocumentStatus.collection_id == collection_id),
1694 )
1695 )
1697 logger.debug(f"Base query for collection {collection_id}: {query}")
1699 # Apply filters based on rag_document_status existence
1700 if filter_type == "indexed":
1701 query = query.filter(RagDocumentStatus.document_id.isnot(None))
1702 elif filter_type == "unindexed":
1703 # Documents in collection but not indexed yet
1704 query = query.filter(RagDocumentStatus.document_id.is_(None))
1706 # Get total count before pagination
1707 total_count = query.count()
1708 logger.info(
1709 f"Found {total_count} total documents for collection {collection_id} with filter {filter_type}"
1710 )
1712 # Apply pagination
1713 results = (
1714 query.order_by(Document.created_at.desc())
1715 .limit(per_page)
1716 .offset((page - 1) * per_page)
1717 .all()
1718 )
1720 documents = [
1721 {
1722 "id": doc.id,
1723 "title": doc.title,
1724 "original_url": doc.original_url,
1725 "rag_indexed": rag_status is not None,
1726 "chunk_count": rag_status.chunk_count if rag_status else 0,
1727 "created_at": doc.created_at.isoformat()
1728 if doc.created_at
1729 else None,
1730 }
1731 for doc, doc_collection, rag_status in results
1732 ]
1734 # Debug logging to help diagnose indexing status issues
1735 indexed_count = sum(1 for d in documents if d["rag_indexed"])
1737 # Additional debug: check rag_document_status for this collection
1738 all_indexed_statuses = (
1739 db_session.query(RagDocumentStatus)
1740 .filter_by(collection_id=collection_id)
1741 .all()
1742 )
1743 logger.info(
1744 f"rag_document_status table shows: {len(all_indexed_statuses)} documents indexed for collection {collection_id}"
1745 )
1747 logger.info(
1748 f"Returning {len(documents)} documents on page {page}: "
1749 f"{indexed_count} indexed, {len(documents) - indexed_count} not indexed"
1750 )
1752 return {
1753 "success": True,
1754 "documents": documents,
1755 "pagination": {
1756 "page": page,
1757 "per_page": per_page,
1758 "total": total_count,
1759 "pages": (total_count + per_page - 1) // per_page,
1760 },
1761 }
1762 except Exception as e:
1763 return handle_api_error("getting documents", e)
1766# Collection Management Routes
1769@router.get("/api/collections")
1770def get_collections(
1771 request: Request, username: Annotated[str, Depends(require_auth)]
1772):
1773 """Get all document collections for the current user."""
1774 from ...database.session_context import get_user_db_session
1776 try:
1777 with get_user_db_session(username) as db_session:
1778 # No need to filter by username - each user has their own database
1779 collections = db_session.query(Collection).all()
1781 # Canonical durable count. Reconciliation maintains these rows from
1782 # actual FAISS membership; the legacy DocumentCollection.indexed flag
1783 # is retained only for compatibility and must not drive user-visible
1784 # truth.
1785 collection_counts = {
1786 collection_id: (
1787 int(document_count or 0),
1788 int(indexed_count or 0),
1789 )
1790 for collection_id, document_count, indexed_count in (
1791 db_session.query(
1792 DocumentCollection.collection_id,
1793 func.count(DocumentCollection.document_id),
1794 func.count(
1795 case(
1796 (
1797 DocumentCollection.indexed.is_(True),
1798 DocumentCollection.document_id,
1799 ),
1800 else_=None,
1801 )
1802 ),
1803 )
1804 .group_by(DocumentCollection.collection_id)
1805 .all()
1806 )
1807 }
1809 result = []
1810 for coll in collections:
1811 document_count, indexed_document_count = collection_counts.get(
1812 coll.id,
1813 (0, 0),
1814 )
1815 collection_data = {
1816 "id": coll.id,
1817 "name": coll.name,
1818 "description": coll.description,
1819 "created_at": coll.created_at.isoformat()
1820 if coll.created_at
1821 else None,
1822 "collection_type": coll.collection_type,
1823 "is_default": coll.is_default
1824 if hasattr(coll, "is_default")
1825 else False,
1826 "is_public": bool(getattr(coll, "is_public", False)),
1827 "agent_enabled": _agent_enabled_default_on(coll),
1828 "document_count": document_count,
1829 "indexed_document_count": indexed_document_count,
1830 "folder_count": len(coll.linked_folders)
1831 if hasattr(coll, "linked_folders")
1832 else 0,
1833 }
1835 # Include embedding metadata if available
1836 if coll.embedding_model:
1837 collection_data["embedding"] = {
1838 "model": coll.embedding_model,
1839 "provider": coll.embedding_model_type.value
1840 if coll.embedding_model_type
1841 else None,
1842 "dimension": coll.embedding_dimension,
1843 "chunk_size": coll.chunk_size,
1844 "chunk_overlap": coll.chunk_overlap,
1845 }
1846 else:
1847 collection_data["embedding"] = None
1849 result.append(collection_data)
1851 return {"success": True, "collections": result}
1852 except Exception as e:
1853 return handle_api_error("getting collections", e)
1856@router.post("/api/collections")
1857async def create_collection(
1858 request: Request, username: Annotated[str, Depends(require_auth)]
1859):
1860 """Create a new document collection."""
1861 data = await request.json()
1862 if not isinstance(data, dict):
1863 return json_body_error("success", "Request body must be valid JSON")
1864 return await run_db_sync(_create_collection_sync, data, username)
1867def _create_collection_sync(data, username):
1868 from ...database.session_context import get_user_db_session
1870 try:
1871 name = data.get("name", "")
1872 if not isinstance(name, str):
1873 # `.get(key, "")` only supplies the default when the key is
1874 # ABSENT; a present-but-wrong-typed value (e.g. `"name": 123`
1875 # or `"name": null`) passes through untouched and blows up on
1876 # `.strip()` with AttributeError -> 500. Reject it explicitly.
1877 return JSONResponse(
1878 {"success": False, "error": "Name must be a string"},
1879 status_code=400,
1880 )
1881 name = name.strip()
1883 description = data.get("description", "")
1884 if not isinstance(description, str):
1885 return JSONResponse(
1886 {"success": False, "error": "Description must be a string"},
1887 status_code=400,
1888 )
1889 description = description.strip()
1891 collection_type = data.get("type", "user_uploads")
1892 if not isinstance(collection_type, str):
1893 # Ported from main's Flask original, which returned
1894 # ``jsonify({...}), 400``. Neither half survives here:
1895 # ``jsonify`` is undefined (NameError -> 500), and a
1896 # ``(body, status)`` tuple is a Flask convention FastAPI does
1897 # not honour. Matches the sibling error returns below.
1898 return JSONResponse(
1899 {
1900 "success": False,
1901 "error": "Collection type must be a string",
1902 },
1903 status_code=400,
1904 )
1905 # Allowlist user-creatable types. System types (notes,
1906 # default_library, research_history) are lazy-created by their
1907 # owning subsystems; a user-crafted impostor (e.g. type="notes")
1908 # would be undeletable under PROTECTED_COLLECTION_TYPES and could
1909 # nondeterministically win _get_or_create_notes_collection's
1910 # .first() lookup, splitting the notes corpus.
1911 allowed_types = {"user_uploads", "user_collection"}
1912 if collection_type not in allowed_types:
1913 return JSONResponse(
1914 {
1915 "success": False,
1916 "error": (
1917 f"Invalid collection type '{collection_type}'. "
1918 f"Allowed: {sorted(allowed_types)}"
1919 ),
1920 },
1921 status_code=400,
1922 )
1923 # Egress classification, default private (the safe choice). A public
1924 # collection counts as a public engine and may use cloud inference.
1925 is_public = data.get("is_public", False)
1926 if not isinstance(is_public, bool):
1927 return JSONResponse(
1928 {
1929 "success": False,
1930 "error": "is_public must be a boolean",
1931 },
1932 status_code=400,
1933 )
1934 # Usability switch (NOT egress): offer this collection to the research
1935 # agent? Default True (available) — behaviour-preserving. An explicit
1936 # JSON null normalizes to True so it matches how NULL is read back
1937 # everywhere else (NULL → available); only an explicit false disables.
1938 agent_enabled_raw = data.get("agent_enabled", True)
1939 if agent_enabled_raw is not None and not isinstance(
1940 agent_enabled_raw, bool
1941 ):
1942 return JSONResponse(
1943 {
1944 "success": False,
1945 "error": "agent_enabled must be a boolean or null",
1946 },
1947 status_code=400,
1948 )
1949 agent_enabled = True if agent_enabled_raw is None else agent_enabled_raw
1951 if not name:
1952 return JSONResponse(
1953 {"success": False, "error": "Name is required"}, status_code=400
1954 )
1956 with get_user_db_session(username) as db_session:
1957 # Check if collection with this name already exists in this user's database
1958 existing = db_session.query(Collection).filter_by(name=name).first()
1960 if existing:
1961 return JSONResponse(
1962 {
1963 "success": False,
1964 "error": f"Collection '{name}' already exists",
1965 },
1966 status_code=400,
1967 )
1969 # Create new collection (no username needed - each user has their own DB)
1970 # Note: created_at uses default=utcnow() in the model, so we don't need to set it manually
1971 collection = Collection(
1972 id=str(uuid.uuid4()), # Generate UUID for collection
1973 name=name,
1974 description=description,
1975 collection_type=collection_type,
1976 is_public=is_public,
1977 agent_enabled=agent_enabled,
1978 )
1980 db_session.add(collection)
1981 db_session.commit()
1983 return {
1984 "success": True,
1985 "collection": {
1986 "id": collection.id,
1987 "name": collection.name,
1988 "description": collection.description,
1989 "created_at": collection.created_at.isoformat(),
1990 "collection_type": collection.collection_type,
1991 "is_public": bool(collection.is_public),
1992 "agent_enabled": _agent_enabled_default_on(collection),
1993 },
1994 }
1995 except Exception as e:
1996 return handle_api_error("creating collection", e)
1999@router.put("/api/collections/{collection_id}")
2000async def update_collection(
2001 request: Request,
2002 collection_id,
2003 username: Annotated[str, Depends(require_auth)],
2004):
2005 """Update a collection's details."""
2006 data = await request.json()
2007 if not isinstance(data, dict):
2008 return json_body_error("success", "Request body must be valid JSON")
2009 return await run_db_sync(
2010 _update_collection_sync, data, collection_id, username
2011 )
2014def _update_collection_sync(data, collection_id, username):
2015 from ...database.session_context import get_user_db_session
2017 try:
2018 name = data.get("name", "")
2019 if not isinstance(name, str):
2020 # See _create_collection_sync: `.get(key, "")` only supplies
2021 # the default when the key is absent, so a present-but-wrong-
2022 # typed value reaches `.strip()` untouched -> AttributeError -> 500.
2023 return JSONResponse(
2024 {"success": False, "error": "Name must be a string"},
2025 status_code=400,
2026 )
2027 name = name.strip()
2029 description = data.get("description", "")
2030 if not isinstance(description, str):
2031 return JSONResponse(
2032 {"success": False, "error": "Description must be a string"},
2033 status_code=400,
2034 )
2035 description = description.strip()
2037 is_public = data.get("is_public")
2038 if "is_public" in data and not isinstance(is_public, bool):
2039 return JSONResponse(
2040 {
2041 "success": False,
2042 "error": "is_public must be a boolean",
2043 },
2044 status_code=400,
2045 )
2047 agent_enabled = data.get("agent_enabled")
2048 if (
2049 "agent_enabled" in data
2050 and agent_enabled is not None
2051 and not isinstance(agent_enabled, bool)
2052 ):
2053 return JSONResponse(
2054 {
2055 "success": False,
2056 "error": "agent_enabled must be a boolean or null",
2057 },
2058 status_code=400,
2059 )
2061 with get_user_db_session(username) as db_session:
2062 # No need to filter by username - each user has their own database
2063 collection = (
2064 db_session.query(Collection).filter_by(id=collection_id).first()
2065 )
2067 if not collection:
2068 return JSONResponse(
2069 {"success": False, "error": "Collection not found"},
2070 status_code=404,
2071 )
2073 # System collections (Notes, Library, Research History) own
2074 # first-class user data whose lifecycle is managed by other
2075 # subsystems. Renaming them via this generic endpoint
2076 # bypasses their intended invariants and confuses every UI
2077 # surface that lists collections. Mirrors PROTECTED_COLLECTION_TYPES
2078 # in the deletion service. Only identity fields (name,
2079 # description) are locked — the is_public / agent_enabled
2080 # toggles below are deliberate per-collection settings that
2081 # must keep working for system collections too.
2082 from ...research_library.deletion.services.collection_deletion import (
2083 PROTECTED_COLLECTION_TYPES,
2084 )
2086 if collection.collection_type in PROTECTED_COLLECTION_TYPES and (
2087 name or "description" in data
2088 ):
2089 return JSONResponse(
2090 {
2091 "success": False,
2092 "error": (
2093 f"Cannot rename or redescribe system "
2094 f"collection '{collection.name}' "
2095 f"(type={collection.collection_type})."
2096 ),
2097 },
2098 status_code=409,
2099 )
2101 if name:
2102 # Check if new name conflicts with existing collection
2103 existing = (
2104 db_session.query(Collection)
2105 .filter(
2106 Collection.name == name,
2107 Collection.id != collection_id,
2108 )
2109 .first()
2110 )
2112 if existing:
2113 return JSONResponse(
2114 {
2115 "success": False,
2116 "error": f"Collection '{name}' already exists",
2117 },
2118 status_code=400,
2119 )
2121 collection.name = name
2123 # Only write when the caller sends the key (sending "" still
2124 # clears) — otherwise toggle-only PUTs wipe the description.
2125 if "description" in data:
2126 collection.description = description
2128 # Egress classification toggle (only when the caller sends it).
2129 if "is_public" in data:
2130 collection.is_public = is_public
2132 # Research-agent availability toggle (only when the caller sends it).
2133 # Explicit null normalizes to True (available) for parity with how
2134 # NULL is read back elsewhere; only an explicit false disables.
2135 if "agent_enabled" in data:
2136 collection.agent_enabled = (
2137 True if agent_enabled is None else agent_enabled
2138 )
2140 db_session.commit()
2142 return {
2143 "success": True,
2144 "collection": {
2145 "id": collection.id,
2146 "name": collection.name,
2147 "description": collection.description,
2148 "created_at": collection.created_at.isoformat()
2149 if collection.created_at
2150 else None,
2151 "collection_type": collection.collection_type,
2152 "is_public": bool(collection.is_public),
2153 "agent_enabled": _agent_enabled_default_on(collection),
2154 },
2155 }
2156 except Exception as e:
2157 return handle_api_error("updating collection", e)
2160def _try_pdf_upgrade(
2161 *,
2162 db_session,
2163 document,
2164 file_content: bytes,
2165 filename: str,
2166 pdf_storage: str,
2167 pdf_storage_manager,
2168) -> bool:
2169 """Check PDF upgrade eligibility and attempt upgrading a Document to include PDF bytes.
2171 Returns True if the document was upgraded, False otherwise. Exceptions
2172 raised during the upgrade attempt are logged and swallowed; the caller
2173 sees False and the upload continues as a non-upgraded success.
2174 """
2175 if pdf_storage != "database" or pdf_storage_manager is None:
2176 return False
2178 # NOTE: Only the PDF magic-byte check is needed here.
2179 # Size limits are already enforced by the async upload wrapper, before
2180 # these bytes are ever buffered.
2181 # Filename sanitization already happens via sanitize_filename() in
2182 # upload_to_collection() before this helper is invoked.
2183 # See PR #3145 review for details.
2184 if file_content[:4] != b"%PDF":
2185 logger.debug(
2186 "Skipping PDF upgrade for {}: not a PDF file",
2187 filename,
2188 )
2189 return False
2191 try:
2192 return bool(
2193 pdf_storage_manager.upgrade_to_pdf(
2194 document=document,
2195 pdf_content=file_content,
2196 session=db_session,
2197 )
2198 )
2199 except Exception:
2200 logger.exception(f"Failed to upgrade PDF for {filename}")
2201 return False
2204@router.post("/api/collections/{collection_id}/upload")
2205@upload_rate_limit_user
2206@upload_rate_limit_ip
2207async def upload_to_collection(
2208 request: Request,
2209 collection_id,
2210 username: Annotated[str, Depends(require_auth)],
2211):
2212 """Upload files to a collection.
2214 The async wrapper consumes the multipart form (the only legitimate
2215 awaits — request.form() and per-file file.read()) and validates
2216 sizes. The sync DB body — opening the SQLCipher session, hashing,
2217 text extraction, and Document/Collection writes — is offloaded to
2218 a threadpool so a multi-file upload doesn't stall the event loop
2219 behind PBKDF2 key derivation and per-file extraction work.
2220 """
2221 # Starlette's form parser yields STARLETTE UploadFile instances;
2222 # fastapi.UploadFile is a SUBCLASS of it on fastapi>=0.113, so
2223 # `isinstance(f, fastapi.UploadFile)` is False for every real
2224 # upload and the filter below silently dropped all files
2225 # (every upload 400'd with "No files provided").
2226 from starlette.datastructures import UploadFile
2228 # Per-file and total upload limits to prevent memory exhaustion.
2229 #
2230 # These MUST come from FileUploadValidator, not a route-local
2231 # constant: GET /api/config/limits (research.py) advertises
2232 # FileUploadValidator.MAX_FILE_SIZE / MAX_FILES_PER_REQUEST to the
2233 # frontend as "the backend's authoritative limits", and
2234 # BodySizeLimitMiddleware (fastapi_app.py) already enforces the same
2235 # constants as the global per-request body cap for every route,
2236 # including this one. A previous hardening pass gave this route its
2237 # own tighter, undocumented cap (100MB / 50 files) instead of reusing
2238 # FileUploadValidator like origin/main's Flask handler did — so a
2239 # file the frontend was told is acceptable could be silently
2240 # rejected here with a different limit. Only the size/count checks
2241 # are reused (not validate_upload's MIME/PDF-structure checks): this
2242 # route indexes arbitrary document types via document_loaders, not
2243 # just PDFs.
2244 from ...security import FileUploadValidator
2246 MAX_FILE_SIZE = FileUploadValidator.MAX_FILE_SIZE
2247 MAX_FILES_PER_UPLOAD = FileUploadValidator.MAX_FILES_PER_REQUEST
2248 MAX_TOTAL_UPLOAD_SIZE = MAX_FILES_PER_UPLOAD * MAX_FILE_SIZE
2250 # Early request size check based on Content-Length, before reading body.
2251 content_length = request.headers.get("content-length")
2252 if content_length:
2253 try:
2254 cl = int(content_length)
2255 if cl > MAX_TOTAL_UPLOAD_SIZE: 2255 ↛ 2256line 2255 didn't jump to line 2256 because the condition on line 2255 was never true
2256 return JSONResponse(
2257 {
2258 "success": False,
2259 "error": f"Request too large. Max {MAX_TOTAL_UPLOAD_SIZE // (1024 * 1024)}MB",
2260 },
2261 status_code=413,
2262 )
2263 except ValueError:
2264 pass
2266 # Parse the multipart form once and reuse — request.form() can only be
2267 # consumed once per request.
2268 form = await request.form()
2269 upload_files: list[UploadFile] = [
2270 f for f in form.getlist("files") if isinstance(f, UploadFile)
2271 ]
2272 if not upload_files:
2273 return JSONResponse(
2274 {"success": False, "error": "No files provided"},
2275 status_code=400,
2276 )
2277 if len(upload_files) > MAX_FILES_PER_UPLOAD:
2278 return JSONResponse(
2279 {
2280 "success": False,
2281 "error": f"Too many files. Max {MAX_FILES_PER_UPLOAD} per upload.",
2282 },
2283 status_code=400,
2284 )
2286 # Buffer every file's bytes here in the async path — UploadFile.read()
2287 # is async (the body may still be streaming from the client), and we
2288 # need the bytes available before we can hand control to the sync
2289 # threadpool worker. Large-file rejection happens up front so we
2290 # don't load anything we'll just throw away.
2291 pdf_storage_form_value = form.get("pdf_storage")
2292 files_data: list[dict] = [] # [{filename, content, oversized}]
2293 for uf in upload_files:
2294 # Preserve empty browser file inputs in the submitted-part count, but
2295 # do not read or process them. The canonical Flask route skipped these
2296 # parts while still reporting them in summary.total.
2297 if not uf.filename:
2298 files_data.append(
2299 {"filename": "", "content": None, "oversized": False}
2300 )
2301 continue
2302 # Per-file size guard before reading into memory (UploadFile.size
2303 # is set when the client sent Content-Length per part).
2304 if uf.size is not None and uf.size > MAX_FILE_SIZE:
2305 files_data.append(
2306 {
2307 "filename": uf.filename,
2308 "content": None,
2309 "oversized": True,
2310 }
2311 )
2312 continue
2313 content = await uf.read()
2314 files_data.append(
2315 {
2316 "filename": uf.filename,
2317 "content": content,
2318 "oversized": len(content) > MAX_FILE_SIZE,
2319 }
2320 )
2322 session_id = request.session.get("session_id")
2323 return await run_db_sync(
2324 _upload_to_collection_sync,
2325 files_data,
2326 pdf_storage_form_value,
2327 collection_id,
2328 username,
2329 session_id,
2330 MAX_FILE_SIZE,
2331 )
2334def _upload_to_collection_sync(
2335 files_data,
2336 pdf_storage_form_value,
2337 collection_id,
2338 username,
2339 session_id,
2340 MAX_FILE_SIZE,
2341):
2342 from ...database.session_context import (
2343 get_user_db_session,
2344 )
2345 from ...utilities.db_utils import get_settings_manager
2346 from ...security import sanitize_filename, UnsafeFilenameError
2347 import hashlib
2348 import uuid
2349 from ...research_library.services.pdf_storage_manager import (
2350 PDFStorageManager,
2351 resolve_pdf_storage_mode,
2352 )
2354 try:
2355 with get_user_db_session(username) as db_session:
2356 settings = get_settings_manager(db_session, username)
2358 # Verify collection exists in this user's database
2359 collection = (
2360 db_session.query(Collection).filter_by(id=collection_id).first()
2361 )
2363 if not collection:
2364 return JSONResponse(
2365 {"success": False, "error": "Collection not found"},
2366 status_code=404,
2367 )
2369 # Get PDF storage mode from form data, falling back to user's setting
2370 default_pdf_storage = settings.get_setting(
2371 "research_library.upload_pdf_storage", "none"
2372 )
2373 # `is None`, not `or`: Flask's form.get(k, default) returned the
2374 # default only when the field was ABSENT. An explicit empty
2375 # `pdf_storage=` fell through to the not-in-(database,none)
2376 # check below. `or` also swallows the empty string, silently
2377 # substituting the user's stored default -- which may be
2378 # "database" -- for a caller who explicitly sent nothing.
2379 pdf_storage = (
2380 default_pdf_storage
2381 if pdf_storage_form_value is None
2382 else pdf_storage_form_value
2383 )
2384 if pdf_storage not in ("database", "none"): 2384 ↛ 2387line 2384 didn't jump to line 2387 because the condition on line 2384 was never true
2385 # Security: user uploads can only use database (encrypted) or none (text-only)
2386 # Filesystem storage is not allowed for user uploads
2387 pdf_storage = "none"
2389 # Initialize PDF storage manager if storing PDFs in database
2390 pdf_storage_manager = None
2391 if pdf_storage == "database":
2392 library_root = settings.get_setting(
2393 "research_library.storage_path",
2394 str(get_library_directory()),
2395 )
2396 library_root = str(
2397 Path(os.path.expandvars(library_root))
2398 .expanduser()
2399 .resolve()
2400 )
2401 shared_library = settings.get_setting(
2402 "research_library.shared_library", False
2403 )
2404 # Per-user library root (issue #5521), mirroring the sibling
2405 # PDFStorageManager construction sites (library.py's
2406 # view_pdf_page, DownloadService.__init__, zotero
2407 # sync_service._library_root()): narrow the shared base to
2408 # this user's own subdirectory so two users' uploads can't
2409 # collide under one shared root. Not exploitable today since
2410 # this instance is only used in "database" mode (bytes go to
2411 # DocumentBlob, never through library_root), but the isolation
2412 # should hold regardless of which methods happen to read it.
2413 #
2414 # Route the mode through resolve_pdf_storage_mode() as well,
2415 # mirroring sync_service's own database-only construction —
2416 # `pdf_storage` is already guaranteed "database" by the branch
2417 # above, so this is a no-op today, but it ties the
2418 # construction to the actual resolved value (rather than a
2419 # bare literal) so it can't silently start bypassing the
2420 # unencrypted-filesystem gate if this code is ever refactored
2421 # to pass `pdf_storage` straight through.
2422 pdf_storage_manager = PDFStorageManager(
2423 library_root=apply_user_subdir(
2424 Path(library_root), username, shared_library
2425 ),
2426 storage_mode=resolve_pdf_storage_mode(pdf_storage),
2427 )
2428 logger.info("PDF storage mode: database (encrypted)")
2429 else:
2430 logger.info("PDF storage mode: none (text-only)")
2432 uploaded_files = []
2433 errors = []
2435 # Track hashes processed earlier in THIS request so we can
2436 # distinguish intra-batch duplicates (two uploaded files with
2437 # identical content) from duplicates against a pre-existing
2438 # library doc. Without this, the second occurrence of an
2439 # intra-batch duplicate was reported as "already_in_collection"
2440 # under the FIRST occurrence's filename -- the warning looked
2441 # like it was skipping books that were actually being added,
2442 # and the skipped entry showed the wrong filename (#5495).
2443 # Maps content hash -> Document so a later identical PDF in the
2444 # same batch can still run the PDF-upgrade path against the
2445 # document kept from the first occurrence.
2446 seen_hashes: dict[str, Document] = {}
2448 for fd in files_data:
2449 # Match main: an empty browser file input counts as a submitted
2450 # part in the summary but is not processed as an upload.
2451 if not fd["filename"]:
2452 continue
2454 # Files were already buffered + size-checked in the async
2455 # wrapper; oversized entries arrive with content=None and
2456 # oversized=True so we can record an error here without
2457 # having held the bytes in memory.
2458 if fd["oversized"]:
2459 # Sanitize BEFORE echoing. Flask sanitized first and only
2460 # ever put the cleaned name into `errors`; the port
2461 # inverted the order, so this branch reflected the raw
2462 # multipart filename — attacker-chosen bytes — straight
2463 # back in the JSON response. Masked today only because the
2464 # single consumer escapes it client-side, which is not a
2465 # property this handler should depend on. Every other
2466 # errors.append below already uses the sanitized name.
2467 try:
2468 safe_name = sanitize_filename(fd["filename"])
2469 except UnsafeFilenameError:
2470 safe_name = "rejected"
2471 errors.append(
2472 {
2473 "filename": safe_name,
2474 "error": f"File too large (max {MAX_FILE_SIZE // (1024 * 1024)}MB)",
2475 }
2476 )
2477 continue
2479 try:
2480 filename = sanitize_filename(fd["filename"])
2481 except UnsafeFilenameError:
2482 errors.append(
2483 {
2484 "filename": "rejected",
2485 "error": "Invalid or unsafe filename",
2486 }
2487 )
2488 continue
2490 # Per-file SAVEPOINT: a failure rolls back only this file so
2491 # earlier successes (and seen_hashes) stay consistent with
2492 # the outer transaction. Commit the savepoint on every
2493 # non-exception exit — including soft validation failures —
2494 # so we never leave nested SAVEPOINTs stacked open across
2495 # the batch (same pattern as research_sources_service).
2496 sp = None
2497 try:
2498 sp = db_session.begin_nested()
2500 # Size limits were already enforced in the async wrapper
2501 # (oversized entries arrive with content=None and are
2502 # rejected above), so main's Content-Length pre-flight and
2503 # post-read re-check have no work left to do here.
2504 file_content = fd["content"]
2505 # Calculate file hash for deduplication
2506 file_hash = hashlib.sha256(file_content).hexdigest()
2508 # Intra-batch duplicate: an earlier file in THIS
2509 # request had identical bytes. Report it as its own
2510 # status so the UI doesn't show it under the first
2511 # occurrence's filename and pretend it was a
2512 # "skipped - already in collection" hit. The first
2513 # occurrence already linked the matching Document to
2514 # this collection (or to the library), so doing the
2515 # Document hash lookup and link check here would be
2516 # misleading at best. Still allow PDF upgrade when
2517 # the kept twin was stored text-only and this copy
2518 # is a PDF with database storage enabled.
2519 if file_hash in seen_hashes:
2520 logger.info(
2521 "Intra-batch duplicate detected: {} (hash {})",
2522 filename,
2523 file_hash,
2524 )
2525 target_doc = seen_hashes[file_hash]
2526 pdf_upgraded = _try_pdf_upgrade(
2527 db_session=db_session,
2528 document=target_doc,
2529 file_content=file_content,
2530 filename=filename,
2531 pdf_storage=pdf_storage,
2532 pdf_storage_manager=pdf_storage_manager,
2533 )
2534 sp.commit()
2535 uploaded_files.append(
2536 {
2537 "filename": filename,
2538 "status": "duplicate_in_batch",
2539 "id": target_doc.id,
2540 "pdf_upgraded": pdf_upgraded,
2541 }
2542 )
2543 continue
2545 # Check if document already exists
2546 existing_doc = (
2547 db_session.query(Document)
2548 .filter_by(document_hash=file_hash)
2549 .first()
2550 )
2552 if existing_doc:
2553 # Document exists, check if we can upgrade to include PDF
2554 pdf_upgraded = _try_pdf_upgrade(
2555 db_session=db_session,
2556 document=existing_doc,
2557 file_content=file_content,
2558 filename=filename,
2559 pdf_storage=pdf_storage,
2560 pdf_storage_manager=pdf_storage_manager,
2561 )
2563 # Check if already in collection
2564 existing_link = (
2565 db_session.query(DocumentCollection)
2566 .filter_by(
2567 document_id=existing_doc.id,
2568 collection_id=collection_id,
2569 )
2570 .first()
2571 )
2573 if not existing_link:
2574 ensure_in_collection(
2575 db_session, existing_doc.id, collection_id
2576 )
2577 status = "added_to_collection"
2578 if pdf_upgraded:
2579 status = "added_to_collection_pdf_upgraded"
2580 sp.commit()
2581 uploaded_files.append(
2582 {
2583 # Report the file the user actually
2584 # uploaded, not the existing doc's
2585 # filename. Otherwise the user sees
2586 # one filename repeated under both
2587 # "added to collection" and
2588 # "new uploads" (or here under
2589 # "already in collection") and
2590 # can't tell which file was which.
2591 "filename": filename,
2592 "status": status,
2593 "id": existing_doc.id,
2594 "pdf_upgraded": pdf_upgraded,
2595 }
2596 )
2597 seen_hashes[file_hash] = existing_doc
2598 else:
2599 status = "already_in_collection"
2600 if pdf_upgraded:
2601 status = "pdf_upgraded"
2602 sp.commit()
2603 uploaded_files.append(
2604 {
2605 "filename": filename,
2606 "status": status,
2607 "id": existing_doc.id,
2608 "pdf_upgraded": pdf_upgraded,
2609 }
2610 )
2611 seen_hashes[file_hash] = existing_doc
2612 else:
2613 # Create new document
2614 from ...document_loaders import (
2615 extract_text_from_bytes,
2616 is_extension_supported,
2617 )
2619 file_extension = Path(filename).suffix.lower()
2621 # Validate extension is supported before extraction
2622 if not is_extension_supported(file_extension):
2623 sp.commit()
2624 errors.append(
2625 {
2626 "filename": filename,
2627 "error": f"Unsupported format: {file_extension}",
2628 }
2629 )
2630 continue
2632 # Use file_type without leading dot for storage
2633 file_type = (
2634 file_extension[1:]
2635 if file_extension.startswith(".")
2636 else file_extension
2637 )
2639 # Extract text using document_loaders module
2640 extracted_text = extract_text_from_bytes(
2641 file_content, file_extension, filename
2642 )
2644 # Clean the extracted text to remove surrogate characters
2645 if extracted_text:
2646 from ...text_processing import remove_surrogates
2648 extracted_text = remove_surrogates(extracted_text)
2650 if not extracted_text:
2651 sp.commit()
2652 errors.append(
2653 {
2654 "filename": filename,
2655 "error": f"Could not extract text from {file_type} file",
2656 }
2657 )
2658 logger.warning(
2659 f"Skipping file {filename} - no text could be extracted"
2660 )
2661 continue
2663 # Get or create the user_upload source type
2664 logger.info(
2665 f"Getting or creating user_upload source type for {filename}"
2666 )
2667 source_type = (
2668 db_session.query(SourceType)
2669 .filter_by(name="user_upload")
2670 .first()
2671 )
2672 if not source_type: 2672 ↛ 2673line 2672 didn't jump to line 2673 because the condition on line 2672 was never true
2673 logger.info("Creating new user_upload source type")
2674 source_type = SourceType(
2675 id=str(uuid.uuid4()),
2676 name="user_upload",
2677 display_name="User Upload",
2678 description="Documents uploaded by users",
2679 icon="fas fa-upload",
2680 )
2681 db_session.add(source_type)
2682 db_session.flush()
2683 logger.info(
2684 f"Created source type with ID: {source_type.id}"
2685 )
2686 else:
2687 logger.info(
2688 f"Found existing source type with ID: {source_type.id}"
2689 )
2691 # Create document with extracted text (no username needed - in user's own database)
2692 # Note: uploaded_at uses default=utcnow() in the model, so we don't need to set it manually
2693 doc_id = str(uuid.uuid4())
2694 logger.info(
2695 f"Creating document {doc_id} for {filename}"
2696 )
2698 # Determine storage mode and file_path
2699 store_pdf_in_db = (
2700 pdf_storage == "database"
2701 and file_type == "pdf"
2702 and pdf_storage_manager is not None
2703 )
2705 new_doc = Document(
2706 id=doc_id,
2707 source_type_id=source_type.id,
2708 filename=filename,
2709 document_hash=file_hash,
2710 file_size=len(file_content),
2711 file_type=file_type,
2712 text_content=extracted_text, # Always store extracted text
2713 file_path=None
2714 if store_pdf_in_db
2715 else FILE_PATH_TEXT_ONLY,
2716 storage_mode="database"
2717 if store_pdf_in_db
2718 else "none",
2719 )
2720 db_session.add(new_doc)
2721 db_session.flush() # Get the ID
2722 logger.info(
2723 f"Document {new_doc.id} created successfully"
2724 )
2726 # Store PDF in encrypted database if requested
2727 pdf_stored = False
2728 if store_pdf_in_db:
2729 try:
2730 pdf_storage_manager.save_pdf(
2731 pdf_content=file_content,
2732 document=new_doc,
2733 session=db_session,
2734 filename=filename,
2735 )
2736 pdf_stored = True
2737 logger.info(
2738 f"PDF stored in encrypted database for {filename}"
2739 )
2740 except Exception:
2741 logger.exception(
2742 f"Failed to store PDF in database for {filename}"
2743 )
2744 # Continue without PDF storage - text is still saved
2746 # Add to collection
2747 ensure_in_collection(
2748 db_session, new_doc.id, collection_id
2749 )
2751 # Release the per-file SAVEPOINT so later failures cannot
2752 # undo this file's writes (and so soft/success paths do
2753 # not leave nested SAVEPOINTs stacked open).
2754 sp.commit()
2756 uploaded_files.append(
2757 {
2758 "filename": filename,
2759 "status": "uploaded",
2760 "id": new_doc.id,
2761 "text_length": len(extracted_text),
2762 "pdf_stored": pdf_stored,
2763 }
2764 )
2765 seen_hashes[file_hash] = new_doc
2767 except Exception:
2768 # A per-file savepoint rollback isolates the failure of this
2769 # file so earlier successful uploads in the batch remain intact
2770 # in the transaction state. Record the error and log the traceback
2771 # FIRST so a rollback failure cannot suppress them or crash the batch.
2772 errors.append(
2773 {
2774 "filename": filename,
2775 "error": "Failed to upload file",
2776 }
2777 )
2778 logger.exception(f"Error uploading file {filename}")
2779 if sp is not None:
2780 try:
2781 sp.rollback()
2782 except Exception:
2783 logger.opt(exception=True).warning(
2784 f"Failed to rollback savepoint for {filename}"
2785 )
2787 db_session.commit()
2789 # Trigger auto-indexing for successfully uploaded documents
2790 document_ids = [
2791 f["id"]
2792 for f in uploaded_files
2793 if f.get("status") in ("uploaded", "added_to_collection")
2794 ]
2795 if document_ids:
2796 from ...database.session_passwords import session_password_store
2798 # session_id was captured by the async wrapper before to_thread
2799 db_password = session_password_store.get_session_password(
2800 username, session_id
2801 )
2802 if db_password:
2803 trigger_auto_index(
2804 document_ids, collection_id, username, db_password
2805 )
2807 return {
2808 "success": True,
2809 "uploaded": uploaded_files,
2810 "errors": errors,
2811 "summary": {
2812 "total": len(files_data),
2813 "successful": len(uploaded_files),
2814 "failed": len(errors),
2815 },
2816 }
2818 except Exception as e:
2819 return handle_api_error("uploading files", e)
2822# Research History Semantic Search Routes have been moved to
2823# web/routers/library_search.py
2826@router.get("/api/collections/{collection_id}/documents")
2827def get_collection_documents(
2828 request: Request,
2829 collection_id,
2830 username: Annotated[str, Depends(require_auth)],
2831):
2832 """Get all documents in a collection."""
2833 from ...database.session_context import get_user_db_session
2835 try:
2836 with get_user_db_session(username) as db_session:
2837 # Verify collection exists in this user's database
2838 collection = (
2839 db_session.query(Collection).filter_by(id=collection_id).first()
2840 )
2842 if not collection:
2843 return JSONResponse(
2844 {"success": False, "error": "Collection not found"},
2845 status_code=404,
2846 )
2848 # Look up note source type to filter notes into separate array
2849 note_source_type = (
2850 db_session.query(SourceType).filter_by(name="note").first()
2851 )
2852 note_source_type_id = (
2853 note_source_type.id if note_source_type else None
2854 )
2856 # Get documents through junction table. Compute "has text in DB"
2857 # at the SQL level and defer the text_content column, so listing a
2858 # collection doesn't pull every document's full text body into
2859 # memory just to test it for truthiness (#4560).
2860 has_text_col = (
2861 Document.text_content.isnot(None)
2862 & (Document.text_content != "")
2863 ).label("has_text_db")
2864 doc_links = (
2865 db_session.query(DocumentCollection, Document, has_text_col)
2866 .join(Document)
2867 .options(defer(Document.text_content))
2868 .filter(DocumentCollection.collection_id == collection_id)
2869 .all()
2870 )
2872 documents = []
2873 for link, doc, has_text_db in doc_links:
2874 # Skip notes — they go in the separate notes array
2875 if (
2876 note_source_type_id
2877 and doc.source_type_id == note_source_type_id
2878 ):
2879 continue
2880 # Check if PDF file is stored
2881 has_pdf = bool(
2882 doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
2883 )
2884 has_text_db = bool(has_text_db)
2886 # Use title if available, otherwise filename
2887 display_title = doc.title or doc.filename or "Untitled"
2889 # Get source type name
2890 source_type_name = (
2891 doc.source_type.name if doc.source_type else "unknown"
2892 )
2894 # Check if document is in other collections
2895 other_collections_count = (
2896 db_session.query(DocumentCollection)
2897 .filter(
2898 DocumentCollection.document_id == doc.id,
2899 DocumentCollection.collection_id != collection_id,
2900 )
2901 .count()
2902 )
2904 documents.append(
2905 {
2906 "id": doc.id,
2907 "filename": display_title,
2908 "title": display_title,
2909 "file_type": doc.file_type,
2910 "file_size": doc.file_size,
2911 "uploaded_at": doc.created_at.isoformat()
2912 if doc.created_at
2913 else None,
2914 "indexed": link.indexed,
2915 "chunk_count": link.chunk_count,
2916 "last_indexed_at": link.last_indexed_at.isoformat()
2917 if link.last_indexed_at
2918 else None,
2919 "has_pdf": has_pdf,
2920 "has_text_db": has_text_db,
2921 "source_type": source_type_name,
2922 "in_other_collections": other_collections_count > 0,
2923 "other_collections_count": other_collections_count,
2924 }
2925 )
2927 # Get notes in this collection (note_source_type resolved above)
2928 notes = []
2929 if note_source_type:
2930 note_links = (
2931 db_session.query(DocumentCollection, Document)
2932 .join(Document)
2933 .filter(
2934 DocumentCollection.collection_id == collection_id,
2935 Document.source_type_id == note_source_type.id,
2936 )
2937 .all()
2938 )
2939 for link, note in note_links:
2940 content = note.text_content or ""
2941 notes.append(
2942 {
2943 "id": note.id,
2944 "title": note.title or "Untitled",
2945 "content_preview": content[:200] + "..."
2946 if len(content) > 200
2947 else content,
2948 "tags": note.tags or [],
2949 "pinned": note.favorite,
2950 "created_at": note.created_at.isoformat()
2951 if note.created_at
2952 else None,
2953 "updated_at": note.updated_at.isoformat()
2954 if note.updated_at
2955 else None,
2956 "indexed": link.indexed,
2957 "chunk_count": link.chunk_count,
2958 "source_type": "note",
2959 }
2960 )
2962 # Get index file size if available
2963 index_file_size = None
2964 index_file_size_bytes = None
2965 collection_name = f"collection_{collection_id}"
2966 rag_index = (
2967 db_session.query(RAGIndex)
2968 .filter_by(collection_name=collection_name)
2969 .first()
2970 )
2971 if rag_index and rag_index.index_path:
2972 from pathlib import Path
2974 index_path = Path(rag_index.index_path)
2975 if index_path.exists(): 2975 ↛ 2986line 2975 didn't jump to line 2986 because the condition on line 2975 was always true
2976 size_bytes = index_path.stat().st_size
2977 index_file_size_bytes = size_bytes
2978 # Format as human-readable
2979 if size_bytes < 1024:
2980 index_file_size = f"{size_bytes} B"
2981 elif size_bytes < 1024 * 1024:
2982 index_file_size = f"{size_bytes / 1024:.1f} KB"
2983 else:
2984 index_file_size = f"{size_bytes / (1024 * 1024):.1f} MB"
2986 return {
2987 "success": True,
2988 "collection": {
2989 "id": collection.id,
2990 "name": collection.name,
2991 "description": collection.description,
2992 "is_public": bool(getattr(collection, "is_public", False)),
2993 "agent_enabled": _agent_enabled_default_on(collection),
2994 "embedding_model": collection.embedding_model,
2995 "embedding_model_type": collection.embedding_model_type.value
2996 if collection.embedding_model_type
2997 else None,
2998 "embedding_dimension": collection.embedding_dimension,
2999 "chunk_size": collection.chunk_size,
3000 "chunk_overlap": collection.chunk_overlap,
3001 # Advanced settings
3002 "splitter_type": collection.splitter_type,
3003 "distance_metric": collection.distance_metric,
3004 "index_type": collection.index_type,
3005 "normalize_vectors": collection.normalize_vectors,
3006 # Index file info
3007 "index_file_size": index_file_size,
3008 "index_file_size_bytes": index_file_size_bytes,
3009 "collection_type": collection.collection_type,
3010 # Deletion policy comes from the server so the UI can't
3011 # drift from PROTECTED_COLLECTION_TYPES: the details page
3012 # hides its Delete button for system collections, whose
3013 # deletion the service refuses with a 409 anyway.
3014 "is_protected": _is_protected_collection(collection),
3015 },
3016 "documents": documents,
3017 "notes": notes,
3018 }
3020 except Exception as e:
3021 return handle_api_error("getting collection documents", e)
3024# =============================================================================
3025# Shared collection-indexing helpers
3026#
3027# Three routes index a collection: the single-collection SSE route
3028# (index_collection), the bulk SSE route (index_all), and the background worker
3029# (_background_index_worker). These helpers hold the logic that must stay
3030# identical across all three so it cannot drift again — the embedding metadata
3031# to persist, the force-reindex cleanup, and the query for documents to index.
3032# The callers differ only in how they report progress (SSE stream vs
3033# TaskMetadata) and so keep their own loops.
3034# =============================================================================
3037def _store_collection_embedding_metadata(collection, rag_service):
3038 """Persist the embedding/index configuration used to index a collection.
3040 Run on first index or force-reindex. Includes the embedding dimension,
3041 probed by embedding a test string (the same way LibraryRAGService derives
3042 it for the RAGIndex row). Previously this read
3043 ``embedding_manager.provider.embedding_dimension``, an attribute the real
3044 LocalEmbeddingManager does not have, so the probe always failed and
3045 ``embedding_dimension`` was persisted as NULL via both indexing paths. Does
3046 not commit — the caller owns the transaction.
3047 """
3048 embedding_dim = None
3049 try:
3050 # Derive the dimension by embedding a test string — mirrors
3051 # LibraryRAGService.* which computes len(embed_query("test")) for the
3052 # RAGIndex row. The embedding manager has no usable dimension attribute.
3053 test_embedding = rag_service.embedding_manager.embeddings.embed_query(
3054 "test"
3055 )
3056 embedding_dim = len(test_embedding)
3057 except Exception:
3058 logger.debug("Could not determine embedding dimension", exc_info=True)
3060 collection.embedding_model = rag_service.embedding_model
3061 collection.embedding_model_type = EmbeddingProvider(
3062 rag_service.embedding_provider
3063 )
3064 collection.embedding_dimension = embedding_dim
3065 collection.chunk_size = rag_service.chunk_size
3066 collection.chunk_overlap = rag_service.chunk_overlap
3067 # Advanced settings
3068 collection.splitter_type = rag_service.splitter_type
3069 collection.text_separators = rag_service.text_separators
3070 collection.distance_metric = rag_service.distance_metric
3071 # Ensure normalize_vectors is a proper boolean for the database
3072 collection.normalize_vectors = bool(rag_service.normalize_vectors)
3073 collection.index_type = rag_service.index_type
3074 logger.info(
3075 f"Stored embedding metadata for collection: provider={rag_service.embedding_provider}"
3076 )
3079def _reset_collection_for_reindex(db_session, collection_id):
3080 """Clear old chunks, FAISS indices, and indexed-state before a force rebuild.
3082 Prevents mixed-model vectors / stale chunks when re-indexing (e.g. if a
3083 previous run was cancelled midway). Shared so a force-reindex behaves the
3084 same whether started via the SSE route or the background worker. Does not
3085 commit — the caller owns the transaction.
3087 Returns the list of on-disk FAISS index paths to unlink AFTER the caller
3088 commits. The RAGIndex ROWS are deleted in this transaction, but the files
3089 must NOT be removed before the commit that could still roll back — a
3090 rollback would restore the rows while the files were already gone, leaving
3091 the collection pointing at missing indices (mirrors collection deletion).
3092 """
3093 # Import stays function-local: hoisting it would pull in the whole
3094 # deletion package (deletion/__init__ imports its services) at
3095 # route-module import time for a path only exercised on force-reindex.
3096 from ...research_library.deletion.utils.cascade_helper import (
3097 CascadeHelper,
3098 )
3100 collection_name = f"collection_{collection_id}"
3102 # Delete all old document chunks from DB
3103 deleted_chunks = CascadeHelper.delete_collection_chunks(
3104 db_session, collection_name
3105 )
3106 logger.info(
3107 f"Cleared {deleted_chunks} old chunks for collection {collection_id}"
3108 )
3110 # Delete old RAGIndex records (DB only). Files unlinked by the caller after
3111 # commit (see docstring). RagDocumentStatus cascade-deletes via FK.
3112 rag_result = CascadeHelper.delete_rag_indices_for_collection(
3113 db_session, collection_name, unlink_files=False
3114 )
3115 logger.info(
3116 f"Cleared old RAG indices for collection {collection_id}: {rag_result}"
3117 )
3119 # Mark all documents as unindexed
3120 db_session.query(DocumentCollection).filter_by(
3121 collection_id=collection_id
3122 ).update(
3123 {
3124 DocumentCollection.indexed: False,
3125 DocumentCollection.chunk_count: 0,
3126 }
3127 )
3128 logger.info(f"Reset indexing state for collection {collection_id}")
3129 return rag_result["index_paths"]
3132def _unlink_reindex_faiss_files(index_paths):
3133 """Unlink FAISS index files returned by _reset_collection_for_reindex, AFTER
3134 the caller's DB commit. Best-effort: the DB delete is already durable."""
3135 if not index_paths:
3136 return
3137 from ...research_library.deletion.utils.cascade_helper import (
3138 CascadeHelper,
3139 )
3140 from ...config.paths import get_cache_directory
3142 # NOTE: allowed_root is the COARSE shared rag_indices/ root, not the
3143 # tighter per-user rag_indices/<sha256(user)>/ subdir. It cannot be
3144 # narrowed to the per-user subdir without risking refused deletes:
3145 # pre-per-user-scoping (legacy) indexes live DIRECTLY in this shared
3146 # root (see library_rag_service._migrate_legacy_index_files), so a
3147 # per-user allowed_root would reject a legacy-layout RAGIndex.index_path
3148 # and orphan its files. The containment check still blocks symlink /
3149 # ancestor escapes outside rag_indices/ entirely.
3150 rag_indices_root = get_cache_directory() / "rag_indices"
3151 for path in index_paths:
3152 CascadeHelper.delete_faiss_index_files(
3153 path, allowed_root=rag_indices_root
3154 )
3157def _query_documents_to_index(db_session, collection_id, force_reindex):
3158 """Return the ``(DocumentCollection, Document)`` pairs to index.
3160 When not force-reindexing, only un-indexed documents are returned.
3161 """
3162 query = (
3163 db_session.query(DocumentCollection, Document)
3164 .join(Document)
3165 # Don't load the large text_content body: the indexing loop only
3166 # reads doc.id/filename/title (index_document re-fetches by id in a
3167 # worker thread), so loading every document's full text here just to
3168 # iterate is what exhausts memory on big collections (#4560).
3169 .options(defer(Document.text_content))
3170 .filter(DocumentCollection.collection_id == collection_id)
3171 )
3173 if not force_reindex:
3174 # IMPORTANT: DO NOT CHANGE THIS LINE! DO NOT REMOVE THIS COMMENT!
3175 # We MUST use `== False` instead of `not DocumentCollection.indexed`
3176 # The Python `not` operator does NOT work correctly in SQLAlchemy filters.
3177 # Using `not` will cause the query to return NO results (zero documents).
3178 # SQLAlchemy requires explicit comparison: `== False` or `== True`
3179 # This has been fixed multiple times - DO NOT change it back to `not`!
3180 query = query.filter(DocumentCollection.indexed == False) # noqa: E712
3182 return query.all()
3185@router.get("/api/collections/{collection_id}/index")
3186def index_collection(
3187 request: Request,
3188 collection_id,
3189 username: Annotated[str, Depends(require_auth)],
3190):
3191 """Index all documents in a collection with Server-Sent Events progress."""
3192 from ...database.session_context import get_user_db_session
3193 from ...database.session_passwords import session_password_store
3195 # Parse as bool — a raw string "false" is truthy, which would silently
3196 # force a full reindex for every caller that asked to skip already-indexed
3197 # documents.
3198 force_reindex = (
3199 request.query_params.get("force_reindex", "false").lower() == "true"
3200 )
3201 session_id = request.session.get("session_id")
3203 logger.info(f"Starting index_collection, force_reindex={force_reindex}")
3205 # Get password for thread access to encrypted database
3206 db_password = None
3207 if session_id:
3208 db_password = session_password_store.get_session_password(
3209 username, session_id
3210 )
3212 # Create RAG service — on force reindex use current default model
3213 rag_service = get_rag_service(
3214 request, username, collection_id, use_defaults=force_reindex
3215 )
3216 logger.info(
3217 f"RAG service created: provider={rag_service.embedding_provider}"
3218 )
3220 # Resolve the parallel worker bound before the generator runs — the
3221 # parallel helper must never touch SettingsManager from a worker thread.
3222 with get_user_db_session(username, db_password) as _settings_session:
3223 _settings = get_settings_manager(_settings_session, username)
3224 try:
3225 _max_workers = int(
3226 _settings.get_setting("rag.indexing_max_parallel_docs", 4)
3227 )
3228 except Exception:
3229 _max_workers = 4
3230 _max_workers = max(1, min(_max_workers, 16))
3232 def generate():
3233 """Generator for SSE progress updates."""
3234 logger.info("SSE generator started")
3235 _sse_cancel = threading.Event()
3236 with _active_sse_indexers_lock:
3237 _active_sse_indexers.setdefault(
3238 (username, collection_id), set()
3239 ).add(_sse_cancel)
3240 worker_thread = None
3241 try:
3242 # All setup DB work happens in a SHORT-LIVED session; the per-doc
3243 # ids/filenames are snapshotted into plain tuples so the long
3244 # indexing loop below (which yields progress + heartbeats) holds NO
3245 # session across a yield — see download_bulk for the Starlette/anyio
3246 # thread-affinity rationale. index_document opens its own session
3247 # per document in a worker thread and commits its own writes, so the
3248 # old outer db_session.commit()/flush() was vestigial.
3249 collection_found = True
3250 collection_name = None
3251 doc_info = []
3252 with get_user_db_session(username, db_password) as db_session:
3253 # Verify collection exists in this user's database
3254 collection = (
3255 db_session.query(Collection)
3256 .filter_by(id=collection_id)
3257 .first()
3258 )
3260 if not collection:
3261 collection_found = False
3262 else:
3263 collection_name = collection.name
3264 # Store embedding metadata AND reset the old index in ONE
3265 # commit (prevents stale / mixed-model vectors). Committing
3266 # the config write separately from the reset leaves a window
3267 # where a crash keeps the new config but the old vectors
3268 # survive.
3269 changed = False
3270 faiss_reset_paths = []
3271 if collection.embedding_model is None or force_reindex:
3272 _store_collection_embedding_metadata(
3273 collection, rag_service
3274 )
3275 changed = True
3276 if force_reindex:
3277 faiss_reset_paths = _reset_collection_for_reindex(
3278 db_session, collection_id
3279 )
3280 changed = True
3281 if changed:
3282 db_session.commit()
3283 _unlink_reindex_faiss_files(faiss_reset_paths)
3285 # Snapshot (id, display name) for each document to index.
3286 doc_info = [
3287 (doc.id, doc.filename or doc.title or "Unknown")
3288 for _link, doc in _query_documents_to_index(
3289 db_session, collection_id, force_reindex
3290 )
3291 ]
3293 if not collection_found:
3294 yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
3295 return
3297 if not doc_info:
3298 logger.info("No documents to index in collection")
3299 yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
3300 return
3302 total = len(doc_info)
3303 logger.info(f"Found {total} documents to index")
3304 results = {
3305 "successful": 0,
3306 "skipped": 0,
3307 "failed": 0,
3308 "errors": [],
3309 }
3311 yield f"data: {json.dumps({'type': 'start', 'message': f'Indexing {total} documents in collection: {collection_name}'})}\n\n"
3313 # Fan out indexing across a bounded worker pool. We run the
3314 # pool in a background thread so the SSE generator's main
3315 # thread can still yield heartbeats during long embedding
3316 # round-trips (matters for nginx / browser SSE timeout
3317 # behaviour). The progress_callback fired by the parallel
3318 # helper pushes per-document events into ``progress_queue``,
3319 # which the generator drains as they arrive.
3320 #
3321 # ``_max_workers`` is resolved from settings before the
3322 # generator starts (closure capture) so the parallel helper
3323 # never touches SettingsManager from inside a worker.
3324 progress_queue: queue.Queue = queue.Queue()
3325 _SENTINEL_COMPLETE = object()
3326 _SENTINEL_ERROR = object()
3328 def _on_progress(
3329 completed: int,
3330 total: int,
3331 title: str,
3332 status: str,
3333 ) -> None:
3334 progress_queue.put(
3335 (
3336 "progress",
3337 completed,
3338 total,
3339 title,
3340 status,
3341 )
3342 )
3344 def _run_parallel():
3345 try:
3346 # ``is_cancelled`` polls the SSE-disconnect event
3347 # so the helper can exit early between completions
3348 # when the client has gone away; the helper's
3349 # ``pool.shutdown(wait=True, ...)`` still drains
3350 # in-flight workers before returning so no worker
3351 # outlives the helper return — see ``index_documents_parallel`` for the
3352 # rationale.
3353 aggregate = rag_service.index_documents_parallel(
3354 doc_info,
3355 collection_id,
3356 force_reindex=force_reindex,
3357 max_workers=_max_workers,
3358 progress_callback=_on_progress,
3359 is_cancelled=_sse_cancel.is_set,
3360 )
3361 progress_queue.put((_SENTINEL_COMPLETE, aggregate))
3362 except Exception as exc:
3363 logger.exception(
3364 "Parallel indexing in SSE generator crashed"
3365 )
3366 progress_queue.put((_SENTINEL_ERROR, exc))
3367 finally:
3368 # Best-effort thread-local DB session cleanup so a
3369 # long-running worker batch doesn't accumulate
3370 # scoped sessions (#3194).
3371 try:
3372 from ...database.thread_local_session import (
3373 cleanup_current_thread,
3374 )
3376 cleanup_current_thread()
3377 except Exception:
3378 logger.debug(
3379 "best-effort thread-local DB session cleanup",
3380 exc_info=True,
3381 )
3383 # Copy the request's contextvars so username-dependent
3384 # fallbacks (e.g. log attribution) inside the worker resolve
3385 # to the authenticated user rather than None.
3386 _par_ctx = contextvars.copy_context()
3387 worker_thread = threading.Thread(
3388 target=_par_ctx.run,
3389 args=(_run_parallel,),
3390 daemon=True,
3391 name="index-collection-parallel",
3392 )
3393 worker_thread.start()
3395 # Drain the queue: yield progress events as they arrive,
3396 # yield heartbeat comments on idle to keep SSE proxies
3397 # from timing out, exit on completion / error sentinels.
3398 heartbeat_interval = 5 # seconds
3399 _aggregate = None
3400 while True:
3401 try:
3402 item = progress_queue.get(timeout=heartbeat_interval)
3403 except queue.Empty:
3404 # No event in 5s. Heartbeat unless the worker
3405 # died without signalling (would hang forever).
3406 if worker_thread.is_alive():
3407 yield f": heartbeat {total}\n\n"
3408 continue
3409 logger.exception(
3410 "Indexer worker exited without completion "
3411 "signal; aborting SSE"
3412 )
3413 yield f"data: {json.dumps({'type': 'error', 'error': 'Indexer stopped unexpectedly'})}\n\n"
3414 return
3416 if item[0] is _SENTINEL_COMPLETE:
3417 _aggregate = item[1]
3418 break
3419 if item[0] is _SENTINEL_ERROR: 3419 ↛ 3420line 3419 didn't jump to line 3420 because the condition on line 3419 was never true
3420 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
3421 return
3423 _kind, completed, total, title, status = item
3424 yield f"data: {json.dumps({'type': 'progress', 'current': completed, 'total': total, 'filename': title, 'percent': int((completed / total) * 100)})}\n\n"
3426 if status == "error":
3427 # Per-doc error surfaced so the browser can show
3428 # failed documents inline. Scrub creds/keys here
3429 # (the full trace stays in server logs above).
3430 # Look the error up from the aggregate's errors
3431 # list — at this point _aggregate is None
3432 # (indexer still running), so we accept a brief
3433 # imprecision on the exact error text and emit
3434 # a generic doc_error; the final ``complete``
3435 # event below will list precise errors.
3436 yield f"data: {json.dumps({'type': 'doc_error', 'filename': title, 'error': 'Indexing failed'})}\n\n"
3438 if _aggregate is not None: 3438 ↛ 3460line 3438 didn't jump to line 3460 because the condition on line 3438 was always true
3439 results["successful"] = _aggregate["successful"]
3440 results["skipped"] = _aggregate["skipped"]
3441 results["failed"] = _aggregate["failed"]
3442 # Replace placeholder errors with the structured ones
3443 # from the parallel helper so the final ``complete``
3444 # event matches what previous serial runs returned.
3445 results["errors"] = [
3446 {
3447 "filename": err.get("title"),
3448 "error": sanitize_error_message(str(err.get("error")))
3449 or "Failed to index document",
3450 }
3451 for err in _aggregate.get("errors", [])
3452 ]
3453 logger.info(
3454 f"Indexing complete: "
3455 f"{results['successful']} successful, "
3456 f"{results['failed']} failed, "
3457 f"{results['skipped']} skipped"
3458 )
3460 yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
3461 logger.info("SSE generator finished successfully")
3463 except Exception:
3464 logger.exception("Error in collection indexing")
3465 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
3466 finally:
3467 # Signal disconnect/cancellation first so any in-flight checks notice it immediately,
3468 # then remove this specific stream's cancel event from the registry under lock.
3469 _sse_cancel.set()
3470 with _active_sse_indexers_lock:
3471 events = _active_sse_indexers.get((username, collection_id))
3472 if events is not None: 3472 ↛ 3484line 3472 didn't jump to line 3484
3473 events.discard(_sse_cancel)
3474 if not events: 3474 ↛ 3484line 3474 didn't jump to line 3484
3475 _active_sse_indexers.pop(
3476 (username, collection_id), None
3477 )
3478 # On client disconnect this finally can run on the event loop
3479 # thread (generator close), so the drain must be bounded: give
3480 # in-flight workers a short grace period, then hand the
3481 # remaining join + service close to a detached daemon thread.
3482 # _sse_cancel is already set, so the parallel helper admits no
3483 # new work and terminates once in-flight docs finish.
3484 if worker_thread is not None and worker_thread.is_alive():
3485 worker_thread.join(timeout=5.0)
3486 if worker_thread is not None and worker_thread.is_alive():
3487 logger.info(
3488 "Indexer still draining after SSE close; deferring "
3489 "cleanup to background thread"
3490 )
3491 _lingering_worker = worker_thread
3493 def _drain_and_close():
3494 _lingering_worker.join()
3495 safe_close(
3496 rag_service,
3497 "rag_service (index-collection SSE, deferred)",
3498 )
3500 threading.Thread(
3501 target=_drain_and_close,
3502 daemon=True,
3503 name="index-collection-drain",
3504 ).start()
3505 else:
3506 safe_close(rag_service, "rag_service (index-collection SSE)")
3508 response = WorkerCleanupStreamingResponse(
3509 generate(), media_type="text/event-stream"
3510 )
3511 # Prevent buffering for proper SSE streaming
3512 response.headers["Cache-Control"] = "no-cache, no-transform"
3513 response.headers["Connection"] = "keep-alive"
3514 response.headers["X-Accel-Buffering"] = "no"
3515 return response
3518# =============================================================================
3519# Background Indexing Endpoints
3520# =============================================================================
3523def _get_rag_service_for_thread(
3524 collection_id: str,
3525 username: str,
3526 db_password: str,
3527 use_defaults: bool = False,
3528) -> LibraryRAGService:
3529 """
3530 Create RAG service for use in background threads (no Flask context).
3532 Delegates settings resolution to the shared rag_service_factory, then
3533 propagates db_password to the embedding manager for thread-safe DB access.
3534 """
3535 from ...research_library.services.rag_service_factory import (
3536 get_rag_service as _get_rag_service,
3537 )
3539 service = _get_rag_service(
3540 username,
3541 collection_id,
3542 use_defaults=use_defaults,
3543 db_password=db_password,
3544 )
3545 # The factory passes db_password to LibraryRAGService, but __init__ stores
3546 # it in the backing field (_db_password) without propagating to sub-managers.
3547 # Re-assign via the property setter to propagate to embedding_manager and
3548 # integrity_manager, which need it for thread-safe session access.
3549 service.db_password = db_password
3550 return service
3553def trigger_auto_index(
3554 document_ids: list[str],
3555 collection_id: str,
3556 username: str,
3557 db_password: str,
3558) -> None:
3559 """
3560 Trigger automatic RAG indexing for documents if auto-indexing is enabled.
3562 This function checks the auto_index_enabled setting and spawns a background
3563 thread to index the specified documents. It does not block the caller.
3565 Args:
3566 document_ids: List of document IDs to index
3567 collection_id: The collection to index into
3568 username: The username for database access
3569 db_password: The user's database password for thread-safe access
3570 """
3571 from ...database.session_context import get_user_db_session
3573 if not document_ids:
3574 logger.debug("No documents to auto-index")
3575 return
3577 # Check if auto-indexing is enabled
3578 try:
3579 with get_user_db_session(username, db_password) as db_session:
3580 settings = SettingsManager(db_session)
3581 auto_index_enabled = settings.get_bool_setting(
3582 "research_library.auto_index_enabled", True
3583 )
3585 if not auto_index_enabled:
3586 logger.debug("Auto-indexing is disabled, skipping")
3587 return
3589 try:
3590 _auto_max_workers = int(
3591 settings.get_setting("rag.indexing_max_parallel_docs", 4)
3592 )
3593 except Exception:
3594 _auto_max_workers = 4
3595 _auto_max_workers = max(1, min(_auto_max_workers, 16))
3596 except Exception:
3597 logger.exception(
3598 "Failed to check auto-index setting, skipping auto-index"
3599 )
3600 return
3602 # Reserve a queue slot before submission. If the indexing queue is
3603 # saturated (too many uploads in flight), drop this auto-index job
3604 # rather than letting the executor's unbounded internal queue grow
3605 # without bound. The user can reindex manually later via the UI.
3606 if not _try_reserve_auto_index_slot():
3607 logger.warning(
3608 "Auto-index queue saturated ({}+ jobs pending); dropping "
3609 "auto-index for {} document(s) in collection {}. "
3610 "Documents are uploaded; trigger a manual reindex if needed.",
3611 _MAX_PENDING_AUTO_INDEX_JOBS,
3612 len(document_ids),
3613 collection_id,
3614 )
3615 return
3617 logger.info(
3618 f"Auto-indexing {len(document_ids)} documents in collection {collection_id}"
3619 )
3621 # Release the reserved slot EXACTLY ONCE per submission. CPython's
3622 # ThreadPoolExecutor.submit() enqueues the work item BEFORE it tries to
3623 # spin up a worker thread; if thread-start raises (e.g.
3624 # RuntimeError("can't start new thread")), the item is already queued, so
3625 # a live worker can run _wrapped_worker (releasing in its finally) AND the
3626 # except block below also fires. A per-call lock + flag makes the release
3627 # idempotent so that double-fire only releases once — otherwise the
3628 # counter under-counts in-flight jobs and erodes the OOM bound (the
3629 # max(0, ...) floor would silently hide the drift).
3630 _release_lock = threading.Lock()
3631 _slot_released = {"done": False}
3633 def _release_slot_once():
3634 with _release_lock:
3635 if _slot_released["done"]:
3636 return
3637 _slot_released["done"] = True
3638 _release_auto_index_slot()
3640 def _wrapped_worker(*args, **kwargs):
3641 try:
3642 _auto_index_documents_worker(*args, **kwargs)
3643 finally:
3644 _release_slot_once()
3646 # Submit to thread pool (bounded concurrency, prevents thread proliferation).
3647 # Build the executor inside the try too: if _get_auto_index_executor()
3648 # itself fails (OS thread/mutex exhaustion), the slot must still be
3649 # released — otherwise a reserved slot leaks permanently.
3650 try:
3651 executor = _get_auto_index_executor()
3652 executor.submit(
3653 _wrapped_worker,
3654 document_ids,
3655 collection_id,
3656 username,
3657 db_password,
3658 _auto_max_workers,
3659 )
3660 except Exception:
3661 # If submit itself fails (executor shutting down, OOM, etc.), the
3662 # wrapped worker may never run, so release the slot here. The release
3663 # is idempotent: if the work item was already enqueued and a worker
3664 # ran (and released) before thread-start failed, this is a no-op. Do
3665 # NOT re-raise: the upload has already been committed by the caller,
3666 # so propagating would turn a successful upload into a 500 (and prompt
3667 # the client to retry, creating duplicates). Auto-indexing is simply
3668 # skipped; the user can trigger a manual reindex later via the UI.
3669 _release_slot_once()
3670 logger.exception(
3671 "Failed to submit auto-index job for {} document(s) in "
3672 "collection {}; upload succeeded, auto-indexing skipped. "
3673 "Trigger a manual reindex if needed.",
3674 len(document_ids),
3675 collection_id,
3676 )
3679@thread_cleanup
3680def _auto_index_documents_worker(
3681 document_ids: list[str],
3682 collection_id: str,
3683 username: str,
3684 db_password: str,
3685 max_workers: int = 4,
3686) -> None:
3687 """
3688 Background worker to index documents automatically.
3690 This is a simpler worker than _background_index_worker - it doesn't track
3691 progress via TaskMetadata since it's meant to be a lightweight auto-indexing
3692 operation.
3694 Documents inside one call are indexed concurrently via
3695 :meth:`LibraryRAGService.index_documents_parallel`. The outer
3696 ``_auto_index_executor`` already parallelises separate jobs at the upload
3697 layer (one job per upload batch) and we additionally fan out inside each
3698 job, so a burst of uploads doesn't serialise within a single upload's
3699 document set.
3701 The ``max_workers`` parameter is resolved on the caller (which owns
3702 the Flask request context and therefore a real
3703 :func:`get_settings_manager` call). Background-thread workers must
3704 NOT call ``get_settings_manager()`` without an explicit
3705 ``db_session=`` — see pre-commit hook
3706 ``check-settings-manager-thread-safety`` and issue #3453.
3707 """
3709 try:
3710 # Create RAG service (thread-safe, no Flask context needed)
3711 with _get_rag_service_for_thread(
3712 collection_id, username, db_password
3713 ) as rag_service:
3714 # The auto-indexer doesn't surface per-doc titles to a user,
3715 # so we stub them with empty strings — the parallel helper
3716 # only reads titles for progress reporting.
3717 doc_info = [(doc_id, "") for doc_id in document_ids]
3718 aggregate = rag_service.index_documents_parallel(
3719 doc_info,
3720 collection_id,
3721 force_reindex=False,
3722 max_workers=max(1, min(int(max_workers), 16)),
3723 )
3724 logger.info(
3725 f"Auto-indexing complete: {aggregate['successful']}"
3726 f"/{len(document_ids)} documents indexed "
3727 f"(skipped={aggregate['skipped']}, "
3728 f"failed={aggregate['failed']})"
3729 )
3731 except Exception:
3732 logger.exception("Auto-indexing worker failed")
3735def _sanitized_indexing_errors(results: dict, limit: int = 50) -> list:
3736 """Return a sanitized, bounded list of per-document errors for the
3737 indexing task metadata. Used by both terminal paths of
3738 :func:`_background_index_worker` so the scrubbing logic is defined once.
3739 """
3740 return [
3741 {
3742 "doc_id": item.get("doc_id"),
3743 "title": item.get("title"),
3744 "error": sanitize_error_message(
3745 str(item.get("error") or "Indexing failed")
3746 ),
3747 }
3748 for item in results.get("errors", [])[:limit]
3749 ]
3752@thread_cleanup
3753def _background_index_worker(
3754 task_id: str,
3755 collection_id: str,
3756 username: str,
3757 db_password: str,
3758 force_reindex: bool,
3759 max_workers: int = 4,
3760):
3761 """
3762 Background worker thread for indexing documents.
3763 Updates TaskMetadata with progress and checks for cancellation.
3765 The ``max_workers`` parameter bounds the per-job fan-out into
3766 :meth:`LibraryRAGService.index_documents_parallel`. It must be
3767 resolved on the route (which has Flask app context), not inside
3768 this worker — background threads cannot call
3769 :func:`get_settings_manager` safely (#3453, pre-commit hook
3770 ``check-settings-manager-thread-safety``).
3771 """
3772 from ...database.session_context import get_user_db_session
3774 try:
3775 # Create RAG service (thread-safe, no Flask context needed)
3776 with _get_rag_service_for_thread(
3777 collection_id, username, db_password, use_defaults=force_reindex
3778 ) as rag_service:
3779 with get_user_db_session(username, db_password) as db_session:
3780 # Get collection
3781 collection = (
3782 db_session.query(Collection)
3783 .filter_by(id=collection_id)
3784 .first()
3785 )
3787 if not collection:
3788 _update_task_status(
3789 username,
3790 db_password,
3791 task_id,
3792 status="failed",
3793 error_message="Collection not found",
3794 )
3795 return
3797 # Store embedding metadata AND reset the old index in ONE commit,
3798 # so a crash between them can't leave the new config committed
3799 # while the old (stale/mixed-model) vectors survive.
3800 changed = False
3801 faiss_reset_paths = []
3802 if collection.embedding_model is None or force_reindex:
3803 _store_collection_embedding_metadata(
3804 collection, rag_service
3805 )
3806 changed = True
3807 if force_reindex:
3808 faiss_reset_paths = _reset_collection_for_reindex(
3809 db_session, collection_id
3810 )
3811 changed = True
3812 if changed:
3813 db_session.commit()
3814 _unlink_reindex_faiss_files(faiss_reset_paths)
3816 # Get documents to index
3817 doc_links = _query_documents_to_index(
3818 db_session, collection_id, force_reindex
3819 )
3821 if not doc_links:
3822 _update_task_status(
3823 username,
3824 db_password,
3825 task_id,
3826 status="completed",
3827 progress_message="No documents to index",
3828 )
3829 return
3831 total = len(doc_links)
3832 results = {"successful": 0, "skipped": 0, "failed": 0}
3834 # Update task with total count
3835 _update_task_status(
3836 username,
3837 db_password,
3838 task_id,
3839 progress_total=total,
3840 progress_message=f"Indexing {total} documents",
3841 )
3843 # The caller (start_background_index, which owns the
3844 # Flask request context) resolved ``max_workers`` from
3845 # ``rag.indexing_max_parallel_docs`` and forwarded it
3846 # here. Workers must NOT call ``get_settings_manager()``
3847 # themselves — no Flask app context, pre-commit
3848 # ``check-settings-manager-thread-safety`` would reject,
3849 # and #3453 documents the data-loss shape that produces.
3850 _max_workers = max(1, min(int(max_workers), 16))
3852 doc_info = [
3853 (doc.id, doc.filename or doc.title or "Unknown")
3854 for _link, doc in doc_links
3855 ]
3857 def _on_progress(
3858 completed: int,
3859 total: int,
3860 title: str,
3861 status: str,
3862 ) -> None:
3863 _update_task_status(
3864 username,
3865 db_password,
3866 task_id,
3867 progress_current=completed,
3868 progress_message=(
3869 f"Indexing {completed}/{total}: {title}"
3870 ),
3871 )
3873 def _is_cancelled() -> bool:
3874 return _is_task_cancelled(username, db_password, task_id)
3876 aggregate = rag_service.index_documents_parallel(
3877 doc_info,
3878 collection_id,
3879 force_reindex=force_reindex,
3880 max_workers=_max_workers,
3881 progress_callback=_on_progress,
3882 is_cancelled=_is_cancelled,
3883 )
3885 results["successful"] = aggregate["successful"]
3886 results["skipped"] = aggregate["skipped"]
3887 results["failed"] = aggregate["failed"]
3888 results["errors"] = aggregate["errors"]
3890 if aggregate["cancelled"]:
3891 completed = (
3892 aggregate["successful"]
3893 + aggregate["skipped"]
3894 + aggregate["failed"]
3895 )
3896 _update_task_status(
3897 username,
3898 db_password,
3899 task_id,
3900 status="cancelled",
3901 progress_message=(
3902 f"Cancelled after {completed}/{total} documents"
3903 ),
3904 )
3905 logger.info(
3906 f"Indexing task {task_id} was cancelled "
3907 f"after {completed}/{total} documents"
3908 )
3909 db_session.commit()
3910 return
3912 db_session.commit()
3914 # Wrap reconciliation so a transient store/DB error does not
3915 # leave the task in "processing" forever (the previous code
3916 # let exceptions escape, so _update_task_status never ran and
3917 # the UI polled indefinitely — see PR #5235 review comment
3918 # 5085604502). On failure we still record a terminal status
3919 # so cleanup_old_tasks can reap the row.
3920 try:
3921 reconciliation = rag_service.reconcile_collection_index(
3922 collection_id
3923 )
3924 except Exception as exc:
3925 logger.exception(
3926 f"Background indexing task {task_id}: reconciliation failed"
3927 )
3928 reconciliation = {
3929 "indexed_documents": results["successful"],
3930 "indexed_chunks": 0,
3931 "live_vectors": 0,
3932 "orphan_vectors": 0,
3933 "reconciliation_skipped": True,
3934 "reconciliation_reason": sanitize_error_message(str(exc)),
3935 }
3936 if not isinstance(reconciliation, dict):
3937 reconciliation = {
3938 "indexed_documents": results["successful"],
3939 "indexed_chunks": 0,
3940 "live_vectors": 0,
3941 "orphan_vectors": 0,
3942 }
3943 # The reconciler returns ``reconciliation_skipped=True`` when
3944 # live_ids is empty but chunk rows exist (refuse-to-shrink
3945 # guard). Surface this as a task failure so the user sees the
3946 # durable-state inconsistency instead of a misleading
3947 # "0 indexed" success.
3948 if isinstance(reconciliation, dict) and reconciliation.get(
3949 "reconciliation_skipped"
3950 ):
3951 reason = reconciliation.get("reconciliation_reason") or (
3952 reconciliation.get("reason")
3953 or "stored vectors could not be loaded"
3954 )
3955 sanitized_reason = sanitize_error_message(reason)
3956 logger.warning(
3957 f"Background indexing task {task_id}: reconciliation "
3958 f"skipped — {sanitized_reason}"
3959 )
3960 _update_task_status(
3961 username,
3962 db_password,
3963 task_id,
3964 status="failed",
3965 progress_current=total,
3966 progress_message=(
3967 f"Completed: {results['successful']} indexed, "
3968 f"{results['failed']} failed, "
3969 f"{results['skipped']} skipped; reconciliation "
3970 f"skipped: {sanitized_reason}"
3971 ),
3972 error_message=(
3973 f"Reconciliation skipped: {sanitized_reason}. "
3974 "Indexed flags preserved; verify the FAISS index "
3975 "before retrying."
3976 ),
3977 result_metadata={
3978 **{k: v for k, v in results.items() if k != "errors"},
3979 # Omit durable_indexed_documents/chunks and
3980 # live_vectors/orphan_vectors: true durable state is
3981 # unverified on the skipped path, and the UI
3982 # suppresses its "Durable vector store: ..." sentence
3983 # when these keys are absent (``Number.isInteger``
3984 # on undefined returns false). Reporting zeros here
3985 # would imply data loss where state is merely
3986 # unverified — see PR #5235 review comment
3987 # 5085604502.
3988 "reconciliation_skipped": True,
3989 "reconciliation_reason": sanitized_reason,
3990 "errors": _sanitized_indexing_errors(results),
3991 },
3992 )
3993 return
3994 durable_documents = reconciliation["indexed_documents"]
3995 durable_chunks = reconciliation["indexed_chunks"]
3996 task_result = {
3997 **results,
3998 "durable_indexed_documents": durable_documents,
3999 "durable_indexed_chunks": durable_chunks,
4000 "live_vectors": reconciliation["live_vectors"],
4001 "orphan_vectors": reconciliation["orphan_vectors"],
4002 "errors": _sanitized_indexing_errors(results),
4003 }
4004 has_failures = results["failed"] > 0
4005 status = "failed" if has_failures else "completed"
4006 message = (
4007 f"Completed: {results['successful']} indexed, "
4008 f"{results['failed']} failed, {results['skipped']} skipped; "
4009 f"durable vector store: {durable_documents} documents, "
4010 f"{durable_chunks} chunks"
4011 )
4012 _update_task_status(
4013 username,
4014 db_password,
4015 task_id,
4016 status=status,
4017 progress_current=total,
4018 progress_message=message,
4019 error_message=(
4020 f"{results['failed']} document(s) failed to index. "
4021 "The collection was reconciled to the durable vector store."
4022 if has_failures
4023 else None
4024 ),
4025 result_metadata=task_result,
4026 )
4027 logger.info(
4028 f"Background indexing task {task_id} completed: {results}"
4029 )
4031 except Exception as e:
4032 logger.exception(f"Background indexing task {task_id} failed")
4033 _update_task_status(
4034 username,
4035 db_password,
4036 task_id,
4037 status="failed",
4038 # Scrub creds/keys: this is later returned to the client by the
4039 # index-status endpoint; full trace stays in server logs above.
4040 error_message=sanitize_error_message(str(e)),
4041 )
4044def _is_database_locked_error(exc: BaseException) -> bool:
4045 """Return True if ``exc`` looks like a SQLite ``database is locked`` error.
4047 Background indexing can hit transient ``database is locked`` errors when
4048 SQLite is contended; those are recoverable via a short backoff retry.
4049 Anything else (connection failure, integrity error, programming mistake)
4050 should surface immediately so the outer exception handler marks the task
4051 as failed and sets ``completed_at`` for reaping.
4052 """
4053 return "database is locked" in str(exc).lower()
4056@retry(
4057 stop=stop_after_attempt(5),
4058 wait=wait_exponential(multiplier=0.05),
4059 retry=retry_if_exception(_is_database_locked_error),
4060)
4061def _do_update_task_status(
4062 username: str,
4063 db_password: str,
4064 task_id: str,
4065 status: str = None,
4066 progress_current: int = None,
4067 progress_total: int = None,
4068 progress_message: str = None,
4069 error_message: str = None,
4070 result_metadata: dict = None,
4071):
4072 """Perform the task-status update under a tenacity-managed retry loop.
4074 Only SQLite ``database is locked`` errors are retried (up to 5 attempts
4075 with 0.05s exponential backoff); any other exception escapes so the
4076 outer caller can log it without retrying. See PR #5235 review comment
4077 3669857779.
4078 """
4079 from ...database.session_context import get_user_db_session
4081 with get_user_db_session(username, db_password) as db_session:
4082 task = db_session.query(TaskMetadata).filter_by(task_id=task_id).first()
4083 if task:
4084 # Preserve terminal states: once cancelled or failed, only
4085 # updates confirming that same state are allowed.
4086 if task.status in ("cancelled", "failed") and status != task.status:
4087 return
4088 if status == "completed" and task.status != "processing": 4088 ↛ 4089line 4088 didn't jump to line 4089 because the condition on line 4088 was never true
4089 return
4090 if status is not None:
4091 task.status = status
4092 # Set completed_at for ALL terminal statuses so
4093 # ``cleanup_old_tasks`` (which filters on
4094 # ``status in ["completed", "failed", "cancelled"] AND
4095 # completed_at < cutoff_date``) can reap them.
4096 # Previously only "completed" set the timestamp,
4097 # leaving failed/cancelled rows permanent.
4098 if status in ("completed", "failed", "cancelled"): 4098 ↛ 4100line 4098 didn't jump to line 4100 because the condition on line 4098 was always true
4099 task.completed_at = datetime.now(UTC)
4100 if progress_current is not None:
4101 task.progress_current = progress_current
4102 if progress_total is not None:
4103 task.progress_total = progress_total
4104 if progress_message is not None:
4105 task.progress_message = progress_message
4106 if error_message is not None:
4107 task.error_message = error_message
4108 if result_metadata is not None:
4109 metadata = dict(task.metadata_json or {})
4110 metadata["result"] = result_metadata
4111 task.metadata_json = metadata
4112 db_session.commit()
4115def _update_task_status(
4116 username: str,
4117 db_password: str,
4118 task_id: str,
4119 status: str = None,
4120 progress_current: int = None,
4121 progress_total: int = None,
4122 progress_message: str = None,
4123 error_message: str = None,
4124 result_metadata: dict = None,
4125):
4126 """Update task metadata in the database."""
4127 try:
4128 _do_update_task_status(
4129 username,
4130 db_password,
4131 task_id,
4132 status=status,
4133 progress_current=progress_current,
4134 progress_total=progress_total,
4135 progress_message=progress_message,
4136 error_message=error_message,
4137 result_metadata=result_metadata,
4138 )
4139 except Exception:
4140 logger.exception(f"Failed to update task status for {task_id}")
4143def _is_task_cancelled(username: str, db_password: str, task_id: str) -> bool:
4144 """Check if a task has been cancelled."""
4145 from ...database.session_context import get_user_db_session
4147 try:
4148 with get_user_db_session(username, db_password) as db_session:
4149 task = (
4150 db_session.query(TaskMetadata)
4151 .filter_by(task_id=task_id)
4152 .first()
4153 )
4154 return task and task.status == "cancelled"
4155 except Exception:
4156 logger.warning(
4157 "Could not check cancellation status for task {}", task_id
4158 )
4159 return False
4162@router.post("/api/collections/{collection_id}/index/start")
4163async def start_background_index(
4164 request: Request,
4165 collection_id,
4166 username: Annotated[str, Depends(require_auth)],
4167):
4168 """Start background indexing for a collection."""
4169 from ...database.session_passwords import session_password_store
4171 session_id = request.session.get("session_id")
4173 # Get password for thread access
4174 db_password = None
4175 if session_id:
4176 db_password = session_password_store.get_session_password(
4177 username, session_id
4178 )
4180 # Parse request body. `await request.json()` raising on malformed bytes
4181 # is left to propagate to the app's registered json.JSONDecodeError ->
4182 # 400 handler (no local except here to shadow it). A truthy non-dict
4183 # body (bare string/int/list) is guarded explicitly below — `or {}`
4184 # alone only catches FALSY bodies (null, [], ""), letting a truthy
4185 # non-dict reach `.get()` and raise AttributeError -> 500.
4186 data = await request.json()
4187 if not isinstance(data, dict):
4188 return json_body_error("success", "Request body must be valid JSON")
4189 force_reindex = data.get("force_reindex", False)
4190 if not isinstance(force_reindex, bool):
4191 return JSONResponse(
4192 {
4193 "success": False,
4194 "error": "force_reindex must be a boolean",
4195 },
4196 status_code=400,
4197 )
4199 # The lock acquisition and SQLCipher session open below are blocking —
4200 # run the whole check-and-create + thread spawn in the threadpool so a
4201 # contended lock or a cold-engine PBKDF2 derivation can't stall the
4202 # event loop (contextvars propagate through to_thread, so the
4203 # copy_context() for the worker still sees this request's user).
4204 return await run_db_sync(
4205 _start_background_index_sync,
4206 collection_id,
4207 username,
4208 db_password,
4209 force_reindex,
4210 )
4213def _start_background_index_sync(
4214 collection_id, username, db_password, force_reindex
4215):
4216 from ...database.session_context import get_user_db_session
4218 # Serialize check-and-create per (user, collection) to close the TOCTOU
4219 # window between scanning for an in-progress task and inserting a new
4220 # row. Two concurrent threads would otherwise both see no match and
4221 # both spawn an indexer on the same collection, racing FAISS writes.
4222 # Single-worker scope only (matches the rest of this module's locking
4223 # model; multi-worker deployments already documented as unsupported
4224 # for this kind of coordination).
4225 lock_key = (username, str(collection_id))
4226 lock = _start_bg_index_locks.setdefault(lock_key, threading.Lock())
4228 try:
4229 with lock, get_user_db_session(username, db_password) as db_session:
4230 # Scan ALL in-progress indexing tasks for one that matches this
4231 # collection_id. The old query did `.first()` and then checked
4232 # — so it missed collision when another collection's task
4233 # happened to sort first, and also raced when two requests for
4234 # the same collection arrived concurrently. Scanning all rows
4235 # is fine: `TaskMetadata.status == "processing"` is bounded by
4236 # the number of in-flight indexes (small).
4237 in_progress = (
4238 db_session.query(TaskMetadata)
4239 .filter(
4240 TaskMetadata.task_type == "indexing",
4241 TaskMetadata.status == "processing",
4242 )
4243 .all()
4244 )
4246 for existing_task in in_progress:
4247 metadata = existing_task.metadata_json or {}
4248 if metadata.get("collection_id") == collection_id:
4249 return JSONResponse(
4250 {
4251 "success": False,
4252 "error": "Indexing is already in progress for this collection",
4253 "task_id": existing_task.task_id,
4254 },
4255 status_code=409,
4256 )
4258 # Create new task
4259 task_id = str(uuid.uuid4())
4260 task = TaskMetadata(
4261 task_id=task_id,
4262 status="processing",
4263 task_type="indexing",
4264 created_at=datetime.now(UTC),
4265 started_at=datetime.now(UTC),
4266 progress_current=0,
4267 progress_total=0,
4268 progress_message="Starting indexing...",
4269 metadata_json={
4270 "collection_id": collection_id,
4271 "force_reindex": force_reindex,
4272 },
4273 )
4274 db_session.add(task)
4275 db_session.commit()
4277 # Resolve per-job parallel fan-out here (request-scoped
4278 # session) and forward into the worker — the worker itself
4279 # runs on a background thread where the no-arg
4280 # ``get_settings_manager()`` is unsafe (#3453).
4281 try:
4282 _bg_settings = get_settings_manager(db_session, username)
4283 _bg_max_workers = int(
4284 _bg_settings.get_setting(
4285 "rag.indexing_max_parallel_docs", 4
4286 )
4287 )
4288 except Exception:
4289 _bg_max_workers = 4
4290 _bg_max_workers = max(1, min(_bg_max_workers, 16))
4292 # Start background thread. Copy the request's contextvars so
4293 # any service code inside the worker that calls
4294 # `get_current_username()` sees the actual user instead of None.
4295 _bg_ctx = contextvars.copy_context()
4297 def _ctx_worker():
4298 _bg_ctx.run(
4299 _background_index_worker,
4300 task_id,
4301 collection_id,
4302 username,
4303 db_password,
4304 force_reindex,
4305 _bg_max_workers,
4306 )
4308 thread = threading.Thread(target=_ctx_worker, daemon=True)
4309 thread.start()
4311 logger.info(
4312 f"Started background indexing task {task_id} for collection {collection_id}"
4313 )
4315 return {
4316 "success": True,
4317 "task_id": task_id,
4318 "message": "Indexing started in background",
4319 }
4321 except Exception:
4322 logger.exception("Failed to start background indexing")
4323 return JSONResponse(
4324 {
4325 "success": False,
4326 "error": "Failed to start indexing. Please try again.",
4327 },
4328 status_code=500,
4329 )
4332@router.get("/api/collections/{collection_id}/index/status")
4333@limiter.exempt
4334def get_index_status(
4335 request: Request,
4336 collection_id,
4337 username: Annotated[str, Depends(require_auth)],
4338):
4339 """Get the current indexing status for a collection."""
4340 from ...database.session_context import get_user_db_session
4341 from ...database.session_passwords import session_password_store
4343 session_id = request.session.get("session_id")
4345 db_password = None
4346 if session_id:
4347 db_password = session_password_store.get_session_password(
4348 username, session_id
4349 )
4351 try:
4352 with get_user_db_session(username, db_password) as db_session:
4353 # Find the most recent indexing task FOR THIS COLLECTION.
4354 #
4355 # collection_id is stored inside the metadata_json JSON column, so
4356 # we can't portably filter on it in SQL (SQLite/SQLCipher JSON
4357 # support varies). Instead, scan recent indexing tasks newest-first
4358 # and return the first whose metadata.collection_id matches. This
4359 # is scoped per collection: a newer indexing task for a DIFFERENT
4360 # collection no longer makes this one falsely report "idle".
4361 #
4362 # This endpoint is polled every ~2s during a reindex and the
4363 # indexing-task table is never pruned, so an unbounded .all() would
4364 # materialize the entire history on every poll (and task_type /
4365 # created_at are unindexed). Bound the scan to the newest N tasks.
4366 # The task we're looking for was just created by the reindex that
4367 # started this poll, so it sits near the top; N=200 is large enough
4368 # that concurrent reindexes of other collections can't push it out
4369 # of the window before it terminates. Trade-off: if more than ~200
4370 # newer indexing tasks for OTHER collections appear while this one
4371 # is still in-flight, the scoped lookup falls back to "idle".
4372 recent_task_scan_limit = 200
4373 recent_tasks = (
4374 db_session.query(TaskMetadata)
4375 .filter(TaskMetadata.task_type == "indexing")
4376 .order_by(TaskMetadata.created_at.desc())
4377 .limit(recent_task_scan_limit)
4378 .all()
4379 )
4381 task = None
4382 for candidate in recent_tasks:
4383 metadata = candidate.metadata_json or {}
4384 if metadata.get("collection_id") == collection_id:
4385 task = candidate
4386 break
4388 if not task:
4389 return {
4390 "status": "idle",
4391 "collection_id": collection_id,
4392 "message": "No indexing task for this collection",
4393 }
4395 return {
4396 "task_id": task.task_id,
4397 "collection_id": collection_id,
4398 "status": task.status,
4399 "progress_current": task.progress_current or 0,
4400 "progress_total": task.progress_total or 0,
4401 "progress_message": task.progress_message,
4402 "error_message": task.error_message,
4403 "result": (task.metadata_json or {}).get("result"),
4404 "created_at": task.created_at.isoformat()
4405 if task.created_at
4406 else None,
4407 "completed_at": task.completed_at.isoformat()
4408 if task.completed_at
4409 else None,
4410 }
4412 except Exception:
4413 logger.exception("Failed to get index status")
4414 return JSONResponse(
4415 {
4416 "status": "error",
4417 "error": "Failed to get indexing status. Please try again.",
4418 },
4419 status_code=500,
4420 )
4423@router.post("/api/collections/{collection_id}/index/cancel")
4424def cancel_indexing(
4425 request: Request,
4426 collection_id,
4427 username: Annotated[str, Depends(require_auth)],
4428):
4429 """Cancel an active indexing task for a collection."""
4430 from ...database.session_context import get_user_db_session
4431 from ...database.session_passwords import session_password_store
4433 session_id = request.session.get("session_id")
4435 db_password = None
4436 if session_id:
4437 db_password = session_password_store.get_session_password(
4438 username, session_id
4439 )
4441 try:
4442 # Signal active process-local SSE generator(s) if present
4443 sse_cancelled = False
4444 with _active_sse_indexers_lock:
4445 sse_events = _active_sse_indexers.get((username, collection_id))
4446 if sse_events:
4447 for sse_event in sse_events:
4448 sse_event.set()
4449 sse_cancelled = True
4450 logger.info(
4451 f"Signalled active SSE indexing event(s) for collection {collection_id}"
4452 )
4454 with get_user_db_session(username, db_password) as db_session:
4455 # Find active indexing task for this collection
4456 task = (
4457 db_session.query(TaskMetadata)
4458 .filter(
4459 TaskMetadata.task_type == "indexing",
4460 TaskMetadata.status == "processing",
4461 )
4462 .first()
4463 )
4465 matched_task = None
4466 if task:
4467 metadata = task.metadata_json or {}
4468 if metadata.get("collection_id") == collection_id:
4469 matched_task = task
4471 if not matched_task and not sse_cancelled:
4472 if task:
4473 return JSONResponse(
4474 {
4475 "success": False,
4476 "error": "No active indexing task for this collection",
4477 },
4478 status_code=404,
4479 )
4480 return JSONResponse(
4481 {
4482 "success": False,
4483 "error": "No active indexing task found",
4484 },
4485 status_code=404,
4486 )
4488 if matched_task:
4489 # This endpoint must surface a failed cancellation write.
4490 # Background workers use the best-effort wrapper, but returning
4491 # success here after it swallowed an error would mislead the
4492 # caller while the task remains active.
4493 task_id_to_cancel = matched_task.task_id
4494 _do_update_task_status(
4495 username,
4496 db_password,
4497 task_id_to_cancel,
4498 status="cancelled",
4499 progress_message="Cancellation requested...",
4500 )
4502 logger.info(
4503 f"Cancelled indexing task {task_id_to_cancel} for collection {collection_id}"
4504 )
4506 return {
4507 "success": True,
4508 "message": "Cancellation requested",
4509 "task_id": matched_task.task_id if matched_task else None,
4510 }
4512 except Exception:
4513 logger.exception("Failed to cancel indexing")
4514 return JSONResponse(
4515 {
4516 "success": False,
4517 "error": "Failed to cancel indexing. Please try again.",
4518 },
4519 status_code=500,
4520 )
4523# Research History Semantic Search Routes have been moved to
4524# web/routers/library_search.py