Coverage for src/local_deep_research/web/routers/rag.py: 96%
1468 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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 Future, 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_message,
67)
68from ...utilities.db_utils import get_settings_manager
69from ...utilities.resource_utils import safe_close
70from ...research_library.services.library_rag_service import LibraryRAGService
71from ...settings.exceptions import SettingsValueError
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 ..dependencies.json_body import json_body_error, read_json_dict
91from .notes import _log_value_preview
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_EMBEDDING_TEST_SENTENCE = (
108 "This is a test sentence to verify the embedding model is working "
109 "correctly."
110)
112# Process-local registry tracking active SSE indexing streams for cancellation.
113# Keyed by (username, collection_id) -> Set[threading.Event]
114# Process-local cancellation and admission require a single worker process.
115# Multiple workers need shared admission and cancellation coordination;
116# sticky sessions alone do not exclude two devices indexing the same collection.
117_active_sse_indexers: Dict[Tuple[str, str], Set[threading.Event]] = {}
118_active_sse_indexers_lock = threading.Lock()
119# A cancelled task can still have workers draining. Keep admission closed
120# until its worker exits, independently of the task's persisted status.
121_active_background_indexers: Dict[Tuple[str, str], str] = {}
123# NOTE: Routes use username (not .get()) intentionally.
124# Depends(require_auth) guarantees the key exists; direct access fails
125# fast if the dependency is ever removed.
128def _agent_enabled_default_on(collection) -> bool:
129 """Read a collection's ``agent_enabled`` flag with NULL → available.
131 A missing attribute (pre-migration object) or a stored NULL both mean
132 "available to the research agent" (default-on); only an explicit ``False``
133 disables. Used by every serializer so the four read sites can't drift on
134 null-handling. ``is not False`` is safe here because the source is a
135 Boolean column (only ever None/True/False) — NOT raw JSON input, where a
136 falsy ``0`` must still mean disabled (handled separately at the input edge).
137 """
138 return getattr(collection, "agent_enabled", True) is not False
141def _is_protected_collection(collection) -> bool:
142 """True when the collection's type is deletion-protected.
144 Single source of truth is PROTECTED_COLLECTION_TYPES in the deletion
145 service (imported lazily, matching update_collection's usage); the
146 serializers expose the result as ``is_protected`` so UI surfaces can
147 hide destructive affordances instead of offering an action the server
148 categorically refuses with a 409.
149 """
150 from ...research_library.deletion.services.collection_deletion import (
151 PROTECTED_COLLECTION_TYPES,
152 )
154 return collection.collection_type in PROTECTED_COLLECTION_TYPES
157# Global ThreadPoolExecutor for auto-indexing to prevent thread proliferation.
158#
159# ThreadPoolExecutor's internal _work_queue is unbounded. Without
160# backpressure, a sustained burst of uploads (possible under the
161# configurable upload rate cap from #3935) could queue thousands of indexing
162# jobs in memory, exhausting RAM. We track pending submissions in a counter
163# and reject new submissions once the queue is saturated.
164_auto_index_executor: ThreadPoolExecutor | None = None
165_auto_index_executor_lock = threading.Lock()
167# Per-(user, collection) locks guarding admission for all collection indexers.
168# Keyed by tuple so two users indexing different
169# collections don't serialize against each other.
170_start_bg_index_locks: dict[tuple[str, str], threading.Lock] = {}
172# Maximum in-flight + queued indexing jobs. Each job is one upload batch
173# (an unbounded list of document IDs), so per-job duration varies with batch
174# size; 100 is a buffer on the number of *queued batches*, not documents.
175# The OOM bound still holds because a queued job holds only a small list of
176# ID strings, not document contents. Tunable if real-world workloads need
177# more headroom.
178_MAX_PENDING_AUTO_INDEX_JOBS = 100
179_pending_auto_index_jobs = 0
180_pending_auto_index_lock = threading.Lock()
183def _get_auto_index_executor() -> ThreadPoolExecutor:
184 """Get or create the global auto-indexing executor (thread-safe)."""
185 global _auto_index_executor
186 with _auto_index_executor_lock:
187 if _auto_index_executor is None:
188 _auto_index_executor = ThreadPoolExecutor(
189 max_workers=4,
190 thread_name_prefix="auto_index_",
191 )
192 return _auto_index_executor
195def _try_reserve_auto_index_slot() -> bool:
196 """Reserve a queue slot if capacity allows. Returns True on success.
198 Caller MUST call ``_release_auto_index_slot`` when the job completes
199 (success OR failure), otherwise the counter leaks and eventually
200 blocks all submissions.
201 """
202 global _pending_auto_index_jobs
203 with _pending_auto_index_lock:
204 if _pending_auto_index_jobs >= _MAX_PENDING_AUTO_INDEX_JOBS:
205 return False
206 _pending_auto_index_jobs += 1
207 return True
210def _release_auto_index_slot() -> None:
211 """Release a previously reserved queue slot."""
212 global _pending_auto_index_jobs
213 with _pending_auto_index_lock:
214 _pending_auto_index_jobs = max(0, _pending_auto_index_jobs - 1)
217def _shutdown_auto_index_executor() -> None:
218 """Shutdown the auto-index executor gracefully.
220 Holds ``_auto_index_executor_lock`` for the read+nullify so two
221 concurrent callers (e.g. atexit + a test teardown) can't race into an
222 ``AttributeError`` on the ``None`` reference; the blocking ``shutdown``
223 runs outside the lock. Resets the pending-jobs counter afterwards so a
224 re-created executor (tests, a dev-server reload) starts from a clean count
225 instead of inheriting a stale, possibly-saturated value.
226 """
227 global _auto_index_executor, _pending_auto_index_jobs
228 with _auto_index_executor_lock:
229 executor = _auto_index_executor
230 _auto_index_executor = None
231 if executor is not None:
232 executor.shutdown(wait=True)
233 with _pending_auto_index_lock:
234 _pending_auto_index_jobs = 0
237atexit.register(_shutdown_auto_index_executor)
240def get_rag_service(
241 request: Request,
242 username: str,
243 collection_id: Optional[str] = None,
244 use_defaults: bool = False,
245) -> LibraryRAGService:
246 """
247 Get RAG service instance with appropriate settings.
249 Args:
250 request: The FastAPI request (for session lookup).
251 username: The authenticated username.
252 collection_id: Optional collection UUID to load stored settings from
253 use_defaults: When True, ignore stored collection settings and use
254 current defaults. Pass True on force-reindex so that the new
255 default embedding model is picked up.
256 """
257 from ...research_library.services.rag_service_factory import (
258 get_rag_service as _get_rag_service,
259 )
260 from ...database.session_passwords import session_password_store
262 session_id = request.session.get("session_id")
263 db_password = None
264 if session_id:
265 db_password = session_password_store.get_session_password(
266 username, session_id
267 )
268 if not db_password:
269 db_password = session_password_store.get_any_session_password(username)
270 return _get_rag_service(
271 username,
272 collection_id,
273 use_defaults=use_defaults,
274 db_password=db_password,
275 )
278def _get_text_separators(settings):
279 """Return configured text separators, parsing string values if needed."""
280 # Load the stored setting. It may already be a list or a JSON string.
281 text_separators = settings.get_setting(
282 "local_search_text_separators",
283 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
284 )
286 # If the setting is stored as a string, parse it into a list. A value that
287 # is not valid JSON (e.g. a not-yet-migrated corrupt row) falls back to the
288 # default separators — migration #4298 heals existing corrupt data.
289 if isinstance(text_separators, str):
290 try:
291 text_separators = json.loads(text_separators)
292 except json.JSONDecodeError:
293 logger.warning(
294 "Invalid JSON for local_search_text_separators setting: {!r}. Using default separators.",
295 text_separators,
296 )
297 text_separators = DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
299 return text_separators
302def _parse_configured_text_separators(value):
303 """Coerce a text-separator setting to a list of strings, or None.
305 Called via ``asyncio.to_thread`` from ``configure_rag``: ``value``
306 comes straight from the request body, so a caller can hand it a
307 string up to that route's 16 MB body cap. ``json.loads`` runs at
308 ~110 ms/MB, which is ~1.7 s at the cap -- fine on a worker thread
309 (Flask charged the same work to a request thread), a whole-instance
310 freeze on the event loop under single-worker uvicorn.
311 """
312 separators = value
313 if isinstance(value, str):
314 try:
315 separators = json.loads(value)
316 except json.JSONDecodeError:
317 return None
318 if not isinstance(separators, list) or not all(
319 isinstance(separator, str) for separator in separators
320 ):
321 return None
322 return separators
325# Module prefix for exceptions DEFINED inside LDR itself. An exception whose
326# class lives under ``local_deep_research`` is something only we can fix, so we
327# steer the user to file a bug instead of second-guessing their model choice.
328#
329# We deliberately do NOT treat bare builtins (KeyError/TypeError/RuntimeError/
330# ...) as internal: ``type(exc).__module__`` is where the class is *defined*,
331# not where it was *raised*, so a builtin tells us nothing about whether LDR or
332# a provider/langchain frame produced it. On this path an OpenAI-compatible
333# server that returns a malformed 200 body can surface a bare TypeError from
334# langchain's response parser; telling the user to "report it on GitHub" for
335# that is exactly the misleading guidance #4208 set out to remove.
336_INTERNAL_MODULE_PREFIX = "local_deep_research"
338# Module prefixes whose exceptions are upstream-provider errors (network,
339# auth, model not found, etc.). Matched against ``type(exc).__module__``.
340_UPSTREAM_MODULE_PREFIXES = (
341 "openai",
342 "httpx",
343 "requests",
344 "urllib3",
345 "anthropic",
346)
348# Upstream exception classes that mean "the request never reached the
349# provider, or never got an answer" as opposed to "the provider answered and
350# refused". Told apart by class NAME across the MRO so no provider SDK has to
351# be imported here, and matched only inside the upstream branch below — a
352# builtins ``ConnectionError`` is still default-denied. The name is used to
353# pick between two fixed messages; it is not part of either one.
354_CONNECTION_FAILURE_CLASS_NAMES = frozenset(
355 {
356 "APIConnectionError", # openai (APITimeoutError subclasses it)
357 "ConnectError", # httpx
358 "ConnectionError", # requests.exceptions
359 "ConnectTimeout", # httpx
360 "ConnectTimeoutError", # urllib3
361 "NewConnectionError", # urllib3
362 "ProxyError", # httpx / requests / urllib3
363 }
364)
367def _module_matches(type_module: str, prefix: str) -> bool:
368 """True if ``type_module`` is ``prefix`` itself or one of its submodules."""
369 return type_module == prefix or type_module.startswith(f"{prefix}.")
372def _format_test_embedding_error(exc: Exception, model: str) -> str:
373 """Build the user-facing error message for /api/rag/test-embedding.
375 Categorizes the exception by the module its *class* is defined in so the
376 UI distinguishes LDR-internal bugs (which the user can't fix and
377 shouldn't be steered into changing their model over) from real
378 upstream/provider errors. The previous implementation ran a keyword-match
379 heuristic over every exception text — see #4208, where
380 ``NoSettingsContextError`` surfaced as a misleading "try a dedicated
381 embedding model" suggestion.
383 Exception text never crosses the HTTP boundary here. The exception's
384 module and class are used only to SELECT one of the fixed messages below;
385 neither is interpolated into the result, and the full detail stays in the
386 server-side log (``logger.exception`` at the call site).
388 Why not simply scrub the text and echo it, which is what
389 :func:`~...security.log_sanitizer.sanitize_error_for_client` is for
390 elsewhere in the tree: that function removes credential *shapes* plus
391 control characters, and caps the length. It does not remove server
392 filesystem paths, SQL text or dependency internals — and those are
393 exactly what the exceptions reachable from this ``try`` block carry. The
394 block opens the user's SQLCipher database and constructs an embedding
395 manager, so a ``sqlalchemy.exc.OperationalError`` (driver text plus
396 ``[SQL: ...]`` and ``[parameters: ...]``), a ``builtins.OSError``
397 (absolute server path and OS account name) or a provider response body
398 are all in range. ``sanitize_error_for_client`` remains the right tool
399 for a message an author already knows to be safe; it is not a filter that
400 makes an arbitrary exception safe, which is why this function does not
401 use it.
402 """
403 type_module = type(exc).__module__ or ""
405 if _module_matches(type_module, _INTERNAL_MODULE_PREFIX):
406 # Internal LDR exceptions can carry filesystem paths / SQL fragments
407 # that sanitize_error_message() does not pattern-match, so we do not
408 # echo their detail to the browser — the full trace is in the server
409 # logs (logger.exception at the call site) which the user is asked to
410 # attach.
411 return (
412 f"Embedding test failed for model '{model}' due to an "
413 "internal LDR error. This is a bug in LDR, "
414 "not your configuration. Please report it on GitHub with "
415 "the server logs."
416 )
418 if any(
419 _module_matches(type_module, prefix)
420 for prefix in _UPSTREAM_MODULE_PREFIXES
421 ):
422 if _CONNECTION_FAILURE_CLASS_NAMES.intersection(
423 klass.__name__ for klass in type(exc).__mro__
424 ):
425 # Nothing was rejected — the connection never completed. Saying
426 # "the provider rejected the request" here sends the user to
427 # check credentials for what is almost always a wrong host, a
428 # stopped Ollama, or a firewall.
429 return (
430 f"Embedding test failed for model '{model}'. Could not "
431 "connect to the provider; check the URL and that it is "
432 "reachable. The full error is in the server logs."
433 )
434 return (
435 f"Embedding test failed for model '{model}'. The request to the "
436 "provider failed. Check its URL, credentials, and model name; "
437 "the full error is in the server logs."
438 )
440 # Default-deny. Everything outside the upstream allowlist reaches here:
441 # sqlalchemy.exc.* (the try block opens the user's SQLCipher database, and
442 # a DBAPIError's str() renders the driver message plus [SQL: ...] and
443 # [parameters: ...]), builtins.OSError/PermissionError/FileNotFoundError
444 # (absolute server paths and the OS account name), chromadb,
445 # sentence_transformers, langchain_core, botocore... None of that is
446 # touched by the credential-shape scrubber, so no exception-derived detail
447 # is returned. The full text is in the server logs (logger.exception at
448 # the call site).
449 #
450 # Worded deliberately unlike the internal-LDR branch above: a bare
451 # TypeError from langchain's response parser is not an LDR bug, and
452 # steering the user to "report it on GitHub" for it is the misleading
453 # guidance #4208 removed.
454 return (
455 f"Embedding test failed for model '{model}' due to an unexpected "
456 "error. Check the provider URL and model name; the full error is "
457 "in the server logs."
458 )
461# Config API Routes
464@router.get("/api/config/supported-formats")
465def get_supported_formats(
466 request: Request, username: Annotated[str, Depends(require_auth)]
467):
468 """Return list of supported file formats for upload.
470 This endpoint provides the single source of truth for supported file
471 extensions, pulling from the document_loaders registry. The UI can
472 use this to dynamically update the file input accept attribute.
473 """
474 from ...document_loaders import get_supported_extensions
476 extensions = get_supported_extensions()
477 # Sort extensions for consistent display
478 extensions = sorted(extensions)
480 return {
481 "extensions": extensions,
482 "accept_string": ",".join(extensions),
483 "count": len(extensions),
484 }
487# Page Routes
490@router.get("/embedding-settings")
491def embedding_settings_page(
492 request: Request, username: Annotated[str, Depends(require_auth)]
493):
494 """Render the Embedding Settings page."""
495 return templates.TemplateResponse(
496 request=request,
497 name="pages/embedding_settings.html",
498 context={"request": request, "active_page": "embedding-settings"},
499 )
502@router.get("/document/{document_id}/chunks")
503def view_document_chunks(
504 request: Request,
505 document_id,
506 username: Annotated[str, Depends(require_auth)],
507):
508 """View all chunks for a document across all collections."""
509 from ..utils.path_ids import validated_uuid_path_param
510 from ...database.session_context import get_user_db_session
511 from ...database.models.library import DocumentChunk
513 document_id = validated_uuid_path_param(document_id, "Document")
515 with get_user_db_session(username) as db_session:
516 # Get document info
517 document = db_session.query(Document).filter_by(id=document_id).first()
519 if not document:
520 # Browser navigation: return text/html, not JSON. These four page
521 # routes are reached from library.html as ordinary <a href> links, so a
522 # stale link showed the user a raw {"error": ...} body in the browser's
523 # JSON viewer. main returned `"Document not found", 404` (text/html)
524 # here; the sibling /api/ routes keep JSON.
525 return HTMLResponse("Document not found", status_code=404)
527 # Get all chunks for this document
528 chunks = (
529 db_session.query(DocumentChunk)
530 .filter(DocumentChunk.source_id == document_id)
531 .order_by(DocumentChunk.collection_name, DocumentChunk.chunk_index)
532 .all()
533 )
535 # Group chunks by collection
536 chunks_by_collection = {}
537 for chunk in chunks:
538 coll_name = chunk.collection_name
539 if coll_name not in chunks_by_collection: 539 ↛ 553line 539 didn't jump to line 553 because the condition on line 539 was always true
540 # Get collection display name
541 collection_id = coll_name.replace("collection_", "")
542 collection = (
543 db_session.query(Collection)
544 .filter_by(id=collection_id)
545 .first()
546 )
547 chunks_by_collection[coll_name] = {
548 "name": collection.name if collection else coll_name,
549 "id": collection_id,
550 "chunks": [],
551 }
553 chunks_by_collection[coll_name]["chunks"].append(
554 {
555 "id": chunk.id,
556 "index": chunk.chunk_index,
557 "text": chunk.chunk_text,
558 "word_count": chunk.word_count,
559 "start_char": chunk.start_char,
560 "end_char": chunk.end_char,
561 "embedding_model": chunk.embedding_model,
562 "embedding_model_type": chunk.embedding_model_type.value
563 if chunk.embedding_model_type
564 else None,
565 "embedding_dimension": chunk.embedding_dimension,
566 "created_at": chunk.created_at,
567 }
568 )
570 return templates.TemplateResponse(
571 request=request,
572 name="pages/document_chunks.html",
573 context={
574 "request": request,
575 "document": document,
576 "chunks_by_collection": chunks_by_collection,
577 "total_chunks": len(chunks),
578 },
579 )
582@router.get("/collections")
583def collections_page(
584 request: Request, username: Annotated[str, Depends(require_auth)]
585):
586 """Render the Collections page."""
587 return templates.TemplateResponse(
588 request=request,
589 name="pages/collections.html",
590 context={"request": request, "active_page": "collections"},
591 )
594# NOTE: this static route MUST be registered before the parameterized
595# /collections/{collection_id} below — FastAPI matches in registration
596# order, so the param route would otherwise swallow /collections/create
597# as collection_id="create" and render the details page instead of the
598# create form (fenced by tests/web/routers/test_route_ordering.py).
599@router.get("/collections/create")
600def collection_create_page(
601 request: Request, username: Annotated[str, Depends(require_auth)]
602):
603 """Render the Create Collection page."""
604 return templates.TemplateResponse(
605 request=request,
606 name="pages/collection_create.html",
607 context={"request": request, "active_page": "collections"},
608 )
611def _validated_collection_id(collection_id: str) -> str:
612 """Return ``collection_id``'s canonical UUID form if well-formed, else 404.
614 Collection ids are server-generated UUID4 strings (`Collection.id` is
615 String(36), assigned from `str(uuid.uuid4())`), so anything else is either
616 a typo or an injection attempt and cannot match a real row.
618 This was the first fix for a stored-reflection XSS: collection_upload.html
619 interpolated this value into an inline `onclick` handler, and
620 HTML-escaping is NOT sufficient there -- the browser entity-decodes
621 attribute content before treating it as JS, so an escaped quote reopens
622 the string. That template now passes the ID as a data attribute read by a
623 script listener instead. This check stays as the second layer: validating
624 the shape at the boundary means no template can receive a value capable of
625 breaking out, regardless of the context it is interpolated into, which
626 also covers collection_details.html and anything added later.
628 (`/collections/create` is a separate static route registered ahead of the
629 parameterised ones, so it never reaches this.)
631 The return value is the *canonical* form of the id (lowercase,
632 hyphenated, no wrapper) rather than the verbatim input, so alias
633 spellings of the same UUID — for example an ``urn:uuid:`` prefix,
634 ``{...}`` braces, hyphen-free hex, or uppercase hex — come back in
635 the same shape stored on disk, which is what makes "well-formed
636 UUID string" above actually true of the return value. ``uuid.UUID``
637 tolerates a few more, such as ``_`` digit separators; all are
638 canonicalised here. Length is the binding reject: a padded input
639 that pushes the total past 32 characters fails the parser and 404s;
640 shorter padded inputs (e.g. leading/trailing whitespace) are
641 accepted by ``int(..., 16)`` and canonicalise with zero-padding,
642 and CPython's base-16 parser also accepts Unicode digit
643 characters (`"𝟏"` / Arabic-Indic digits), so this validator is
644 NOT a strict whitelist of ASCII hex — the canonical output is
645 pure ASCII hex, which is what makes it safe to interpolate in
646 any template context (#6602; the shared helper
647 ``web/utils/path_ids.py::validated_uuid_path_param`` has the same
648 shape, this local twin stays in ``rag.py`` so the route-table
649 parity scanner keeps seeing its 404).
650 """
651 try:
652 canonical = str(uuid.UUID(str(collection_id)))
653 except (ValueError, AttributeError, TypeError):
654 logger.warning(
655 f"Rejected malformed collection_id: {_log_value_preview(collection_id)}"
656 )
657 # `from None` mirrors `validated_uuid_path_param`: the only
658 # caller of this guard sees a clean 404 detail, and the
659 # `uuid.UUID` parse-failure traceback adds nothing the
660 # `_log_value_preview` line above doesn't already carry.
661 raise HTTPException(
662 status_code=404, detail="Collection not found"
663 ) from None
664 return canonical
667@router.get("/collections/{collection_id}")
668def collection_details_page(
669 request: Request,
670 collection_id,
671 username: Annotated[str, Depends(require_auth)],
672):
673 """Render the Collection Details page."""
674 collection_id = _validated_collection_id(collection_id)
675 return templates.TemplateResponse(
676 request=request,
677 name="pages/collection_details.html",
678 context={
679 "active_page": "collections",
680 "collection_id": collection_id,
681 },
682 )
685@router.get("/collections/{collection_id}/upload")
686def collection_upload_page(
687 request: Request,
688 collection_id,
689 username: Annotated[str, Depends(require_auth)],
690):
691 """Render the Collection Upload page."""
692 collection_id = _validated_collection_id(collection_id)
694 from ...database.session_context import get_user_db_session
695 from ...utilities.db_utils import get_settings_manager
697 # Get the upload PDF storage setting
698 with get_user_db_session(username) as db_session:
699 settings = get_settings_manager(db_session, username)
700 upload_pdf_storage = settings.get_setting(
701 "research_library.upload_pdf_storage", "none"
702 )
703 # Only allow valid values for uploads (no filesystem)
704 if upload_pdf_storage not in ("database", "none"):
705 upload_pdf_storage = "none"
707 return templates.TemplateResponse(
708 request=request,
709 name="pages/collection_upload.html",
710 context={
711 "active_page": "collections",
712 "collection_id": collection_id,
713 "collection_name": None,
714 "upload_pdf_storage": upload_pdf_storage,
715 },
716 )
719# API Routes
722@router.get("/api/rag/settings")
723def get_current_settings(
724 request: Request,
725 username: Annotated[str, Depends(require_auth)],
726):
727 """Get current RAG configuration from settings."""
728 from ...database.session_context import get_user_db_session
730 try:
731 with get_user_db_session(username) as db_session:
732 settings = get_settings_manager(db_session, username)
734 return {
735 "success": True,
736 "settings": {
737 "embedding_provider": settings.get_setting(
738 "local_search_embedding_provider",
739 DEFAULT_LOCAL_SEARCH_PROVIDER,
740 ),
741 "embedding_model": settings.get_setting(
742 "local_search_embedding_model",
743 DEFAULT_LOCAL_SEARCH_MODEL,
744 ),
745 "chunk_size": settings.get_setting(
746 "local_search_chunk_size",
747 DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
748 ),
749 "chunk_overlap": settings.get_setting(
750 "local_search_chunk_overlap",
751 DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
752 ),
753 "splitter_type": settings.get_setting(
754 "local_search_splitter_type",
755 DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
756 ),
757 "text_separators": _get_text_separators(settings),
758 "distance_metric": settings.get_setting(
759 "local_search_distance_metric",
760 DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
761 ),
762 "normalize_vectors": settings.get_setting(
763 "local_search_normalize_vectors",
764 DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
765 ),
766 "index_type": settings.get_setting(
767 "local_search_index_type",
768 DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
769 ),
770 },
771 }
772 except Exception as e:
773 return handle_api_error("getting RAG settings", e)
776@router.post("/api/rag/test-embedding")
777async def test_embedding(
778 request: Request,
779 username: Annotated[str, Depends(require_auth)],
780):
781 """Test an embedding configuration by generating a test embedding."""
782 from ...database.session_context import get_user_db_session
783 from ...utilities.db_utils import get_settings_manager
785 # Pre-bound: the except block below formats an error using these, so a
786 # malformed/empty body would otherwise raise UnboundLocalError from
787 # inside the handler's own error path.
788 provider = None
789 model = None
790 try:
791 data = await request.json()
792 if not isinstance(data, dict):
793 return json_body_error("success", "Request body must be valid JSON")
794 provider = data.get("provider")
795 model = data.get("model")
797 if not provider or not model:
798 return JSONResponse(
799 {"success": False, "error": "Provider and model are required"},
800 status_code=400,
801 )
803 from ...embeddings.embeddings_config import get_embeddings
805 logger.info(
806 "Testing embedding with provider={}, model={}",
807 _log_value_preview(provider),
808 _log_value_preview(model),
809 )
811 # Get user's settings so provider URLs (e.g. Ollama) are resolved correctly.
812 # Sync SQLAlchemy + a network-bound embedding call — both must run in a
813 # thread, never on the uvicorn event loop.
814 def _run_embedding_test() -> tuple[Any, int]:
815 with get_user_db_session(username) as db_session:
816 settings = get_settings_manager(db_session, username)
817 settings_snapshot = (
818 settings.get_all_settings()
819 if hasattr(settings, "get_all_settings")
820 else {}
821 )
822 start = time.time()
823 # get_embeddings (not get_embedding_function) so the owning
824 # instance is reachable here for cleanup: the wrapped callable
825 # from get_embedding_function discards it, leaking the
826 # per-instance httpx client(s) built for this one-shot test.
827 embeddings = get_embeddings(
828 provider=provider,
829 model=model,
830 settings_snapshot=settings_snapshot,
831 )
832 try:
833 emb = embeddings.embed_documents([_EMBEDDING_TEST_SENTENCE])[0]
834 finally:
835 from ...utilities.llm_utils import _close_base_llm
837 try:
838 _close_base_llm(embeddings)
839 except Exception:
840 logger.warning("Failed to close test embedding resources")
841 return emb, int((time.time() - start) * 1000)
843 # run_db_sync (not raw to_thread): the thunk opens a
844 # get_user_db_session block, and to_thread's reused workers would
845 # keep the user's thread-local session attached after the task.
846 embedding, response_time_ms = await run_db_sync(_run_embedding_test)
848 # Get embedding dimension
849 dimension = len(embedding) if hasattr(embedding, "__len__") else None
851 return {
852 "success": True,
853 "dimension": dimension,
854 "response_time_ms": response_time_ms,
855 "provider": provider,
856 "model": model,
857 }
859 except json.JSONDecodeError:
860 # Caught here (ahead of the broad `except Exception` below) so a
861 # malformed body gets the route's own "success"-shaped 400 instead
862 # of falling into `_format_test_embedding_error`, which would echo
863 # the raw decoder message (e.g. "Expecting value: line 1 column 1")
864 # back to the client as part of a hardcoded 500.
865 return json_body_error("success", "Request body must be valid JSON")
866 except Exception as e:
867 # Egress policy denial → clean 4xx (the embedding never fired).
868 from ...security.egress.policy import PolicyDeniedError
870 if isinstance(e, PolicyDeniedError):
871 reason = e.decision.reason
872 if reason == "embeddings_model_not_curated":
873 return JSONResponse(
874 {
875 "success": False,
876 "error": f"Sentence Transformers model '{model}' is not in LDR's vetted model list. Choose one of the listed models, or use Ollama for custom locally managed embedding models. Direct arbitrary Hugging Face downloads are disabled.",
877 "reason_code": reason,
878 },
879 status_code=400,
880 )
881 if reason == "embeddings_cache_revision_unknown":
882 return JSONResponse(
883 {
884 "success": False,
885 "error": f"The existing Sentence Transformers cache for '{model}' does not identify a single immutable model revision, so a new download is refused: it could change the model behind an existing index. To start over, delete that model's cache directory (the 'models--...' folder under the Hugging Face cache, by default ~/.cache/huggingface/hub) and run this test again under Public only — do this only if no collection was indexed with it. To keep an existing index, put a trusted copy of the model under the application's models directory and select it by that name instead, or select a different model.",
886 "reason_code": reason,
887 },
888 status_code=400,
889 )
890 if reason == "embeddings_model_not_cached": 890 ↛ 899line 890 didn't jump to line 899 because the condition on line 890 was always true
891 return JSONResponse(
892 {
893 "success": False,
894 "error": f"Sentence Transformers model '{model}' has not been downloaded yet. Set Egress Scope to Public only, or use Adaptive with a public primary and no private collection selected. If 'Require local embeddings' is enabled, turn it off for this download, then run this test again. After the download completes, you can return to your preferred restrictive scope.",
895 "reason_code": reason,
896 },
897 status_code=400,
898 )
899 return JSONResponse(
900 {
901 "success": False,
902 "error": f"Embedding provider '{provider}' refused by egress policy ({reason}). Disable 'Require local embeddings' or pick a local provider.",
903 "reason_code": reason,
904 },
905 status_code=400,
906 )
907 logger.exception("Error during testing embedding")
908 user_message = _format_test_embedding_error(e, model)
909 # The formatter uses exception type metadata only to select a fixed
910 # message. Raw exception text stays in the server log above.
911 return JSONResponse(
912 {"success": False, "error": user_message}, status_code=500
913 )
916@router.get("/api/rag/models")
917def get_available_models(
918 request: Request,
919 username: Annotated[str, Depends(require_auth)],
920):
921 """Get list of available embedding providers and models."""
922 from ...database.session_context import get_user_db_session
923 from ...utilities.db_utils import get_settings_manager
925 try:
926 from ...embeddings.embeddings_config import _get_provider_classes
927 from ...security.egress.policy import (
928 Decision,
929 PolicyDeniedError,
930 )
932 # Get current settings for providers. STRICT snapshot is mandatory
933 # at this policy-sensitive seam: a non-strict read silently falls
934 # back to JSON defaults when the underlying settings query fails
935 # (SQLAlchemyError / stale enum row), and a defaults-only snapshot
936 # can lack the operator-selected scope/provider inputs that the
937 # local-only posture is supposed to enforce. We refuse before any
938 # provider discovery, credential read, or network probe instead of
939 # admitting a cloud embedder under a permissive fallback.
940 with get_user_db_session(username) as db_session:
941 settings = get_settings_manager(db_session, username)
942 try:
943 settings_snapshot = (
944 settings.get_all_settings(strict=True)
945 if hasattr(settings, "get_all_settings")
946 else {}
947 )
948 except PolicyDeniedError:
949 raise
950 except Exception as exc:
951 raise PolicyDeniedError(
952 Decision(False, "settings_unavailable"),
953 target="available_rag_models",
954 ) from exc
956 # Get provider classes
957 provider_classes = _get_provider_classes()
959 # Egress policy: availability checks and model discovery can open
960 # connections to a provider. Under an effective
961 # local-only embeddings posture (PRIVATE_ONLY / adaptive-private) we
962 # must NOT probe remote-classified providers — mirror the PEP in
963 # embeddings_config.get_embeddings(). Fail closed before either
964 # provider call if the policy can't evaluate.
965 def _embeddings_model_fetch_allowed(provider_key: str) -> bool:
966 try:
967 from ...security.egress.policy import (
968 context_from_snapshot,
969 evaluate_embeddings,
970 resolve_run_primary_engine,
971 )
973 # Thread username so a per-user private retriever primary
974 # resolves PRIVATE_ONLY (forcing local embeddings) and this
975 # route can't probe a remote embedder for that user.
976 ctx = context_from_snapshot(
977 settings_snapshot,
978 resolve_run_primary_engine(settings_snapshot),
979 username=username,
980 )
981 if not ctx.require_local_embeddings:
982 return True
983 decision = evaluate_embeddings(
984 provider_key, ctx, settings_snapshot=settings_snapshot
985 )
986 if not decision.allowed:
987 logger.bind(policy_audit=True).info(
988 "skipping embeddings model-list probe "
989 "(local-only egress posture)",
990 provider=provider_key,
991 reason=decision.reason,
992 )
993 return decision.allowed
994 except Exception:
995 # Fail closed: a policy/snapshot error must not open a cloud
996 # probe under a local-only posture.
997 logger.bind(policy_audit=True).warning(
998 "embeddings model-list egress check failed; skipping probe",
999 provider=provider_key,
1000 exc_info=True,
1001 )
1002 return False
1004 # Provider display names
1005 provider_labels = {
1006 "sentence_transformers": "Sentence Transformers (Local)",
1007 "ollama": "Ollama (Local)",
1008 "openai": "OpenAI API",
1009 }
1011 # Get provider options and models by looping through providers
1012 provider_options = []
1013 providers = {}
1015 for provider_key, provider_class in provider_classes.items():
1016 policy_allowed = _embeddings_model_fetch_allowed(provider_key)
1017 available = policy_allowed and provider_class.is_available(
1018 settings_snapshot
1019 )
1021 # Always show the provider in the dropdown so users can
1022 # configure its settings (e.g. fix a wrong Ollama URL).
1023 provider_options.append(
1024 {
1025 "value": provider_key,
1026 "label": provider_labels.get(provider_key, provider_key),
1027 "available": available,
1028 "policy_allowed": policy_allowed,
1029 }
1030 )
1032 # Availability is checked only after the policy permits probing.
1033 if available:
1034 models = provider_class.get_available_models(settings_snapshot)
1035 providers[provider_key] = [
1036 {
1037 "value": m["value"],
1038 "label": m["label"],
1039 "provider": provider_key,
1040 **(
1041 {"is_embedding": m["is_embedding"]}
1042 if "is_embedding" in m
1043 else {}
1044 ),
1045 }
1046 for m in models
1047 ]
1048 else:
1049 providers[provider_key] = []
1051 return {
1052 "success": True,
1053 "provider_options": provider_options,
1054 "providers": providers,
1055 }
1057 except Exception as e:
1058 # Egress policy denial (incl. settings_unavailable) → clean 4xx/5xx
1059 # so the caller distinguishes a refused probe from a server crash.
1060 from ...security.egress.policy import PolicyDeniedError as _PDE
1062 if isinstance(e, _PDE):
1063 reason = e.decision.reason
1064 status_code = 503 if reason == "settings_unavailable" else 400
1065 logger.bind(policy_audit=True).info(
1066 "available-models refused by egress policy",
1067 reason=reason,
1068 target=e.target,
1069 )
1070 return JSONResponse(
1071 {
1072 "success": False,
1073 "error": (
1074 "Settings are currently unavailable; cannot safely "
1075 "enumerate embedding models."
1076 )
1077 if reason == "settings_unavailable"
1078 else f"Available-models request refused ({reason}).",
1079 },
1080 status_code=status_code,
1081 )
1082 return handle_api_error("getting available models", e)
1085@router.get("/api/rag/info")
1086def get_index_info(
1087 request: Request, username: Annotated[str, Depends(require_auth)]
1088):
1089 """Get information about the current RAG index."""
1090 from ...database.library_init import get_default_library_id
1092 try:
1093 # Get collection_id from request or use default Library collection
1094 collection_id = request.query_params.get("collection_id")
1095 if not collection_id:
1096 collection_id = get_default_library_id(username)
1098 logger.info(
1099 f"Getting RAG index info for collection_id: {_log_value_preview(collection_id)}"
1100 )
1102 # `with`: LocalEmbeddingManager eagerly opens httpx client pairs for
1103 # Ollama embeddings, so an unclosed service leaks file descriptors on
1104 # every call (#4407). This is a hot endpoint — the collection page
1105 # hits it on each load.
1106 with get_rag_service(request, username, collection_id) as rag_service:
1107 info = rag_service.get_current_index_info(collection_id)
1109 if info is None:
1110 logger.info(
1111 f"No RAG index found for collection_id: {_log_value_preview(collection_id)}"
1112 )
1113 return {"success": True, "info": None, "message": "No index found"}
1115 logger.info(
1116 f"Found RAG index for collection_id: {_log_value_preview(collection_id)}"
1117 )
1118 return {"success": True, "info": info}
1119 except Exception as e:
1120 return handle_api_error("getting index info", e)
1123@router.get("/api/rag/stats")
1124def get_rag_stats(
1125 request: Request, username: Annotated[str, Depends(require_auth)]
1126):
1127 """Get RAG statistics for a collection."""
1128 from ...database.library_init import get_default_library_id
1130 try:
1131 # Get collection_id from request or use default Library collection
1132 collection_id = request.query_params.get("collection_id")
1133 if not collection_id:
1134 collection_id = get_default_library_id(username)
1136 with get_rag_service(request, username, collection_id) as rag_service:
1137 stats = rag_service.get_rag_stats(collection_id)
1139 return {"success": True, "stats": stats}
1140 except Exception as e:
1141 return handle_api_error("getting RAG stats", e)
1144@router.post("/api/rag/index-document")
1145async def index_document(
1146 request: Request, username: Annotated[str, Depends(require_auth)]
1147):
1148 """Index a single document in a collection."""
1149 from ...database.library_init import get_default_library_id
1151 text_doc_id = None
1152 try:
1153 data = await request.json()
1154 if not isinstance(data, dict):
1155 return json_body_error("success", "Request body must be valid JSON")
1156 text_doc_id = data.get("text_doc_id")
1157 force_reindex = data.get("force_reindex", False)
1158 collection_id = data.get("collection_id")
1160 if not isinstance(force_reindex, bool):
1161 return JSONResponse(
1162 {
1163 "success": False,
1164 "error": "force_reindex must be a boolean",
1165 },
1166 status_code=400,
1167 )
1169 if not text_doc_id:
1170 return JSONResponse(
1171 {"success": False, "error": "text_doc_id is required"},
1172 status_code=400,
1173 )
1175 if not collection_id:
1176 # Opens the user's SQLCipher session (sync) — threadpool it.
1177 collection_id = await run_db_sync(get_default_library_id, username)
1179 def _index_sync():
1180 with get_rag_service(
1181 request, username, collection_id
1182 ) as rag_service:
1183 return rag_service.index_document(
1184 text_doc_id, collection_id, force_reindex
1185 )
1187 result = await run_db_sync(_index_sync)
1189 if result["status"] == "error":
1190 return JSONResponse(
1191 {
1192 "success": False,
1193 "error": _client_indexing_error(result.get("error")),
1194 },
1195 status_code=400,
1196 )
1198 return {"success": True, "result": result}
1199 except json.JSONDecodeError:
1200 return json_body_error("success", "Request body must be valid JSON")
1201 except Exception as e:
1202 return handle_api_error(f"indexing document {text_doc_id}", e)
1205@router.post("/api/rag/remove-document")
1206async def remove_document(
1207 request: Request, username: Annotated[str, Depends(require_auth)]
1208):
1209 """Remove a document from RAG in a collection."""
1210 from ...database.library_init import get_default_library_id
1212 # Pre-bound for the same reason as test_embedding above: the except
1213 # block interpolates text_doc_id into its error message.
1214 text_doc_id = None
1215 try:
1216 data = await request.json()
1217 if not isinstance(data, dict):
1218 return json_body_error("success", "Request body must be valid JSON")
1219 text_doc_id = data.get("text_doc_id")
1220 collection_id = data.get("collection_id")
1222 if not text_doc_id:
1223 return JSONResponse(
1224 {"success": False, "error": "text_doc_id is required"},
1225 status_code=400,
1226 )
1228 if not collection_id: 1228 ↛ 1232line 1228 didn't jump to line 1232 because the condition on line 1228 was always true
1229 # Opens the user's SQLCipher session (sync) — threadpool it.
1230 collection_id = await run_db_sync(get_default_library_id, username)
1232 def _remove_sync():
1233 with get_rag_service(
1234 request, username, collection_id
1235 ) as rag_service:
1236 return rag_service.remove_document_from_rag(
1237 text_doc_id, collection_id
1238 )
1240 result = await run_db_sync(_remove_sync)
1242 if result["status"] == "error":
1243 return JSONResponse(
1244 {
1245 "success": False,
1246 "error": _client_indexing_error(result.get("error")),
1247 },
1248 status_code=400,
1249 )
1251 return {"success": True, "result": result}
1252 except json.JSONDecodeError:
1253 return json_body_error("success", "Request body must be valid JSON")
1254 except Exception as e:
1255 return handle_api_error(f"removing document {text_doc_id}", e)
1258@router.post("/api/rag/index-all")
1259def index_all(
1260 request: Request,
1261 username: Annotated[str, Depends(require_auth)],
1262):
1263 """Index all documents in a collection with Server-Sent Events progress.
1265 POST, not GET, because it mutates: it resolves the caller's collection,
1266 writes embedding metadata, and runs the embedding/index pipeline over
1267 every document in it (``?force_reindex=true`` stages and replaces
1268 vectors that already exist). CSRFMiddleware keys enforcement on the
1269 method, so declaring it here is what puts it behind the token check —
1270 a GET would be outside the middleware entirely and startable from
1271 same-site markup riding the victim's session cookie. Progress still
1272 streams as Server-Sent Events; the caller is a fetch client, which can
1273 send both the X-CSRFToken header and a POST, so nothing about the
1274 streaming response changes. Query parameters (``collection_id``,
1275 ``force_reindex``) are unchanged.
1276 """
1277 from ...database.session_context import get_user_db_session
1278 from ...database.library_init import get_default_library_id
1279 from ...utilities.db_utils import get_settings_manager
1281 # Parse as bool — a raw string "false" is truthy, which would silently
1282 # force a full reindex for every caller that asked to skip already-indexed
1283 # documents.
1284 force_reindex = (
1285 request.query_params.get("force_reindex", "false").lower() == "true"
1286 )
1288 # Get collection_id from request or use default Library collection
1289 collection_id = request.query_params.get("collection_id")
1290 if not collection_id:
1291 collection_id = get_default_library_id(username)
1293 if force_reindex:
1294 # A replacement is one transaction for the complete collection. Use
1295 # the heartbeat-capable stream instead of publishing batch by batch.
1296 return index_collection(request, collection_id, username)
1298 logger.info(
1299 f"Starting index-all for collection_id: {_log_value_preview(collection_id)}, force_reindex: {force_reindex}"
1300 )
1302 # Create RAG service in request context before generator runs.
1303 # use_defaults=force_reindex mirrors index_collection / the background
1304 # worker so a force-reindex resolves the same embedding config.
1305 rag_service = get_rag_service(
1306 request, username, collection_id, use_defaults=force_reindex
1307 )
1309 # Import before building the SSE generator. Importing inside its try
1310 # would leave the exception handler unbound if this import failed.
1311 # Keep it below module scope because it pulls in the deletion package.
1312 from ...research_library.deletion.utils.cascade_helper import (
1313 RemoteVectorStoreDropError,
1314 remote_drop_failure_message,
1315 )
1317 # Fetch batch size from settings in this thread before generator runs
1318 with get_user_db_session(username) as db_session:
1319 settings = get_settings_manager(db_session, username)
1320 batch_size = int(settings.get_setting("rag.indexing_batch_size", 15))
1321 try:
1322 _max_workers = int(
1323 settings.get_setting("rag.indexing_max_parallel_docs", 4)
1324 )
1325 except SettingsValueError:
1326 raise
1327 except Exception:
1328 _max_workers = 4
1329 _max_workers = max(1, min(_max_workers, 16))
1331 def generate():
1332 """Generator function for SSE progress updates."""
1333 _sse_cancel = threading.Event()
1334 try:
1335 if not _register_sse_indexer( 1335 ↛ 1338line 1335 didn't jump to line 1338 because the condition on line 1335 was never true
1336 username, collection_id, None, _sse_cancel
1337 ):
1338 yield f"data: {json.dumps({'type': 'error', 'error': 'Indexing is already in progress for this collection'})}\n\n"
1339 return
1340 # Send initial status
1341 yield f"data: {json.dumps({'type': 'start', 'message': 'Starting bulk indexing...'})}\n\n"
1343 # Get document IDs to index from DocumentCollection. All DB work
1344 # happens inside this SHORT-LIVED session; NO ``yield`` runs while
1345 # it is held (the "collection not found" error is yielded AFTER the
1346 # block) — keeping a get_user_db_session scope open across a yield
1347 # risks the Starlette/anyio thread-affinity corruption fixed in
1348 # download_bulk (__enter__/__exit__ can land on different threads).
1349 collection_found = True
1350 doc_info = []
1351 with get_user_db_session(username) as db_session:
1352 # Persist embedding metadata + clean up on force-reindex via the
1353 # shared helpers, so this bulk route behaves like the
1354 # single-collection SSE route and the background worker
1355 # (previously it stored no embedding_dimension and never reset
1356 # stale chunks/indices on force-reindex).
1357 collection = (
1358 db_session.query(Collection)
1359 .filter_by(id=collection_id)
1360 .first()
1361 )
1363 if not collection:
1364 collection_found = False
1365 else:
1366 # Ordinary indexing records its initial configuration.
1367 # Forced rebuilds publish metadata with the new generation.
1368 if collection.embedding_model is None and not force_reindex:
1369 _store_collection_embedding_metadata(
1370 collection, rag_service
1371 )
1372 db_session.commit()
1374 doc_info = [
1375 (doc.id, doc.title)
1376 for _dc, doc in _query_documents_to_index(
1377 db_session, collection_id, force_reindex
1378 )
1379 ]
1381 if not collection_found:
1382 yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
1383 return
1385 if not doc_info:
1386 yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
1387 return
1389 results = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
1390 total = len(doc_info)
1392 # Process documents in batches to pace SSE progress events.
1393 # The batch boundary is purely cosmetic now — documents within
1394 # a batch run in parallel via the bounded worker pool below.
1395 # batch_size and _max_workers are read from settings before the
1396 # generator starts (closure capture) — no DB work in-stream.
1397 processed = 0
1399 for i in range(0, len(doc_info), batch_size):
1400 batch = doc_info[i : i + batch_size]
1402 # Run the per-batch docs through the bounded parallel
1403 # helper. ``as_completed`` order means progress events
1404 # fire in completion order, not submission order — the
1405 # `processed` counter below keeps the SSE payload
1406 # monotonically increasing so the UI percent stays sane.
1407 # ``is_cancelled`` polls the SSE-disconnect event so an
1408 # early-disconnect watcher (or this generator's own
1409 # ``finally``) can short-circuit the batch before all
1410 # queued futures run. The helper still drains in-flight
1411 # workers via ``pool.shutdown(wait=True, ...)`` so no
1412 # worker outlives the helper return — see
1413 # ``index_documents_parallel`` for the rationale.
1414 parallel_result = rag_service.index_documents_parallel(
1415 batch,
1416 collection_id,
1417 force_reindex=force_reindex,
1418 max_workers=_max_workers,
1419 is_cancelled=_sse_cancel.is_set,
1420 )
1421 batch_results = parallel_result["results"]
1423 for doc_id, title in batch:
1424 processed += 1
1425 result = batch_results[doc_id]
1427 # Send progress update
1428 yield f"data: {json.dumps({'type': 'progress', 'current': processed, 'total': total, 'title': title, 'percent': int((processed / total) * 100)})}\n\n"
1430 if result["status"] == "success":
1431 results["successful"] += 1
1432 elif result["status"] in ("skipped", "cleared"):
1433 # "cleared" (empty-text purge) is a handled non-failure.
1434 results["skipped"] += 1
1435 else:
1436 results["failed"] += 1
1437 results["errors"].append(
1438 {
1439 "doc_id": doc_id,
1440 "title": title,
1441 "error": _client_indexing_error(
1442 result.get("error")
1443 ),
1444 }
1445 )
1447 # Send completion status
1448 yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
1450 # Log final status for debugging
1451 logger.info(
1452 f"Bulk indexing complete: {results['successful']} successful, {results['skipped']} skipped, {results['failed']} failed"
1453 )
1455 except RemoteVectorStoreDropError as exc:
1456 # Caught ahead of the catch-all below: this is an actionable
1457 # vector-store failure, not an internal error. Force-reindex rolls
1458 # back, keeping the RAGIndex rows (the remote collections' only
1459 # handle); the message says whether a retry can help and whether
1460 # some collections were already dropped (#5765). Logged from the
1461 # error's sanitized fields, with no traceback (a traceback would
1462 # add frame state; the backend's error text can carry secrets).
1463 drop_detail = exc.log_detail()
1464 logger.warning(
1465 "Force-reindex aborted: could not drop the vector-store "
1466 f"collection before rebuilding it: {drop_detail}"
1467 )
1468 drop_error = {
1469 "type": "error",
1470 "error": remote_drop_failure_message(
1471 exc, "indexing was aborted"
1472 ),
1473 }
1474 yield f"data: {json.dumps(drop_error)}\n\n"
1475 except Exception:
1476 logger.exception("Error in bulk indexing")
1477 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
1478 finally:
1479 # Generator ``finally`` runs at stream completion (or client
1480 # disconnect via ``GeneratorExit``) — the safe place to release
1481 # the RAG service's embedding-manager httpx clients. Closing at
1482 # the outer route scope would tear it down before the streamed
1483 # generator runs. ``safe_close`` swallows close-time errors so
1484 # a broken Ollama doesn't mask the original generator outcome.
1485 #
1486 # Signal cancellation BEFORE closing: the helper has already
1487 # drained in-flight workers (its ``shutdown(wait=True, ...)``
1488 # guarantees no worker outlives the helper return), so setting
1489 # the event here is for any in-progress batch loop iteration
1490 # and any future code path that bypasses the helper.
1491 _sse_cancel.set()
1492 _unregister_sse_indexer(username, collection_id, _sse_cancel)
1493 safe_close(rag_service, "rag_service (index-all SSE)")
1495 return WorkerCleanupStreamingResponse(
1496 generate(),
1497 media_type="text/event-stream",
1498 headers={
1499 "Cache-Control": "no-cache, no-transform",
1500 "X-Accel-Buffering": "no",
1501 },
1502 )
1505@router.post("/api/rag/configure")
1506async def configure_rag(
1507 request: Request,
1508 username: Annotated[str, Depends(require_auth)],
1509):
1510 """
1511 Change RAG configuration (embedding model, chunk size, etc.).
1512 This will create a new index with the new configuration.
1513 """
1514 from ...database.session_context import get_user_db_session
1515 from ...utilities.db_utils import get_settings_manager
1517 try:
1518 data = await request.json()
1519 if not isinstance(data, dict):
1520 return json_body_error("success", "Request body must be valid JSON")
1521 embedding_model = data.get("embedding_model")
1522 embedding_provider = data.get("embedding_provider")
1523 chunk_size = data.get("chunk_size")
1524 chunk_overlap = data.get("chunk_overlap")
1525 collection_id = data.get("collection_id")
1527 # Get new advanced settings (with defaults)
1528 splitter_type = data.get(
1529 "splitter_type", DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE
1530 )
1531 text_separators = data.get(
1532 "text_separators", DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
1533 )
1534 distance_metric = data.get(
1535 "distance_metric", DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC
1536 )
1537 normalize_vectors = data.get(
1538 "normalize_vectors", DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS
1539 )
1540 index_type = data.get("index_type", DEFAULT_LOCAL_SEARCH_INDEX_TYPE)
1542 if (
1543 not embedding_model
1544 or not embedding_provider
1545 or chunk_size is None
1546 or chunk_overlap is None
1547 ):
1548 return JSONResponse(
1549 {
1550 "success": False,
1551 "error": "All configuration parameters are required (embedding_model, embedding_provider, chunk_size, chunk_overlap",
1552 },
1553 status_code=400,
1554 )
1556 if type(chunk_size) is not int or not (
1557 _RAG_CHUNK_SIZE_MIN <= chunk_size <= _RAG_CHUNK_SIZE_MAX
1558 ):
1559 return JSONResponse(
1560 {
1561 "success": False,
1562 "error": (
1563 "chunk_size must be an integer between "
1564 f"{_RAG_CHUNK_SIZE_MIN} and {_RAG_CHUNK_SIZE_MAX}"
1565 ),
1566 },
1567 status_code=400,
1568 )
1570 if type(chunk_overlap) is not int or not (
1571 _RAG_CHUNK_OVERLAP_MIN <= chunk_overlap <= _RAG_CHUNK_OVERLAP_MAX
1572 ):
1573 return JSONResponse(
1574 {
1575 "success": False,
1576 "error": (
1577 "chunk_overlap must be an integer between "
1578 f"{_RAG_CHUNK_OVERLAP_MIN} and "
1579 f"{_RAG_CHUNK_OVERLAP_MAX}"
1580 ),
1581 },
1582 status_code=400,
1583 )
1585 if chunk_overlap > chunk_size:
1586 return JSONResponse(
1587 {
1588 "success": False,
1589 "error": (
1590 "chunk_overlap must be less than or equal to chunk_size"
1591 ),
1592 },
1593 status_code=400,
1594 )
1596 if not isinstance(normalize_vectors, bool):
1597 return JSONResponse(
1598 {
1599 "success": False,
1600 "error": "normalize_vectors must be a boolean",
1601 },
1602 status_code=400,
1603 )
1605 # The local_search_text_separators setting is registered with
1606 # ui_element "json", so store the separators as a proper list. A
1607 # string payload (e.g. a textarea value) is parsed into a list
1608 # first; anything that isn't a JSON array of strings is rejected
1609 # outright (400) rather than silently coerced to defaults, so a
1610 # malformed payload can't get written into settings or fed into
1611 # the chunker.
1612 text_separators = await asyncio.to_thread(
1613 _parse_configured_text_separators, text_separators
1614 )
1615 if text_separators is None:
1616 return JSONResponse(
1617 {
1618 "success": False,
1619 "error": "text_separators must be a JSON array of strings",
1620 },
1621 status_code=400,
1622 )
1624 # All settings writes + the collection reconfiguration (FAISS index
1625 # creation) run sync I/O. Offload to a worker thread so the uvicorn
1626 # event loop is free to serve other requests during the potentially
1627 # slow index-create path.
1628 def _persist_configuration():
1629 """Persist the settings and (optionally) create/promote the
1630 collection's index in ONE database transaction.
1632 Settings and the selected index must land atomically: a
1633 mid-way failure (locked/environment-locked setting, DB error)
1634 must leave BOTH unchanged, never commit the settings while the
1635 old index stays "current" (or vice versa) — that mismatch is
1636 exactly what would let a subsequent search silently run
1637 against a stale/inconsistent configuration.
1639 Returns a ``JSONResponse`` for an error that should short-
1640 circuit the request, or the new/reused index hash (``None``
1641 when no ``collection_id`` was supplied) on success.
1642 """
1643 with get_user_db_session(username) as db_session:
1644 settings = get_settings_manager(db_session, username)
1646 requested_settings = (
1647 ("local_search_embedding_model", embedding_model),
1648 ("local_search_embedding_provider", embedding_provider),
1649 ("local_search_chunk_size", int(chunk_size)),
1650 ("local_search_chunk_overlap", int(chunk_overlap)),
1651 ("local_search_splitter_type", splitter_type),
1652 ("local_search_text_separators", text_separators),
1653 ("local_search_distance_metric", distance_metric),
1654 (
1655 "local_search_normalize_vectors",
1656 normalize_vectors,
1657 ),
1658 ("local_search_index_type", index_type),
1659 )
1660 requested_setting_keys = [
1661 key for key, _value in requested_settings
1662 ]
1664 if settings.settings_locked:
1665 return JSONResponse(
1666 {
1667 "success": False,
1668 "error": "RAG configuration is locked by app.lock_settings",
1669 },
1670 status_code=403,
1671 )
1673 # Environment-locked settings (LDR_* env vars) must never be
1674 # overwritten through a settings-mutation endpoint. Refuse
1675 # the WHOLE request up front, before any write, rather than
1676 # silently skipping just the locked keys — that would both
1677 # mislead the caller with a 200 and partially apply an
1678 # inconsistent configuration.
1679 environment_locked_keys = [
1680 key
1681 for key, _value in requested_settings
1682 if check_env_setting(key) is not None
1683 ]
1684 if environment_locked_keys:
1685 return JSONResponse(
1686 {
1687 "success": False,
1688 "error": (
1689 "Settings "
1690 f"{', '.join(environment_locked_keys)} "
1691 "are environment-locked"
1692 ),
1693 },
1694 status_code=403,
1695 )
1697 for key, value in requested_settings:
1698 if not settings.set_setting(key, value, commit=False):
1699 db_session.rollback()
1700 return JSONResponse(
1701 {
1702 "success": False,
1703 "error": "Unable to save RAG configuration. No changes were applied.",
1704 },
1705 status_code=500,
1706 )
1708 index_hash = None
1709 if collection_id:
1710 with LibraryRAGService(
1711 username=username,
1712 embedding_model=embedding_model,
1713 embedding_provider=embedding_provider,
1714 chunk_size=int(chunk_size),
1715 chunk_overlap=int(chunk_overlap),
1716 splitter_type=splitter_type,
1717 text_separators=text_separators,
1718 distance_metric=distance_metric,
1719 normalize_vectors=normalize_vectors,
1720 index_type=index_type,
1721 ) as new_rag_service:
1722 rag_index = new_rag_service._get_or_create_rag_index(
1723 collection_id, db_session=db_session, commit=False
1724 )
1725 index_hash = rag_index.index_hash
1727 db_session.commit()
1728 settings.emit_settings_changed_after_commit(
1729 requested_setting_keys
1730 )
1731 return index_hash
1733 # run_db_sync (not raw to_thread): get_user_db_session/LibraryRAGService
1734 # open the user's DB session; reused to_thread workers must not retain it.
1735 result = await run_db_sync(_persist_configuration)
1736 if isinstance(result, JSONResponse):
1737 return result
1738 index_hash = result
1740 if collection_id:
1741 return {
1742 "success": True,
1743 "message": "Configuration updated for collection. You can now index documents with the new settings.",
1744 "index_hash": index_hash,
1745 }
1747 # Just saving default settings without updating a specific collection
1748 return {
1749 "success": True,
1750 "message": "Default embedding settings saved successfully. New collections will use these settings.",
1751 }
1753 except json.JSONDecodeError:
1754 return json_body_error("success", "Request body must be valid JSON")
1755 except Exception as e:
1756 return handle_api_error("configuring RAG", e)
1759@router.get("/api/rag/documents")
1760def get_documents(
1761 request: Request, username: Annotated[str, Depends(require_auth)]
1762):
1763 """Get library documents with their RAG status for the default Library collection (paginated).
1765 Query parameters:
1766 page (int): 1-indexed page number, clamped to 1-10000 (default 1;
1767 non-numeric input falls back to 1).
1768 per_page (int): rows per page, clamped to 1-100 (default 50;
1769 non-numeric input falls back to 50).
1770 """
1771 from ...database.session_context import get_user_db_session
1772 from ...database.library_init import get_default_library_id
1774 try:
1775 # Get pagination parameters
1776 try:
1777 page = int(request.query_params.get("page", "1"))
1778 except (TypeError, ValueError):
1779 page = 1
1780 try:
1781 per_page = int(request.query_params.get("per_page", "50"))
1782 except (TypeError, ValueError):
1783 per_page = 50
1784 page = max(1, page)
1785 filter_type = request.query_params.get(
1786 "filter", "all"
1787 ) # all, indexed, unindexed
1789 # Validate pagination parameters
1790 # Upper-bound the page as well. A numerically valid but astronomical
1791 # ?page=10**40 reaches .offset() and raises OverflowError from
1792 # SQLite's 64-bit integer conversion, surfacing as a 500. The
1793 # try/except above only rescues NON-numeric input. Mirrors
1794 # metrics.py's _MAX_PAGE, which already handles this correctly.
1795 page = max(1, min(page, _MAX_PAGE))
1796 per_page = min(max(1, per_page), 100)
1798 # Close current thread's session to force fresh connection
1799 from ...database.thread_local_session import cleanup_current_thread
1801 cleanup_current_thread()
1803 # Get collection_id from request or use default Library collection
1804 collection_id = request.query_params.get("collection_id")
1805 if not collection_id:
1806 collection_id = get_default_library_id(username)
1808 logger.info(
1809 f"Getting documents for collection_id: {_log_value_preview(collection_id)}, filter: {_log_value_preview(filter_type)}, page: {page}"
1810 )
1812 with get_user_db_session(username) as db_session:
1813 # Expire all cached objects to ensure we get fresh data from DB
1814 db_session.expire_all()
1816 # Import RagDocumentStatus model
1817 from ...database.models.library import RagDocumentStatus
1819 # Build base query - join Document with DocumentCollection for the collection
1820 # LEFT JOIN with rag_document_status to check indexed status
1821 query = (
1822 db_session.query(
1823 Document, DocumentCollection, RagDocumentStatus
1824 )
1825 .join(
1826 DocumentCollection,
1827 (DocumentCollection.document_id == Document.id)
1828 & (DocumentCollection.collection_id == collection_id),
1829 )
1830 .outerjoin(
1831 RagDocumentStatus,
1832 (RagDocumentStatus.document_id == Document.id)
1833 & (RagDocumentStatus.collection_id == collection_id),
1834 )
1835 )
1837 logger.opt(lazy=True).debug(
1838 "Base query for collection {}: {}",
1839 lambda: _log_value_preview(collection_id),
1840 lambda: query,
1841 )
1843 # Apply filters based on rag_document_status existence
1844 if filter_type == "indexed":
1845 query = query.filter(RagDocumentStatus.document_id.isnot(None))
1846 elif filter_type == "unindexed":
1847 # Documents in collection but not indexed yet
1848 query = query.filter(RagDocumentStatus.document_id.is_(None))
1850 # Get total count before pagination
1851 total_count = query.count()
1852 logger.info(
1853 f"Found {total_count} total documents for collection {_log_value_preview(collection_id)} with filter {_log_value_preview(filter_type)}"
1854 )
1856 # Apply pagination
1857 results = (
1858 query.order_by(Document.created_at.desc())
1859 .limit(per_page)
1860 .offset((page - 1) * per_page)
1861 .all()
1862 )
1864 documents = [
1865 {
1866 "id": doc.id,
1867 "title": doc.title,
1868 "original_url": doc.original_url,
1869 "rag_indexed": rag_status is not None,
1870 "chunk_count": rag_status.chunk_count if rag_status else 0,
1871 "created_at": doc.created_at.isoformat()
1872 if doc.created_at
1873 else None,
1874 }
1875 for doc, doc_collection, rag_status in results
1876 ]
1878 # Debug logging to help diagnose indexing status issues
1879 indexed_count = sum(1 for d in documents if d["rag_indexed"])
1881 # Additional debug: check rag_document_status for this collection
1882 all_indexed_statuses = (
1883 db_session.query(RagDocumentStatus)
1884 .filter_by(collection_id=collection_id)
1885 .all()
1886 )
1887 logger.info(
1888 f"rag_document_status table shows: {len(all_indexed_statuses)} documents indexed for collection {_log_value_preview(collection_id)}"
1889 )
1891 logger.info(
1892 f"Returning {len(documents)} documents on page {page}: "
1893 f"{indexed_count} indexed, {len(documents) - indexed_count} not indexed"
1894 )
1896 return {
1897 "success": True,
1898 "documents": documents,
1899 "pagination": {
1900 "page": page,
1901 "per_page": per_page,
1902 "total": total_count,
1903 "pages": (total_count + per_page - 1) // per_page,
1904 },
1905 }
1906 except Exception as e:
1907 return handle_api_error("getting documents", e)
1910# Collection Management Routes
1913@router.get("/api/collections")
1914def get_collections(
1915 request: Request, username: Annotated[str, Depends(require_auth)]
1916):
1917 """Get all document collections for the current user."""
1918 from ...database.session_context import get_user_db_session
1920 try:
1921 with get_user_db_session(username) as db_session:
1922 # No need to filter by username - each user has their own database
1923 collections = db_session.query(Collection).all()
1925 # Canonical durable count. Reconciliation maintains these rows from
1926 # actual FAISS membership; the legacy DocumentCollection.indexed flag
1927 # is retained only for compatibility and must not drive user-visible
1928 # truth.
1929 collection_counts = {
1930 collection_id: (
1931 int(document_count or 0),
1932 int(indexed_count or 0),
1933 )
1934 for collection_id, document_count, indexed_count in (
1935 db_session.query(
1936 DocumentCollection.collection_id,
1937 func.count(DocumentCollection.document_id),
1938 func.count(
1939 case(
1940 (
1941 DocumentCollection.indexed.is_(True),
1942 DocumentCollection.document_id,
1943 ),
1944 else_=None,
1945 )
1946 ),
1947 )
1948 .group_by(DocumentCollection.collection_id)
1949 .all()
1950 )
1951 }
1953 result = []
1954 for coll in collections:
1955 document_count, indexed_document_count = collection_counts.get(
1956 coll.id,
1957 (0, 0),
1958 )
1959 collection_data = {
1960 "id": coll.id,
1961 "name": coll.name,
1962 "description": coll.description,
1963 "created_at": coll.created_at.isoformat()
1964 if coll.created_at
1965 else None,
1966 "collection_type": coll.collection_type,
1967 "is_default": coll.is_default
1968 if hasattr(coll, "is_default")
1969 else False,
1970 "is_public": bool(getattr(coll, "is_public", False)),
1971 "agent_enabled": _agent_enabled_default_on(coll),
1972 "document_count": document_count,
1973 "indexed_document_count": indexed_document_count,
1974 "folder_count": len(coll.linked_folders)
1975 if hasattr(coll, "linked_folders")
1976 else 0,
1977 }
1979 # Include embedding metadata if available
1980 if coll.embedding_model:
1981 collection_data["embedding"] = {
1982 "model": coll.embedding_model,
1983 "provider": coll.embedding_model_type.value
1984 if coll.embedding_model_type
1985 else None,
1986 "dimension": coll.embedding_dimension,
1987 "chunk_size": coll.chunk_size,
1988 "chunk_overlap": coll.chunk_overlap,
1989 }
1990 else:
1991 collection_data["embedding"] = None
1993 result.append(collection_data)
1995 return {"success": True, "collections": result}
1996 except Exception as e:
1997 return handle_api_error("getting collections", e)
2000@router.post("/api/collections")
2001async def create_collection(
2002 request: Request, username: Annotated[str, Depends(require_auth)]
2003):
2004 """Create a new document collection."""
2005 data, body_error = await read_json_dict(
2006 request, "success", "Request body must be valid JSON"
2007 )
2008 if body_error is not None:
2009 return body_error
2010 return await run_db_sync(_create_collection_sync, data, username)
2013def _create_collection_sync(data, username):
2014 from ...database.session_context import get_user_db_session
2016 try:
2017 name = data.get("name", "")
2018 if not isinstance(name, str):
2019 # `.get(key, "")` only supplies the default when the key is
2020 # ABSENT; a present-but-wrong-typed value (e.g. `"name": 123`
2021 # or `"name": null`) passes through untouched and blows up on
2022 # `.strip()` with AttributeError -> 500. Reject it explicitly.
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 collection_type = data.get("type", "user_uploads")
2038 if not isinstance(collection_type, str):
2039 # Ported from main's Flask original, which returned
2040 # ``jsonify({...}), 400``. Neither half survives here:
2041 # ``jsonify`` is undefined (NameError -> 500), and a
2042 # ``(body, status)`` tuple is a Flask convention FastAPI does
2043 # not honour. Matches the sibling error returns below.
2044 return JSONResponse(
2045 {
2046 "success": False,
2047 "error": "Collection type must be a string",
2048 },
2049 status_code=400,
2050 )
2051 # Allowlist user-creatable types. System types (notes,
2052 # default_library, research_history) are lazy-created by their
2053 # owning subsystems; a user-crafted impostor (e.g. type="notes")
2054 # would be undeletable under PROTECTED_COLLECTION_TYPES and could
2055 # nondeterministically win _get_or_create_notes_collection's
2056 # .first() lookup, splitting the notes corpus.
2057 allowed_types = {"user_uploads", "user_collection"}
2058 if collection_type not in allowed_types:
2059 return JSONResponse(
2060 {
2061 "success": False,
2062 "error": (
2063 f"Invalid collection type '{collection_type}'. "
2064 f"Allowed: {sorted(allowed_types)}"
2065 ),
2066 },
2067 status_code=400,
2068 )
2069 # Egress classification, default private (the safe choice). A public
2070 # collection counts as a public engine and may use cloud inference.
2071 is_public = data.get("is_public", False)
2072 if not isinstance(is_public, bool):
2073 return JSONResponse(
2074 {
2075 "success": False,
2076 "error": "is_public must be a boolean",
2077 },
2078 status_code=400,
2079 )
2080 # Usability switch (NOT egress): offer this collection to the research
2081 # agent? Default True (available) — behaviour-preserving. An explicit
2082 # JSON null normalizes to True so it matches how NULL is read back
2083 # everywhere else (NULL → available); only an explicit false disables.
2084 agent_enabled_raw = data.get("agent_enabled", True)
2085 if agent_enabled_raw is not None and not isinstance(
2086 agent_enabled_raw, bool
2087 ):
2088 return JSONResponse(
2089 {
2090 "success": False,
2091 "error": "agent_enabled must be a boolean or null",
2092 },
2093 status_code=400,
2094 )
2095 agent_enabled = True if agent_enabled_raw is None else agent_enabled_raw
2097 if not name:
2098 return JSONResponse(
2099 {"success": False, "error": "Name is required"}, status_code=400
2100 )
2102 with get_user_db_session(username) as db_session:
2103 # Check if collection with this name already exists in this user's database
2104 existing = db_session.query(Collection).filter_by(name=name).first()
2106 if existing:
2107 return JSONResponse(
2108 {
2109 "success": False,
2110 "error": f"Collection '{name}' already exists",
2111 },
2112 status_code=400,
2113 )
2115 # Create new collection (no username needed - each user has their own DB)
2116 # Note: created_at uses default=utcnow() in the model, so we don't need to set it manually
2117 collection = Collection(
2118 id=str(uuid.uuid4()), # Generate UUID for collection
2119 name=name,
2120 description=description,
2121 collection_type=collection_type,
2122 is_public=is_public,
2123 agent_enabled=agent_enabled,
2124 )
2126 db_session.add(collection)
2127 db_session.commit()
2129 from ...web_search_engines.search_engines_config import (
2130 invalidate_collection_engines_cache,
2131 )
2133 invalidate_collection_engines_cache(username)
2135 return {
2136 "success": True,
2137 "collection": {
2138 "id": collection.id,
2139 "name": collection.name,
2140 "description": collection.description,
2141 "created_at": collection.created_at.isoformat()
2142 if collection.created_at
2143 else None,
2144 "collection_type": collection.collection_type,
2145 "is_public": bool(collection.is_public),
2146 "agent_enabled": _agent_enabled_default_on(collection),
2147 },
2148 }
2149 except Exception as e:
2150 return handle_api_error("creating collection", e)
2153@router.put("/api/collections/{collection_id}")
2154async def update_collection(
2155 request: Request,
2156 collection_id,
2157 username: Annotated[str, Depends(require_auth)],
2158):
2159 """Update a collection's details."""
2160 data, body_error = await read_json_dict(
2161 request, "success", "Request body must be valid JSON"
2162 )
2163 if body_error is not None:
2164 return body_error
2165 return await run_db_sync(
2166 _update_collection_sync, data, collection_id, username
2167 )
2170def _update_collection_sync(data, collection_id, username):
2171 from ...database.session_context import get_user_db_session
2173 try:
2174 name = data.get("name", "")
2175 if not isinstance(name, str):
2176 # See _create_collection_sync: `.get(key, "")` only supplies
2177 # the default when the key is absent, so a present-but-wrong-
2178 # typed value reaches `.strip()` untouched -> AttributeError -> 500.
2179 return JSONResponse(
2180 {"success": False, "error": "Name must be a string"},
2181 status_code=400,
2182 )
2183 name = name.strip()
2185 description = data.get("description", "")
2186 if not isinstance(description, str):
2187 return JSONResponse(
2188 {"success": False, "error": "Description must be a string"},
2189 status_code=400,
2190 )
2191 description = description.strip()
2193 is_public = data.get("is_public")
2194 if "is_public" in data and not isinstance(is_public, bool):
2195 return JSONResponse(
2196 {
2197 "success": False,
2198 "error": "is_public must be a boolean",
2199 },
2200 status_code=400,
2201 )
2203 agent_enabled = data.get("agent_enabled")
2204 if (
2205 "agent_enabled" in data
2206 and agent_enabled is not None
2207 and not isinstance(agent_enabled, bool)
2208 ):
2209 return JSONResponse(
2210 {
2211 "success": False,
2212 "error": "agent_enabled must be a boolean or null",
2213 },
2214 status_code=400,
2215 )
2217 with get_user_db_session(username) as db_session:
2218 # No need to filter by username - each user has their own database
2219 collection = (
2220 db_session.query(Collection).filter_by(id=collection_id).first()
2221 )
2223 if not collection:
2224 return JSONResponse(
2225 {"success": False, "error": "Collection not found"},
2226 status_code=404,
2227 )
2229 # System collections (Notes, Library, Research History) own
2230 # first-class user data whose lifecycle is managed by other
2231 # subsystems. Renaming them via this generic endpoint
2232 # bypasses their intended invariants and confuses every UI
2233 # surface that lists collections. Mirrors PROTECTED_COLLECTION_TYPES
2234 # in the deletion service. Only identity fields (name,
2235 # description) are locked — the is_public / agent_enabled
2236 # toggles below are deliberate per-collection settings that
2237 # must keep working for system collections too.
2238 from ...research_library.deletion.services.collection_deletion import (
2239 PROTECTED_COLLECTION_TYPES,
2240 )
2242 if collection.collection_type in PROTECTED_COLLECTION_TYPES and (
2243 name or "description" in data
2244 ):
2245 return JSONResponse(
2246 {
2247 "success": False,
2248 "error": (
2249 f"Cannot rename or redescribe system "
2250 f"collection '{collection.name}' "
2251 f"(type={collection.collection_type})."
2252 ),
2253 },
2254 status_code=409,
2255 )
2257 if name:
2258 # Check if new name conflicts with existing collection
2259 existing = (
2260 db_session.query(Collection)
2261 .filter(
2262 Collection.name == name,
2263 Collection.id != collection_id,
2264 )
2265 .first()
2266 )
2268 if existing:
2269 return JSONResponse(
2270 {
2271 "success": False,
2272 "error": f"Collection '{name}' already exists",
2273 },
2274 status_code=400,
2275 )
2277 collection.name = name
2279 # Only write when the caller sends the key (sending "" still
2280 # clears) — otherwise toggle-only PUTs wipe the description.
2281 if "description" in data:
2282 collection.description = description
2284 # Egress classification toggle (only when the caller sends it).
2285 if "is_public" in data:
2286 collection.is_public = is_public
2288 # Research-agent availability toggle (only when the caller sends it).
2289 # Explicit null normalizes to True (available) for parity with how
2290 # NULL is read back elsewhere; only an explicit false disables.
2291 if "agent_enabled" in data:
2292 collection.agent_enabled = (
2293 True if agent_enabled is None else agent_enabled
2294 )
2296 db_session.commit()
2298 from ...web_search_engines.search_engines_config import (
2299 invalidate_collection_engines_cache,
2300 )
2302 invalidate_collection_engines_cache(username)
2304 return {
2305 "success": True,
2306 "collection": {
2307 "id": collection.id,
2308 "name": collection.name,
2309 "description": collection.description,
2310 "created_at": collection.created_at.isoformat()
2311 if collection.created_at
2312 else None,
2313 "collection_type": collection.collection_type,
2314 "is_public": bool(collection.is_public),
2315 "agent_enabled": _agent_enabled_default_on(collection),
2316 },
2317 }
2318 except Exception as e:
2319 return handle_api_error("updating collection", e)
2322def _try_pdf_upgrade(
2323 *,
2324 db_session,
2325 document,
2326 file_content: bytes,
2327 filename: str,
2328 pdf_storage: str,
2329 pdf_storage_manager,
2330) -> bool:
2331 """Check PDF upgrade eligibility and attempt upgrading a Document to include PDF bytes.
2333 Returns True if the document was upgraded, False otherwise. Exceptions
2334 raised during the upgrade attempt are logged and swallowed; the caller
2335 sees False and the upload continues as a non-upgraded success.
2336 """
2337 if pdf_storage != "database" or pdf_storage_manager is None:
2338 return False
2340 # NOTE: Only the PDF magic-byte check is needed here.
2341 # Size limits are already enforced by the async upload wrapper, before
2342 # these bytes are ever buffered.
2343 # Filename sanitization already happens via sanitize_filename() in
2344 # upload_to_collection() before this helper is invoked.
2345 # See PR #3145 review for details.
2346 if file_content[:4] != b"%PDF":
2347 logger.debug(
2348 "Skipping PDF upgrade for {}: not a PDF file",
2349 filename,
2350 )
2351 return False
2353 try:
2354 return bool(
2355 pdf_storage_manager.upgrade_to_pdf(
2356 document=document,
2357 pdf_content=file_content,
2358 session=db_session,
2359 )
2360 )
2361 except Exception:
2362 logger.exception(f"Failed to upgrade PDF for {filename}")
2363 return False
2366@router.post("/api/collections/{collection_id}/upload")
2367@upload_rate_limit_user
2368@upload_rate_limit_ip
2369async def upload_to_collection(
2370 request: Request,
2371 collection_id,
2372 username: Annotated[str, Depends(require_auth)],
2373):
2374 """Upload files to a collection.
2376 The async wrapper consumes the multipart form (the only legitimate
2377 awaits — request.form() and per-file file.read()) and validates
2378 sizes. The sync DB body — opening the SQLCipher session, hashing,
2379 text extraction, and Document/Collection writes — is offloaded to
2380 a threadpool so a multi-file upload doesn't stall the event loop
2381 behind PBKDF2 key derivation and per-file extraction work.
2382 """
2383 # Starlette's form parser yields STARLETTE UploadFile instances;
2384 # fastapi.UploadFile is a SUBCLASS of it on fastapi>=0.113, so
2385 # `isinstance(f, fastapi.UploadFile)` is False for every real
2386 # upload and the filter below silently dropped all files
2387 # (every upload 400'd with "No files provided").
2388 from starlette.datastructures import UploadFile
2390 # Per-file and total upload limits to prevent memory exhaustion.
2391 #
2392 # These MUST come from FileUploadValidator, not a route-local
2393 # constant: GET /api/config/limits (research.py) advertises
2394 # FileUploadValidator.MAX_FILE_SIZE / MAX_FILES_PER_REQUEST to the
2395 # frontend as "the backend's authoritative limits", and
2396 # BodySizeLimitMiddleware (fastapi_app.py) already enforces the same
2397 # constants as the global per-request body cap for every route,
2398 # including this one. A previous hardening pass gave this route its
2399 # own tighter, undocumented cap (100MB / 50 files) instead of reusing
2400 # FileUploadValidator like origin/main's Flask handler did — so a
2401 # file the frontend was told is acceptable could be silently
2402 # rejected here with a different limit. Only the size/count checks
2403 # are reused (not validate_upload's MIME/PDF-structure checks): this
2404 # route indexes arbitrary document types via document_loaders, not
2405 # just PDFs.
2406 from ...security import FileUploadValidator
2408 MAX_FILE_SIZE = FileUploadValidator.MAX_FILE_SIZE
2409 MAX_FILES_PER_UPLOAD = FileUploadValidator.MAX_FILES_PER_REQUEST
2410 MAX_TOTAL_UPLOAD_SIZE = MAX_FILES_PER_UPLOAD * MAX_FILE_SIZE
2412 # Early request size check based on Content-Length, before reading body.
2413 content_length = request.headers.get("content-length")
2414 if content_length:
2415 try:
2416 cl = int(content_length)
2417 if cl > MAX_TOTAL_UPLOAD_SIZE: 2417 ↛ 2418line 2417 didn't jump to line 2418 because the condition on line 2417 was never true
2418 return JSONResponse(
2419 {
2420 "success": False,
2421 "error": f"Request too large. Max {MAX_TOTAL_UPLOAD_SIZE // (1024 * 1024)}MB",
2422 },
2423 status_code=413,
2424 )
2425 except ValueError:
2426 pass
2428 # Parse the multipart form once and reuse — request.form() can only be
2429 # consumed once per request.
2430 form = await request.form()
2431 upload_files: list[UploadFile] = [
2432 f for f in form.getlist("files") if isinstance(f, UploadFile)
2433 ]
2434 if not upload_files:
2435 return JSONResponse(
2436 {"success": False, "error": "No files provided"},
2437 status_code=400,
2438 )
2439 if len(upload_files) > MAX_FILES_PER_UPLOAD:
2440 return JSONResponse(
2441 {
2442 "success": False,
2443 "error": f"Too many files. Max {MAX_FILES_PER_UPLOAD} per upload.",
2444 },
2445 status_code=400,
2446 )
2448 # Buffer every file's bytes here in the async path — UploadFile.read()
2449 # is async (the body may still be streaming from the client), and we
2450 # need the bytes available before we can hand control to the sync
2451 # threadpool worker. Large-file rejection happens up front so we
2452 # don't load anything we'll just throw away.
2453 pdf_storage_form_value = form.get("pdf_storage")
2454 files_data: list[dict] = [] # [{filename, content, oversized}]
2455 for uf in upload_files:
2456 # Preserve empty browser file inputs in the submitted-part count, but
2457 # do not read or process them. The canonical Flask route skipped these
2458 # parts while still reporting them in summary.total.
2459 if not uf.filename:
2460 files_data.append(
2461 {"filename": "", "content": None, "oversized": False}
2462 )
2463 continue
2464 # Per-file size guard before reading into memory. On the pinned
2465 # Starlette range (1.3.x), UploadFile.size is the byte count
2466 # request.form() actually spooled for this part, not a
2467 # client-declared value, so it is never None here.
2468 if uf.size is not None and uf.size > MAX_FILE_SIZE:
2469 files_data.append(
2470 {
2471 "filename": uf.filename,
2472 "content": None,
2473 "oversized": True,
2474 }
2475 )
2476 continue
2477 content = await uf.read()
2478 files_data.append(
2479 {
2480 "filename": uf.filename,
2481 "content": content,
2482 "oversized": len(content) > MAX_FILE_SIZE,
2483 }
2484 )
2486 session_id = request.session.get("session_id")
2487 return await run_db_sync(
2488 _upload_to_collection_sync,
2489 files_data,
2490 pdf_storage_form_value,
2491 collection_id,
2492 username,
2493 session_id,
2494 MAX_FILE_SIZE,
2495 )
2498def _upload_to_collection_sync(
2499 files_data,
2500 pdf_storage_form_value,
2501 collection_id,
2502 username,
2503 session_id,
2504 MAX_FILE_SIZE,
2505):
2506 from ...database.session_context import (
2507 get_user_db_session,
2508 )
2509 from ...utilities.db_utils import get_settings_manager
2510 from ...security import sanitize_filename, UnsafeFilenameError
2511 import hashlib
2512 import uuid
2513 from ...research_library.services.pdf_storage_manager import (
2514 PDFStorageManager,
2515 resolve_pdf_storage_mode,
2516 )
2518 try:
2519 with get_user_db_session(username) as db_session:
2520 settings = get_settings_manager(db_session, username)
2522 # Verify collection exists in this user's database
2523 collection = (
2524 db_session.query(Collection).filter_by(id=collection_id).first()
2525 )
2527 if not collection:
2528 return JSONResponse(
2529 {"success": False, "error": "Collection not found"},
2530 status_code=404,
2531 )
2533 # Get PDF storage mode from form data, falling back to user's setting
2534 default_pdf_storage = settings.get_setting(
2535 "research_library.upload_pdf_storage", "none"
2536 )
2537 # `is None`, not `or`: Flask's form.get(k, default) returned the
2538 # default only when the field was ABSENT. An explicit empty
2539 # `pdf_storage=` fell through to the not-in-(database,none)
2540 # check below. `or` also swallows the empty string, silently
2541 # substituting the user's stored default -- which may be
2542 # "database" -- for a caller who explicitly sent nothing.
2543 pdf_storage = (
2544 default_pdf_storage
2545 if pdf_storage_form_value is None
2546 else pdf_storage_form_value
2547 )
2548 if pdf_storage not in ("database", "none"): 2548 ↛ 2551line 2548 didn't jump to line 2551 because the condition on line 2548 was never true
2549 # Security: user uploads can only use database (encrypted) or none (text-only)
2550 # Filesystem storage is not allowed for user uploads
2551 pdf_storage = "none"
2553 # Initialize PDF storage manager if storing PDFs in database
2554 pdf_storage_manager = None
2555 if pdf_storage == "database":
2556 library_root = settings.get_setting(
2557 "research_library.storage_path",
2558 str(get_library_directory()),
2559 )
2560 library_root = str(
2561 Path(os.path.expandvars(library_root))
2562 .expanduser()
2563 .resolve()
2564 )
2565 shared_library = settings.get_setting(
2566 "research_library.shared_library", False
2567 )
2568 # Per-user library root (issue #5521), mirroring the sibling
2569 # PDFStorageManager construction sites (library.py's
2570 # view_pdf_page, DownloadService.__init__, zotero
2571 # sync_service._library_root()): narrow the shared base to
2572 # this user's own subdirectory so two users' uploads can't
2573 # collide under one shared root. Not exploitable today since
2574 # this instance is only used in "database" mode (bytes go to
2575 # DocumentBlob, never through library_root), but the isolation
2576 # should hold regardless of which methods happen to read it.
2577 #
2578 # Route the mode through resolve_pdf_storage_mode() as well,
2579 # mirroring sync_service's own database-only construction —
2580 # `pdf_storage` is already guaranteed "database" by the branch
2581 # above, so this is a no-op today, but it ties the
2582 # construction to the actual resolved value (rather than a
2583 # bare literal) so it can't silently start bypassing the
2584 # unencrypted-filesystem gate if this code is ever refactored
2585 # to pass `pdf_storage` straight through.
2586 pdf_storage_manager = PDFStorageManager(
2587 library_root=apply_user_subdir(
2588 Path(library_root), username, shared_library
2589 ),
2590 storage_mode=resolve_pdf_storage_mode(pdf_storage),
2591 )
2592 logger.info("PDF storage mode: database (encrypted)")
2593 else:
2594 logger.info("PDF storage mode: none (text-only)")
2596 uploaded_files = []
2597 errors = []
2599 # Track hashes processed earlier in THIS request so we can
2600 # distinguish intra-batch duplicates (two uploaded files with
2601 # identical content) from duplicates against a pre-existing
2602 # library doc. Without this, the second occurrence of an
2603 # intra-batch duplicate was reported as "already_in_collection"
2604 # under the FIRST occurrence's filename -- the warning looked
2605 # like it was skipping books that were actually being added,
2606 # and the skipped entry showed the wrong filename (#5495).
2607 # Maps content hash -> Document so a later identical PDF in the
2608 # same batch can still run the PDF-upgrade path against the
2609 # document kept from the first occurrence.
2610 seen_hashes: dict[str, Document] = {}
2612 for fd in files_data:
2613 # Match main: an empty browser file input counts as a submitted
2614 # part in the summary but is not processed as an upload.
2615 if not fd["filename"]:
2616 continue
2618 # Files were already buffered + size-checked in the async
2619 # wrapper; oversized entries arrive with content=None and
2620 # oversized=True so we can record an error here without
2621 # having held the bytes in memory.
2622 if fd["oversized"]:
2623 # Sanitize BEFORE echoing. Flask sanitized first and only
2624 # ever put the cleaned name into `errors`; the port
2625 # inverted the order, so this branch reflected the raw
2626 # multipart filename — attacker-chosen bytes — straight
2627 # back in the JSON response. Masked today only because the
2628 # single consumer escapes it client-side, which is not a
2629 # property this handler should depend on. Every other
2630 # errors.append below already uses the sanitized name.
2631 try:
2632 safe_name = sanitize_filename(fd["filename"])
2633 except UnsafeFilenameError:
2634 safe_name = "rejected"
2635 errors.append(
2636 {
2637 "filename": safe_name,
2638 "error": f"File too large (max {MAX_FILE_SIZE // (1024 * 1024)}MB)",
2639 }
2640 )
2641 continue
2643 try:
2644 filename = sanitize_filename(fd["filename"])
2645 except UnsafeFilenameError:
2646 errors.append(
2647 {
2648 "filename": "rejected",
2649 "error": "Invalid or unsafe filename",
2650 }
2651 )
2652 continue
2654 # Per-file SAVEPOINT: a failure rolls back only this file so
2655 # earlier successes (and seen_hashes) stay consistent with
2656 # the outer transaction. Commit the savepoint on every
2657 # non-exception exit — including soft validation failures —
2658 # so we never leave nested SAVEPOINTs stacked open across
2659 # the batch (same pattern as research_sources_service).
2660 sp = None
2661 try:
2662 sp = db_session.begin_nested()
2664 # Size limits were already enforced in the async wrapper
2665 # (oversized entries arrive with content=None and are
2666 # rejected above), so main's Content-Length pre-flight and
2667 # post-read re-check have no work left to do here.
2668 file_content = fd["content"]
2669 # Calculate file hash for deduplication
2670 file_hash = hashlib.sha256(file_content).hexdigest()
2672 # Intra-batch duplicate: an earlier file in THIS
2673 # request had identical bytes. Report it as its own
2674 # status so the UI doesn't show it under the first
2675 # occurrence's filename and pretend it was a
2676 # "skipped - already in collection" hit. The first
2677 # occurrence already linked the matching Document to
2678 # this collection (or to the library), so doing the
2679 # Document hash lookup and link check here would be
2680 # misleading at best. Still allow PDF upgrade when
2681 # the kept twin was stored text-only and this copy
2682 # is a PDF with database storage enabled.
2683 if file_hash in seen_hashes:
2684 logger.info(
2685 "Intra-batch duplicate detected: {} (hash {})",
2686 filename,
2687 file_hash,
2688 )
2689 target_doc = seen_hashes[file_hash]
2690 pdf_upgraded = _try_pdf_upgrade(
2691 db_session=db_session,
2692 document=target_doc,
2693 file_content=file_content,
2694 filename=filename,
2695 pdf_storage=pdf_storage,
2696 pdf_storage_manager=pdf_storage_manager,
2697 )
2698 sp.commit()
2699 uploaded_files.append(
2700 {
2701 "filename": filename,
2702 "status": "duplicate_in_batch",
2703 "id": target_doc.id,
2704 "pdf_upgraded": pdf_upgraded,
2705 }
2706 )
2707 continue
2709 # Check if document already exists
2710 existing_doc = (
2711 db_session.query(Document)
2712 .filter_by(document_hash=file_hash)
2713 .first()
2714 )
2716 if existing_doc:
2717 # Document exists, check if we can upgrade to include PDF
2718 pdf_upgraded = _try_pdf_upgrade(
2719 db_session=db_session,
2720 document=existing_doc,
2721 file_content=file_content,
2722 filename=filename,
2723 pdf_storage=pdf_storage,
2724 pdf_storage_manager=pdf_storage_manager,
2725 )
2727 # Check if already in collection
2728 existing_link = (
2729 db_session.query(DocumentCollection)
2730 .filter_by(
2731 document_id=existing_doc.id,
2732 collection_id=collection_id,
2733 )
2734 .first()
2735 )
2737 if not existing_link:
2738 ensure_in_collection(
2739 db_session, existing_doc.id, collection_id
2740 )
2741 status = "added_to_collection"
2742 if pdf_upgraded:
2743 status = "added_to_collection_pdf_upgraded"
2744 sp.commit()
2745 uploaded_files.append(
2746 {
2747 # Report the file the user actually
2748 # uploaded, not the existing doc's
2749 # filename. Otherwise the user sees
2750 # one filename repeated under both
2751 # "added to collection" and
2752 # "new uploads" (or here under
2753 # "already in collection") and
2754 # can't tell which file was which.
2755 "filename": filename,
2756 "status": status,
2757 "id": existing_doc.id,
2758 "pdf_upgraded": pdf_upgraded,
2759 }
2760 )
2761 seen_hashes[file_hash] = existing_doc
2762 else:
2763 status = "already_in_collection"
2764 if pdf_upgraded:
2765 status = "pdf_upgraded"
2766 sp.commit()
2767 uploaded_files.append(
2768 {
2769 "filename": filename,
2770 "status": status,
2771 "id": existing_doc.id,
2772 "pdf_upgraded": pdf_upgraded,
2773 }
2774 )
2775 seen_hashes[file_hash] = existing_doc
2776 else:
2777 # Create new document
2778 from ...document_loaders import (
2779 extract_text_from_bytes,
2780 is_extension_supported,
2781 )
2783 file_extension = Path(filename).suffix.lower()
2785 # Validate extension is supported before extraction
2786 if not is_extension_supported(file_extension):
2787 sp.commit()
2788 errors.append(
2789 {
2790 "filename": filename,
2791 "error": f"Unsupported format: {file_extension}",
2792 }
2793 )
2794 continue
2796 # Use file_type without leading dot for storage
2797 file_type = (
2798 file_extension[1:]
2799 if file_extension.startswith(".")
2800 else file_extension
2801 )
2803 # Parsing can import large libraries or spend minutes
2804 # extracting a document. Release this read snapshot and
2805 # any earlier file's write lock before that work. A
2806 # background writer may otherwise make the snapshot
2807 # impossible to upgrade (SQLITE_BUSY_SNAPSHOT).
2808 sp.commit()
2809 sp = None
2810 db_session.commit()
2811 extracted_text = extract_text_from_bytes(
2812 file_content, file_extension, filename
2813 )
2814 # Keep write failures isolated to this file after the
2815 # parser returns, using a fresh database snapshot.
2816 sp = db_session.begin_nested()
2818 # Clean the extracted text to remove surrogate characters
2819 if extracted_text:
2820 from ...text_processing import remove_surrogates
2822 extracted_text = remove_surrogates(extracted_text)
2824 if not extracted_text:
2825 sp.commit()
2826 errors.append(
2827 {
2828 "filename": filename,
2829 "error": f"Could not extract text from {file_type} file",
2830 }
2831 )
2832 logger.warning(
2833 f"Skipping file {filename} - no text could be extracted"
2834 )
2835 continue
2837 # Get or create the user_upload source type
2838 logger.info(
2839 f"Getting or creating user_upload source type for {filename}"
2840 )
2841 source_type = (
2842 db_session.query(SourceType)
2843 .filter_by(name="user_upload")
2844 .first()
2845 )
2846 if not source_type: 2846 ↛ 2847line 2846 didn't jump to line 2847 because the condition on line 2846 was never true
2847 logger.info("Creating new user_upload source type")
2848 source_type = SourceType(
2849 id=str(uuid.uuid4()),
2850 name="user_upload",
2851 display_name="User Upload",
2852 description="Documents uploaded by users",
2853 icon="fas fa-upload",
2854 )
2855 db_session.add(source_type)
2856 db_session.flush()
2857 logger.info(
2858 f"Created source type with ID: {source_type.id}"
2859 )
2860 else:
2861 logger.info(
2862 f"Found existing source type with ID: {source_type.id}"
2863 )
2865 # Create document with extracted text (no username needed - in user's own database)
2866 # Note: uploaded_at uses default=utcnow() in the model, so we don't need to set it manually
2867 doc_id = str(uuid.uuid4())
2868 logger.info(
2869 f"Creating document {doc_id} for {filename}"
2870 )
2872 # Determine storage mode and file_path
2873 store_pdf_in_db = (
2874 pdf_storage == "database"
2875 and file_type == "pdf"
2876 and pdf_storage_manager is not None
2877 )
2879 new_doc = Document(
2880 id=doc_id,
2881 source_type_id=source_type.id,
2882 filename=filename,
2883 document_hash=file_hash,
2884 file_size=len(file_content),
2885 file_type=file_type,
2886 text_content=extracted_text, # Always store extracted text
2887 file_path=None
2888 if store_pdf_in_db
2889 else FILE_PATH_TEXT_ONLY,
2890 storage_mode="database"
2891 if store_pdf_in_db
2892 else "none",
2893 )
2894 db_session.add(new_doc)
2895 db_session.flush() # Get the ID
2896 logger.info(
2897 f"Document {new_doc.id} created successfully"
2898 )
2900 # Store PDF in encrypted database if requested
2901 pdf_stored = False
2902 if store_pdf_in_db:
2903 try:
2904 pdf_storage_manager.save_pdf(
2905 pdf_content=file_content,
2906 document=new_doc,
2907 session=db_session,
2908 filename=filename,
2909 )
2910 pdf_stored = True
2911 logger.info(
2912 f"PDF stored in encrypted database for {filename}"
2913 )
2914 except Exception:
2915 logger.exception(
2916 f"Failed to store PDF in database for {filename}"
2917 )
2918 # Continue without PDF storage - text is still saved
2920 # Add to collection
2921 ensure_in_collection(
2922 db_session, new_doc.id, collection_id
2923 )
2925 # Release the per-file SAVEPOINT so later failures cannot
2926 # undo this file's writes (and so soft/success paths do
2927 # not leave nested SAVEPOINTs stacked open).
2928 sp.commit()
2930 uploaded_files.append(
2931 {
2932 "filename": filename,
2933 "status": "uploaded",
2934 "id": new_doc.id,
2935 "text_length": len(extracted_text),
2936 "pdf_stored": pdf_stored,
2937 }
2938 )
2939 seen_hashes[file_hash] = new_doc
2941 except Exception:
2942 # A per-file savepoint rollback isolates the failure of this
2943 # file so earlier successful uploads in the batch remain intact
2944 # in the transaction state. Record the error and log the traceback
2945 # FIRST so a rollback failure cannot suppress them or crash the batch.
2946 errors.append(
2947 {
2948 "filename": filename,
2949 "error": "Failed to upload file",
2950 }
2951 )
2952 logger.exception(f"Error uploading file {filename}")
2953 if sp is not None:
2954 try:
2955 sp.rollback()
2956 except Exception:
2957 logger.opt(exception=True).warning(
2958 f"Failed to rollback savepoint for {filename}"
2959 )
2961 db_session.commit()
2963 # Trigger auto-indexing for successfully uploaded documents
2964 document_ids = [
2965 f["id"]
2966 for f in uploaded_files
2967 if f.get("status") in ("uploaded", "added_to_collection")
2968 ]
2969 if document_ids:
2970 from ...database.session_passwords import session_password_store
2972 # session_id was captured by the async wrapper before to_thread
2973 db_password = session_password_store.get_session_password(
2974 username, session_id
2975 )
2976 if db_password:
2977 trigger_auto_index(
2978 document_ids, collection_id, username, db_password
2979 )
2981 return {
2982 "success": True,
2983 "uploaded": uploaded_files,
2984 "errors": errors,
2985 "summary": {
2986 "total": len(files_data),
2987 "successful": len(uploaded_files),
2988 "failed": len(errors),
2989 },
2990 }
2992 except Exception as e:
2993 return handle_api_error("uploading files", e)
2996# Research History Semantic Search Routes have been moved to
2997# web/routers/library_search.py
3000def _collection_rebuild_info(collection, index):
3001 """Report durable rebuild state, independently of expiring task rows."""
3002 pending = collection.rebuild_pending_since if collection else None
3003 updated = index.last_updated_at if index else None
3004 return {
3005 "rebuild_pending_since": pending.isoformat()
3006 if isinstance(pending, datetime)
3007 else None,
3008 "has_published_index": index is not None,
3009 "index_last_updated_at": updated.isoformat()
3010 if isinstance(updated, datetime)
3011 else None,
3012 }
3015@router.get("/api/collections/{collection_id}/documents")
3016def get_collection_documents(
3017 request: Request,
3018 collection_id,
3019 username: Annotated[str, Depends(require_auth)],
3020):
3021 """Get all documents in a collection."""
3022 from ...database.session_context import get_user_db_session
3024 try:
3025 with get_user_db_session(username) as db_session:
3026 # Verify collection exists in this user's database
3027 collection = (
3028 db_session.query(Collection).filter_by(id=collection_id).first()
3029 )
3031 if not collection:
3032 return JSONResponse(
3033 {"success": False, "error": "Collection not found"},
3034 status_code=404,
3035 )
3037 # Look up note source type to filter notes into separate array
3038 note_source_type = (
3039 db_session.query(SourceType).filter_by(name="note").first()
3040 )
3041 note_source_type_id = (
3042 note_source_type.id if note_source_type else None
3043 )
3045 # Get documents through junction table. Compute "has text in DB"
3046 # at the SQL level and defer the text_content column, so listing a
3047 # collection doesn't pull every document's full text body into
3048 # memory just to test it for truthiness (#4560).
3049 has_text_col = (
3050 Document.text_content.isnot(None)
3051 & (Document.text_content != "")
3052 ).label("has_text_db")
3053 doc_links = (
3054 db_session.query(DocumentCollection, Document, has_text_col)
3055 .join(Document)
3056 .options(defer(Document.text_content))
3057 .filter(DocumentCollection.collection_id == collection_id)
3058 .all()
3059 )
3061 documents = []
3062 for link, doc, has_text_db in doc_links:
3063 # Skip notes — they go in the separate notes array
3064 if (
3065 note_source_type_id
3066 and doc.source_type_id == note_source_type_id
3067 ):
3068 continue
3069 # Check if PDF file is stored
3070 has_pdf = bool(
3071 doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
3072 )
3073 has_text_db = bool(has_text_db)
3075 # Use title if available, otherwise filename
3076 display_title = doc.title or doc.filename or "Untitled"
3078 # Get source type name
3079 source_type_name = (
3080 doc.source_type.name if doc.source_type else "unknown"
3081 )
3083 # Check if document is in other collections
3084 other_collections_count = (
3085 db_session.query(DocumentCollection)
3086 .filter(
3087 DocumentCollection.document_id == doc.id,
3088 DocumentCollection.collection_id != collection_id,
3089 )
3090 .count()
3091 )
3093 documents.append(
3094 {
3095 "id": doc.id,
3096 "filename": display_title,
3097 "title": display_title,
3098 "file_type": doc.file_type,
3099 "file_size": doc.file_size,
3100 "uploaded_at": doc.created_at.isoformat()
3101 if doc.created_at
3102 else None,
3103 "indexed": link.indexed,
3104 "chunk_count": link.chunk_count,
3105 "last_indexed_at": link.last_indexed_at.isoformat()
3106 if link.last_indexed_at
3107 else None,
3108 "has_pdf": has_pdf,
3109 "has_text_db": has_text_db,
3110 "source_type": source_type_name,
3111 "in_other_collections": other_collections_count > 0,
3112 "other_collections_count": other_collections_count,
3113 }
3114 )
3116 # Get notes in this collection (note_source_type resolved above)
3117 notes = []
3118 if note_source_type:
3119 note_links = (
3120 db_session.query(DocumentCollection, Document)
3121 .join(Document)
3122 .filter(
3123 DocumentCollection.collection_id == collection_id,
3124 Document.source_type_id == note_source_type.id,
3125 )
3126 .all()
3127 )
3128 for link, note in note_links:
3129 content = note.text_content or ""
3130 notes.append(
3131 {
3132 "id": note.id,
3133 "title": note.title or "Untitled",
3134 "content_preview": content[:200] + "..."
3135 if len(content) > 200
3136 else content,
3137 "tags": note.tags or [],
3138 "pinned": note.favorite,
3139 "created_at": note.created_at.isoformat()
3140 if note.created_at
3141 else None,
3142 "updated_at": note.updated_at.isoformat()
3143 if note.updated_at
3144 else None,
3145 "indexed": link.indexed,
3146 "chunk_count": link.chunk_count,
3147 "source_type": "note",
3148 }
3149 )
3151 # Get index file size if available
3152 index_file_size = None
3153 index_file_size_bytes = None
3154 collection_name = f"collection_{collection_id}"
3155 rag_index = (
3156 db_session.query(RAGIndex)
3157 .filter_by(collection_name=collection_name, is_current=True)
3158 .first()
3159 )
3160 if rag_index and rag_index.index_path:
3161 from pathlib import Path
3163 index_path = Path(rag_index.index_path)
3164 if index_path.exists(): 3164 ↛ 3175line 3164 didn't jump to line 3175 because the condition on line 3164 was always true
3165 size_bytes = index_path.stat().st_size
3166 index_file_size_bytes = size_bytes
3167 # Format as human-readable
3168 if size_bytes < 1024:
3169 index_file_size = f"{size_bytes} B"
3170 elif size_bytes < 1024 * 1024:
3171 index_file_size = f"{size_bytes / 1024:.1f} KB"
3172 else:
3173 index_file_size = f"{size_bytes / (1024 * 1024):.1f} MB"
3175 return {
3176 "success": True,
3177 "collection": {
3178 **_collection_rebuild_info(collection, rag_index),
3179 "id": collection.id,
3180 "name": collection.name,
3181 "description": collection.description,
3182 "is_public": bool(getattr(collection, "is_public", False)),
3183 "agent_enabled": _agent_enabled_default_on(collection),
3184 "embedding_model": collection.embedding_model,
3185 "embedding_model_type": collection.embedding_model_type.value
3186 if collection.embedding_model_type
3187 else None,
3188 "embedding_dimension": collection.embedding_dimension,
3189 "chunk_size": collection.chunk_size,
3190 "chunk_overlap": collection.chunk_overlap,
3191 # Advanced settings
3192 "splitter_type": collection.splitter_type,
3193 "distance_metric": collection.distance_metric,
3194 "index_type": collection.index_type,
3195 "normalize_vectors": collection.normalize_vectors,
3196 # Index file info
3197 "index_file_size": index_file_size,
3198 "index_file_size_bytes": index_file_size_bytes,
3199 "collection_type": collection.collection_type,
3200 # Deletion policy comes from the server so the UI can't
3201 # drift from PROTECTED_COLLECTION_TYPES: the details page
3202 # hides its Delete button for system collections, whose
3203 # deletion the service refuses with a 409 anyway.
3204 "is_protected": _is_protected_collection(collection),
3205 },
3206 "documents": documents,
3207 "notes": notes,
3208 }
3210 except Exception as e:
3211 return handle_api_error("getting collection documents", e)
3214# =============================================================================
3215# Shared collection-indexing helpers
3216#
3217# Three routes index a collection: the single-collection SSE route
3218# (index_collection), the bulk SSE route (index_all), and the background worker
3219# (_background_index_worker). These helpers hold the logic that must stay
3220# identical across all three so it cannot drift again — the embedding metadata
3221# to persist, the force-reindex cleanup, and the query for documents to index.
3222# The callers differ only in how they report progress (SSE stream vs
3223# TaskMetadata) and so keep their own loops.
3224# =============================================================================
3227def _indexing_conflict(db_session, username, collection_id):
3228 """Check while holding the collection's admission lock."""
3229 key = (username, str(collection_id))
3230 conflict = {
3231 "success": False,
3232 "error": "Indexing is already in progress for this collection",
3233 }
3234 with _active_sse_indexers_lock:
3235 if _active_sse_indexers.get(key):
3236 return conflict
3237 task_id = _active_background_indexers.get(key)
3238 if task_id is not None:
3239 return {**conflict, "task_id": task_id}
3241 in_progress = (
3242 db_session.query(TaskMetadata)
3243 .filter(
3244 TaskMetadata.task_type == "indexing",
3245 TaskMetadata.status == "processing",
3246 )
3247 .all()
3248 )
3249 for task in in_progress:
3250 if (task.metadata_json or {}).get("collection_id") == collection_id:
3251 return {**conflict, "task_id": task.task_id}
3252 return None
3255def _register_sse_indexer(username, collection_id, db_password, cancel_event):
3256 """Admit an SSE worker atomically with SSE and background requests."""
3257 from ...database.session_context import get_user_db_session
3259 key = (username, str(collection_id))
3260 lock = _start_bg_index_locks.setdefault(key, threading.Lock())
3261 with lock, get_user_db_session(username, db_password) as db_session:
3262 if _indexing_conflict(db_session, username, collection_id):
3263 return False
3264 with _active_sse_indexers_lock:
3265 _active_sse_indexers.setdefault(key, set()).add(cancel_event)
3266 return True
3269def _unregister_sse_indexer(username, collection_id, cancel_event):
3270 """Release admission only once every worker belonging to this stream stopped."""
3271 key = (username, str(collection_id))
3272 with _active_sse_indexers_lock:
3273 events = _active_sse_indexers.get(key)
3274 if events is not None:
3275 events.discard(cancel_event)
3276 if not events:
3277 _active_sse_indexers.pop(key, None)
3280def _unregister_background_indexer(username, collection_id, task_id):
3281 key = (username, str(collection_id))
3282 with _active_sse_indexers_lock:
3283 if _active_background_indexers.get(key) == task_id: 3283 ↛ exitline 3283 didn't jump to the function exit
3284 _active_background_indexers.pop(key, None)
3287def _store_collection_embedding_metadata(collection, rag_service):
3288 """Persist the embedding/index configuration used to index a collection.
3290 Run on first incremental index. Forced rebuilds publish their metadata
3291 together with the replacement index. Includes the embedding dimension,
3292 probed by embedding a test string (the same way LibraryRAGService derives
3293 it for the RAGIndex row). Previously this read
3294 ``embedding_manager.provider.embedding_dimension``, an attribute the real
3295 LocalEmbeddingManager does not have, so the probe always failed and
3296 ``embedding_dimension`` was persisted as NULL via both indexing paths. Does
3297 not commit — the caller owns the transaction.
3298 """
3299 embedding_dim = None
3300 try:
3301 # Derive the dimension by embedding a test string — mirrors
3302 # LibraryRAGService.* which computes len(embed_query("test")) for the
3303 # RAGIndex row. The embedding manager has no usable dimension attribute.
3304 test_embedding = rag_service.embedding_manager.embeddings.embed_query(
3305 "test"
3306 )
3307 embedding_dim = len(test_embedding)
3308 except Exception:
3309 logger.debug("Could not determine embedding dimension", exc_info=True)
3311 collection.embedding_model = rag_service.embedding_model
3312 collection.embedding_model_type = EmbeddingProvider(
3313 rag_service.embedding_provider
3314 )
3315 collection.embedding_dimension = embedding_dim
3316 collection.chunk_size = rag_service.chunk_size
3317 collection.chunk_overlap = rag_service.chunk_overlap
3318 # Advanced settings
3319 collection.splitter_type = rag_service.splitter_type
3320 collection.text_separators = rag_service.text_separators
3321 collection.distance_metric = rag_service.distance_metric
3322 # Ensure normalize_vectors is a proper boolean for the database
3323 collection.normalize_vectors = bool(rag_service.normalize_vectors)
3324 collection.index_type = rag_service.index_type
3325 logger.info(
3326 f"Stored embedding metadata for collection: provider={rag_service.embedding_provider}"
3327 )
3330def _query_documents_to_index(db_session, collection_id, force_reindex):
3331 """Return the ``(DocumentCollection, Document)`` pairs to index.
3333 When not force-reindexing, only un-indexed documents are returned.
3334 """
3335 query = (
3336 db_session.query(DocumentCollection, Document)
3337 .join(Document)
3338 # Don't load the large text_content body: the indexing loop only
3339 # reads doc.id/filename/title (index_document re-fetches by id in a
3340 # worker thread), so loading every document's full text here just to
3341 # iterate is what exhausts memory on big collections (#4560).
3342 .options(defer(Document.text_content))
3343 .filter(DocumentCollection.collection_id == collection_id)
3344 )
3346 if not force_reindex:
3347 # IMPORTANT: DO NOT CHANGE THIS LINE! DO NOT REMOVE THIS COMMENT!
3348 # We MUST use `== False` instead of `not DocumentCollection.indexed`
3349 # The Python `not` operator does NOT work correctly in SQLAlchemy filters.
3350 # Using `not` will cause the query to return NO results (zero documents).
3351 # SQLAlchemy requires explicit comparison: `== False` or `== True`
3352 # This has been fixed multiple times - DO NOT change it back to `not`!
3353 query = query.filter(DocumentCollection.indexed == False) # noqa: E712
3355 return query.all()
3358@router.post("/api/collections/{collection_id}/index")
3359def index_collection(
3360 request: Request,
3361 collection_id,
3362 username: Annotated[str, Depends(require_auth)],
3363):
3364 """Index all documents in a collection with Server-Sent Events progress.
3366 POST, not GET, because it mutates — same reasoning as index-all:
3367 a first index writes embedding metadata, and it runs the
3368 embedding/index pipeline over every document in the collection
3369 (``?force_reindex=true`` stages a complete replacement and publishes
3370 it in one transaction, keeping the existing vectors until then), and
3371 CSRFMiddleware keys enforcement on the method, so the verb is what
3372 puts it behind the token check. A GET would sit outside the
3373 middleware entirely and be startable from same-site markup riding
3374 the victim's session cookie. Nothing in the shipped frontend called
3375 it — the history page's Index button POSTs COLLECTION_INDEX_START.
3376 Progress still streams as SSE; a fetch client can send both a POST
3377 and the header. Query parameters are unchanged."""
3378 from ...database.session_context import get_user_db_session
3379 from ...database.session_passwords import session_password_store
3381 # Parse as bool — a raw string "false" is truthy, which would silently
3382 # force a full reindex for every caller that asked to skip already-indexed
3383 # documents.
3384 force_reindex = (
3385 request.query_params.get("force_reindex", "false").lower() == "true"
3386 )
3387 session_id = request.session.get("session_id")
3389 logger.info(f"Starting index_collection, force_reindex={force_reindex}")
3391 # Get password for thread access to encrypted database
3392 db_password = None
3393 if session_id:
3394 db_password = session_password_store.get_session_password(
3395 username, session_id
3396 )
3398 # Create RAG service — on force reindex use current default model
3399 rag_service = get_rag_service(
3400 request, username, collection_id, use_defaults=force_reindex
3401 )
3402 logger.info(
3403 f"RAG service created: provider={rag_service.embedding_provider}"
3404 )
3406 # Import before building the SSE generator. Importing inside its try
3407 # would leave the exception handler unbound if this import failed.
3408 # Keep it below module scope because it pulls in the deletion package.
3409 from ...research_library.deletion.utils.cascade_helper import (
3410 RemoteVectorStoreDropError,
3411 remote_drop_failure_message,
3412 )
3414 # Resolve the parallel worker bound before the generator runs — the
3415 # parallel helper must never touch SettingsManager from a worker thread.
3416 with get_user_db_session(username, db_password) as _settings_session:
3417 _settings = get_settings_manager(_settings_session, username)
3418 try:
3419 _max_workers = int(
3420 _settings.get_setting("rag.indexing_max_parallel_docs", 4)
3421 )
3422 except SettingsValueError:
3423 raise
3424 except Exception:
3425 _max_workers = 4
3426 _max_workers = max(1, min(_max_workers, 16))
3428 def generate():
3429 """Generator for SSE progress updates."""
3430 logger.info("SSE generator started")
3431 _sse_cancel = threading.Event()
3432 worker_thread = None
3433 worker_finished: Future[None] = Future()
3434 try:
3435 if not _register_sse_indexer(
3436 username, collection_id, db_password, _sse_cancel
3437 ):
3438 yield f"data: {json.dumps({'type': 'error', 'error': 'Indexing is already in progress for this collection'})}\n\n"
3439 return
3440 # All setup DB work happens in a SHORT-LIVED session; the per-doc
3441 # ids/filenames are snapshotted into plain tuples so the long
3442 # indexing loop below (which yields progress + heartbeats) holds NO
3443 # session across a yield — see download_bulk for the Starlette/anyio
3444 # thread-affinity rationale. index_document opens its own session
3445 # per document in a worker thread and commits its own writes, so the
3446 # old outer db_session.commit()/flush() was vestigial.
3447 collection_found = True
3448 collection_name = None
3449 doc_info = []
3450 with get_user_db_session(username, db_password) as db_session:
3451 # Verify collection exists in this user's database
3452 collection = (
3453 db_session.query(Collection)
3454 .filter_by(id=collection_id)
3455 .first()
3456 )
3458 if not collection:
3459 collection_found = False
3460 else:
3461 collection_name = collection.name
3462 # Keep the published configuration during a rebuild.
3463 # Its replacement is committed with the complete new index.
3464 if collection.embedding_model is None and not force_reindex:
3465 _store_collection_embedding_metadata(
3466 collection, rag_service
3467 )
3468 db_session.commit()
3470 # Snapshot (id, display name) for each document to index.
3471 doc_info = [
3472 (doc.id, doc.filename or doc.title or "Unknown")
3473 for _link, doc in _query_documents_to_index(
3474 db_session, collection_id, force_reindex
3475 )
3476 ]
3478 if not collection_found:
3479 yield f"data: {json.dumps({'type': 'error', 'error': 'Collection not found'})}\n\n"
3480 return
3482 if not doc_info and not force_reindex:
3483 logger.info("No documents to index in collection")
3484 yield f"data: {json.dumps({'type': 'complete', 'results': {'successful': 0, 'skipped': 0, 'failed': 0, 'message': 'No documents to index'}})}\n\n"
3485 return
3487 total = len(doc_info)
3488 logger.info(f"Found {total} documents to index")
3489 results = {
3490 "successful": 0,
3491 "skipped": 0,
3492 "failed": 0,
3493 "errors": [],
3494 }
3496 yield f"data: {json.dumps({'type': 'start', 'message': f'Indexing {total} documents in collection: {collection_name}'})}\n\n"
3498 # Fan out indexing across a bounded worker pool. We run the
3499 # pool in a background thread so the SSE generator's main
3500 # thread can still yield heartbeats during long embedding
3501 # round-trips (matters for nginx / browser SSE timeout
3502 # behaviour). The progress_callback fired by the parallel
3503 # helper pushes per-document events into ``progress_queue``,
3504 # which the generator drains as they arrive.
3505 #
3506 # ``_max_workers`` is resolved from settings before the
3507 # generator starts (closure capture) so the parallel helper
3508 # never touches SettingsManager from inside a worker.
3509 progress_queue: queue.Queue = queue.Queue()
3510 _SENTINEL_COMPLETE = object()
3511 _SENTINEL_ERROR = object()
3513 def _on_progress(
3514 completed: int,
3515 total: int,
3516 title: str,
3517 status: str,
3518 ) -> None:
3519 progress_queue.put(
3520 (
3521 "progress",
3522 completed,
3523 total,
3524 title,
3525 status,
3526 )
3527 )
3529 def _run_parallel():
3530 try:
3531 # ``is_cancelled`` polls the SSE-disconnect event
3532 # so the helper can exit early between completions
3533 # when the client has gone away; the helper's
3534 # ``pool.shutdown(wait=True, ...)`` still drains
3535 # in-flight workers before returning so no worker
3536 # outlives the helper return — see ``index_documents_parallel`` for the
3537 # rationale.
3538 index_documents = (
3539 rag_service.rebuild_collection
3540 if force_reindex
3541 else rag_service.index_documents_parallel
3542 )
3543 aggregate = index_documents(
3544 doc_info,
3545 collection_id,
3546 max_workers=_max_workers,
3547 progress_callback=_on_progress,
3548 is_cancelled=_sse_cancel.is_set,
3549 )
3550 progress_queue.put((_SENTINEL_COMPLETE, aggregate))
3551 except Exception as exc:
3552 logger.exception(
3553 "Parallel indexing in SSE generator crashed"
3554 )
3555 progress_queue.put((_SENTINEL_ERROR, exc))
3556 finally:
3557 # Best-effort thread-local DB session cleanup so a
3558 # long-running worker batch doesn't accumulate
3559 # scoped sessions (#3194).
3560 try:
3561 from ...database.thread_local_session import (
3562 cleanup_current_thread,
3563 )
3565 cleanup_current_thread()
3566 except Exception:
3567 logger.debug(
3568 "best-effort thread-local DB session cleanup",
3569 exc_info=True,
3570 )
3571 finally:
3572 # All indexing writes and thread-local cleanup have
3573 # finished. A disconnected stream can release its
3574 # admission and service here without another thread.
3575 worker_finished.set_result(None)
3577 # Copy the request's contextvars so username-dependent
3578 # fallbacks (e.g. log attribution) inside the worker resolve
3579 # to the authenticated user rather than None.
3580 _par_ctx = contextvars.copy_context()
3581 worker_thread = threading.Thread(
3582 target=_par_ctx.run,
3583 args=(_run_parallel,),
3584 daemon=True,
3585 name="index-collection-parallel",
3586 )
3587 worker_thread.start()
3589 # Drain the queue: yield progress events as they arrive,
3590 # yield heartbeat comments on idle to keep SSE proxies
3591 # from timing out, exit on completion / error sentinels.
3592 heartbeat_interval = 5 # seconds
3593 _aggregate = None
3594 while True:
3595 try:
3596 item = progress_queue.get(timeout=heartbeat_interval)
3597 except queue.Empty:
3598 # No event in 5s. Heartbeat unless the worker
3599 # died without signalling (would hang forever).
3600 if worker_thread.is_alive():
3601 yield f": heartbeat {total}\n\n"
3602 continue
3603 logger.exception(
3604 "Indexer worker exited without completion "
3605 "signal; aborting SSE"
3606 )
3607 yield f"data: {json.dumps({'type': 'error', 'error': 'Indexer stopped unexpectedly'})}\n\n"
3608 return
3610 if item[0] is _SENTINEL_COMPLETE:
3611 _aggregate = item[1]
3612 break
3613 if item[0] is _SENTINEL_ERROR:
3614 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
3615 return
3617 _kind, completed, total, title, status = item
3618 yield f"data: {json.dumps({'type': 'progress', 'current': completed, 'total': total, 'filename': title, 'percent': int((completed / total) * 100)})}\n\n"
3620 if status == "error":
3621 # Per-doc error surfaced so the browser can show
3622 # failed documents inline. Scrub creds/keys here
3623 # (the full trace stays in server logs above).
3624 # Look the error up from the aggregate's errors
3625 # list — at this point _aggregate is None
3626 # (indexer still running), so we accept a brief
3627 # imprecision on the exact error text and emit
3628 # a generic doc_error; the final ``complete``
3629 # event below will list precise errors.
3630 yield f"data: {json.dumps({'type': 'doc_error', 'filename': title, 'error': 'Indexing failed'})}\n\n"
3632 if _aggregate is not None: 3632 ↛ 3652line 3632 didn't jump to line 3652 because the condition on line 3632 was always true
3633 results["successful"] = _aggregate["successful"]
3634 results["skipped"] = _aggregate["skipped"]
3635 results["failed"] = _aggregate["failed"]
3636 # Keep document identity for the UI, but never put an
3637 # extension's arbitrary error detail in the SSE event.
3638 results["errors"] = [
3639 {
3640 "filename": err.get("title"),
3641 "error": _client_indexing_error(err.get("error")),
3642 }
3643 for err in _aggregate.get("errors", [])
3644 ]
3645 logger.info(
3646 f"Indexing complete: "
3647 f"{results['successful']} successful, "
3648 f"{results['failed']} failed, "
3649 f"{results['skipped']} skipped"
3650 )
3652 yield f"data: {json.dumps({'type': 'complete', 'results': results})}\n\n"
3653 logger.info("SSE generator finished successfully")
3655 except RemoteVectorStoreDropError as exc:
3656 # Caught ahead of the catch-all below: this is an actionable
3657 # vector-store failure, not an internal error. Force-reindex rolls
3658 # back, keeping the RAGIndex rows (the remote collections' only
3659 # handle); the message says whether a retry can help and whether
3660 # some collections were already dropped (#5765). Logged from the
3661 # error's sanitized fields, with no traceback (a traceback would
3662 # add frame state; the backend's error text can carry secrets).
3663 drop_detail = exc.log_detail()
3664 logger.warning(
3665 "Force-reindex aborted: could not drop the vector-store "
3666 f"collection before rebuilding it: {drop_detail}"
3667 )
3668 drop_error = {
3669 "type": "error",
3670 "error": remote_drop_failure_message(
3671 exc, "indexing was aborted"
3672 ),
3673 }
3674 yield f"data: {json.dumps(drop_error)}\n\n"
3675 except Exception:
3676 logger.exception("Error in collection indexing")
3677 yield f"data: {json.dumps({'type': 'error', 'error': 'An internal error occurred during indexing'})}\n\n"
3678 finally:
3679 # Signal cancellation immediately, but retain admission until
3680 # every in-flight writer has stopped, including deferred drains.
3681 _sse_cancel.set()
3682 # On client disconnect this finally can run on the event loop
3683 # thread (generator close), so the drain must be bounded: give
3684 # in-flight workers a short grace period, then hand the
3685 # remaining cleanup to the existing worker's completion callback.
3686 # _sse_cancel is already set, so the parallel helper admits no
3687 # new work and terminates once in-flight docs finish.
3688 if worker_thread is not None and worker_thread.is_alive():
3689 worker_thread.join(timeout=5.0)
3690 if worker_thread is not None and worker_thread.is_alive():
3691 logger.info(
3692 "Indexer still draining after SSE close; deferring "
3693 "cleanup until the worker finishes"
3694 )
3696 def _close_after_worker(_finished):
3697 _unregister_sse_indexer(
3698 username, collection_id, _sse_cancel
3699 )
3700 safe_close(
3701 rag_service,
3702 "rag_service (index-collection SSE, deferred)",
3703 )
3705 # Future invokes the callback exactly once, even if the
3706 # worker finished between is_alive() and registration.
3707 # Starting a separate drain thread could fail and strand
3708 # admission after the indexing worker has already exited.
3709 worker_finished.add_done_callback(_close_after_worker)
3710 else:
3711 _unregister_sse_indexer(username, collection_id, _sse_cancel)
3712 safe_close(rag_service, "rag_service (index-collection SSE)")
3714 response = WorkerCleanupStreamingResponse(
3715 generate(), media_type="text/event-stream"
3716 )
3717 # Prevent buffering for proper SSE streaming
3718 response.headers["Cache-Control"] = "no-cache, no-transform"
3719 response.headers["Connection"] = "keep-alive"
3720 response.headers["X-Accel-Buffering"] = "no"
3721 return response
3724# =============================================================================
3725# Background Indexing Endpoints
3726# =============================================================================
3729def _get_rag_service_for_thread(
3730 collection_id: str,
3731 username: str,
3732 db_password: str,
3733 use_defaults: bool = False,
3734) -> LibraryRAGService:
3735 """
3736 Create RAG service for use in background threads (no Flask context).
3738 Delegates settings resolution to the shared rag_service_factory, then
3739 propagates db_password to the embedding manager for thread-safe DB access.
3740 """
3741 from ...research_library.services.rag_service_factory import (
3742 get_rag_service as _get_rag_service,
3743 )
3745 service = _get_rag_service(
3746 username,
3747 collection_id,
3748 use_defaults=use_defaults,
3749 db_password=db_password,
3750 )
3751 # The factory passes db_password to LibraryRAGService, but __init__ stores
3752 # it in the backing field (_db_password) without propagating to sub-managers.
3753 # Re-assign via the property setter to propagate to embedding_manager and
3754 # integrity_manager, which need it for thread-safe session access.
3755 service.db_password = db_password
3756 return service
3759def trigger_auto_index(
3760 document_ids: list[str],
3761 collection_id: str,
3762 username: str,
3763 db_password: str,
3764) -> None:
3765 """
3766 Trigger automatic RAG indexing for documents if auto-indexing is enabled.
3768 This function checks the auto_index_enabled setting and spawns a background
3769 thread to index the specified documents. It does not block the caller.
3771 Args:
3772 document_ids: List of document IDs to index
3773 collection_id: The collection to index into
3774 username: The username for database access
3775 db_password: The user's database password for thread-safe access
3776 """
3777 from ...database.session_context import get_user_db_session
3779 if not document_ids:
3780 logger.debug("No documents to auto-index")
3781 return
3783 # Check if auto-indexing is enabled
3784 try:
3785 with get_user_db_session(username, db_password) as db_session:
3786 settings = SettingsManager(db_session)
3787 auto_index_enabled = settings.get_bool_setting(
3788 "research_library.auto_index_enabled", True
3789 )
3791 if not auto_index_enabled:
3792 logger.debug("Auto-indexing is disabled, skipping")
3793 return
3795 try:
3796 _auto_max_workers = int(
3797 settings.get_setting("rag.indexing_max_parallel_docs", 4)
3798 )
3799 except SettingsValueError:
3800 raise
3801 except Exception:
3802 _auto_max_workers = 4
3803 _auto_max_workers = max(1, min(_auto_max_workers, 16))
3804 except Exception:
3805 logger.exception(
3806 "Failed to check auto-index setting, skipping auto-index"
3807 )
3808 return
3810 # Reserve a queue slot before submission. If the indexing queue is
3811 # saturated (too many uploads in flight), drop this auto-index job
3812 # rather than letting the executor's unbounded internal queue grow
3813 # without bound. The user can reindex manually later via the UI.
3814 if not _try_reserve_auto_index_slot():
3815 logger.warning(
3816 "Auto-index queue saturated ({}+ jobs pending); dropping "
3817 "auto-index for {} document(s) in collection {}. "
3818 "Documents are uploaded; trigger a manual reindex if needed.",
3819 _MAX_PENDING_AUTO_INDEX_JOBS,
3820 len(document_ids),
3821 collection_id,
3822 )
3823 return
3825 logger.info(
3826 f"Auto-indexing {len(document_ids)} documents in collection {collection_id}"
3827 )
3829 # Release the reserved slot EXACTLY ONCE per submission. CPython's
3830 # ThreadPoolExecutor.submit() enqueues the work item BEFORE it tries to
3831 # spin up a worker thread; if thread-start raises (e.g.
3832 # RuntimeError("can't start new thread")), the item is already queued, so
3833 # a live worker can run _wrapped_worker (releasing in its finally) AND the
3834 # except block below also fires. A per-call lock + flag makes the release
3835 # idempotent so that double-fire only releases once — otherwise the
3836 # counter under-counts in-flight jobs and erodes the OOM bound (the
3837 # max(0, ...) floor would silently hide the drift).
3838 _release_lock = threading.Lock()
3839 _slot_released = {"done": False}
3841 def _release_slot_once():
3842 with _release_lock:
3843 if _slot_released["done"]:
3844 return
3845 _slot_released["done"] = True
3846 _release_auto_index_slot()
3848 def _wrapped_worker(*args, **kwargs):
3849 try:
3850 _auto_index_documents_worker(*args, **kwargs)
3851 finally:
3852 _release_slot_once()
3854 # Submit to thread pool (bounded concurrency, prevents thread proliferation).
3855 # Build the executor inside the try too: if _get_auto_index_executor()
3856 # itself fails (OS thread/mutex exhaustion), the slot must still be
3857 # released — otherwise a reserved slot leaks permanently.
3858 try:
3859 executor = _get_auto_index_executor()
3860 executor.submit(
3861 _wrapped_worker,
3862 document_ids,
3863 collection_id,
3864 username,
3865 db_password,
3866 _auto_max_workers,
3867 )
3868 except Exception:
3869 # If submit itself fails (executor shutting down, OOM, etc.), the
3870 # wrapped worker may never run, so release the slot here. The release
3871 # is idempotent: if the work item was already enqueued and a worker
3872 # ran (and released) before thread-start failed, this is a no-op. Do
3873 # NOT re-raise: the upload has already been committed by the caller,
3874 # so propagating would turn a successful upload into a 500 (and prompt
3875 # the client to retry, creating duplicates). Auto-indexing is simply
3876 # skipped; the user can trigger a manual reindex later via the UI.
3877 _release_slot_once()
3878 logger.exception(
3879 "Failed to submit auto-index job for {} document(s) in "
3880 "collection {}; upload succeeded, auto-indexing skipped. "
3881 "Trigger a manual reindex if needed.",
3882 len(document_ids),
3883 collection_id,
3884 )
3887@thread_cleanup
3888def _auto_index_documents_worker(
3889 document_ids: list[str],
3890 collection_id: str,
3891 username: str,
3892 db_password: str,
3893 max_workers: int = 4,
3894) -> None:
3895 """
3896 Background worker to index documents automatically.
3898 This is a simpler worker than _background_index_worker - it doesn't track
3899 progress via TaskMetadata since it's meant to be a lightweight auto-indexing
3900 operation.
3902 Documents inside one call are indexed concurrently via
3903 :meth:`LibraryRAGService.index_documents_parallel`. The outer
3904 ``_auto_index_executor`` already parallelises separate jobs at the upload
3905 layer (one job per upload batch) and we additionally fan out inside each
3906 job, so a burst of uploads doesn't serialise within a single upload's
3907 document set.
3909 The ``max_workers`` parameter is resolved on the caller (which owns
3910 the Flask request context and therefore a real
3911 :func:`get_settings_manager` call). Background-thread workers must
3912 NOT call ``get_settings_manager()`` without an explicit
3913 ``db_session=`` — see pre-commit hook
3914 ``check-settings-manager-thread-safety`` and issue #3453.
3915 """
3917 try:
3918 # Create RAG service (thread-safe, no Flask context needed)
3919 with _get_rag_service_for_thread(
3920 collection_id, username, db_password
3921 ) as rag_service:
3922 # The auto-indexer doesn't surface per-doc titles to a user,
3923 # so we stub them with empty strings — the parallel helper
3924 # only reads titles for progress reporting.
3925 doc_info = [(doc_id, "") for doc_id in document_ids]
3926 aggregate = rag_service.index_documents_parallel(
3927 doc_info,
3928 collection_id,
3929 force_reindex=False,
3930 max_workers=max(1, min(int(max_workers), 16)),
3931 )
3932 logger.info(
3933 f"Auto-indexing complete: {aggregate['successful']}"
3934 f"/{len(document_ids)} documents indexed "
3935 f"(skipped={aggregate['skipped']}, "
3936 f"failed={aggregate['failed']})"
3937 )
3939 except Exception:
3940 logger.exception("Auto-indexing worker failed")
3943# Fixed per-document guidance that LibraryRAGService returns as a result's
3944# "error". An exact match reaches the client; _client_indexing_error replaces
3945# any other text. TestIndexingErrorAllowlistMatchesService in
3946# tests/security/test_library_rag_security_fastapi.py checks this set, and
3947# _EMPTY_VECTOR_STORE_REASON below, against the text the service builds.
3948_SAFE_INDEXING_ERRORS = frozenset(
3949 {
3950 "Document not found",
3951 "Document not found in collection",
3952 "Document has no text content",
3953 "Document deleted during indexing",
3954 "Document changed; retry indexing",
3955 "Collection configuration changed; retry indexing",
3956 }
3957)
3958_EMPTY_VECTOR_STORE_REASON = (
3959 "live vector store reported zero ids while DocumentChunk rows exist; "
3960 "refusing to clear indexed flags"
3961)
3962_COLLECTION_NOT_FOUND = "Collection not found"
3963_REMOTE_DROP_OUTCOME = "indexing was aborted"
3964# Appended when a cancel arrived after a rebuild began publishing.
3965_LATE_CANCEL_NOTE = (
3966 "The cancel request arrived after the rebuild began publishing, "
3967 "so the new index is in use."
3968)
3969# Codes for ``metadata_json["failure"]``. The indexing worker records one next
3970# to an ``error_message`` it built only from fixed phrases and counts, and the
3971# status endpoint rebuilds that message from the record instead of serving the
3972# stored text.
3973_FAILURE_COLLECTION_NOT_FOUND = "collection_not_found"
3974_FAILURE_REMOTE_DROP = "remote_vector_store_drop"
3977def _client_indexing_error(error: Any) -> str:
3978 """Return only messages known to be safe for indexing clients.
3980 Service extensions and exception handlers can supply arbitrary error text.
3981 Credential-shape scrubbing does not remove paths, SQL, or provider URLs.
3982 """
3983 if error is None or (isinstance(error, str) and not error):
3984 return "Indexing failed"
3985 if isinstance(error, str) and error in _SAFE_INDEXING_ERRORS:
3986 return error
3987 return "Indexing failed. Check the server logs for details."
3990def _client_reconciliation_reason(reason: Any) -> str:
3991 """Preserve the known durable-state warning, never arbitrary detail."""
3992 if isinstance(reason, str) and reason == _EMPTY_VECTOR_STORE_REASON:
3993 return _EMPTY_VECTOR_STORE_REASON
3994 return (
3995 "Could not verify the vector store. Check the server logs for details."
3996 )
3999def _reconciliation_skipped_message(reason: str) -> str:
4000 """Task error for a skipped reconciliation, given a client-safe reason.
4002 The worker stores it and the status endpoint rebuilds it, both through
4003 this function, so the two texts cannot drift apart.
4004 """
4005 return (
4006 f"Reconciliation skipped: {reason.rstrip('.')}. Indexed flags "
4007 "preserved; verify the FAISS index before retrying."
4008 )
4011def _reconciled_failures_message(failed: int) -> str:
4012 """Task error when documents failed and reconciliation then completed."""
4013 return (
4014 f"{failed} document(s) failed to index. The collection was "
4015 "reconciled to the durable vector store."
4016 )
4019def _remote_drop_failure(exc: Any) -> dict:
4020 """The failure record for a force-reindex whose remote drop failed.
4022 It keeps only the two fields ``remote_drop_failure_message`` reads: a
4023 flag and a count. The error's own text is never recorded. A count that is
4024 not a non-negative int is stored as None, so the status endpoint withholds
4025 the message rather than rebuild it from a bad value.
4026 """
4027 dropped = exc.dropped
4028 return {
4029 "code": _FAILURE_REMOTE_DROP,
4030 "retryable": bool(exc.retryable),
4031 "dropped": dropped if type(dropped) is int and dropped >= 0 else None,
4032 }
4035def _client_task_failure_message(failure: Any) -> Optional[str]:
4036 """Rebuild the client text for a worker failure record.
4038 Returns None for anything else: no record, an unknown code, or a field of
4039 the wrong type. The caller then withholds the stored message.
4040 """
4041 if not isinstance(failure, dict):
4042 return None
4043 code = failure.get("code")
4044 if code == _FAILURE_COLLECTION_NOT_FOUND:
4045 return _COLLECTION_NOT_FOUND
4046 if code == _FAILURE_REMOTE_DROP:
4047 retryable = failure.get("retryable")
4048 dropped = failure.get("dropped")
4049 if type(retryable) is bool and type(dropped) is int and dropped >= 0:
4050 from ...research_library.deletion.utils.cascade_helper import (
4051 RemoteVectorStoreDropError,
4052 remote_drop_failure_message,
4053 )
4055 # The same template the worker used, fed only the recorded
4056 # flag and count.
4057 return remote_drop_failure_message(
4058 RemoteVectorStoreDropError(
4059 "recorded remote-drop failure",
4060 retryable=retryable,
4061 dropped=dropped,
4062 ),
4063 _REMOTE_DROP_OUTCOME,
4064 )
4065 return None
4068def _sanitized_indexing_errors(results: dict, limit: int = 50) -> list:
4069 """Return a bounded list of client-safe per-document errors."""
4070 errors = results.get("errors")
4071 if not isinstance(errors, list):
4072 return []
4073 return [
4074 {
4075 "doc_id": item.get("doc_id"),
4076 "title": item.get("title"),
4077 "error": _client_indexing_error(item.get("error")),
4078 }
4079 for item in errors[: max(0, limit)]
4080 if isinstance(item, dict)
4081 ]
4084def _client_indexing_summary(results: dict) -> dict:
4085 """Keep only the non-negative integer counts and the ``cancelled`` flag.
4087 Every other key is dropped.
4088 """
4089 summary = {}
4090 for key in ("successful", "skipped", "failed", "total"):
4091 value = results.get(key)
4092 if type(value) is int and value >= 0:
4093 summary[key] = value
4094 if type(results.get("cancelled")) is bool: 4094 ↛ 4095line 4094 didn't jump to line 4095 because the condition on line 4094 was never true
4095 summary["cancelled"] = results["cancelled"]
4096 return summary
4099def _client_indexing_result(result: Any) -> Optional[dict]:
4100 """Filter old and new task rows at the HTTP boundary."""
4101 if not isinstance(result, dict):
4102 return None
4103 safe = _client_indexing_summary(result)
4104 for key in (
4105 "durable_indexed_documents",
4106 "durable_indexed_chunks",
4107 "live_vectors",
4108 "orphan_vectors",
4109 ):
4110 value = result.get(key)
4111 if type(value) is int and value >= 0:
4112 safe[key] = value
4113 if result.get("reconciliation_skipped") is True:
4114 safe["reconciliation_skipped"] = True
4115 safe["reconciliation_reason"] = _client_reconciliation_reason(
4116 result.get("reconciliation_reason")
4117 )
4118 if "errors" in result: 4118 ↛ 4120line 4118 didn't jump to line 4120 because the condition on line 4118 was always true
4119 safe["errors"] = _sanitized_indexing_errors(result)
4120 return safe
4123def _client_indexing_task_error(
4124 message: Any, result: Optional[dict], failure: Any = None
4125) -> Optional[str]:
4126 """Rebuild known task errors rather than trusting persisted text.
4128 The client text comes from the worker's failure record (``failure``) or
4129 from the result counts (``result``, already filtered by
4130 ``_client_indexing_result``). Any other stored message gets the generic
4131 text. That includes rows written before these rules, whose text can hold
4132 exception detail.
4134 A stored message may be served unchanged only through an exact match on
4135 text that this module writes to an indexing task row.
4136 ``TestIndexingTaskErrorPassThroughs`` in
4137 tests/security/test_library_rag_security_fastapi.py fails on a match
4138 that nothing writes, and on fixed text written without a failure record
4139 that is neither matched nor listed there as withheld.
4140 """
4141 if message is None:
4142 return None
4143 rebuilt = _client_task_failure_message(failure)
4144 if rebuilt is not None:
4145 return rebuilt
4146 if message == "Failed to start indexing. Please try again.":
4147 return message
4148 if result is not None:
4149 if result.get("reconciliation_skipped") is True:
4150 return _reconciliation_skipped_message(
4151 _client_reconciliation_reason(
4152 result.get("reconciliation_reason")
4153 )
4154 )
4155 failed = result.get("failed")
4156 if type(failed) is int and failed > 0: 4156 ↛ 4160line 4156 didn't jump to line 4160 because the condition on line 4156 was always true
4157 if "durable_indexed_documents" in result:
4158 return _reconciled_failures_message(failed)
4159 return f"{failed} document(s) failed to index. Check the server logs for details."
4160 return "Indexing failed. Check the server logs for details."
4163@thread_cleanup
4164def _background_index_worker(
4165 task_id: str,
4166 collection_id: str,
4167 username: str,
4168 db_password: str,
4169 force_reindex: bool,
4170 max_workers: int = 4,
4171 settings_snapshot: Optional[dict] = None,
4172):
4173 """
4174 Background worker thread for indexing documents.
4175 Updates TaskMetadata with progress and checks for cancellation.
4177 The ``max_workers`` parameter bounds the per-job fan-out into
4178 :meth:`LibraryRAGService.index_documents_parallel`. It must be
4179 resolved on the route (which has Flask app context), not inside
4180 this worker — background threads cannot call
4181 :func:`get_settings_manager` safely (#3453, pre-commit hook
4182 ``check-settings-manager-thread-safety``).
4184 ``settings_snapshot`` remains accepted for caller compatibility. Staged
4185 rebuilds currently require exclusively local-file vector-store providers;
4186 remote providers need a separate atomic publication protocol.
4187 """
4188 from ...database.session_context import get_user_db_session
4189 from ...research_library.deletion.utils.cascade_helper import (
4190 RemoteVectorStoreDropError,
4191 remote_drop_failure_message,
4192 )
4194 try:
4195 # Create RAG service (thread-safe, no Flask context needed)
4196 with _get_rag_service_for_thread(
4197 collection_id, username, db_password, use_defaults=force_reindex
4198 ) as rag_service:
4199 with get_user_db_session(username, db_password) as db_session:
4200 # Get collection
4201 collection = (
4202 db_session.query(Collection)
4203 .filter_by(id=collection_id)
4204 .first()
4205 )
4207 if not collection:
4208 _update_task_status(
4209 username,
4210 db_password,
4211 task_id,
4212 status="failed",
4213 error_message=_COLLECTION_NOT_FOUND,
4214 failure={"code": _FAILURE_COLLECTION_NOT_FOUND},
4215 )
4216 return
4218 # A rebuild retains the published configuration until its
4219 # complete replacement can be committed atomically.
4220 if collection.embedding_model is None and not force_reindex:
4221 _store_collection_embedding_metadata(
4222 collection, rag_service
4223 )
4224 db_session.commit()
4226 # Keep only scalar values after the setup scope exits. The
4227 # parallel coordinator owns its per-document write scopes.
4228 doc_info = [
4229 (doc.id, doc.filename or doc.title or "Unknown")
4230 for _link, doc in _query_documents_to_index(
4231 db_session, collection_id, force_reindex
4232 )
4233 ]
4235 if not doc_info and not force_reindex:
4236 _update_task_status(
4237 username,
4238 db_password,
4239 task_id,
4240 status="completed",
4241 progress_message="No documents to index",
4242 )
4243 return
4245 total = len(doc_info)
4246 results = {"successful": 0, "skipped": 0, "failed": 0}
4248 # Update task with total count
4249 _update_task_status(
4250 username,
4251 db_password,
4252 task_id,
4253 progress_total=total,
4254 progress_message=f"Indexing {total} documents",
4255 )
4257 # The caller (start_background_index, which owns the
4258 # Flask request context) resolved ``max_workers`` from
4259 # ``rag.indexing_max_parallel_docs`` and forwarded it
4260 # here. Workers must NOT call ``get_settings_manager()``
4261 # themselves — no Flask app context, pre-commit
4262 # ``check-settings-manager-thread-safety`` would reject,
4263 # and #3453 documents the data-loss shape that produces.
4264 _max_workers = max(1, min(int(max_workers), 16))
4266 def _on_progress(
4267 completed: int,
4268 total: int,
4269 title: str,
4270 status: str,
4271 ) -> None:
4272 _update_task_status(
4273 username,
4274 db_password,
4275 task_id,
4276 progress_current=completed,
4277 progress_message=(f"Indexing {completed}/{total}: {title}"),
4278 )
4280 def _is_cancelled() -> bool:
4281 return _is_task_cancelled(username, db_password, task_id)
4283 index_documents = (
4284 rag_service.rebuild_collection
4285 if force_reindex
4286 else rag_service.index_documents_parallel
4287 )
4288 aggregate = index_documents(
4289 doc_info,
4290 collection_id,
4291 max_workers=_max_workers,
4292 progress_callback=_on_progress,
4293 is_cancelled=_is_cancelled,
4294 )
4296 results["successful"] = aggregate["successful"]
4297 results["skipped"] = aggregate["skipped"]
4298 results["failed"] = aggregate["failed"]
4299 results["errors"] = aggregate["errors"]
4301 if aggregate["cancelled"]:
4302 completed = (
4303 aggregate["successful"]
4304 + aggregate["skipped"]
4305 + aggregate["failed"]
4306 )
4307 _update_task_status(
4308 username,
4309 db_password,
4310 task_id,
4311 status="cancelled",
4312 progress_message=(
4313 f"Cancelled after {completed}/{total} documents"
4314 ),
4315 )
4316 logger.info(
4317 f"Indexing task {task_id} was cancelled "
4318 f"after {completed}/{total} documents"
4319 )
4320 return
4322 # Wrap reconciliation so a transient store/DB error does not
4323 # leave the task in "processing" forever (the previous code
4324 # let exceptions escape, so _update_task_status never ran and
4325 # the UI polled indefinitely — see PR #5235 review comment
4326 # 5085604502). On failure we still record a terminal status
4327 # so cleanup_old_tasks can reap the row.
4328 try:
4329 reconciliation = rag_service.reconcile_collection_index(
4330 collection_id
4331 )
4332 except Exception as exc:
4333 logger.exception(
4334 f"Background indexing task {task_id}: reconciliation failed"
4335 )
4336 reconciliation = {
4337 "indexed_documents": results["successful"],
4338 "indexed_chunks": 0,
4339 "live_vectors": 0,
4340 "orphan_vectors": 0,
4341 "reconciliation_skipped": True,
4342 "reconciliation_reason": _client_reconciliation_reason(exc),
4343 }
4344 if not isinstance(reconciliation, dict):
4345 reconciliation = {
4346 "indexed_documents": results["successful"],
4347 "indexed_chunks": 0,
4348 "live_vectors": 0,
4349 "orphan_vectors": 0,
4350 }
4351 # The reconciler returns ``reconciliation_skipped=True`` when
4352 # live_ids is empty but chunk rows exist (refuse-to-shrink
4353 # guard). Surface this as a task failure so the user sees the
4354 # durable-state inconsistency instead of a misleading
4355 # "0 indexed" success.
4356 if isinstance(reconciliation, dict) and reconciliation.get(
4357 "reconciliation_skipped"
4358 ):
4359 reason = reconciliation.get("reconciliation_reason") or (
4360 reconciliation.get("reason")
4361 or "stored vectors could not be loaded"
4362 )
4363 sanitized_reason = _client_reconciliation_reason(reason)
4364 logger.warning(
4365 f"Background indexing task {task_id}: reconciliation "
4366 f"skipped — {sanitize_error_message(str(reason))}"
4367 )
4368 _update_task_status(
4369 username,
4370 db_password,
4371 task_id,
4372 status="failed",
4373 progress_current=total,
4374 progress_message=(
4375 f"Completed: {results['successful']} indexed, "
4376 f"{results['failed']} failed, "
4377 f"{results['skipped']} skipped; reconciliation "
4378 f"skipped: {sanitized_reason}"
4379 ),
4380 error_message=_reconciliation_skipped_message(
4381 sanitized_reason
4382 ),
4383 result_metadata={
4384 **_client_indexing_summary(results),
4385 # Omit durable_indexed_documents/chunks and
4386 # live_vectors/orphan_vectors: true durable state is
4387 # unverified on the skipped path, and the UI
4388 # suppresses its "Durable vector store: ..." sentence
4389 # when these keys are absent (``Number.isInteger``
4390 # on undefined returns false). Reporting zeros here
4391 # would imply data loss where state is merely
4392 # unverified — see PR #5235 review comment
4393 # 5085604502.
4394 "reconciliation_skipped": True,
4395 "reconciliation_reason": sanitized_reason,
4396 "errors": _sanitized_indexing_errors(results),
4397 },
4398 published_rebuild=force_reindex,
4399 )
4400 return
4401 durable_documents = reconciliation["indexed_documents"]
4402 durable_chunks = reconciliation["indexed_chunks"]
4403 task_result = {
4404 **_client_indexing_summary(results),
4405 "durable_indexed_documents": durable_documents,
4406 "durable_indexed_chunks": durable_chunks,
4407 "live_vectors": reconciliation["live_vectors"],
4408 "orphan_vectors": reconciliation["orphan_vectors"],
4409 "errors": _sanitized_indexing_errors(results),
4410 }
4411 has_failures = results["failed"] > 0
4412 status = "failed" if has_failures else "completed"
4413 message = (
4414 f"Completed: {results['successful']} indexed, "
4415 f"{results['failed']} failed, {results['skipped']} skipped; "
4416 f"durable vector store: {durable_documents} documents, "
4417 f"{durable_chunks} chunks"
4418 )
4419 _update_task_status(
4420 username,
4421 db_password,
4422 task_id,
4423 status=status,
4424 progress_current=total,
4425 progress_message=message,
4426 error_message=(
4427 _reconciled_failures_message(results["failed"])
4428 if has_failures
4429 else None
4430 ),
4431 result_metadata=task_result,
4432 # A forced rebuild only gets here after it has published.
4433 published_rebuild=force_reindex,
4434 )
4435 logger.info(
4436 f"Background indexing task {task_id} completed: {results}"
4437 )
4439 except RemoteVectorStoreDropError as exc:
4440 # Caught ahead of the catch-all below: this is an actionable
4441 # vector-store failure, not an internal error (see index_all's
4442 # handler), and logged the same way.
4443 drop_detail = exc.log_detail()
4444 logger.warning(
4445 f"Background indexing task {task_id}: force-reindex aborted, "
4446 f"could not drop the vector-store collection before rebuilding "
4447 f"it: {drop_detail}"
4448 )
4449 _update_task_status(
4450 username,
4451 db_password,
4452 task_id,
4453 status="failed",
4454 error_message=remote_drop_failure_message(
4455 exc, _REMOTE_DROP_OUTCOME
4456 ),
4457 # Lets the status endpoint rebuild this fixed-phrase message.
4458 failure=_remote_drop_failure(exc),
4459 )
4460 except Exception as e:
4461 logger.exception(f"Background indexing task {task_id} failed")
4462 _update_task_status(
4463 username,
4464 db_password,
4465 task_id,
4466 status="failed",
4467 # Always the generic text for an exception, which is also what
4468 # the status endpoint serves for a message it cannot rebuild. The
4469 # full exception stays in the server log above.
4470 error_message=_client_indexing_error(e),
4471 )
4474def _is_database_locked_error(exc: BaseException) -> bool:
4475 """Return True if ``exc`` looks like a SQLite ``database is locked`` error.
4477 Background indexing can hit transient ``database is locked`` errors when
4478 SQLite is contended; those are recoverable via a short backoff retry.
4479 Anything else (connection failure, integrity error, programming mistake)
4480 should surface immediately so the outer exception handler marks the task
4481 as failed and sets ``completed_at`` for reaping.
4482 """
4483 return "database is locked" in str(exc).lower()
4486@retry(
4487 stop=stop_after_attempt(5),
4488 wait=wait_exponential(multiplier=0.05),
4489 retry=retry_if_exception(_is_database_locked_error),
4490)
4491def _do_update_task_status(
4492 username: str,
4493 db_password: str,
4494 task_id: str,
4495 status: str = None,
4496 progress_current: int = None,
4497 progress_total: int = None,
4498 progress_message: str = None,
4499 error_message: str = None,
4500 result_metadata: dict = None,
4501 failure: Optional[dict] = None,
4502 published_rebuild: bool = False,
4503):
4504 """Perform the task-status update under a tenacity-managed retry loop.
4506 Only SQLite ``database is locked`` errors are retried (up to 5 attempts
4507 with 0.05s exponential backoff); any other exception escapes so the
4508 outer caller can log it without retrying. See PR #5235 review comment
4509 3669857779.
4511 ``result_metadata`` is stored as ``metadata_json["result"]`` and
4512 ``failure`` as ``metadata_json["failure"]`` (see
4513 ``_client_task_failure_message``).
4515 ``published_rebuild`` marks the final update of a forced rebuild whose
4516 replacement index has been committed. Only that update may replace
4517 ``cancelled``: a cancel that arrives after publication starts cannot
4518 stop it, and ``cancelled`` would report the previous index as kept.
4519 """
4520 from ...database.session_context import get_user_db_session
4522 with get_user_db_session(username, db_password) as db_session:
4523 task_query = db_session.query(TaskMetadata).filter_by(task_id=task_id)
4524 # Serialize the guard and update across cancellation and worker
4525 # connections. A SELECT alone does not start SQLite's deferred write
4526 # transaction, and a nested cancellation scope may cache an old row.
4527 task_query.update(
4528 {TaskMetadata.task_id: TaskMetadata.task_id},
4529 synchronize_session=False,
4530 )
4531 task = task_query.first()
4532 if task:
4533 db_session.refresh(task)
4534 if (
4535 published_rebuild
4536 and task.status == "cancelled"
4537 and status in ("completed", "failed")
4538 ):
4539 progress_message = " ".join(
4540 filter(None, (progress_message, _LATE_CANCEL_NOTE))
4541 )
4542 # Preserve terminal states: once completed, cancelled or failed, only
4543 # updates confirming that same state are allowed.
4544 elif task.status in ("completed", "cancelled", "failed") and (
4545 status != task.status
4546 ):
4547 db_session.commit()
4548 return
4549 elif status == "completed" and task.status != "processing":
4550 db_session.commit()
4551 return
4552 if status is not None:
4553 task.status = status
4554 # Set completed_at for ALL terminal statuses so
4555 # ``cleanup_old_tasks`` (which filters on
4556 # ``status in ["completed", "failed", "cancelled"] AND
4557 # completed_at < cutoff_date``) can reap them.
4558 # Previously only "completed" set the timestamp,
4559 # leaving failed/cancelled rows permanent.
4560 if status in ("completed", "failed", "cancelled"): 4560 ↛ 4562line 4560 didn't jump to line 4562 because the condition on line 4560 was always true
4561 task.completed_at = datetime.now(UTC)
4562 if progress_current is not None:
4563 task.progress_current = progress_current
4564 if progress_total is not None:
4565 task.progress_total = progress_total
4566 if progress_message is not None:
4567 task.progress_message = progress_message
4568 if error_message is not None:
4569 task.error_message = error_message
4570 if result_metadata is not None or failure is not None:
4571 metadata = dict(task.metadata_json or {})
4572 if result_metadata is not None:
4573 metadata["result"] = result_metadata
4574 if failure is not None:
4575 metadata["failure"] = failure
4576 task.metadata_json = metadata
4577 db_session.commit()
4580def _update_task_status(
4581 username: str,
4582 db_password: str,
4583 task_id: str,
4584 status: str = None,
4585 progress_current: int = None,
4586 progress_total: int = None,
4587 progress_message: str = None,
4588 error_message: str = None,
4589 result_metadata: dict = None,
4590 failure: Optional[dict] = None,
4591 published_rebuild: bool = False,
4592):
4593 """Update task metadata in the database."""
4594 try:
4595 _do_update_task_status(
4596 username,
4597 db_password,
4598 task_id,
4599 status=status,
4600 progress_current=progress_current,
4601 progress_total=progress_total,
4602 progress_message=progress_message,
4603 error_message=error_message,
4604 result_metadata=result_metadata,
4605 failure=failure,
4606 published_rebuild=published_rebuild,
4607 )
4608 except Exception:
4609 logger.exception(f"Failed to update task status for {task_id}")
4612def _is_task_cancelled(username: str, db_password: str, task_id: str) -> bool:
4613 """Check if a task has been cancelled."""
4614 from ...database.session_context import get_user_db_session
4616 try:
4617 with get_user_db_session(username, db_password) as db_session:
4618 task = (
4619 db_session.query(TaskMetadata)
4620 .filter_by(task_id=task_id)
4621 .first()
4622 )
4623 return task and task.status == "cancelled"
4624 except Exception:
4625 logger.warning(
4626 "Could not check cancellation status for task {}", task_id
4627 )
4628 return False
4631@router.post("/api/collections/{collection_id}/index/start")
4632async def start_background_index(
4633 request: Request,
4634 collection_id,
4635 username: Annotated[str, Depends(require_auth)],
4636):
4637 """Start background indexing for a collection."""
4638 from ...database.session_passwords import session_password_store
4640 session_id = request.session.get("session_id")
4642 # Get password for thread access
4643 db_password = None
4644 if session_id:
4645 db_password = session_password_store.get_session_password(
4646 username, session_id
4647 )
4649 # The shared parser rejects both malformed and non-object bodies before
4650 # they reach the indexing controls below.
4651 data, body_error = await read_json_dict(
4652 request, "success", "Request body must be valid JSON"
4653 )
4654 if body_error is not None:
4655 return body_error
4656 force_reindex = data.get("force_reindex", False)
4657 if not isinstance(force_reindex, bool):
4658 return JSONResponse(
4659 {
4660 "success": False,
4661 "error": "force_reindex must be a boolean",
4662 },
4663 status_code=400,
4664 )
4666 # The lock acquisition and SQLCipher session open below are blocking —
4667 # run the whole check-and-create + thread spawn in the threadpool so a
4668 # contended lock or a cold-engine PBKDF2 derivation can't stall the
4669 # event loop (contextvars propagate through to_thread, so the
4670 # copy_context() for the worker still sees this request's user).
4671 return await run_db_sync(
4672 _start_background_index_sync,
4673 collection_id,
4674 username,
4675 db_password,
4676 force_reindex,
4677 )
4680def _start_background_index_sync(
4681 collection_id, username, db_password, force_reindex
4682):
4683 from ...database.session_context import get_user_db_session
4685 # Serialize check-and-create per (user, collection) to close the TOCTOU
4686 # window between scanning for an in-progress task and inserting a new
4687 # row. Two concurrent threads would otherwise both see no match and
4688 # both spawn an indexer on the same collection, racing FAISS writes.
4689 # Single-worker scope only (matches the rest of this module's locking
4690 # model; multi-worker deployments already documented as unsupported
4691 # for this kind of coordination).
4692 lock_key = (username, str(collection_id))
4693 lock = _start_bg_index_locks.setdefault(lock_key, threading.Lock())
4695 task_id = None
4696 try:
4697 with lock, get_user_db_session(username, db_password) as db_session:
4698 conflict = _indexing_conflict(db_session, username, collection_id)
4699 if conflict:
4700 return JSONResponse(conflict, status_code=409)
4702 # Preserve the caller's provider snapshot for force-reindex.
4703 # Staged rebuilds currently refuse remote providers, but the
4704 # background-worker interface still accepts this context.
4705 # Captured here, on the request-scoped session, because
4706 # the worker thread must not build one -- and BEFORE the task row
4707 # is committed, so a failure cannot leave a stuck 'processing'
4708 # task behind. Read strictly and not swallowed: a defaults-only
4709 # snapshot (or None) would silently resolve the global default
4710 # provider instead of the configured backend (see
4711 # read_vector_cleanup_settings_snapshot).
4712 from ...research_library.deletion.utils.cascade_helper import (
4713 read_vector_cleanup_settings_snapshot,
4714 )
4716 _bg_settings_snapshot = (
4717 read_vector_cleanup_settings_snapshot(
4718 get_settings_manager(db_session, username)
4719 )
4720 if force_reindex
4721 else None
4722 )
4724 # Resolve per-job parallel fan-out before committing the task
4725 # (request-scoped session) and forward into the worker, which
4726 # runs on a background thread where the no-arg
4727 # ``get_settings_manager()`` is unsafe (#3453).
4728 try:
4729 _bg_settings = get_settings_manager(db_session, username)
4730 _bg_max_workers = int(
4731 _bg_settings.get_setting(
4732 "rag.indexing_max_parallel_docs", 4
4733 )
4734 )
4735 except SettingsValueError:
4736 raise
4737 except Exception:
4738 _bg_max_workers = 4
4739 _bg_max_workers = max(1, min(_bg_max_workers, 16))
4741 # Create new task
4742 task_id = str(uuid.uuid4())
4743 task = TaskMetadata(
4744 task_id=task_id,
4745 status="processing",
4746 task_type="indexing",
4747 created_at=datetime.now(UTC),
4748 started_at=datetime.now(UTC),
4749 progress_current=0,
4750 progress_total=0,
4751 progress_message="Starting collection rebuild..."
4752 if force_reindex
4753 else "Starting indexing...",
4754 metadata_json={
4755 "collection_id": collection_id,
4756 "force_reindex": force_reindex,
4757 },
4758 )
4759 db_session.add(task)
4760 db_session.commit()
4761 with _active_sse_indexers_lock:
4762 _active_background_indexers[lock_key] = task_id
4764 # Start background thread. Copy the request's contextvars so
4765 # any service code inside the worker that calls
4766 # `get_current_username()` sees the actual user instead of None.
4767 _bg_ctx = contextvars.copy_context()
4769 def _ctx_worker():
4770 try:
4771 _bg_ctx.run(
4772 _background_index_worker,
4773 task_id,
4774 collection_id,
4775 username,
4776 db_password,
4777 force_reindex,
4778 _bg_max_workers,
4779 settings_snapshot=_bg_settings_snapshot,
4780 )
4781 finally:
4782 _unregister_background_indexer(username, collection_id, task_id)
4784 thread = threading.Thread(target=_ctx_worker, daemon=True)
4785 thread.start()
4787 logger.info(
4788 f"Started background indexing task {task_id} for collection {_log_value_preview(collection_id)}"
4789 )
4791 return {
4792 "success": True,
4793 "task_id": task_id,
4794 "message": "Indexing started in background",
4795 }
4797 except Exception:
4798 logger.exception("Failed to start background indexing")
4799 if task_id is not None:
4800 _update_task_status(
4801 username,
4802 db_password,
4803 task_id,
4804 status="failed",
4805 error_message="Failed to start indexing. Please try again.",
4806 )
4807 _unregister_background_indexer(username, collection_id, task_id)
4808 return JSONResponse(
4809 {
4810 "success": False,
4811 "error": "Failed to start indexing. Please try again.",
4812 },
4813 status_code=500,
4814 )
4817@router.get("/api/collections/{collection_id}/index/status")
4818@limiter.exempt
4819def get_index_status(
4820 request: Request,
4821 collection_id,
4822 username: Annotated[str, Depends(require_auth)],
4823):
4824 """Get the current indexing status for a collection."""
4825 from ...database.session_context import get_user_db_session
4826 from ...database.session_passwords import session_password_store
4828 session_id = request.session.get("session_id")
4830 db_password = None
4831 if session_id:
4832 db_password = session_password_store.get_session_password(
4833 username, session_id
4834 )
4836 # Observe admission before opening the read transaction. If the worker
4837 # finishes between these reads, report active for one extra poll rather
4838 # than combine an old cancelled task snapshot with a now-inactive worker.
4839 with _active_sse_indexers_lock:
4840 active_task_id = _active_background_indexers.get(
4841 (username, str(collection_id))
4842 )
4844 try:
4845 with get_user_db_session(username, db_password) as db_session:
4846 # Find the most recent indexing task FOR THIS COLLECTION.
4847 #
4848 # collection_id is stored inside the metadata_json JSON column, so
4849 # we can't portably filter on it in SQL (SQLite/SQLCipher JSON
4850 # support varies). Instead, scan recent indexing tasks newest-first
4851 # and return the first whose metadata.collection_id matches. This
4852 # is scoped per collection: a newer indexing task for a DIFFERENT
4853 # collection no longer makes this one falsely report "idle".
4854 #
4855 # This endpoint is polled every ~2s during a reindex and the
4856 # indexing-task table is never pruned, so an unbounded .all() would
4857 # materialize the entire history on every poll (and task_type /
4858 # created_at are unindexed). Bound the scan to the newest N tasks.
4859 # The task we're looking for was just created by the reindex that
4860 # started this poll, so it sits near the top; N=200 is large enough
4861 # that concurrent reindexes of other collections can't push it out
4862 # of the window before it terminates. Trade-off: if more than ~200
4863 # newer indexing tasks for OTHER collections appear while this one
4864 # is still in-flight, the scoped lookup falls back to "idle".
4865 recent_task_scan_limit = 200
4866 recent_tasks = (
4867 db_session.query(TaskMetadata)
4868 .filter(TaskMetadata.task_type == "indexing")
4869 .order_by(TaskMetadata.created_at.desc())
4870 .limit(recent_task_scan_limit)
4871 .all()
4872 )
4874 task = None
4875 for candidate in recent_tasks:
4876 metadata = candidate.metadata_json or {}
4877 if metadata.get("collection_id") == collection_id:
4878 task = candidate
4879 break
4881 collection = db_session.get(Collection, collection_id)
4882 index = (
4883 db_session.query(RAGIndex)
4884 .filter_by(
4885 collection_name=f"collection_{collection_id}",
4886 is_current=True,
4887 )
4888 .first()
4889 )
4890 rebuild_info = _collection_rebuild_info(collection, index)
4892 if not task:
4893 return {
4894 **rebuild_info,
4895 "status": "idle",
4896 "collection_id": collection_id,
4897 "message": "No indexing task for this collection",
4898 }
4900 task_metadata = task.metadata_json or {}
4901 safe_result = _client_indexing_result(task_metadata.get("result"))
4902 worker_active = active_task_id == task.task_id
4903 return {
4904 **rebuild_info,
4905 "task_id": task.task_id,
4906 "force_reindex": task_metadata.get("force_reindex") is True,
4907 # A cancellation request can precede drain or an indivisible
4908 # publication. Clients must observe the worker's final result.
4909 "worker_active": worker_active,
4910 "collection_id": collection_id,
4911 "status": task.status,
4912 "progress_current": task.progress_current or 0,
4913 "progress_total": task.progress_total or 0,
4914 "progress_message": (
4915 "Indexing failed"
4916 if task.status == "failed"
4917 else task.progress_message
4918 ),
4919 "error_message": _client_indexing_task_error(
4920 task.error_message,
4921 safe_result,
4922 task_metadata.get("failure"),
4923 ),
4924 "result": safe_result,
4925 "created_at": task.created_at.isoformat()
4926 if task.created_at
4927 else None,
4928 "completed_at": task.completed_at.isoformat()
4929 if task.completed_at
4930 else None,
4931 }
4933 except Exception:
4934 logger.exception("Failed to get index status")
4935 return JSONResponse(
4936 {
4937 "status": "error",
4938 "error": "Failed to get indexing status. Please try again.",
4939 },
4940 status_code=500,
4941 )
4944@router.post("/api/collections/{collection_id}/index/cancel")
4945def cancel_indexing(
4946 request: Request,
4947 collection_id,
4948 username: Annotated[str, Depends(require_auth)],
4949):
4950 """Cancel an active indexing task for a collection."""
4951 from ...database.session_context import get_user_db_session
4952 from ...database.session_passwords import session_password_store
4954 session_id = request.session.get("session_id")
4956 db_password = None
4957 if session_id:
4958 db_password = session_password_store.get_session_password(
4959 username, session_id
4960 )
4962 try:
4963 # Signal active process-local SSE generator(s) if present
4964 sse_cancelled = False
4965 with _active_sse_indexers_lock:
4966 sse_events = _active_sse_indexers.get((username, collection_id))
4967 if sse_events:
4968 for sse_event in sse_events:
4969 sse_event.set()
4970 sse_cancelled = True
4971 logger.info(
4972 f"Signalled active SSE indexing event(s) for collection {collection_id}"
4973 )
4975 with get_user_db_session(username, db_password) as db_session:
4976 # Find active indexing task for this collection
4977 tasks = (
4978 db_session.query(TaskMetadata)
4979 .filter(
4980 TaskMetadata.task_type == "indexing",
4981 TaskMetadata.status == "processing",
4982 )
4983 .all()
4984 )
4986 # Another collection's earlier task must not hide this one.
4987 # Match the same collection metadata used by index/start.
4988 matched_task = next(
4989 (
4990 task
4991 for task in tasks
4992 if (task.metadata_json or {}).get("collection_id")
4993 == collection_id
4994 ),
4995 None,
4996 )
4998 if not matched_task and not sse_cancelled:
4999 if tasks:
5000 return JSONResponse(
5001 {
5002 "success": False,
5003 "error": "No active indexing task for this collection",
5004 },
5005 status_code=404,
5006 )
5007 return JSONResponse(
5008 {
5009 "success": False,
5010 "error": "No active indexing task found",
5011 },
5012 status_code=404,
5013 )
5015 if matched_task:
5016 # This endpoint must surface a failed cancellation write.
5017 # Background workers use the best-effort wrapper, but returning
5018 # success here after it swallowed an error would mislead the
5019 # caller while the task remains active.
5020 task_id_to_cancel = matched_task.task_id
5021 _do_update_task_status(
5022 username,
5023 db_password,
5024 task_id_to_cancel,
5025 status="cancelled",
5026 progress_message="Cancellation requested...",
5027 )
5029 logger.info(
5030 f"Requested cancellation of indexing task {task_id_to_cancel} for collection {collection_id}"
5031 )
5033 return {
5034 "success": True,
5035 "message": "Cancellation requested",
5036 "task_id": matched_task.task_id if matched_task else None,
5037 }
5039 except Exception:
5040 logger.exception("Failed to cancel indexing")
5041 return JSONResponse(
5042 {
5043 "success": False,
5044 "error": "Failed to cancel indexing. Please try again.",
5045 },
5046 status_code=500,
5047 )
5050# Research History Semantic Search Routes have been moved to
5051# web/routers/library_search.py