Coverage for src/local_deep_research/web/routers/research.py: 91%
878 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1from fastapi import APIRouter, Depends, Request
2from fastapi.responses import (
3 JSONResponse,
4 RedirectResponse,
5)
6from ..dependencies.auth import require_auth
7from ..dependencies.rate_limit import (
8 _api_exempt,
9 _api_user_key,
10 api_rate_limit,
11 limiter,
12 upload_rate_limit_ip,
13 upload_rate_limit_user,
14)
15from ..dependencies.threadpool import (
16 WorkerCleanupStreamingResponse,
17 run_db_sync,
18)
19from ..template_config import templates
21import asyncio
22import json
23import re
24from datetime import datetime, UTC
25from pathlib import Path
27from loguru import logger
28from ...settings.logger import log_settings
29from sqlalchemy import case, func
30from sqlalchemy.orm import Session
31from typing import Optional, Annotated
33# Security imports
34from ...config.constants import DEFAULT_OLLAMA_URL
35from ...exceptions import DuplicateResearchError, SystemAtCapacityError
36from ...constants import HISTORY_LOGS_HARD_CAP, ResearchStatus
37from ...security import (
38 FileUploadValidator,
39 UnsafeFilenameError,
40 filter_research_metadata,
41 sanitize_filename,
42 strip_settings_snapshot,
43)
44from ...utilities.type_utils import to_bool
45from ...utilities.url_utils import is_safe_custom_llm_endpoint
48from ...config.paths import get_config_directory
50# Services imports
51from ..services.pdf_extraction_service import get_pdf_extraction_service
52from ..services.pdf_service import (
53 MissingPDFDependencyError,
54 get_weasyprint_install_instructions,
55)
57from ...database.models import (
58 QueuedResearch,
59 ResearchHistory,
60 UserActiveResearch,
61)
62from ..models.database import (
63 ResearchLog,
64 calculate_duration,
65)
66from ...database.models.library import Document as Document
67from ...database.session_context import get_user_db_session
68from ...llm.providers.base import normalize_provider
69from ..auth.password_utils import resolve_user_password
70from ..services.research_service import (
71 clamp_user_max_concurrent,
72 export_report_to_memory,
73 run_research_process,
74 start_research_process,
75)
77from ..routes.research_validation import (
78 validate_research_query_length,
79 validate_search_overrides,
80)
81from ..research_state import (
82 append_research_log,
83 get_active_research_ids,
84 get_research_field,
85 get_user_research_start_lock,
86 is_research_active,
87 set_termination_flag,
88 user_research_start_gate,
89)
90from ...constants import (
91 DEFAULT_SEARCH_TOOL,
92 LANGGRAPH_STRATEGY_ALIASES,
93 LANGGRAPH_STRATEGY_NAME as LANGGRAPH_STRATEGY_NAME,
94)
95from ..dependencies.json_body import json_body_error
97# Create the router for the research application
98router = APIRouter(tags=["research"])
100# NOTE: Routes use username (not .get()) intentionally.
101# require_auth guarantees the value is present; direct access fails fast
102# if the dependency is ever removed.
105# Add static route at the root level
106# include_in_schema=False matches the sibling `/static/{path:path}` mount in
107# fastapi_app.py. A `:path` converter cannot be represented in OpenAPI, so a
108# schema-included route using one fails test_route_contracts.py's
109# "every schema-included APIRoute appears in OpenAPI" check. This is a legacy
110# redirect shim for bookmarked URLs, not part of the documented API surface,
111# so excluding it is correct rather than a workaround.
112@router.get("/redirect-static/{path:path}", include_in_schema=False)
113def redirect_static(path: str):
114 """Redirect old static URLs to new static URLs.
116 Two things here are load-bearing and were both wrong in the initial port,
117 which left the route enumerable (so route-parity checks passed) but
118 functionally dead:
120 * ``{path:path}`` — Flask's ``<path:path>`` matches slashes; Starlette's
121 plain ``{path}`` does not. Every realistic legacy URL has at least one
122 (``css/styles.css``), so they all 404'd.
123 * the captured ``path`` must actually be used. The port ignored it and
124 redirected to a bare ``/static``, dropping the filename even in the
125 single-segment case.
127 Flask did ``redirect(url_for("static", filename=path))``.
128 """
129 # Quote so a `?` or `#` in a legacy filename cannot truncate the target
130 # into a query or fragment. `safe="/"` keeps directory separators intact.
131 # The result always begins with the literal "/static/", so it cannot be
132 # turned into a protocol-relative (`//host`) off-site redirect.
133 from urllib.parse import quote
135 return RedirectResponse(
136 url=f"/static/{quote(path.lstrip('/'), safe='/')}", status_code=302
137 )
140@router.get("/progress/{research_id}")
141def progress_page(
142 request: Request,
143 research_id,
144 username: Annotated[str, Depends(require_auth)],
145):
146 """Render the research progress page"""
147 return templates.TemplateResponse(
148 request=request,
149 name="pages/progress.html",
150 context={"request": request},
151 )
154@router.get("/details/{research_id}")
155def research_details_page(
156 request: Request,
157 research_id,
158 username: Annotated[str, Depends(require_auth)],
159):
160 """Render the research details page"""
161 return templates.TemplateResponse(
162 request=request, name="pages/details.html", context={"request": request}
163 )
166@router.get("/results/{research_id}")
167def results_page(
168 request: Request,
169 research_id,
170 username: Annotated[str, Depends(require_auth)],
171):
172 """Render the research results page"""
173 return templates.TemplateResponse(
174 request=request, name="pages/results.html", context={"request": request}
175 )
178@router.get("/history")
179def history_page(
180 request: Request, username: Annotated[str, Depends(require_auth)]
181):
182 """Render the history page"""
183 return templates.TemplateResponse(
184 request=request, name="pages/history.html", context={"request": request}
185 )
188def _extract_research_params(data, settings_manager):
189 """Extract and resolve research parameters from request data and settings.
191 Returns a dict with keys: model_provider, model, custom_endpoint,
192 ollama_url, search_engine, max_results, time_period, iterations,
193 questions_per_iteration, strategy.
194 """
195 model_provider = data.get("model_provider")
196 if not model_provider:
197 model_provider = settings_manager.get_setting("llm.provider", "ollama")
198 logger.debug(
199 f"No model_provider in request, using database setting: {model_provider}"
200 )
201 else:
202 logger.debug(f"Using model_provider from request: {model_provider}")
203 # Normalize provider to lowercase canonical form (main fix #3348) —
204 # the uppercase comparisons below would otherwise never match the
205 # lowercase values stored in settings / sent by the UI, silently
206 # skipping the custom_endpoint and ollama_url settings fallbacks.
207 model_provider = normalize_provider(model_provider)
209 model = data.get("model")
210 if not model:
211 model = settings_manager.get_setting("llm.model", None)
212 logger.debug(f"No model in request, using database setting: {model}")
213 else:
214 logger.debug(f"Using model from request: {model}")
216 custom_endpoint = None
217 if model_provider == "openai_endpoint":
218 custom_endpoint = data.get("custom_endpoint")
219 if not custom_endpoint:
220 custom_endpoint = settings_manager.get_setting(
221 "llm.openai_endpoint.url", None
222 )
223 logger.debug(
224 f"No custom_endpoint in request, using database setting: {custom_endpoint}"
225 )
227 # SSRF guard: a user-supplied custom_endpoint goes straight to the
228 # OpenAI client as base_url. Without validation, an authenticated user
229 # could point the LLM client at internal services (cloud metadata,
230 # localhost, etc.) and read responses through completions output.
231 if custom_endpoint and isinstance(custom_endpoint, str):
232 # Delegate to the shared guard so the endpoint is normalized the way
233 # the OpenAI-compatible provider normalizes it — a bare
234 # ``localhost:11434`` is a legitimate local backend and must not be
235 # rejected here. Private IPs stay allowed; cloud-metadata and bad
236 # schemes are still blocked.
237 if not is_safe_custom_llm_endpoint(custom_endpoint):
238 raise ValueError(
239 "custom_endpoint is not a valid HTTP(S) URL or points to a blocked address"
240 )
242 ollama_url = data.get("ollama_url")
243 if not ollama_url and model_provider == "ollama":
244 ollama_url = settings_manager.get_setting(
245 "llm.ollama.url", DEFAULT_OLLAMA_URL
246 )
247 logger.debug(
248 f"No ollama_url in request, using database setting: {ollama_url}"
249 )
251 search_engine = data.get("search_engine") or data.get("search_tool")
252 if not search_engine:
253 search_engine = settings_manager.get_setting(
254 "search.tool", DEFAULT_SEARCH_TOOL
255 )
257 max_results = data.get("max_results")
258 time_period = data.get("time_period")
260 iterations = data.get("iterations")
261 if iterations is None:
262 iterations = settings_manager.get_setting("search.iterations", 5)
264 questions_per_iteration = data.get("questions_per_iteration")
265 if questions_per_iteration is None:
266 questions_per_iteration = settings_manager.get_setting(
267 "search.questions_per_iteration", 5
268 )
270 strategy = data.get("strategy")
271 if not strategy:
272 strategy = settings_manager.get_setting(
273 "search.search_strategy", "source-based"
274 )
276 # Egress policy per-research overrides. Mirror the
277 # model/search_engine pattern: missing values fall back to saved
278 # settings; supplied values override JUST FOR THIS RUN. They do
279 # NOT persist to the user's settings DB.
280 policy_egress_scope = data.get("policy_egress_scope")
281 llm_require_local_endpoint = data.get("llm_require_local_endpoint")
282 embeddings_require_local = data.get("embeddings_require_local")
284 return {
285 "model_provider": model_provider,
286 "model": model,
287 "custom_endpoint": custom_endpoint,
288 "ollama_url": ollama_url,
289 "search_engine": search_engine,
290 "max_results": max_results,
291 "time_period": time_period,
292 "iterations": iterations,
293 "questions_per_iteration": questions_per_iteration,
294 "strategy": strategy,
295 "policy_egress_scope": policy_egress_scope,
296 "llm_require_local_endpoint": llm_require_local_endpoint,
297 "embeddings_require_local": embeddings_require_local,
298 }
301def _precheck_collection_agent_enabled(
302 search_engine: str,
303 strategy: str,
304 username: str,
305) -> Optional[JSONResponse]:
306 """Reject a run that pairs a LangGraph strategy with an agent-hidden
307 collection.
309 Ported from the Flask ``research_routes`` (#5221). Returns a
310 ``JSONResponse`` when the run should be rejected, or ``None`` to
311 continue — the FastAPI shape, where Flask returned a
312 ``(jsonify, status)`` tuple.
314 The per-collection ``agent_enabled`` flag is exclusive to the LangGraph
315 research agent: it gates which collection engines the agent offers as
316 specialized tools (see ``AdvancedSearchSystem`` /
317 ``LangGraphAgentStrategy._load_specialized_engine_tools``). Public
318 collections also reach the agent via their adapter, but ONLY as
319 specialized tools — the primary engine path is unaffected, so a
320 collection marked ``agent_enabled=False`` is effectively unusable for a
321 LangGraph run: the agent will not pick it as a tool, and the user cannot
322 pick anything else because the primary was the only way to surface it.
324 Mirrors ``_precheck_engine_policy`` below. Fails OPEN (returns None) on
325 any internal error so a transient DB blip doesn't block the run — the
326 frontend is the primary UX guarantee and this is the second backstop.
327 Only fires for ``collection_<uuid>`` engines (the only engines carrying
328 the flag today), and only when ``strategy`` maps to the LangGraph agent;
329 other strategies never read the flag.
330 """
331 if not search_engine or not search_engine.startswith("collection_"):
332 return None
333 if not strategy or strategy.lower() not in LANGGRAPH_STRATEGY_ALIASES:
334 return None
335 try:
336 from ...database.models.library import Collection
338 # Imported locally rather than using the module-level binding, so a
339 # late patch of ``database.session_context.get_user_db_session``
340 # takes effect. Main's version does the same and its tests rely on
341 # it; binding at module import time would silently defeat them.
342 from ...database.session_context import (
343 get_user_db_session as _get_user_db_session,
344 )
346 # Engine id format: ``collection_<uuid>``. The split is safe because
347 # the id is constructed the same way in
348 # ``search_engines_config.search_config``.
349 collection_id = search_engine[len("collection_") :]
350 with _get_user_db_session(username) as db_session:
351 row = (
352 db_session.query(Collection)
353 .filter(Collection.id == collection_id)
354 .first()
355 )
356 if row is None:
357 # Unknown collection id — let the factory PEP produce a clearer
358 # error rather than masking it with a 400 here.
359 return None
360 # Default-on for unrecognised / NULL values keeps the
361 # behaviour-preserving contract used elsewhere in the codebase.
362 agent_enabled = getattr(row, "agent_enabled", True) is not False
363 if agent_enabled:
364 return None
365 display_name = getattr(row, "name", None) or search_engine
366 logger.bind(policy_audit=True).warning(
367 "POST /api/start_research collection agent_enabled refused",
368 search_engine=search_engine,
369 strategy=strategy,
370 )
371 return JSONResponse(
372 {
373 "status": "error",
374 "message": (
375 f"Collection '{display_name}' is hidden from the "
376 "LangGraph research agent (the 'Available to the "
377 "research agent' flag is off). Re-enable it on the "
378 "collection's page, or pick a different search "
379 "engine / strategy."
380 ),
381 "reason": "collection_agent_disabled",
382 "field": "strategy",
383 },
384 status_code=400,
385 )
386 except Exception: # pragma: no cover - defensive
387 logger.exception("collection agent_enabled pre-check skipped")
388 return None
391def _precheck_engine_policy(settings_manager, params, search_engine, username):
392 """Validate the requested search engine against the saved egress
393 policy at the request boundary.
395 Returns a ``JSONResponse`` when the request should be rejected, or
396 ``None`` to continue. Falls through (returns None) when there's no
397 real dict snapshot to validate or the policy module errors — the
398 factory PEP still enforces at engine-instantiation time.
399 """
400 try:
401 from ...security.egress.policy import (
402 PolicyDeniedError,
403 context_from_snapshot,
404 evaluate_engine,
405 resolve_run_primary_engine,
406 )
408 policy_snapshot = settings_manager.get_settings_snapshot()
409 # Only validate against a real dict snapshot. A test double or
410 # an unavailable settings backend hands back something else;
411 # skip rather than misfire.
412 if not isinstance(policy_snapshot, dict):
413 return None
415 # Overlay per-research form overrides so the snapshot reflects
416 # what the user picked for THIS run (per-research overrides, not
417 # a global settings save).
418 # Resolve the primary the SAME way the worker does (single source of
419 # truth) so the precheck and the background worker agree on accept vs.
420 # refuse — including the fail-closed missing-primary case, which the
421 # ValueError handler below maps to a 400. (Previously this substituted
422 # searxng and accepted runs the worker then refused.)
423 try:
424 _apply_policy_overrides(policy_snapshot, params)
425 primary = resolve_run_primary_engine(policy_snapshot)
426 policy_ctx = context_from_snapshot(
427 policy_snapshot,
428 primary,
429 username=username,
430 )
431 except PolicyDeniedError as exc:
432 return JSONResponse(
433 {
434 "status": "error",
435 "message": (
436 f"Egress policy refused this run: {exc.decision.reason}"
437 ),
438 },
439 status_code=400,
440 )
441 except ValueError as exc:
442 # An invalid policy config is unrecoverable. Previously this
443 # raised, fell through to the outer ``except Exception``
444 # below, and silently returned None — so the run started
445 # successfully at the precheck and only failed at a
446 # downstream PEP. Surface it here as a 400 instead.
447 logger.bind(policy_audit=True).warning(
448 "POST /api/start_research policy precheck rejected",
449 reason=str(exc),
450 )
451 # Return a generic reason to the client — the raw ValueError text
452 # can carry policy-config internals and is already captured in the
453 # policy_audit warning above (CWE-209). Unlike the PolicyDeniedError
454 # branch, this path has no curated, user-safe decision reason.
455 return JSONResponse(
456 {
457 "status": "error",
458 "message": (
459 "Egress policy refused this run due to an "
460 "invalid policy configuration."
461 ),
462 },
463 status_code=400,
464 )
466 decision = evaluate_engine(
467 search_engine,
468 policy_ctx,
469 settings_snapshot=policy_snapshot,
470 )
471 if not decision.allowed:
472 logger.bind(policy_audit=True).warning(
473 "POST /api/start_research search_engine refused",
474 engine=search_engine,
475 reason=decision.reason,
476 )
477 # Local import: this module imports the egress policy lazily inside
478 # this function to avoid a circular import at load time; keep the
479 # guidance import with it.
480 from ...security.egress.guidance import denial_guidance
482 # Only the scope-mismatch and strict-not-primary denials have a
483 # form field the user can toggle to fix the error (the Egress
484 # Scope dropdown). For engine_unknown / unclassified the user has
485 # to pick a different search engine — flagging the egress-scope
486 # dropdown there would be misleading and cause the wrong field to
487 # turn red. internal_error / no_snapshot are server-side issues
488 # with no form-level fix at all, so we return ``field: null`` and
489 # let the frontend render just the alert (no inline field error).
490 field_for_reason = {
491 "scope_mismatch_private_only": "policy_egress_scope",
492 "scope_mismatch_public_only": "policy_egress_scope",
493 "strict_not_primary": "policy_egress_scope",
494 }.get(decision.reason)
496 return JSONResponse(
497 {
498 "status": "error",
499 # Clear, actionable message (what + why + how to allow),
500 # plus the raw reason code for support/logs. The
501 # guidance uses ``primary_engine`` to decide whether
502 # Adaptive (default) is a reliable remedy — see
503 # ``_scope_mismatch_how``.
504 "message": denial_guidance(
505 decision.reason,
506 target=f"Search engine '{search_engine}'",
507 # Raw engine id, used by the guidance to decide
508 # whether Adaptive (default) is a reliable
509 # remedy for a scope-mismatch denial (see
510 # ``_scope_mismatch_how`` — kept separate from
511 # ``target`` so the formatted display string
512 # above doesn't break the equality check).
513 target_id=search_engine,
514 primary_engine=policy_ctx.primary_engine,
515 ),
516 "reason": decision.reason,
517 # The frontend uses this to highlight the offending
518 # form field with an inline error (FormValidator's
519 # .ldr-field-error + .ldr-field-invalid convention)
520 # and to anchor an alert near the submit button.
521 # ``null`` means "no form field fixes this — show the
522 # alert without an inline field error".
523 "field": field_for_reason,
524 },
525 status_code=400,
526 )
528 # Stage C (ADR-0007): enforce the two-axis (sensitivity x exposure)
529 # admissibility as defense-in-depth on top of the scope check. This is
530 # a fast pre-start 400 for the API path; the same rule is also enforced
531 # at the shared run chokepoint in run_research_process, so follow-up /
532 # chat / queue runs are covered too. Under the UNPROTECTED escape hatch
533 # audit_run evaluates in permissive mode (always allowed), so an
534 # explicitly-unprotected run is never blocked. On an internal error
535 # audit_run fails CLOSED (reason "audit_error") so the run is refused
536 # visibly rather than silently degrading to the scope PEPs.
537 #
538 # NOTE: this sees the run's PRIMARY engine only. The per-engine scope
539 # PEP still runs for expanded engines, but it enforces SCOPE, not the
540 # two-axis rule — so a non-primary sensitive engine reaching a cloud
541 # sink under a permissive scope (legacy `both`) is not caught here.
542 # Widening to the full resolved engine set is a follow-up.
543 from ...security.egress.run_classification import (
544 audit_run_from_snapshot,
545 )
547 # Same shared entry point the worker chokepoint uses, so both resolve
548 # providers identically (LLM override wins over the snapshot; embeddings
549 # only for RAG primaries) and fail closed the same way.
550 two_axis = audit_run_from_snapshot(
551 policy_snapshot,
552 policy_ctx,
553 search_engine,
554 llm_provider=params.get("model_provider"),
555 )
556 if not two_axis.allowed:
557 logger.bind(policy_audit=True).warning(
558 "POST /api/start_research two-axis egress refused",
559 reason=two_axis.reason,
560 offending=two_axis.offending,
561 )
562 if two_axis.reason == "audit_error": 562 ↛ 563line 562 didn't jump to line 563 because the condition on line 562 was never true
563 message = (
564 "Egress policy could not verify this run, so it was "
565 "refused (fail-closed). To override, ask the server "
566 "operator to enable the Unprotected escape hatch "
567 "(LDR_POLICY_ALLOW_UNPROTECTED_EGRESS), then select the "
568 "'Unprotected' scope yourself."
569 )
570 else:
571 message = (
572 "Egress policy refused this run: a sensitive source would "
573 f"reach an exposing destination ({two_axis.reason}). Use "
574 "local inference, mark the collection public, or — to "
575 "override — ask the server operator to enable the "
576 "Unprotected escape hatch "
577 "(LDR_POLICY_ALLOW_UNPROTECTED_EGRESS), then set Egress "
578 "Scope to 'Unprotected' yourself."
579 )
580 return JSONResponse(
581 {
582 "status": "error",
583 "message": message,
584 "reason": two_axis.reason,
585 },
586 status_code=400,
587 )
588 return None
589 except Exception:
590 # Policy module unavailable / internal error → log and fall
591 # through; the factory PEP will catch any actual violation.
592 logger.exception("egress policy pre-check skipped")
593 return None
596def _apply_policy_overrides(settings_snapshot, params):
597 """Overlay form-supplied egress policy values onto the snapshot.
599 Per-research overrides: the user picked these values for this
600 specific run and they do NOT persist to the settings DB. Mirrors
601 how model / search_engine overrides work today.
602 """
603 if not isinstance(settings_snapshot, dict):
604 return
605 from ...settings.manager import check_env_setting
606 from ...security.egress.policy import (
607 parse_user_egress_scope,
608 )
610 scope_override = params.get("policy_egress_scope")
611 if (
612 scope_override is not None
613 and check_env_setting("policy.egress_scope") is None
614 ):
615 settings_snapshot["policy.egress_scope"] = parse_user_egress_scope(
616 scope_override
617 ).value
618 if (
619 params.get("llm_require_local_endpoint") is not None
620 and check_env_setting("llm.require_local_endpoint") is None
621 ):
622 settings_snapshot["llm.require_local_endpoint"] = to_bool(
623 params["llm_require_local_endpoint"]
624 )
625 if (
626 params.get("embeddings_require_local") is not None
627 and check_env_setting("embeddings.require_local") is None
628 ):
629 settings_snapshot["embeddings.require_local"] = to_bool(
630 params["embeddings_require_local"]
631 )
632 # NOTE: the per-run `custom_endpoint` form value is deliberately NOT
633 # overlaid onto llm.openai_endpoint.url — that kwarg is non-functional
634 # (config/llm_config.py) and the run reads the URL from the DB setting, so
635 # the snapshot already carries the endpoint the run actually uses.
638def _research_not_found(research_id, message="Research not found"):
639 """Return a consistent 404 JSON for a missing research.
641 Emits ``status``, ``error`` and ``message`` so the body is a strict
642 superset of both historical 404 shapes — every frontend reader and
643 existing test keeps working without changes:
644 - Shape A readers read ``data.error``
645 - Shape B readers read ``data.status`` (``== "error"``) and/or
646 ``data.message``
648 ``research_id`` is used only for a debug log identifying which research
649 was missing; it is intentionally never echoed in the response body.
650 """
651 logger.debug(f"404 for research {research_id}: {message}")
652 return JSONResponse(
653 {"status": "error", "error": message, "message": message},
654 status_code=404,
655 )
658def _queue_research(
659 db_session: Session,
660 username,
661 research_id,
662 query,
663 mode,
664 research_settings,
665 params,
666 session_id,
667 reason="",
668 research=None,
669):
670 """Add research to queue and notify processor. Returns a JSON response.
672 Args:
673 reason: Optional prefix explaining why the research was queued
674 (e.g. "due to concurrent limit").
675 research: Optional ResearchHistory object whose status should be set
676 to QUEUED atomically with the queue record insertion.
677 """
678 max_position = (
679 db_session.query(func.max(QueuedResearch.position))
680 .filter_by(username=username)
681 .scalar()
682 or 0
683 )
685 queued_record = QueuedResearch(
686 username=username,
687 research_id=research_id,
688 query=query,
689 mode=mode,
690 settings_snapshot=research_settings,
691 position=max_position + 1,
692 )
693 db_session.add(queued_record)
694 if research is not None:
695 research.status = ResearchStatus.QUEUED # type: ignore[assignment]
696 db_session.commit()
697 logger.info(
698 f"Queued research {research_id} at position {max_position + 1} for user {username}"
699 )
701 from ..queue.processor_v2 import queue_processor
703 queue_processor.notify_research_queued(
704 username,
705 research_id,
706 session_id=session_id,
707 query=query,
708 mode=mode,
709 settings_snapshot=research_settings,
710 model_provider=params["model_provider"],
711 model=params["model"],
712 custom_endpoint=params["custom_endpoint"],
713 search_engine=params["search_engine"],
714 max_results=params["max_results"],
715 time_period=params["time_period"],
716 iterations=params["iterations"],
717 questions_per_iteration=params["questions_per_iteration"],
718 strategy=params["strategy"],
719 )
721 position = max_position + 1
722 reason_text = f" {reason}" if reason else ""
723 message = f"Your research has been queued{reason_text}. Position in queue: {position}"
724 return {
725 "status": ResearchStatus.QUEUED,
726 "research_id": research_id,
727 "queue_position": position,
728 "message": message,
729 }
732@router.post("/api/start_research")
733# Primary research-submission endpoint — must be rate-limited (per-user,
734# via the shared api_rate_limit bucket) or a single account can flood the
735# research queue (#3135).
736@api_rate_limit
737async def start_research(
738 request: Request, username: Annotated[str, Depends(require_auth)]
739):
740 """
741 Async wrapper that does the only legitimate await (parsing the JSON
742 body) on the event loop, then offloads the synchronous body — which
743 opens four ``get_user_db_session`` blocks (each potentially driving
744 PBKDF2 key derivation), threads, and other blocking work — to a
745 threadpool. Without this, ``async def`` + sync DB blocks would
746 serialise the entire server behind the slowest /start_research call.
747 """
749 # A malformed or non-object JSON body would otherwise reach
750 # _start_research_sync's data.get(...) calls and 500 (AttributeError).
751 # Reject at the boundary with a clean 400.
752 try:
753 data = await request.json()
754 except Exception:
755 return JSONResponse(
756 {"status": "error", "message": "Request body must be JSON"},
757 status_code=400,
758 )
759 if not isinstance(data, dict):
760 return JSONResponse(
761 {
762 "status": "error",
763 "message": "Request body must be a JSON object",
764 },
765 status_code=400,
766 )
767 query = data.get("query")
768 if query is not None and not isinstance(query, str):
769 return JSONResponse(
770 {"status": "error", "message": "query must be a string"},
771 status_code=400,
772 )
773 metadata = data.get("metadata", {})
774 if not isinstance(metadata, dict):
775 return JSONResponse(
776 {"status": "error", "message": "metadata must be an object"},
777 status_code=400,
778 )
779 validation_error = validate_search_overrides(data)
780 if validation_error is not None:
781 return JSONResponse(
782 {"status": "error", "message": validation_error},
783 status_code=400,
784 )
785 base_url = str(request.base_url)
786 session_id = request.session.get("session_id")
787 return await run_db_sync(
788 _start_research_sync, data, username, base_url, session_id
789 )
792def _start_research_sync(
793 data: dict, username: str, base_url: str, session_id: str | None
794):
795 # Debug logging to trace model parameter
796 logger.debug(f"Request data keys: {list(data.keys())}")
798 # Check if this is a news search
799 metadata = data.get("metadata", {})
800 if metadata.get("is_news_search"): 800 ↛ 801line 800 didn't jump to line 801 because the condition on line 800 was never true
801 logger.info(
802 f"News search request received: triggered_by={metadata.get('triggered_by', 'unknown')}"
803 )
805 query = data.get("query")
806 mode = data.get("mode", "quick")
808 # Replace date placeholders if they exist
809 if query and "YYYY-MM-DD" in query:
810 # Use local system time
811 current_date = datetime.now(UTC).strftime("%Y-%m-%d")
813 original_query = query
814 query = query.replace("YYYY-MM-DD", current_date)
815 logger.info(
816 f"Replaced date placeholder in query: {original_query[:100]}... -> {query[:100]}..."
817 )
818 logger.info(f"Using date: {current_date}")
820 # Update metadata to track the replacement
821 if not metadata: 821 ↛ 823line 821 didn't jump to line 823 because the condition on line 821 was always true
822 metadata = {}
823 metadata["original_query"] = original_query
824 metadata["processed_query"] = query
825 metadata["date_replaced"] = current_date
826 data["metadata"] = metadata
828 # Get parameters from request or use database settings
829 from ...settings import SettingsManager
831 with get_user_db_session(username) as db_session:
832 settings_manager = SettingsManager(db_session=db_session)
833 try:
834 params = _extract_research_params(data, settings_manager)
835 except ValueError:
836 # SSRF guard in _extract_research_params rejected the
837 # custom_endpoint — return main's clean 400 contract
838 # instead of an unhandled 500.
839 return JSONResponse(
840 {"status": "error", "message": "Invalid custom endpoint URL"},
841 status_code=400,
842 )
844 model_provider = params["model_provider"]
845 model = params["model"]
846 custom_endpoint = params["custom_endpoint"]
847 search_engine = params["search_engine"]
848 max_results = params["max_results"]
849 time_period = params["time_period"]
850 iterations = params["iterations"]
851 questions_per_iteration = params["questions_per_iteration"]
852 strategy = params["strategy"]
854 # Egress policy: server-side check on the requested search
855 # engine BEFORE we enqueue the research. Defends against an API
856 # client posting an engine name that violates the saved policy
857 # (e.g. ``{"search_engine": "pubmed"}`` under STRICT+primary=arxiv).
858 # The factory PEP catches the same case at instantiation time,
859 # but rejecting at the request boundary lets us return a clean
860 # 4xx instead of an opaque background failure.
861 policy_error = _precheck_engine_policy(
862 settings_manager, params, search_engine, username
863 )
864 if policy_error is not None:
865 return policy_error
867 # Per-collection ``agent_enabled`` is exclusive to the LangGraph
868 # research agent. When the user picks LangGraph + a collection
869 # hidden from the agent, the run is effectively broken (the
870 # agent won't pick it as a tool, and the primary was the only
871 # way to surface it). The frontend already disables the
872 # combination in the dropdown; this is the second backstop for
873 # direct API callers and stale-form submissions.
874 agent_enabled_error = _precheck_collection_agent_enabled(
875 search_engine, strategy, username
876 )
877 if agent_enabled_error is not None: 877 ↛ 878line 877 didn't jump to line 878 because the condition on line 877 was never true
878 return agent_enabled_error
880 # This thread-local session survives the ``with`` block. End its
881 # read transaction before the admission gate below; otherwise this
882 # request can wait for the gate while retaining a pooled connection
883 # (and, under rollback-journal SQLite, a shared database lock).
884 db_session.rollback()
886 # Debug logging for model parameter specifically
887 logger.debug(
888 f"Extracted model value: '{model}' (type: {type(model).__name__})"
889 )
891 # Log the selections for troubleshooting
892 logger.info(
893 f"Starting research with provider: {model_provider}, model: {model}, search engine: {search_engine}"
894 )
895 logger.info(
896 f"Additional parameters: max_results={max_results}, time_period={time_period}, iterations={iterations}, questions={questions_per_iteration}, strategy={strategy}"
897 )
899 if not query:
900 return JSONResponse(
901 {"status": "error", "message": "Query is required"}, status_code=400
902 )
904 # Cap query length to defend against accidental prompt blowups that would
905 # bloat ResearchHistory storage and run up LLM context. The queue replay
906 # path uses the same validator before dispatching a persisted row.
907 query_length_error = validate_research_query_length(query)
908 if query_length_error is not None:
909 return JSONResponse(
910 {
911 "status": "error",
912 "message": query_length_error,
913 },
914 status_code=400,
915 )
917 # Validate required parameters based on provider.
918 # model_provider is normalize_provider()'d to lowercase in
919 # _extract_research_params, so this MUST compare to the lowercase
920 # canonical form — the old "OPENAI_ENDPOINT" literal never matched and
921 # left this required-field check dead.
922 if model_provider == "openai_endpoint" and not custom_endpoint:
923 return JSONResponse(
924 {
925 "status": "error",
926 "message": "Custom endpoint URL is required for OpenAI endpoint provider",
927 },
928 status_code=400,
929 )
931 # SSRF pre-flight on the user-supplied LLM endpoint: reject metadata /
932 # link-local targets at the request boundary, before any research thread
933 # is spawned. This is fail-fast defense-in-depth — the OpenAI-compatible
934 # provider's assert_base_url_safe re-validates the same URL before the
935 # client is built. Private IPs / localhost are permitted so local LLMs
936 # (vLLM, Ollama, LM Studio) work, including scheme-less endpoints
937 # (the helper normalizes exactly as the provider does).
938 if model_provider == "openai_endpoint" and not is_safe_custom_llm_endpoint( 938 ↛ 941line 938 didn't jump to line 941 because the condition on line 938 was never true
939 custom_endpoint
940 ):
941 return JSONResponse(
942 {"status": "error", "message": "Invalid custom endpoint URL"},
943 status_code=400,
944 )
946 if not model:
947 logger.error(
948 f"No model specified or configured. Provider: {model_provider}"
949 )
950 return JSONResponse(
951 {
952 "status": "error",
953 "message": "Model is required. Please configure a model in the settings.",
954 },
955 status_code=400,
956 )
958 # Check if the user has too many active researches
959 from ...settings import SettingsManager
961 should_queue = False
962 # Pre-bind: the post-commit race check below reads this, so if the block
963 # under try raises before the setting is read (a DB hiccup), that check
964 # would raise UnboundLocalError and turn a recoverable failure into a 500.
965 max_concurrent_researches = 3
966 try:
967 # Share one stable per-user admission lock with queue replay. The
968 # preliminary count and the post-commit claim below both use this gate;
969 # the second guarded section closes any race during snapshot capture.
970 with user_research_start_gate(username):
971 with get_user_db_session(username) as db_session:
972 settings_manager = SettingsManager(db_session)
973 # Clamp to the server-wide semaphore ceiling (#5549).
974 # app.max_concurrent_researches is a per-user, user-editable
975 # setting, but the semaphore it is checked against is global —
976 # so without the clamp a user can raise their own limit past
977 # the server's and monopolise it.
978 max_concurrent_researches = clamp_user_max_concurrent(
979 settings_manager.get_setting(
980 "app.max_concurrent_researches", 3
981 )
982 )
984 # First, clean up stale entries where the research thread died.
985 from ..routes.globals import reclaim_stale_user_active_research
987 if reclaim_stale_user_active_research(
988 db_session, username, logger=logger
989 ):
990 db_session.commit()
992 active_count = (
993 db_session.query(UserActiveResearch)
994 .filter_by(
995 username=username,
996 status=ResearchStatus.IN_PROGRESS,
997 )
998 .count()
999 )
1001 logger.info(
1002 f"Active research count for {username}: "
1003 f"{active_count}/{max_concurrent_researches}"
1004 )
1006 should_queue = active_count >= max_concurrent_researches
1007 logger.info(f"Should queue new research: {should_queue}")
1009 # This context reuses a thread-local Session rather than
1010 # closing it on exit. End the count query's read transaction
1011 # before releasing the admission gate; otherwise this request
1012 # can retain a pooled connection and later wait for the gate
1013 # while a queue tick holds the gate and waits for a connection.
1014 db_session.rollback()
1015 except Exception:
1016 logger.exception("Failed to check active researches")
1017 # Default to not queueing if we can't check
1018 should_queue = False
1020 # For non-queued research, verify password is available BEFORE creating DB records
1021 # (queued research gets password later via queue processor)
1022 user_password = None
1023 if not should_queue:
1024 user_password, session_expired = resolve_user_password(username)
1025 if session_expired:
1026 return JSONResponse(
1027 {
1028 "status": "error",
1029 "message": "Your session has expired. Please log out and log back in to start research.",
1030 },
1031 status_code=401,
1032 )
1034 # Create a record in the database with explicit UTC timestamp
1035 import uuid
1036 import threading
1038 created_at = datetime.now(UTC).isoformat()
1039 research_id = str(uuid.uuid4())
1041 # Create organized research metadata with settings snapshot
1042 submission = {
1043 "model_provider": model_provider,
1044 "model": model,
1045 "custom_endpoint": custom_endpoint,
1046 "search_engine": search_engine,
1047 "max_results": max_results,
1048 "time_period": time_period,
1049 "iterations": iterations,
1050 "questions_per_iteration": questions_per_iteration,
1051 "strategy": strategy,
1052 }
1053 submission_overrides = []
1054 if data.get("model_provider"):
1055 submission_overrides.append("model_provider")
1056 if data.get("model"):
1057 submission_overrides.append("model")
1058 if data.get("custom_endpoint"):
1059 submission_overrides.append("custom_endpoint")
1060 if data.get("search_engine") or data.get("search_tool"):
1061 submission_overrides.append("search_engine")
1062 if data.get("max_results") is not None:
1063 submission_overrides.append("max_results")
1064 if data.get("time_period") is not None:
1065 submission_overrides.append("time_period")
1066 if data.get("iterations") is not None:
1067 submission_overrides.append("iterations")
1068 if data.get("questions_per_iteration") is not None:
1069 submission_overrides.append("questions_per_iteration")
1070 if data.get("strategy"):
1071 submission_overrides.append("strategy")
1073 research_settings = {
1074 # Direct submission parameters
1075 "submission": submission,
1076 # Only these request-supplied values remain fixed if queued work is
1077 # dispatched after operator environment settings change.
1078 "submission_overrides": submission_overrides,
1079 # System information
1080 "system": {
1081 "timestamp": created_at,
1082 "user": username,
1083 "version": "1.0", # Track metadata version for future migrations
1084 "server_url": base_url, # Server URL for link generation
1085 },
1086 }
1088 # Add any additional metadata from request
1089 additional_metadata = data.get("metadata", {})
1090 if additional_metadata:
1091 reserved_metadata_keys = {
1092 "submission",
1093 "submission_overrides",
1094 "system",
1095 "settings_snapshot",
1096 }
1097 research_settings.update(
1098 {
1099 key: value
1100 for key, value in additional_metadata.items()
1101 if key not in reserved_metadata_keys
1102 }
1103 )
1104 # Get complete settings snapshot for this research
1105 try:
1106 from ...settings import SettingsManager
1108 with get_user_db_session(username) as db_session:
1109 # Ensure any pending changes are committed
1110 try:
1111 db_session.commit()
1112 except Exception:
1113 db_session.rollback()
1114 settings_manager = SettingsManager(db_session, owns_session=False)
1115 # Get all current settings as a snapshot (bypass cache to ensure fresh data)
1116 all_settings = settings_manager.get_all_settings(
1117 bypass_cache=True,
1118 include_environment_overrides=False,
1119 )
1120 # Apply per-research egress policy overrides (form-supplied
1121 # values override saved settings JUST FOR THIS RUN; they do
1122 # not persist to the user's settings DB).
1123 _apply_policy_overrides(all_settings, params)
1125 # Add settings snapshot to metadata
1126 research_settings["settings_snapshot"] = all_settings
1127 logger.info(
1128 f"Captured {len(all_settings)} settings for research {research_id}"
1129 )
1130 # As above, release the cached session's read transaction before
1131 # taking the second admission gate for the durable slot claim.
1132 db_session.rollback()
1133 except Exception:
1134 logger.exception("Failed to capture settings snapshot")
1135 # Cannot continue without settings snapshot for thread-based research
1136 return JSONResponse(
1137 {
1138 "status": "error",
1139 "message": "Failed to capture settings for research. Please try again.",
1140 },
1141 status_code=500,
1142 )
1144 # Maintain the process-wide order used by queue replay and rekey/logout:
1145 # admission gate before database/session work. The narrower gate below is
1146 # deliberately retained and re-enters this RLock to document the exact
1147 # count -> claim section.
1148 admission_lock = get_user_research_start_lock(username)
1149 admission_lock.acquire()
1150 try:
1151 with get_user_db_session(username) as db_session:
1152 # Determine initial status based on whether we need to queue
1153 initial_status = (
1154 ResearchStatus.QUEUED
1155 if should_queue
1156 else ResearchStatus.IN_PROGRESS
1157 )
1159 research = ResearchHistory(
1160 id=research_id, # Set UUID as primary key
1161 query=query,
1162 mode=mode,
1163 status=initial_status,
1164 created_at=created_at,
1165 progress_log=[{"time": created_at, "progress": 0}],
1166 research_meta=research_settings,
1167 )
1168 db_session.add(research)
1169 db_session.commit()
1170 logger.info(
1171 f"Created research entry with UUID: {research_id}, status: {initial_status}"
1172 )
1174 if should_queue:
1175 return _queue_research(
1176 db_session,
1177 username,
1178 research_id,
1179 query,
1180 mode,
1181 research_settings,
1182 params,
1183 session_id,
1184 )
1185 # Atomically claim a user slot against queue replay. A queue tick
1186 # may have started work while this request captured its settings;
1187 # inserting and re-counting under the shared gate makes one side
1188 # demote before either can exceed the user's cap.
1189 with user_research_start_gate(username):
1190 import threading
1192 active_record = UserActiveResearch(
1193 username=username,
1194 research_id=research_id,
1195 status=ResearchStatus.IN_PROGRESS,
1196 thread_id=str(threading.current_thread().ident),
1197 settings_snapshot=research_settings,
1198 )
1199 db_session.add(active_record)
1200 db_session.commit()
1201 logger.info(
1202 f"Created active research record for user {username}"
1203 )
1205 # Double-check after committing to catch starts that landed
1206 # between the preliminary count and this guarded claim.
1207 try:
1208 final_count = (
1209 db_session.query(UserActiveResearch)
1210 .filter_by(
1211 username=username,
1212 status=ResearchStatus.IN_PROGRESS,
1213 )
1214 .count()
1215 )
1216 logger.info(
1217 f"Final active count after commit: "
1218 f"{final_count}/{max_concurrent_researches}"
1219 )
1221 if final_count > max_concurrent_researches:
1222 logger.warning(
1223 f"Race condition detected: {final_count} > "
1224 f"{max_concurrent_researches}, moving to queue"
1225 )
1226 db_session.delete(active_record)
1227 db_session.commit()
1229 return _queue_research(
1230 db_session,
1231 username,
1232 research_id,
1233 query,
1234 mode,
1235 research_settings,
1236 params,
1237 session_id,
1238 reason="due to concurrent limit",
1239 research=research,
1240 )
1241 except Exception:
1242 logger.warning("Could not recheck active count")
1244 except Exception:
1245 logger.exception("Failed to create research entry")
1246 return JSONResponse(
1247 {"status": "error", "message": "Failed to create research entry"},
1248 status_code=500,
1249 )
1250 finally:
1251 admission_lock.release()
1253 # Only start the research if not queued
1254 if not should_queue: 1254 ↛ 1398line 1254 didn't jump to line 1398 because the condition on line 1254 was always true
1255 # Save the research strategy to the database before starting the thread
1256 try:
1257 from ..services.research_service import save_research_strategy
1259 save_research_strategy(research_id, strategy, username=username)
1260 except Exception:
1261 logger.warning("Could not save research strategy")
1263 # Debug logging for settings snapshot
1264 snapshot_data = research_settings.get("settings_snapshot", {})
1265 log_settings(snapshot_data, "Settings snapshot being passed to thread")
1266 if "search.tool" in snapshot_data:
1267 logger.debug(
1268 f"search.tool in snapshot: {snapshot_data['search.tool']}"
1269 )
1270 else:
1271 logger.debug("search.tool NOT in snapshot")
1273 # Start the research process with the selected parameters.
1274 # If the spawn raises, the UserActiveResearch + IN_PROGRESS
1275 # ResearchHistory rows persisted above would otherwise be
1276 # permanently orphaned (no thread, no cleanup path). Catch any
1277 # exception, mark the research FAILED, delete the active row,
1278 # and return 500 — same contract as the queue processor's
1279 # terminal-failure branch introduced in #3481.
1280 try:
1281 research_thread = start_research_process(
1282 research_id,
1283 query,
1284 mode,
1285 run_research_process,
1286 username=username, # Pass username to the thread
1287 user_password=user_password, # Pass password for database access
1288 model_provider=model_provider,
1289 model=model,
1290 custom_endpoint=custom_endpoint,
1291 search_engine=search_engine,
1292 max_results=max_results,
1293 time_period=time_period,
1294 iterations=iterations,
1295 questions_per_iteration=questions_per_iteration,
1296 strategy=strategy,
1297 settings_snapshot=snapshot_data, # Pass complete settings
1298 )
1299 except DuplicateResearchError:
1300 # A live thread already owns this research_id. Do NOT delete
1301 # the UserActiveResearch row or mark ResearchHistory FAILED —
1302 # that state belongs to the live thread, and mutating it
1303 # would terminate a running research from the user's
1304 # perspective while it keeps executing. Same contract as the
1305 # queue processor's dedicated dup branch (#3506).
1306 logger.warning(
1307 f"Duplicate live thread detected for {research_id} "
1308 "on direct submission; leaving state intact"
1309 )
1310 return JSONResponse(
1311 content={
1312 "status": "error",
1313 "message": "Research is already running.",
1314 },
1315 status_code=409,
1316 )
1317 except SystemAtCapacityError:
1318 # System at concurrent-research capacity. Roll back the rows
1319 # committed above (UserActiveResearch + IN_PROGRESS history)
1320 # and return 429 so the client can retry shortly.
1321 logger.warning(
1322 f"SystemAtCapacityError on direct submission for "
1323 f"{research_id}; rolling back orphan rows"
1324 )
1325 try:
1326 with get_user_db_session(username) as cleanup_session:
1327 stale_active = (
1328 cleanup_session.query(UserActiveResearch)
1329 .filter_by(username=username, research_id=research_id)
1330 .first()
1331 )
1332 if stale_active:
1333 cleanup_session.delete(stale_active)
1334 cleanup_session.query(ResearchHistory).filter_by(
1335 id=research_id
1336 ).delete()
1337 cleanup_session.commit()
1338 except Exception:
1339 logger.exception(
1340 "Cleanup after SystemAtCapacityError raised; "
1341 "leaving orphan rows for the reconciler"
1342 )
1343 return JSONResponse(
1344 content={
1345 "status": "error",
1346 "message": "Server is at research capacity. Please retry shortly.",
1347 },
1348 status_code=429,
1349 )
1350 except Exception:
1351 logger.exception(
1352 f"Failed to spawn research thread for {research_id}"
1353 )
1354 try:
1355 with get_user_db_session(username) as cleanup_session:
1356 stale_active = (
1357 cleanup_session.query(UserActiveResearch)
1358 .filter_by(username=username, research_id=research_id)
1359 .first()
1360 )
1361 if stale_active: 1361 ↛ 1363line 1361 didn't jump to line 1363 because the condition on line 1361 was always true
1362 cleanup_session.delete(stale_active)
1363 research_row = (
1364 cleanup_session.query(ResearchHistory)
1365 .filter_by(id=research_id)
1366 .first()
1367 )
1368 if research_row: 1368 ↛ 1370line 1368 didn't jump to line 1370 because the condition on line 1368 was always true
1369 research_row.status = ResearchStatus.FAILED
1370 cleanup_session.commit()
1371 except Exception:
1372 logger.exception(
1373 "Cleanup after spawn failure raised; leaving "
1374 "orphan rows for the reconciler to handle"
1375 )
1376 return JSONResponse(
1377 content={
1378 "status": "error",
1379 "message": "Failed to start research. Please try again.",
1380 },
1381 status_code=500,
1382 )
1384 # Update the active research record with the actual thread ID.
1385 try:
1386 with get_user_db_session(username) as thread_session:
1387 active_record = (
1388 thread_session.query(UserActiveResearch)
1389 .filter_by(username=username, research_id=research_id)
1390 .first()
1391 )
1392 if active_record: 1392 ↛ 1398line 1392 didn't jump to line 1398
1393 active_record.thread_id = str(research_thread.ident)
1394 thread_session.commit()
1395 except Exception:
1396 logger.warning("Could not update thread ID")
1398 return {"status": "success", "research_id": research_id}
1401@router.post("/api/terminate/{research_id}")
1402def terminate_research(
1403 request: Request,
1404 research_id,
1405 username: Annotated[str, Depends(require_auth)],
1406):
1407 """Terminate an in-progress research process"""
1409 # Check if the research exists and is in progress
1410 try:
1411 with get_user_db_session(username) as db_session:
1412 research = (
1413 db_session.query(ResearchHistory)
1414 .filter_by(id=research_id)
1415 .first()
1416 )
1418 if not research:
1419 return _research_not_found(research_id)
1421 status = research.status
1423 # If it's already in a terminal state, return success
1424 if status in (
1425 ResearchStatus.COMPLETED,
1426 ResearchStatus.SUSPENDED,
1427 ResearchStatus.FAILED,
1428 ResearchStatus.ERROR,
1429 ):
1430 return {
1431 "status": "success",
1432 "message": f"Research already {status}",
1433 "research_status": status,
1434 }
1436 # Check if it's in the active_research dict
1437 if not is_research_active(research_id):
1438 # The worker may not be registered in _active_research yet: a
1439 # just-submitted research commits its IN_PROGRESS row before the
1440 # worker thread registers itself (spawn-grace window). Set the
1441 # termination flag anyway so a worker that starts right after
1442 # this still sees it and aborts at its first checkpoint —
1443 # otherwise the user's Stop is silently ignored and the research
1444 # runs to completion (overwriting this SUSPENDED status). The
1445 # flag is harmless if no worker ever starts.
1446 set_termination_flag(research_id)
1447 if status == ResearchStatus.QUEUED:
1448 claimed_queue_row = (
1449 db_session.query(QueuedResearch.id)
1450 .filter(
1451 QueuedResearch.research_id == research_id,
1452 QueuedResearch.is_processing.is_(True),
1453 )
1454 .exists()
1455 )
1456 suspended = (
1457 db_session.query(ResearchHistory)
1458 .filter(
1459 ResearchHistory.id == research_id,
1460 ResearchHistory.status == ResearchStatus.QUEUED,
1461 ~claimed_queue_row,
1462 )
1463 .update(
1464 {ResearchHistory.status: ResearchStatus.SUSPENDED},
1465 synchronize_session=False,
1466 )
1467 )
1468 if suspended: 1468 ↛ 1476line 1468 didn't jump to line 1476 because the condition on line 1468 was always true
1469 from ..queue.lifecycle_cleanup import (
1470 cleanup_queued_research_state,
1471 )
1473 cleanup_queued_research_state(db_session, [research_id])
1474 else:
1475 research.status = ResearchStatus.SUSPENDED
1476 db_session.commit()
1477 return {"status": "success", "message": "Research terminated"}
1479 # Set the termination flag
1480 set_termination_flag(research_id)
1482 # Log the termination request - using UTC timestamp
1483 timestamp = datetime.now(UTC).isoformat()
1484 termination_message = "Research termination requested by user"
1485 current_progress = get_research_field(research_id, "progress", 0)
1487 # Create log entry
1488 log_entry = {
1489 "time": timestamp,
1490 "message": termination_message,
1491 "progress": current_progress,
1492 "metadata": {"phase": "termination"},
1493 }
1495 # Add to in-memory log
1496 append_research_log(research_id, log_entry)
1498 # Add to database log
1499 logger.log("MILESTONE", f"Research ended: {termination_message}")
1501 # Update the log in the database
1502 if research.progress_log: 1502 ↛ 1511line 1502 didn't jump to line 1511 because the condition on line 1502 was always true
1503 try:
1504 if isinstance(research.progress_log, str):
1505 current_log = json.loads(research.progress_log)
1506 else:
1507 current_log = research.progress_log
1508 except Exception:
1509 current_log = []
1510 else:
1511 current_log = []
1513 current_log.append(log_entry)
1514 research.progress_log = current_log
1515 research.status = ResearchStatus.SUSPENDED
1516 db_session.commit()
1518 # Emit a socket event for the termination request. Must be
1519 # scoped to subscribers of this research_id — a roomless emit
1520 # broadcasts the user's SUSPENDED status to every connected
1521 # client on the default namespace.
1522 try:
1523 event_data = {
1524 "status": ResearchStatus.SUSPENDED,
1525 "message": "Research was suspended by user request",
1526 }
1528 from ..services.socketio_asgi import emit_to_subscribers
1530 emit_to_subscribers(
1531 "research_progress",
1532 research_id,
1533 event_data,
1534 owner=username,
1535 )
1537 except Exception:
1538 logger.exception("Socket emit error (non-critical)")
1540 return {
1541 "status": "success",
1542 "message": "Research termination requested",
1543 }
1545 except Exception:
1546 logger.exception("Error terminating research")
1547 return JSONResponse(
1548 {"status": "error", "message": "Failed to terminate research"},
1549 status_code=500,
1550 )
1553@router.delete("/api/delete/{research_id}")
1554def delete_research(
1555 request: Request,
1556 research_id,
1557 username: Annotated[str, Depends(require_auth)],
1558):
1559 """Delete a research record"""
1561 try:
1562 with get_user_db_session(username) as db_session:
1563 research = (
1564 db_session.query(ResearchHistory)
1565 .filter_by(id=research_id)
1566 .first()
1567 )
1569 if not research:
1570 return _research_not_found(research_id)
1572 status = research.status
1573 report_path = research.report_path
1575 # Don't allow deleting research in progress
1576 # Don't allow deleting research in progress.
1577 if status == ResearchStatus.IN_PROGRESS:
1578 return JSONResponse(
1579 {
1580 "status": "error",
1581 "message": "Cannot delete research that is in progress",
1582 },
1583 status_code=400,
1584 )
1586 # Delete the DB record atomically, but never out from under an
1587 # active run: skip rows still IN_PROGRESS or claimed by a
1588 # processing queue row (#5074), so a queued/running research can't
1589 # be deleted from beneath the worker.
1590 claimed_queue_row = (
1591 db_session.query(QueuedResearch.id)
1592 .filter(
1593 QueuedResearch.research_id == research_id,
1594 QueuedResearch.is_processing.is_(True),
1595 )
1596 .exists()
1597 )
1598 deleted = (
1599 db_session.query(ResearchHistory)
1600 .filter(
1601 ResearchHistory.id == research_id,
1602 ResearchHistory.status != ResearchStatus.IN_PROGRESS,
1603 ~claimed_queue_row,
1604 )
1605 .delete(synchronize_session=False)
1606 )
1607 if not deleted:
1608 return JSONResponse(
1609 {
1610 "status": "error",
1611 "message": "Cannot delete research that is in progress",
1612 },
1613 status_code=400,
1614 )
1616 # Delete report file if it exists. `report_path` comes from the
1617 # DB column — normally written by the research worker pointing at
1618 # the user's reports dir, but a corrupted row (or future buggy
1619 # writer) could set it to `../../etc/passwd`. Resolve and confirm
1620 # the path is inside the reports root before unlinking.
1621 if report_path: 1621 ↛ 1622line 1621 didn't jump to line 1622 because the condition on line 1621 was never true
1622 try:
1623 from ...config.paths import (
1624 get_research_outputs_directory,
1625 )
1627 reports_root = get_research_outputs_directory().resolve()
1628 resolved = Path(report_path).resolve()
1629 if resolved.exists() and resolved.is_relative_to(
1630 reports_root
1631 ):
1632 resolved.unlink()
1633 else:
1634 logger.warning(
1635 "Refusing to unlink report_path outside reports root: {}",
1636 report_path,
1637 )
1638 except Exception:
1639 logger.exception("Error removing report file")
1641 from ..queue.lifecycle_cleanup import cleanup_queued_research_state
1643 cleanup_queued_research_state(db_session, [research_id])
1644 db_session.commit()
1646 return {"status": "success"}
1647 except Exception:
1648 logger.exception("Error deleting research")
1649 return JSONResponse(
1650 {"status": "error", "message": "Failed to delete research"},
1651 status_code=500,
1652 )
1655@router.post("/api/clear_history")
1656def clear_history(
1657 request: Request, username: Annotated[str, Depends(require_auth)]
1658):
1659 """Clear all research history"""
1661 try:
1662 with get_user_db_session(username) as db_session:
1663 # Get all research records first to clean up files. Select
1664 # only id + report_path (the columns the cleanup loop uses)
1665 # so clearing history doesn't load every report body into
1666 # memory (#4560).
1667 research_records = db_session.query(
1668 ResearchHistory.id,
1669 ResearchHistory.report_path,
1670 ResearchHistory.status,
1671 ).all()
1673 # Get IDs of currently active research (snapshot)
1674 protected_ids = set(get_active_research_ids())
1675 protected_ids.update(
1676 research.id
1677 for research in research_records
1678 if research.status == ResearchStatus.IN_PROGRESS
1679 )
1680 deletable_ids = [
1681 research.id
1682 for research in research_records
1683 if research.id not in protected_ids
1684 ]
1686 deleted_ids = []
1687 for research_id in deletable_ids:
1688 claimed_queue_row = (
1689 db_session.query(QueuedResearch.id)
1690 .filter(
1691 QueuedResearch.research_id == research_id,
1692 QueuedResearch.is_processing.is_(True),
1693 )
1694 .exists()
1695 )
1696 deleted = (
1697 db_session.query(ResearchHistory)
1698 .filter(
1699 ResearchHistory.id == research_id,
1700 ResearchHistory.status != ResearchStatus.IN_PROGRESS,
1701 ~claimed_queue_row,
1702 )
1703 .delete(synchronize_session=False)
1704 )
1705 if deleted:
1706 deleted_ids.append(research_id)
1708 if deleted_ids:
1709 for research in research_records:
1710 if research.id not in deleted_ids:
1711 continue
1713 if ( 1713 ↛ 1717line 1713 didn't jump to line 1717 because the condition on line 1713 was never true
1714 research.report_path
1715 and Path(research.report_path).exists()
1716 ):
1717 try:
1718 Path(research.report_path).unlink()
1719 except Exception:
1720 logger.exception("Error removing report file")
1722 from ..queue.lifecycle_cleanup import (
1723 cleanup_queued_research_state,
1724 )
1726 cleanup_queued_research_state(db_session, deleted_ids)
1728 db_session.commit()
1730 return {"status": "success"}
1731 except Exception:
1732 logger.exception("Error clearing history")
1733 return JSONResponse(
1734 {"status": "error", "message": "Failed to process request"},
1735 status_code=500,
1736 )
1739@router.post("/open_file_location")
1740def open_file_location(
1741 request: Request, username: Annotated[str, Depends(require_auth)]
1742):
1743 """Open a file location in the system file explorer.
1745 Security: This endpoint is disabled for server deployments.
1746 It only makes sense for desktop usage where the server and client are on the same machine.
1747 """
1748 return JSONResponse(
1749 {
1750 "status": "error",
1751 "message": "This feature is disabled. It is only available in desktop mode.",
1752 },
1753 status_code=403,
1754 )
1757@router.post("/api/save_raw_config")
1758async def save_raw_config(
1759 request: Request, username: Annotated[str, Depends(require_auth)]
1760):
1761 """Save raw configuration"""
1762 data = await request.json()
1763 if not isinstance(data, dict):
1764 return json_body_error("success", "Request body must be valid JSON")
1765 raw_config = data.get("raw_config")
1767 if not raw_config:
1768 return JSONResponse(
1769 {"success": False, "error": "Raw configuration is required"},
1770 status_code=400,
1771 )
1773 # Security: Parse and validate the TOML to block dangerous keys
1774 try:
1775 import tomllib
1776 except ImportError:
1777 import tomli as tomllib # type: ignore[no-redef]
1779 try:
1780 # Offloaded: tomllib is PURE PYTHON, and this body is capped at 16 MB
1781 # (fastapi_app.py's body-size limit, whose own note budgets "roughly
1782 # 60 ms" — a figure calibrated for C-speed json.loads). Measured here,
1783 # adversarial-but-valid TOML parses at roughly 100 ms/MB, so a 16 MB
1784 # document is ~5.5 s of straight-line CPU. On the event loop, with the
1785 # single-worker deployment this ships as, that stalls EVERY request for
1786 # the duration, repeatably, for any authenticated user.
1787 #
1788 # asyncio.to_thread rather than run_db_sync: this opens no DB session,
1789 # and run_db_sync's docstring says to prefer to_thread for purely
1790 # CPU-bound work. The GIL is released periodically during the parse, so
1791 # the loop interleaves instead of stopping dead.
1792 #
1793 # Note this runs BEFORE the system.allow_config_write gate (enforced
1794 # inside write_file_verified below), so it is reachable even on
1795 # deployments that have config writing turned off. Offloading is what
1796 # makes that ordering safe; it is not an argument for keeping it.
1797 parsed_config = await asyncio.to_thread(tomllib.loads, raw_config)
1798 except Exception:
1799 logger.warning("Invalid TOML configuration")
1800 # Don't expose internal exception details to users (CWE-209)
1801 return JSONResponse(
1802 {
1803 "success": False,
1804 "error": "Invalid TOML syntax. Please check your configuration format.",
1805 },
1806 status_code=400,
1807 )
1809 # Security: Check for dangerous keys that could enable code execution
1810 # These patterns match keys used for dynamic module imports
1811 BLOCKED_KEY_PATTERNS = ["module_path", "class_name", "module", "class"]
1813 def find_blocked_keys(obj, path=""):
1814 """Recursively find any blocked keys in the config."""
1815 blocked = []
1816 if isinstance(obj, dict):
1817 for key, value in obj.items():
1818 current_path = f"{path}.{key}" if path else key
1819 key_lower = key.lower()
1820 for pattern in BLOCKED_KEY_PATTERNS:
1821 if pattern in key_lower:
1822 blocked.append(current_path)
1823 break
1824 # Recurse into nested dicts
1825 blocked.extend(find_blocked_keys(value, current_path))
1826 elif isinstance(obj, list):
1827 for i, item in enumerate(obj):
1828 blocked.extend(find_blocked_keys(item, f"{path}[{i}]"))
1829 return blocked
1831 blocked_keys = find_blocked_keys(parsed_config)
1832 if blocked_keys:
1833 logger.warning(
1834 f"Security: Blocked attempt to write config with dangerous keys: {blocked_keys}"
1835 )
1836 return JSONResponse(
1837 {
1838 "success": False,
1839 "error": "Configuration contains protected keys that cannot be modified",
1840 "blocked_keys": blocked_keys,
1841 },
1842 status_code=403,
1843 )
1845 try:
1846 from ...security.file_write_verifier import write_file_verified
1848 # Get the config file path (uses centralized path config, respects LDR_DATA_DIR)
1849 config_dir = get_config_directory()
1850 config_path = config_dir / "config.toml"
1852 # Write the configuration to file. Threadpooled: the verifier
1853 # resolves the allow-flag setting (may open the user's DB session)
1854 # and does sync disk IO — neither belongs on the event loop.
1855 await run_db_sync(
1856 write_file_verified,
1857 config_path,
1858 raw_config,
1859 "system.allow_config_write",
1860 context="system configuration file",
1861 )
1863 return {"success": True}
1864 except Exception:
1865 logger.exception("Error saving configuration file")
1866 return JSONResponse(
1867 {"success": False, "error": "Failed to process request"},
1868 status_code=500,
1869 )
1872@router.get("/api/history")
1873def get_history(
1874 request: Request, username: Annotated[str, Depends(require_auth)]
1875):
1876 """Get research history.
1878 NOTE: Returns a raw list of items. The newer canonical endpoint
1879 ``GET /history/api`` (in routers/history.py) returns the wrapped
1880 shape ``{"status": "success", "items": [...]}`` — JS callers
1881 should prefer that one. Kept here because Puppeteer tests still
1882 target this path; consolidating to a single endpoint with a
1883 consistent shape is tracked as a follow-up cleanup.
1884 """
1886 # Bound the result set. Without a limit this endpoint loaded every
1887 # research row (and its research_meta JSON, which can hold a settings
1888 # snapshot) into memory at once (#4560). Mirrors the clamp used by the
1889 # symmetric /history/api endpoint.
1890 try:
1891 limit = int(request.query_params.get("limit", 200))
1892 except (TypeError, ValueError):
1893 limit = 200
1894 limit = max(1, min(limit, 500))
1895 try:
1896 offset = max(0, int(request.query_params.get("offset", 0)))
1897 except (TypeError, ValueError):
1898 offset = 0
1900 try:
1901 with get_user_db_session(username) as db_session:
1902 # Query research history ordered by created_at. Project
1903 # only the metadata columns the loop below consumes — never
1904 # the large ``report_content`` Text body — so this listing
1905 # doesn't pull every report into memory (#4560). This mirrors
1906 # the projection used by the symmetric /history/api endpoint.
1907 research_records = (
1908 db_session.query(
1909 ResearchHistory.id,
1910 ResearchHistory.title,
1911 ResearchHistory.query,
1912 ResearchHistory.mode,
1913 ResearchHistory.status,
1914 ResearchHistory.created_at,
1915 ResearchHistory.completed_at,
1916 ResearchHistory.research_meta,
1917 ResearchHistory.chat_session_id,
1918 )
1919 .order_by(ResearchHistory.created_at.desc())
1920 .limit(limit)
1921 .offset(offset)
1922 .all()
1923 )
1925 # Pre-compute Document counts in a single GROUP BY query
1926 # to avoid an N+1 SELECT-COUNT-per-row inside the loop. The
1927 # symmetric /history/api endpoint in web/routers/history.py
1928 # already uses an outerjoin + group_by — this brings
1929 # /api/history to parity for users with deep history.
1930 research_ids = [r.id for r in research_records]
1931 if research_ids:
1932 doc_count_rows = (
1933 db_session.query(
1934 Document.research_id, func.count(Document.id)
1935 )
1936 .filter(Document.research_id.in_(research_ids))
1937 .group_by(Document.research_id)
1938 .all()
1939 )
1940 doc_counts = dict(doc_count_rows)
1941 else:
1942 doc_counts = {}
1944 # Build history items while session is active to avoid
1945 # DetachedInstanceError on ORM attribute access
1946 history_items = []
1947 for research in research_records:
1948 # Calculate duration if completed
1949 duration_seconds = None
1950 if research.completed_at and research.created_at:
1951 try:
1952 duration_seconds = calculate_duration(
1953 research.created_at, research.completed_at
1954 )
1955 except Exception:
1956 logger.exception("Error calculating duration")
1958 # Look up the pre-computed document count.
1959 doc_count = doc_counts.get(research.id, 0)
1961 # Create a history item
1962 item = {
1963 "id": research.id,
1964 "query": research.query,
1965 "mode": research.mode,
1966 "status": research.status,
1967 "created_at": research.created_at,
1968 "completed_at": research.completed_at,
1969 "duration_seconds": duration_seconds,
1970 "metadata": filter_research_metadata(
1971 research.research_meta
1972 ),
1973 "document_count": doc_count,
1974 }
1975 if research.chat_session_id is not None: 1975 ↛ 1976line 1975 didn't jump to line 1976 because the condition on line 1975 was never true
1976 item["metadata"]["chat_session_id"] = (
1977 research.chat_session_id
1978 )
1980 # Add title if it exists
1981 if hasattr(research, "title") and research.title is not None:
1982 item["title"] = research.title
1984 history_items.append(item)
1986 return {"status": "success", "items": history_items}
1987 except Exception:
1988 logger.exception("Error getting history")
1989 return JSONResponse(
1990 {"status": "error", "message": "Failed to process request"},
1991 status_code=500,
1992 )
1995@router.get("/api/research/{research_id}")
1996def get_research_details(
1997 request: Request,
1998 research_id,
1999 username: Annotated[str, Depends(require_auth)],
2000):
2001 """Get full details of a research using ORM"""
2003 try:
2004 with get_user_db_session(username) as db_session:
2005 research = (
2006 db_session.query(ResearchHistory)
2007 .filter(ResearchHistory.id == research_id)
2008 .first()
2009 )
2011 if not research:
2012 return _research_not_found(research_id)
2014 return {
2015 "id": research.id,
2016 "query": research.query,
2017 "status": research.status,
2018 "progress": research.progress,
2019 "progress_percentage": research.progress or 0,
2020 "mode": research.mode,
2021 "created_at": research.created_at,
2022 "completed_at": research.completed_at,
2023 "report_path": research.report_path,
2024 "metadata": strip_settings_snapshot(research.research_meta),
2025 }
2026 except Exception:
2027 logger.exception("Error getting research details")
2028 return JSONResponse(
2029 {"error": "An internal error has occurred"}, status_code=500
2030 )
2033@router.get("/api/research/{research_id}/logs")
2034def get_research_logs(
2035 request: Request,
2036 research_id,
2037 username: Annotated[str, Depends(require_auth)],
2038):
2039 """Get logs for a specific research.
2041 Accepts an optional ``?limit=N`` that bounds the response to ``N`` rows
2042 (returned oldest-first, matching the default ordering) and is
2043 clamped to ``[1, HISTORY_LOGS_HARD_CAP]`` so a client cannot force an
2044 unbounded load — a long langgraph run can persist thousands of rows. When
2045 ``?limit`` is absent or not a valid integer (``type=int`` yields
2046 ``None``) the historical contract is preserved: every row is returned. The
2047 frontend log panel always sends a valid limit and
2048 ``priority=diagnostic``. That mode does a best-effort newest-first
2049 selection prioritizing errors/CRITICAL/FATAL, then warnings, then
2050 milestones/SUCCESS, before filling the remaining window with routine rows.
2051 If the number of higher-priority rows exceeds the limit, the oldest
2052 diagnostics in those categories and all lower-priority/routine rows are
2053 dropped.
2054 Direct API callers retain newest-N behavior unless they opt in.
2056 Each ``timestamp`` is a raw ``datetime`` that FastAPI's encoder
2057 serializes as ISO 8601, matching the streaming ``/logs/export``
2058 endpoint. (Under Flask this endpoint emitted RFC 822 HTTP-dates via
2059 ``jsonify`` and the two wire formats disagreed; the migration to
2060 FastAPI aligned them.)
2061 """
2063 _limit_raw = request.query_params.get("limit")
2064 try:
2065 limit = int(_limit_raw) if _limit_raw is not None else None
2066 except (TypeError, ValueError):
2067 limit = None
2068 if limit is not None:
2069 limit = max(1, min(limit, HISTORY_LOGS_HARD_CAP))
2070 prioritize_diagnostics = (
2071 request.query_params.get("priority") == "diagnostic"
2072 )
2074 try:
2075 # First check if the research exists
2076 with get_user_db_session(username) as db_session:
2077 research = (
2078 db_session.query(ResearchHistory)
2079 .filter_by(id=research_id)
2080 .first()
2081 )
2082 if not research:
2083 return _research_not_found(research_id)
2085 # Get logs from research_logs table
2086 log_query = db_session.query(ResearchLog).filter_by(
2087 research_id=research_id
2088 )
2089 if limit is None:
2090 log_results = log_query.order_by(
2091 ResearchLog.timestamp, ResearchLog.id
2092 ).all()
2093 elif prioritize_diagnostics:
2094 normalized_level = func.lower(ResearchLog.level)
2095 diagnostic_priority = case(
2096 (
2097 normalized_level.in_(("error", "critical", "fatal")),
2098 0,
2099 ),
2100 (normalized_level == "warning", 1),
2101 (
2102 normalized_level.in_(("milestone", "success")),
2103 2,
2104 ),
2105 else_=3,
2106 )
2107 log_results = (
2108 log_query.order_by(
2109 diagnostic_priority,
2110 ResearchLog.timestamp.desc(),
2111 ResearchLog.id.desc(),
2112 )
2113 .limit(limit)
2114 .all()
2115 )
2116 log_results.sort(key=lambda row: (row.timestamp, row.id))
2117 else:
2118 # Take the newest ``limit`` rows at the SQL layer, then flip
2119 # back to oldest-first so the response ordering is unchanged.
2120 # ``id`` is the tie-break: timestamps are not unique, so without
2121 # it the rows that survive ``.limit()`` at a shared-timestamp
2122 # boundary would be SQL-undefined.
2123 log_results = list(
2124 reversed(
2125 log_query.order_by(
2126 ResearchLog.timestamp.desc(),
2127 ResearchLog.id.desc(),
2128 )
2129 .limit(limit)
2130 .all()
2131 )
2132 )
2134 # Extract log attributes while session is active
2135 # to avoid DetachedInstanceError on ORM attribute access
2136 logs = []
2137 for row in log_results:
2138 logs.append(
2139 {
2140 "id": row.id,
2141 "message": row.message,
2142 "timestamp": row.timestamp,
2143 "log_type": row.level,
2144 }
2145 )
2147 return logs
2149 except Exception:
2150 logger.exception("Error getting research logs")
2151 return JSONResponse(
2152 {"error": "An internal error has occurred"}, status_code=500
2153 )
2156def _log_export_exempt(request: Request) -> bool:
2157 """Keep HEAD pre-flights outside the GET export quota.
2159 Port of main's ``_is_log_export_rate_limit_exempt`` (#5369). Main's
2160 version read Flask's ``request`` global, which does not exist here;
2161 slowapi inspects the callable's signature and passes the Starlette
2162 ``Request`` when it declares exactly one parameter (see
2163 ``slowapi/wrappers.py``'s ``_exempt_when_takes_request``), so the
2164 method is read off the live request instead.
2166 Still needed after the migration: Starlette answers HEAD on a route
2167 registered with ``@router.get``, so a browser or proxy pre-flight
2168 would otherwise burn a slot of the caller's 10/min export budget
2169 without transferring a byte.
2170 """
2171 return request.method == "HEAD" or _api_exempt()
2174_log_export_limit = limiter.shared_limit(
2175 "10 per minute",
2176 scope="log_export",
2177 key_func=_api_user_key,
2178 exempt_when=_log_export_exempt,
2179)
2182# GET *and* HEAD. The log panel's "Download Logs" button issues a HEAD
2183# pre-flight (static/js/components/logpanel.js:1854) so the browser never
2184# saves a JSON error body as a .jsonl file, and it bails out on a non-ok
2185# response before creating the download anchor. Under Flask that worked
2186# for free: werkzeug adds HEAD to any rule that serves GET
2187# (routing/rules.py: `if "HEAD" not in methods and "GET" in methods`).
2188# FastAPI takes `methods` literally, so a bare @router.get answers HEAD
2189# with 405 -- the pre-flight fails, the function returns early, and the
2190# export silently never starts. Serving both verbs restores the Flask
2191# behaviour; Starlette runs the handler and discards the body for HEAD,
2192# exactly as werkzeug did.
2193# Registered as two routes rather than one api_route(methods=["GET","HEAD"]):
2194# a single route with both verbs makes FastAPI emit two OpenAPI operations
2195# that share one generated operationId, which
2196# test_route_contracts.py::test_in_process_openapi_covers_schema_included_routes
2197# correctly rejects as a duplicate. GET stays the documented operation; the
2198# HEAD twin is include_in_schema=False because it is an implementation detail
2199# of the pre-flight, exactly as werkzeug's auto-added HEAD never appeared in
2200# any schema either.
2201@router.head("/api/research/{research_id}/logs/export", include_in_schema=False)
2202@router.get("/api/research/{research_id}/logs/export")
2203@_log_export_limit
2204def export_research_logs(
2205 request: Request,
2206 research_id: str,
2207 username: Annotated[str, Depends(require_auth)],
2208):
2209 """Stream every persisted log row for a research as newline-delimited JSON.
2211 This endpoint deliberately bypasses ``HISTORY_LOGS_HARD_CAP``: a long
2212 langgraph run can persist tens of thousands of rows and the on-screen
2213 log panel rightly caps at ``?limit=5000`` to bound DOM and parsing
2214 memory. A *download*, by contrast, is short-lived — the browser's
2215 download manager streams the response body straight to disk — so the
2216 cap does not protect anything here and only hides data.
2218 Memory layout for very large exports:
2220 * **Server**: ``get_user_db_session`` opens a fresh session inside
2221 the generator so it survives until the response finishes flushing;
2222 ``.yield_per(500)`` pulls rows in 500-row batches from the SQLite
2223 cursor so the full set is never resident in process memory.
2224 * **Wire**: each yielded row is a single ``\\n``-terminated JSON
2225 line, so the server flushes incrementally and the browser receives
2226 chunks as they arrive.
2227 * **Client**: ``Content-Disposition: attachment`` plus the
2228 ``log-download-button`` JS trigger a native browser download, so
2229 the response body streams to disk without first being buffered
2230 into a JS Blob.
2232 The NDJSON format is the standard streaming interchange for
2233 line-oriented records. Each line emits ``id``, ``timestamp``, ``message``,
2234 ``level`` (log level), ``log_type`` (aligned with ``/logs`` endpoint),
2235 ``module``, and ``line_no``. Consumers can ``for line in file: json.loads(line)``
2236 without parsing a wrapper array. ``timestamp`` is serialized explicitly
2237 with ``datetime.isoformat()`` (ISO 8601), which matches how FastAPI's
2238 encoder renders the datetimes returned by the sibling ``/logs`` endpoint.
2239 """
2240 # Verify the research exists (and belongs to this user, since
2241 # ``get_user_db_session(username)`` scopes queries to the user's
2242 # encrypted DB). Failing fast here avoids opening a streaming session
2243 # for a 404.
2244 try:
2245 with get_user_db_session(username) as db_session:
2246 research = (
2247 db_session.query(ResearchHistory)
2248 .filter_by(id=research_id)
2249 .first()
2250 )
2251 if not research:
2252 return _research_not_found(research_id)
2253 except Exception:
2254 logger.exception("Error verifying research for log export")
2255 return JSONResponse(
2256 {"error": "An internal error has occurred"}, status_code=500
2257 )
2259 def generate():
2260 # NEVER hold a DB session across a yield: Starlette iterates sync
2261 # generators via anyio's threadpool, so each next() can land on a
2262 # different OS thread and a session entered on one thread would be
2263 # exited on another (see download_bulk for the thread-affinity
2264 # rationale). Snapshot the ordered id list in one short-lived
2265 # session, then hydrate rows in per-batch sessions, serializing
2266 # each batch fully BEFORE yielding it.
2267 yielded_count = 0
2268 try:
2269 with get_user_db_session(username) as db_session:
2270 ordered_ids = [
2271 row_id
2272 for (row_id,) in db_session.query(ResearchLog.id)
2273 .filter_by(research_id=research_id)
2274 .order_by(ResearchLog.timestamp.asc(), ResearchLog.id.asc())
2275 .all()
2276 ]
2278 batch_size = 500
2279 for start in range(0, len(ordered_ids), batch_size):
2280 batch_ids = ordered_ids[start : start + batch_size]
2281 lines = []
2282 with get_user_db_session(username) as db_session:
2283 rows = (
2284 db_session.query(ResearchLog)
2285 .filter(ResearchLog.id.in_(batch_ids))
2286 .all()
2287 )
2288 by_id = {row.id: row for row in rows}
2289 # Serialize inside the session (ORM attribute access
2290 # may lazy-load); emit in snapshot order so the
2291 # timestamp-then-id ordering is preserved exactly.
2292 for row_id in batch_ids:
2293 row = by_id.get(row_id)
2294 if row is None:
2295 continue
2296 yielded_count += 1
2297 lines.append(
2298 json.dumps(
2299 {
2300 "id": row.id,
2301 "timestamp": row.timestamp.isoformat()
2302 if row.timestamp is not None
2303 else None,
2304 "message": row.message,
2305 "level": row.level,
2306 "log_type": row.level,
2307 "module": row.module,
2308 "line_no": row.line_no,
2309 },
2310 default=str,
2311 )
2312 + "\n"
2313 )
2314 yield "".join(lines)
2315 except Exception:
2316 # If the DB blows up mid-stream we have already sent a 200 +
2317 # partial body, so we can't recover with a JSON error here.
2318 # Log and let the iterator end; the client will see a
2319 # truncated file, which is the best we can do without
2320 # buffering the whole result.
2321 logger.exception(
2322 f"Error streaming logs for export (research_id={research_id}, yielded={yielded_count})"
2323 )
2325 safe_id = re.sub(r"[^a-zA-Z0-9_-]", "", research_id)
2326 filename = f"research_logs_{safe_id or 'export'}.jsonl"
2327 return WorkerCleanupStreamingResponse(
2328 generate(),
2329 media_type="application/x-ndjson",
2330 headers={
2331 "Content-Disposition": f'attachment; filename="{filename}"',
2332 # Disable caching: this is a one-shot download, and a stale
2333 # partial body from a previous run could otherwise confuse a
2334 # retry after a mid-stream failure.
2335 "Cache-Control": "no-store",
2336 "X-Content-Type-Options": "nosniff",
2337 },
2338 )
2341@router.get("/api/report/{research_id}")
2342def get_research_report(
2343 request: Request,
2344 research_id,
2345 username: Annotated[str, Depends(require_auth)],
2346):
2347 """Get the research report content"""
2349 try:
2350 with get_user_db_session(username) as db_session:
2351 # Query using ORM
2352 research = (
2353 db_session.query(ResearchHistory)
2354 .filter_by(id=research_id)
2355 .first()
2356 )
2358 if research is None:
2359 return _research_not_found(research_id)
2361 # Parse metadata if it exists
2362 metadata = research.research_meta
2364 # research.report_content holds the answer-only string;
2365 # rebuild the legacy display shape (answer + Sources from
2366 # research_resources + Metrics from research_meta) on demand.
2367 from ..services.report_assembly_service import (
2368 assemble_full_report,
2369 get_research_source_links_batch,
2370 )
2372 content = assemble_full_report(research, db_session)
2373 # Only None means "research not found" — guarded above.
2374 # Empty-but-found rows return "" and are valid responses.
2375 if content is None:
2376 return _research_not_found(
2377 research_id, message="Report not found"
2378 )
2380 # Sources live in the research_resources table, not research_meta.
2381 # The post-refactor save path never writes the legacy
2382 # `all_links_of_system` metadata key, so reading it here returned
2383 # [] for every research created since chat-mode-v2 (#3665). Read
2384 # the structured table instead — the same source of truth the
2385 # assembled `content` and the news feed already use. limit=None
2386 # returns every source (this field was never top-N), matching the
2387 # full list the assembled `content` renders for the same research.
2388 sources = get_research_source_links_batch(
2389 [research.id], db_session, limit=None
2390 ).get(research.id, [])
2392 # Return the report data with backwards-compatible fields
2393 # Examples expect 'summary', 'sources', 'findings' at top level
2394 safe_metadata = strip_settings_snapshot(metadata)
2395 return {
2396 "content": content,
2397 # Backwards-compatible fields for examples
2398 "summary": content, # The markdown report is the summary
2399 "sources": sources,
2400 "findings": safe_metadata.get("findings", []),
2401 "metadata": {
2402 "title": research.title if research.title else None,
2403 "query": research.query,
2404 "mode": research.mode if research.mode else None,
2405 "created_at": research.created_at
2406 if research.created_at
2407 else None,
2408 "completed_at": research.completed_at
2409 if research.completed_at
2410 else None,
2411 "report_path": research.report_path,
2412 **safe_metadata,
2413 },
2414 }
2416 except Exception:
2417 logger.exception("Error getting research report")
2418 return JSONResponse(
2419 {"error": "An internal error has occurred"}, status_code=500
2420 )
2423@router.post("/api/v1/research/{research_id}/export/{format}")
2424def export_research_report(
2425 request: Request,
2426 research_id,
2427 format,
2428 username: Annotated[str, Depends(require_auth)],
2429):
2430 """Export research report to different formats (LaTeX, Quarto, RIS, PDF, ODT, etc.)"""
2431 try:
2432 # Use the exporter registry to validate format
2433 from ...exporters import ExporterRegistry
2435 if not ExporterRegistry.is_format_supported(format):
2436 available = ExporterRegistry.get_available_formats()
2437 return JSONResponse(
2438 {
2439 "error": f"Invalid format. Available formats: {', '.join(available)}"
2440 },
2441 status_code=400,
2442 )
2444 # Get research from database
2446 try:
2447 with get_user_db_session(username) as db_session:
2448 research = (
2449 db_session.query(ResearchHistory)
2450 .filter_by(id=research_id)
2451 .first()
2452 )
2453 if not research:
2454 return _research_not_found(research_id)
2456 # Build the full assembled report (answer + Sources +
2457 # Metrics) so exporters get the same shape they did
2458 # before the report_content refactor.
2459 from ..services.report_assembly_service import (
2460 assemble_full_report,
2461 )
2463 report_content = assemble_full_report(research, db_session)
2464 if report_content is None: 2464 ↛ 2465line 2464 didn't jump to line 2465 because the condition on line 2464 was never true
2465 return _research_not_found(
2466 research_id, message="Report content not found"
2467 )
2469 # Export to requested format (all in memory)
2470 try:
2471 # Use title or query for the PDF title
2472 pdf_title = research.title or research.query
2474 # Generate export content in memory
2475 export_content, filename, mimetype = (
2476 export_report_to_memory(
2477 report_content, format, title=pdf_title
2478 )
2479 )
2481 # Send the file directly from memory.
2482 #
2483 # A plain ``Response``, not ``StreamingResponse``:
2484 # ``export_content`` is an in-memory bytes object that is
2485 # already fully materialised, and iterating a ``BytesIO``
2486 # yields one chunk per ``0x0A`` byte — for binary PDF/ODT
2487 # payloads that is thousands of tiny ASGI sends, each a
2488 # threadpool dispatch, and it also suppresses
2489 # ``Content-Length`` so the browser gets no download
2490 # progress. Same fix already applied to library.py's PDF
2491 # route; this is its sibling call site.
2492 from urllib.parse import quote
2493 from fastapi.responses import Response
2495 # The filename derives from the research title, which
2496 # is user-supplied and may contain non-latin-1 text
2497 # (CJK, Cyrillic, ...). Raw interpolation would raise
2498 # at header encoding and 500 the export — use RFC 5987
2499 # filename* encoding, same as library.py's PDF route.
2500 safe_filename = quote(filename, safe="")
2501 return Response(
2502 content=export_content,
2503 media_type=mimetype,
2504 headers={
2505 "Content-Disposition": (
2506 f"attachment; filename*=UTF-8''{safe_filename}"
2507 ),
2508 },
2509 )
2510 except MissingPDFDependencyError:
2511 logger.exception(
2512 "PDF export failed: WeasyPrint unavailable"
2513 )
2514 return JSONResponse(
2515 {"error": get_weasyprint_install_instructions()},
2516 status_code=500,
2517 )
2518 except Exception:
2519 logger.exception("Error exporting report")
2520 return JSONResponse(
2521 {
2522 "error": f"Failed to export to {format}. Please try again later."
2523 },
2524 status_code=500,
2525 )
2527 except Exception:
2528 logger.exception("Error in export endpoint")
2529 return JSONResponse(
2530 {"error": "An internal error has occurred"}, status_code=500
2531 )
2533 except Exception:
2534 logger.exception("Unexpected error in export endpoint")
2535 return JSONResponse(
2536 {"error": "An internal error has occurred"}, status_code=500
2537 )
2540@router.get("/api/research/{research_id}/status")
2541@limiter.exempt
2542def get_research_status(
2543 request: Request,
2544 research_id,
2545 username: Annotated[str, Depends(require_auth)],
2546):
2547 """Get the status of a research process"""
2549 try:
2550 with get_user_db_session(username) as db_session:
2551 research = (
2552 db_session.query(ResearchHistory)
2553 .filter_by(id=research_id)
2554 .first()
2555 )
2557 if research is None:
2558 return _research_not_found(research_id)
2560 status = research.status
2561 progress = research.progress
2562 completed_at = research.completed_at
2563 report_path = research.report_path
2564 metadata = research.research_meta or {}
2566 # Extract and format error information for better UI display
2567 error_info = {}
2568 if metadata and "error" in metadata:
2569 error_msg = metadata["error"]
2570 error_type = "unknown"
2572 # Detect specific error types
2573 if "timeout" in error_msg.lower():
2574 error_type = "timeout"
2575 error_info = {
2576 "type": "timeout",
2577 "message": "LLM service timed out during synthesis. This may be due to high server load or connectivity issues.",
2578 "suggestion": "Try again later or use a smaller query scope.",
2579 }
2580 elif (
2581 "token limit" in error_msg.lower()
2582 or "context length" in error_msg.lower()
2583 ):
2584 error_type = "token_limit"
2585 error_info = {
2586 "type": "token_limit",
2587 "message": "The research query exceeded the AI model's token limit during synthesis.",
2588 "suggestion": "Try using a more specific query or reduce the research scope.",
2589 }
2590 elif (
2591 "final answer synthesis fail" in error_msg.lower()
2592 or "llm error" in error_msg.lower()
2593 ):
2594 error_type = "llm_error"
2595 error_info = {
2596 "type": "llm_error",
2597 "message": "The AI model encountered an error during final answer synthesis.",
2598 "suggestion": "Check that your LLM service is running correctly or try a different model.",
2599 }
2600 elif "ollama" in error_msg.lower():
2601 error_type = "ollama_error"
2602 error_info = {
2603 "type": "ollama_error",
2604 "message": "The Ollama service is not responding properly.",
2605 "suggestion": "Make sure Ollama is running with 'ollama serve' and the model is downloaded.",
2606 }
2607 elif "connection" in error_msg.lower():
2608 error_type = "connection"
2609 error_info = {
2610 "type": "connection",
2611 "message": "Connection error with the AI service.",
2612 "suggestion": "Check your internet connection and AI service status.",
2613 }
2614 elif metadata.get("solution"):
2615 # Use the solution provided in metadata if available
2616 error_info = {
2617 "type": error_type,
2618 "message": error_msg,
2619 "suggestion": str(metadata.get("solution")),
2620 }
2621 else:
2622 # Generic error with the original message
2623 error_info = {
2624 "type": error_type,
2625 "message": error_msg,
2626 "suggestion": "Try again with a different query or check the application logs.",
2627 }
2629 # Get the latest milestone log for this research
2630 latest_milestone = None
2631 try:
2632 milestone_log = (
2633 db_session.query(ResearchLog)
2634 .filter_by(research_id=research_id, level="MILESTONE")
2635 # id tie-breaks equal timestamps so "latest" is
2636 # deterministic (the most recently inserted milestone).
2637 .order_by(
2638 ResearchLog.timestamp.desc(), ResearchLog.id.desc()
2639 )
2640 .first()
2641 )
2642 if milestone_log:
2643 latest_milestone = {
2644 "message": milestone_log.message,
2645 "time": milestone_log.timestamp.isoformat()
2646 if milestone_log.timestamp
2647 else None,
2648 "type": "MILESTONE",
2649 }
2650 logger.debug(
2651 f"Found latest milestone for research {research_id}: {milestone_log.message}"
2652 )
2653 else:
2654 logger.debug(
2655 f"No milestone logs found for research {research_id}"
2656 )
2657 except Exception:
2658 logger.warning("Error fetching latest milestone")
2660 filtered_metadata = strip_settings_snapshot(metadata)
2661 if error_info:
2662 filtered_metadata["error_info"] = error_info
2664 response_data = {
2665 "status": status,
2666 "progress": progress,
2667 "completed_at": completed_at,
2668 "report_path": report_path,
2669 "metadata": filtered_metadata,
2670 }
2672 # Include latest milestone as a log_entry for frontend compatibility
2673 if latest_milestone:
2674 response_data["log_entry"] = latest_milestone
2676 return response_data
2677 except Exception:
2678 logger.exception("Error getting research status")
2679 return JSONResponse(
2680 {"error": "Error checking research status"}, status_code=500
2681 )
2684@router.get("/api/queue/status")
2685def get_queue_status(
2686 request: Request, username: Annotated[str, Depends(require_auth)]
2687):
2688 """Get the current queue status for the user"""
2690 from ..queue import QueueManager
2692 try:
2693 queue_items = QueueManager.get_user_queue(username)
2695 return {
2696 "status": "success",
2697 "queue": queue_items,
2698 "total": len(queue_items),
2699 }
2700 except Exception:
2701 logger.exception("Error getting queue status")
2702 return JSONResponse(
2703 {"status": "error", "message": "Failed to process request"},
2704 status_code=500,
2705 )
2708@router.get("/api/queue/{research_id}/position")
2709def get_queue_position(
2710 request: Request,
2711 research_id,
2712 username: Annotated[str, Depends(require_auth)],
2713):
2714 """Get the queue position for a specific research"""
2716 from ..queue import QueueManager
2718 try:
2719 position = QueueManager.get_queue_position(username, research_id)
2721 if position is None:
2722 return _research_not_found(
2723 research_id, message="Research not found in queue"
2724 )
2726 return {"status": "success", "position": position}
2727 except Exception:
2728 logger.exception("Error getting queue position")
2729 return JSONResponse(
2730 {"status": "error", "message": "Failed to process request"},
2731 status_code=500,
2732 )
2735@router.get("/api/config/limits")
2736def get_upload_limits(
2737 request: Request, username: Annotated[str, Depends(require_auth)]
2738):
2739 """
2740 Get file upload configuration limits.
2742 Returns the backend's authoritative limits for file uploads,
2743 allowing the frontend to stay in sync without hardcoding values.
2745 Static constants only, but auth-gated to match the pre-migration
2746 Flask route (@login_required) — no endpoint enumeration for free.
2747 """
2748 return {
2749 "max_file_size": FileUploadValidator.MAX_FILE_SIZE,
2750 "max_files": FileUploadValidator.MAX_FILES_PER_REQUEST,
2751 "allowed_mime_types": list(FileUploadValidator.ALLOWED_MIME_TYPES),
2752 }
2755@router.post("/api/upload/pdf")
2756@upload_rate_limit_user
2757@upload_rate_limit_ip
2758async def upload_pdf(
2759 request: Request,
2760 username: Annotated[str, Depends(require_auth)],
2761 files: list = None,
2762):
2763 """
2764 Upload and extract text from PDF files with comprehensive security validation.
2766 Security features:
2767 - Rate limiting per user and per IP (``security.rate_limit_upload_user``
2768 and ``security.rate_limit_upload_ip``)
2769 - File size validation (``FileUploadValidator.MAX_FILE_SIZE``,
2770 configurable via ``security.upload_max_file_size_mb`` /
2771 ``LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB``)
2772 - File count validation (``FileUploadValidator.MAX_FILES_PER_REQUEST``)
2773 - PDF structure validation
2774 - MIME type validation
2776 Performance improvements:
2777 - Single-pass PDF processing (text + metadata)
2778 - Optimized extraction service
2779 """
2780 # Starlette's form parser yields STARLETTE UploadFile instances;
2781 # fastapi.UploadFile is a SUBCLASS of it on fastapi>=0.113, so
2782 # `isinstance(f, fastapi.UploadFile)` is False for every real
2783 # upload and the filter below silently dropped all files
2784 # (every upload 400'd with "No files provided").
2785 from starlette.datastructures import UploadFile
2787 try:
2788 # Early request size validation (before reading any files)
2789 # This prevents memory exhaustion from chunked encoding attacks
2790 max_request_size = (
2791 FileUploadValidator.MAX_FILES_PER_REQUEST
2792 * FileUploadValidator.MAX_FILE_SIZE
2793 )
2794 content_length = request.headers.get("content-length")
2795 if content_length and int(content_length) > max_request_size: 2795 ↛ 2796line 2795 didn't jump to line 2796 because the condition on line 2795 was never true
2796 return JSONResponse(
2797 {
2798 "error": f"Request too large. Maximum size is {max_request_size // (1024 * 1024)}MB"
2799 },
2800 status_code=413,
2801 )
2803 # Parse multipart form data — FastAPI's UploadFile parameter binding
2804 # doesn't work cleanly with the `files: list` shape we want, so we
2805 # parse the form manually and pull out everything named "files".
2806 form = await request.form()
2807 files: list[UploadFile] = [
2808 f for f in form.getlist("files") if isinstance(f, UploadFile)
2809 ]
2811 if not files:
2812 return JSONResponse({"error": "No files provided"}, status_code=400)
2813 if all(not f.filename for f in files):
2814 return JSONResponse({"error": "No files selected"}, status_code=400)
2816 # Validate file count
2817 is_valid, error_msg = FileUploadValidator.validate_file_count(
2818 len(files)
2819 )
2820 if not is_valid:
2821 return JSONResponse({"error": error_msg}, status_code=400)
2823 # Get PDF extraction service
2824 pdf_service = get_pdf_extraction_service()
2826 extracted_texts = []
2827 total_files = len(files)
2828 processed_files = 0
2829 errors = []
2831 for file in files:
2832 if not file or not file.filename: 2832 ↛ 2833line 2832 didn't jump to line 2833 because the condition on line 2832 was never true
2833 errors.append("Unnamed file: Skipped")
2834 continue
2836 try:
2837 filename = sanitize_filename(
2838 file.filename, allowed_extensions={".pdf"}
2839 )
2840 except UnsafeFilenameError:
2841 errors.append("Rejected file: invalid or disallowed filename")
2842 continue
2844 try:
2845 # Read file content (UploadFile spools large bodies to disk)
2846 pdf_content = await file.read()
2848 # Validation + PDF text extraction are CPU-bound (up to
2849 # FileUploadValidator.MAX_FILE_SIZE per file ×
2850 # MAX_FILES_PER_REQUEST files) — run them in the
2851 # threadpool so they don't freeze the event loop (and
2852 # with it every other request and WebSocket) for the
2853 # duration of the parse. No DB session is opened here,
2854 # so a plain to_thread (not run_db_sync) is correct.
2855 def _validate_and_extract(content=pdf_content, name=filename):
2856 ok, err = FileUploadValidator.validate_upload(
2857 filename=name,
2858 file_content=content,
2859 content_length=file.size,
2860 )
2861 if not ok:
2862 return None, err
2863 return (
2864 pdf_service.extract_text_and_metadata(content, name),
2865 None,
2866 )
2868 result, error_msg = await asyncio.to_thread(
2869 _validate_and_extract
2870 )
2872 if result is None:
2873 errors.append(f"{filename}: {error_msg}")
2874 continue
2876 if result["success"]:
2877 extracted_texts.append(
2878 {
2879 "filename": result["filename"],
2880 "text": result["text"],
2881 "size": result["size"],
2882 "pages": result["pages"],
2883 }
2884 )
2885 processed_files += 1
2886 else:
2887 errors.append(f"{filename}: {result['error']}")
2889 except Exception:
2890 logger.exception(f"Error processing {filename}")
2891 errors.append(f"{filename}: Error processing file")
2892 finally:
2893 # Close the file stream to release resources
2894 try:
2895 await file.close()
2896 except Exception:
2897 logger.debug("best-effort file stream close", exc_info=True)
2899 # Prepare response
2900 response_data = {
2901 "status": "success",
2902 "processed_files": processed_files,
2903 "total_files": total_files,
2904 "extracted_texts": extracted_texts,
2905 "combined_text": "\n\n".join(
2906 [
2907 f"--- From {item['filename']} ---\n{item['text']}"
2908 for item in extracted_texts
2909 ]
2910 ),
2911 "errors": errors,
2912 }
2914 if processed_files == 0:
2915 return JSONResponse(
2916 {
2917 "status": "error",
2918 "message": "No files were processed successfully",
2919 "errors": errors,
2920 },
2921 status_code=400,
2922 )
2924 return response_data
2926 except Exception:
2927 logger.exception("Error processing PDF upload")
2928 return JSONResponse(
2929 {"error": "Failed to process PDF files"}, status_code=500
2930 )