Coverage for src/local_deep_research/web/routers/research.py: 94%
1051 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1from fastapi import APIRouter, Depends, HTTPException, Request
2from fastapi.responses import (
3 JSONResponse,
4)
5from python_multipart.exceptions import FormParserError
6from starlette.exceptions import HTTPException as StarletteHTTPException
7from ..dependencies.auth import require_auth
8from ..dependencies.rate_limit import (
9 _api_exempt,
10 _api_user_key,
11 api_rate_limit,
12 limiter,
13 upload_rate_limit_ip,
14 upload_rate_limit_user,
15)
16from ..dependencies.threadpool import (
17 WorkerCleanupStreamingResponse,
18 run_db_sync,
19)
20from ..template_config import templates
21from ..utils.download_headers import build_content_disposition
23# Bounded preview for untrusted request-body values in log lines (#6191).
24from .notes import _log_value_preview
26import asyncio
27import json
28import re
29from datetime import datetime, UTC
30from pathlib import Path
32from loguru import logger
33from ...settings.exceptions import SettingsValueError
34from ...settings.logger import log_settings
35from sqlalchemy import case, func
36from sqlalchemy.orm import Session
37from typing import Optional, Annotated
39# Security imports
40from ...config.constants import DEFAULT_OLLAMA_URL
41from ...exceptions import DuplicateResearchError, SystemAtCapacityError
42from ...error_handling.error_messages import get_known_research_error_message
43from ...constants import HISTORY_LOGS_HARD_CAP, ResearchStatus
44from ...security import (
45 FileUploadValidator,
46 UnsafeFilenameError,
47 filter_research_metadata,
48 sanitize_filename,
49 strip_settings_snapshot,
50)
51from ...security.log_sanitizer import redact_and_bound_for_log
52from ...security.url_redaction import redact_url_credentials
53from ...utilities.type_utils import overlay_runtime_settings, to_bool
54from ...utilities.url_utils import is_safe_custom_llm_endpoint
57from ...config.paths import get_config_directory
59# Services imports
60from ..services.pdf_extraction_service import get_pdf_extraction_service
61from ..services.pdf_service import (
62 MissingPDFDependencyError,
63 get_weasyprint_install_instructions,
64)
66from ...database.models import (
67 QueuedResearch,
68 ResearchHistory,
69 UserActiveResearch,
70)
71from ..models.database import (
72 ResearchLog,
73 calculate_duration,
74)
75from ...database.models.library import Document as Document
76from ...database.research_cascade import (
77 collect_research_document_rag_targets,
78 SharedDownloadIndex,
79 delete_research_with_dependents as _delete_research_with_dependents,
80)
81from ...database.session_context import get_user_db_session
82from ...database.sqlcipher_utils import retry_on_db_lock
83from ...llm.providers.base import normalize_provider
84from ..auth.password_utils import resolve_user_password
85from ..services.research_service import (
86 _parse_research_metadata,
87 clamp_user_max_concurrent,
88 export_report_to_memory,
89 run_research_process,
90 start_research_process,
91)
93from ..routes.research_validation import (
94 ITERATIONS_MAX,
95 ITERATIONS_MIN,
96 QUESTIONS_PER_ITERATION_MAX,
97 QUESTIONS_PER_ITERATION_MIN,
98 validate_research_query_length,
99 validate_search_overrides,
100)
101from ..research_state import (
102 append_research_log,
103 get_active_research_ids,
104 get_research_field,
105 get_user_research_start_lock,
106 is_research_active,
107 set_termination_flag,
108 user_research_start_gate,
109)
110from ...constants import (
111 DEFAULT_SEARCH_TOOL,
112 LANGGRAPH_STRATEGY_ALIASES,
113 LANGGRAPH_STRATEGY_NAME as LANGGRAPH_STRATEGY_NAME,
114)
115from ..dependencies.json_body import (
116 is_json_request,
117 json_body_error,
118 read_json_dict,
119)
120from ..utils.pagination import OFFSET_LIMIT_ERROR, parse_bounded_offset
122# Create the router for the research application
123router = APIRouter(tags=["research"])
125# NOTE: Routes use username (not .get()) intentionally.
126# require_auth guarantees the value is present; direct access fails fast
127# if the dependency is ever removed.
130# Add static route at the root level
131# include_in_schema=False matches the sibling `/static/{path:path}` mount in
132# fastapi_app.py. A `:path` converter cannot be represented in OpenAPI, so a
133# schema-included route using one fails test_route_contracts.py's
134# "every schema-included APIRoute appears in OpenAPI" check. This is a legacy
135# shim for bookmarked URLs, not part of the documented API surface,
136# so excluding it is correct rather than a workaround.
137@router.get("/redirect-static/{path:path}", include_in_schema=False)
138def redirect_static(path: str):
139 """Serve legacy URLs with the same asset and cache policy as /static/.
141 Serving in place removes the user-controlled Location header (CodeQL
142 alert #8204). The shared resolver confines files to the static root,
143 prefers built assets and rejects symlink escapes and missing files.
144 """
145 from ..static_files import static_file_response
147 path = path.lstrip("/")
148 segments = path.split("/")
149 if (
150 not path
151 or "\\" in path
152 or any(segment in {"", ".", ".."} for segment in segments)
153 or any(
154 ord(character) < 32 or ord(character) == 127 for character in path
155 )
156 ):
157 # Raise rather than returning a body: the app's `@app.exception_
158 # handler(404)` (fastapi_app.py) is what renders a missing resource,
159 # and it serves HTML to browser navigations and JSON only to API
160 # callers. This route exists for bookmarked browser URLs, so a
161 # hand-rolled JSON envelope here would show a raw JSON body in the
162 # browser to the one audience the route has.
163 raise HTTPException(status_code=404, detail="Not found")
165 response = static_file_response(path)
166 if response is None:
167 raise HTTPException(status_code=404, detail="Not found")
168 return response
171@router.get("/progress/{research_id}")
172def progress_page(
173 request: Request,
174 research_id,
175 username: Annotated[str, Depends(require_auth)],
176):
177 """Render the research progress page"""
178 return templates.TemplateResponse(
179 request=request,
180 name="pages/progress.html",
181 context={"request": request},
182 )
185@router.get("/details/{research_id}")
186def research_details_page(
187 request: Request,
188 research_id,
189 username: Annotated[str, Depends(require_auth)],
190):
191 """Render the research details page"""
192 return templates.TemplateResponse(
193 request=request, name="pages/details.html", context={"request": request}
194 )
197@router.get("/results/{research_id}")
198def results_page(
199 request: Request,
200 research_id,
201 username: Annotated[str, Depends(require_auth)],
202):
203 """Render the research results page"""
204 return templates.TemplateResponse(
205 request=request, name="pages/results.html", context={"request": request}
206 )
209@router.get("/history")
210def history_page(
211 request: Request, username: Annotated[str, Depends(require_auth)]
212):
213 """Render the history page"""
214 return templates.TemplateResponse(
215 request=request, name="pages/history.html", context={"request": request}
216 )
219# Request-body fields the start-research path uses with str methods:
220# ``model_provider`` goes through ``normalize_provider`` (``.lower()``), the
221# engine (``search_engine``, or its ``search_tool`` alias) through
222# ``_precheck_collection_agent_enabled`` (``.startswith``), and ``strategy``
223# through that precheck's ``.lower()``. JSON null and "" still count as
224# unset, so the usual fallbacks apply; any other non-string is a client
225# error. The async ``start_research`` wrapper rejects it before dispatch,
226# and ``_extract_research_params`` rejects it again for in-process callers
227# that skip the wrapper (the news run-now and overdue paths).
228_STRING_REQUEST_FIELDS = (
229 "model_provider",
230 "search_engine",
231 "search_tool",
232 "strategy",
233)
236def _string_field_error(field):
237 """Return the client-facing 400 message for a non-string ``field``."""
238 return f"{field} must be a string"
241def _non_string_request_field(data):
242 """Return the first ``_STRING_REQUEST_FIELDS`` name in ``data`` whose
243 value is not null and not a ``str``, or None when there is none."""
244 for field in _STRING_REQUEST_FIELDS:
245 value = data.get(field)
246 if value is not None and not isinstance(value, str):
247 return field
248 return None
251def _settings_value_error_response(error):
252 """Refuse a research start because a configured setting is invalid.
254 This is a server-side configuration error, not a request error, so it
255 is a 500 rather than one of the 400 messages for request fields. The
256 message names the key, its source and the reason, never the value.
257 """
258 logger.error("Research start refused: {}", error.diagnostic)
259 return JSONResponse(
260 {"status": "error", "message": error.diagnostic},
261 status_code=500,
262 )
265def _extract_research_params(data, settings_manager):
266 """Extract and resolve research parameters from request data and settings.
268 Returns a dict with keys: model_provider, model, custom_endpoint,
269 ollama_url, search_engine, max_results, time_period, iterations,
270 questions_per_iteration, strategy, policy_egress_scope,
271 llm_require_local_endpoint, embeddings_require_local.
273 Raises ``ValueError`` when a ``_STRING_REQUEST_FIELDS`` value is neither
274 null nor a string (checked first, before any settings lookup), when a
275 string ``custom_endpoint`` (from the request or the saved setting) fails
276 the SSRF guard, or when ``iterations`` / ``questions_per_iteration`` is
277 not an integer in range. ``_start_research_sync`` maps each to a fixed
278 400 message. A settings read raises ``SettingsValueError`` (also a
279 ``ValueError``) for an invalid configured value; that one is returned
280 as a 500 naming the setting instead.
281 """
282 bad_field = _non_string_request_field(data)
283 if bad_field is not None:
284 raise ValueError(_string_field_error(bad_field))
285 model_provider = data.get("model_provider")
286 if not model_provider:
287 model_provider = settings_manager.get_setting("llm.provider", "ollama")
288 logger.debug(
289 f"No model_provider in request, using database setting: {model_provider}"
290 )
291 else:
292 # Raw request-body value: log it through the bounded preview so a
293 # wide dict/list posted for it does not cost a full repr per
294 # request (#6305, same shape as #6191). Same for ``model`` below.
295 logger.opt(lazy=True).debug(
296 "Using model_provider from request: {}",
297 lambda: _log_value_preview(model_provider),
298 )
299 # Normalize provider to lowercase canonical form (main fix #3348) —
300 # the uppercase comparisons below would otherwise never match the
301 # lowercase values stored in settings / sent by the UI, silently
302 # skipping the custom_endpoint and ollama_url settings fallbacks.
303 model_provider = normalize_provider(model_provider)
305 model = data.get("model")
306 if not model:
307 model = settings_manager.get_setting("llm.model", None)
308 logger.debug(f"No model in request, using database setting: {model}")
309 else:
310 logger.opt(lazy=True).debug(
311 "Using model from request: {}", lambda: _log_value_preview(model)
312 )
314 custom_endpoint = None
315 if model_provider == "openai_endpoint":
316 custom_endpoint = data.get("custom_endpoint")
317 if not custom_endpoint:
318 custom_endpoint = settings_manager.get_setting(
319 "llm.openai_endpoint.url", None
320 )
321 logger.debug(
322 "No custom_endpoint in request, using database setting: "
323 f"{redact_url_credentials(custom_endpoint) if custom_endpoint else None}"
324 )
326 # SSRF guard: a user-supplied custom_endpoint goes straight to the
327 # OpenAI client as base_url. Without validation, an authenticated user
328 # could point the LLM client at internal services (cloud metadata,
329 # localhost, etc.) and read responses through completions output.
330 if custom_endpoint and isinstance(custom_endpoint, str):
331 # Delegate to the shared guard so the endpoint is normalized the way
332 # the OpenAI-compatible provider normalizes it — a bare
333 # ``localhost:11434`` is a legitimate local backend and must not be
334 # rejected here. Private IPs stay allowed; cloud-metadata and bad
335 # schemes are still blocked.
336 if not is_safe_custom_llm_endpoint(custom_endpoint):
337 raise ValueError("Invalid custom endpoint URL")
339 ollama_url = data.get("ollama_url")
340 if not ollama_url and model_provider == "ollama":
341 ollama_url = settings_manager.get_setting(
342 "llm.ollama.url", DEFAULT_OLLAMA_URL
343 )
344 logger.debug(
345 "No ollama_url in request, using database setting: "
346 f"{redact_url_credentials(ollama_url) if ollama_url else None}"
347 )
349 search_engine = data.get("search_engine") or data.get("search_tool")
350 if not search_engine:
351 search_engine = settings_manager.get_setting(
352 "search.tool", DEFAULT_SEARCH_TOOL
353 )
355 max_results = data.get("max_results")
356 time_period = data.get("time_period")
358 iterations = data.get("iterations")
359 if iterations is not None:
360 if (
361 type(iterations) is not int
362 or not ITERATIONS_MIN <= iterations <= ITERATIONS_MAX
363 ):
364 raise ValueError(
365 f"iterations must be an integer between {ITERATIONS_MIN} and {ITERATIONS_MAX}"
366 )
367 else:
368 iterations = settings_manager.get_setting("search.iterations", 5)
370 questions_per_iteration = data.get("questions_per_iteration")
371 if questions_per_iteration is not None:
372 if (
373 type(questions_per_iteration) is not int
374 or not QUESTIONS_PER_ITERATION_MIN
375 <= questions_per_iteration
376 <= QUESTIONS_PER_ITERATION_MAX
377 ):
378 raise ValueError(
379 f"questions_per_iteration must be an integer between "
380 f"{QUESTIONS_PER_ITERATION_MIN} and {QUESTIONS_PER_ITERATION_MAX}"
381 )
382 else:
383 questions_per_iteration = settings_manager.get_setting(
384 "search.questions_per_iteration", 5
385 )
387 strategy = data.get("strategy")
388 if not strategy:
389 strategy = settings_manager.get_setting(
390 "search.search_strategy", "source-based"
391 )
393 # Egress policy per-research overrides. Mirror the
394 # model/search_engine pattern: missing values fall back to saved
395 # settings; supplied values override JUST FOR THIS RUN. They do
396 # NOT persist to the user's settings DB.
397 policy_egress_scope = data.get("policy_egress_scope")
398 llm_require_local_endpoint = data.get("llm_require_local_endpoint")
399 embeddings_require_local = data.get("embeddings_require_local")
401 return {
402 "model_provider": model_provider,
403 "model": model,
404 "custom_endpoint": custom_endpoint,
405 "ollama_url": ollama_url,
406 "search_engine": search_engine,
407 "max_results": max_results,
408 "time_period": time_period,
409 "iterations": iterations,
410 "questions_per_iteration": questions_per_iteration,
411 "strategy": strategy,
412 "policy_egress_scope": policy_egress_scope,
413 "llm_require_local_endpoint": llm_require_local_endpoint,
414 "embeddings_require_local": embeddings_require_local,
415 }
418def _precheck_collection_agent_enabled(
419 search_engine: str,
420 strategy: str,
421 username: str,
422) -> Optional[JSONResponse]:
423 """Reject a run that pairs a LangGraph strategy with an agent-hidden
424 collection.
426 Ported from the Flask ``research_routes`` (#5221). Returns a
427 ``JSONResponse`` when the run should be rejected, or ``None`` to
428 continue — the FastAPI shape, where Flask returned a
429 ``(jsonify, status)`` tuple.
431 The per-collection ``agent_enabled`` flag is exclusive to the LangGraph
432 research agent: it gates which collection engines the agent offers as
433 specialized tools (see ``AdvancedSearchSystem`` /
434 ``LangGraphAgentStrategy._load_specialized_engine_tools``). Public
435 collections also reach the agent via their adapter, but ONLY as
436 specialized tools — the primary engine path is unaffected, so a
437 collection marked ``agent_enabled=False`` is effectively unusable for a
438 LangGraph run: the agent will not pick it as a tool, and the user cannot
439 pick anything else because the primary was the only way to surface it.
441 Mirrors ``_precheck_engine_policy`` below. Fails OPEN (returns None) on
442 any internal error so a transient DB blip doesn't block the run — the
443 frontend is the primary UX guarantee and this is the second backstop.
444 Only fires for ``collection_<uuid>`` engines (the only engines carrying
445 the flag today), and only when ``strategy`` maps to the LangGraph agent;
446 other strategies never read the flag.
447 """
448 if not search_engine or not search_engine.startswith("collection_"):
449 return None
450 # ``strategy`` comes straight from the JSON body, so it may be any JSON
451 # type. A non-string cannot name the LangGraph agent; leave it to the
452 # constructor guard downstream instead of crashing on ``.lower()``.
453 if (
454 not isinstance(strategy, str)
455 or strategy.lower() not in LANGGRAPH_STRATEGY_ALIASES
456 ):
457 return None
458 try:
459 from ...database.models.library import Collection
461 # Imported locally rather than using the module-level binding, so a
462 # late patch of ``database.session_context.get_user_db_session``
463 # takes effect. Main's version does the same and its tests rely on
464 # it; binding at module import time would silently defeat them.
465 from ...database.session_context import (
466 get_user_db_session as _get_user_db_session,
467 )
469 # Engine id format: ``collection_<uuid>``. The split is safe because
470 # the id is constructed the same way in
471 # ``search_engines_config.search_config``.
472 collection_id = search_engine[len("collection_") :]
473 with _get_user_db_session(username) as db_session:
474 row = (
475 db_session.query(Collection)
476 .filter(Collection.id == collection_id)
477 .first()
478 )
479 if row is None:
480 # Unknown collection id — let the factory PEP produce a clearer
481 # error rather than masking it with a 400 here.
482 return None
483 # Default-on for unrecognised / NULL values keeps the
484 # behaviour-preserving contract used elsewhere in the codebase.
485 agent_enabled = getattr(row, "agent_enabled", True) is not False
486 if agent_enabled:
487 return None
488 display_name = getattr(row, "name", None) or search_engine
489 logger.bind(policy_audit=True).warning(
490 "POST /api/start_research collection agent_enabled refused",
491 search_engine=search_engine,
492 strategy=strategy,
493 )
494 return JSONResponse(
495 {
496 "status": "error",
497 "message": (
498 f"Collection '{display_name}' is hidden from the "
499 "LangGraph research agent (the 'Available to the "
500 "research agent' flag is off). Re-enable it on the "
501 "collection's page, or pick a different search "
502 "engine / strategy."
503 ),
504 "reason": "collection_agent_disabled",
505 "field": "strategy",
506 },
507 status_code=400,
508 )
509 except Exception: # pragma: no cover - defensive
510 logger.exception("collection agent_enabled pre-check skipped")
511 return None
514def _precheck_engine_policy(settings_manager, params, search_engine, username):
515 """Validate the requested search engine against this run's egress policy.
517 Under non-STRICT scopes the requested engine is overlaid as
518 ``search.tool`` before resolving the primary, matching the worker,
519 so ADAPTIVE scope follows the run engine rather than the saved
520 database default. Under STRICT there is no overlay: the saved
521 primary is resolved first and a blank/missing saved primary fails
522 closed (400), so the request can never choose its own STRICT
523 primary.
525 Returns a ``JSONResponse`` when the request should be rejected, or
526 ``None`` to continue. Falls through (returns None) when there's no
527 real dict snapshot to validate or the policy module errors — the
528 factory PEP still enforces at engine-instantiation time. An invalid
529 configured setting is not such an error: it rejects the request with
530 the setting diagnostic instead of skipping the check.
531 """
532 try:
533 from ...security.egress.policy import (
534 DEFAULT_EGRESS_SCOPE,
535 EgressScope,
536 PolicyDeniedError,
537 _get_setting_value,
538 context_from_snapshot,
539 evaluate_engine,
540 resolve_run_primary_engine,
541 )
543 policy_snapshot = settings_manager.get_settings_snapshot()
544 # Only validate against a real dict snapshot. A test double or
545 # an unavailable settings backend hands back something else;
546 # skip rather than misfire.
547 if not isinstance(policy_snapshot, dict):
548 return None
550 # Overlay per-research form overrides so the snapshot reflects
551 # what the user picked for THIS run (per-research overrides, not
552 # a global settings save).
553 # Resolve the primary the SAME way the worker does (single source of
554 # truth) so the precheck and the background worker agree on accept vs.
555 # refuse — including the fail-closed missing-primary case, which the
556 # ValueError handler below maps to a 400. (Previously this substituted
557 # searxng and accepted runs the worker then refused.)
558 try:
559 _apply_policy_overrides(policy_snapshot, params)
561 scope_raw = _get_setting_value(
562 policy_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
563 )
564 from ...settings.manager import check_env_setting
566 env_scope = check_env_setting("policy.egress_scope")
567 if env_scope is not None:
568 scope_raw = env_scope
569 scope_str = str(scope_raw).strip().lower()
571 if scope_str == EgressScope.STRICT.value:
572 # STRICT must never fall back to the requested engine: a
573 # blank/missing saved primary is a wiring error and must
574 # fail closed (ValueError -> 400 below), not let the
575 # request choose its own primary.
576 saved_primary = resolve_run_primary_engine(policy_snapshot)
577 primary = saved_primary
578 else:
579 saved_primary = resolve_run_primary_engine(
580 policy_snapshot, default=search_engine or None
581 )
582 primary = search_engine or saved_primary
584 overlay_runtime_settings(
585 policy_snapshot,
586 search_engine=search_engine,
587 model_provider=params.get("model_provider"),
588 model=params.get("model"),
589 custom_endpoint=params.get("custom_endpoint"),
590 iterations=params.get("iterations"),
591 questions_per_iteration=params.get("questions_per_iteration"),
592 strategy=params.get("strategy"),
593 max_results=params.get("max_results"),
594 time_period=params.get("time_period"),
595 )
596 policy_ctx = context_from_snapshot(
597 policy_snapshot,
598 primary,
599 username=username,
600 )
601 except PolicyDeniedError as exc:
602 return JSONResponse(
603 {
604 "status": "error",
605 "message": (
606 f"Egress policy refused this run: {exc.decision.reason}"
607 ),
608 },
609 status_code=400,
610 )
611 except SettingsValueError:
612 # A ValueError subclass, but not a policy decision: the outer
613 # handler reports which setting is invalid.
614 raise
615 except ValueError as exc:
616 # An invalid policy config is unrecoverable. Previously this
617 # raised, fell through to the outer ``except Exception``
618 # below, and silently returned None — so the run started
619 # successfully at the precheck and only failed at a
620 # downstream PEP. Surface it here as a 400 instead.
621 logger.bind(policy_audit=True).warning(
622 "POST /api/start_research policy precheck rejected",
623 reason=str(exc),
624 )
625 # Return a generic reason to the client — the raw ValueError text
626 # can carry policy-config internals and is already captured in the
627 # policy_audit warning above (CWE-209). Unlike the PolicyDeniedError
628 # branch, this path has no curated, user-safe decision reason.
629 return JSONResponse(
630 {
631 "status": "error",
632 "message": (
633 "Egress policy refused this run due to an "
634 "invalid policy configuration."
635 ),
636 },
637 status_code=400,
638 )
640 target_engine = search_engine or primary
641 decision = evaluate_engine(
642 target_engine,
643 policy_ctx,
644 settings_snapshot=policy_snapshot,
645 )
646 if not decision.allowed:
647 logger.bind(policy_audit=True).warning(
648 "POST /api/start_research search_engine refused",
649 engine=target_engine,
650 reason=decision.reason,
651 )
652 # Local import: this module imports the egress policy lazily inside
653 # this function to avoid a circular import at load time; keep the
654 # guidance import with it.
655 from ...security.egress.guidance import denial_guidance
657 # Only the scope-mismatch and strict-not-primary denials have a
658 # form field the user can toggle to fix the error (the Egress
659 # Scope dropdown). For engine_unknown / unclassified the user has
660 # to pick a different search engine — flagging the egress-scope
661 # dropdown there would be misleading and cause the wrong field to
662 # turn red. internal_error / no_snapshot are server-side issues
663 # with no form-level fix at all, so we return ``field: null`` and
664 # let the frontend render just the alert (no inline field error).
665 field_for_reason = {
666 "scope_mismatch_private_only": "policy_egress_scope",
667 "scope_mismatch_public_only": "policy_egress_scope",
668 "strict_not_primary": "policy_egress_scope",
669 }.get(decision.reason)
671 return JSONResponse(
672 {
673 "status": "error",
674 # Clear, actionable message (what + why + how to allow),
675 # plus the raw reason code for support/logs. The
676 # guidance uses ``primary_engine`` to decide whether
677 # Adaptive (default) is a reliable remedy — see
678 # ``_scope_mismatch_how``.
679 "message": denial_guidance(
680 decision.reason,
681 target=f"Search engine '{target_engine}'",
682 # Raw engine id, used by the guidance to decide
683 # whether Adaptive (default) is a reliable
684 # remedy for a scope-mismatch denial (see
685 # ``_scope_mismatch_how`` — kept separate from
686 # ``target`` so the formatted display string
687 # above doesn't break the equality check).
688 target_id=target_engine,
689 primary_engine=saved_primary,
690 ),
691 "reason": decision.reason,
692 # The frontend uses this to highlight the offending
693 # form field with an inline error (FormValidator's
694 # .ldr-field-error + .ldr-field-invalid convention)
695 # and to anchor an alert near the submit button.
696 # ``null`` means "no form field fixes this — show the
697 # alert without an inline field error".
698 "field": field_for_reason,
699 },
700 status_code=400,
701 )
703 # Stage C (ADR-0007): enforce the two-axis (sensitivity x exposure)
704 # admissibility as defense-in-depth on top of the scope check. This is
705 # a fast pre-start 400 for the API path; the same rule is also enforced
706 # at the shared run chokepoint in run_research_process, so follow-up /
707 # chat / queue runs are covered too. Under the UNPROTECTED escape hatch
708 # audit_run evaluates in permissive mode (always allowed), so an
709 # explicitly-unprotected run is never blocked. On an internal error
710 # audit_run fails CLOSED (reason "audit_error") so the run is refused
711 # visibly rather than silently degrading to the scope PEPs.
712 #
713 # NOTE: this sees the run's PRIMARY engine only. The per-engine scope
714 # PEP still runs for expanded engines, but it enforces SCOPE, not the
715 # two-axis rule — so a non-primary sensitive engine reaching a cloud
716 # sink under a permissive scope (legacy `both`) is not caught here.
717 # Widening to the full resolved engine set is a follow-up.
718 from ...security.egress.run_classification import (
719 audit_run_from_snapshot,
720 )
722 # Same shared entry point the worker chokepoint uses, so both resolve
723 # providers identically (LLM override wins over the snapshot; embeddings
724 # only for RAG primaries) and fail closed the same way.
725 two_axis = audit_run_from_snapshot(
726 policy_snapshot,
727 policy_ctx,
728 search_engine,
729 llm_provider=params.get("model_provider"),
730 )
731 if not two_axis.allowed:
732 logger.bind(policy_audit=True).warning(
733 "POST /api/start_research two-axis egress refused",
734 reason=two_axis.reason,
735 offending=two_axis.offending,
736 )
737 if two_axis.reason == "audit_error": 737 ↛ 738line 737 didn't jump to line 738 because the condition on line 737 was never true
738 message = (
739 "Egress policy could not verify this run, so it was "
740 "refused (fail-closed). To override, ask the server "
741 "operator to enable the Unprotected escape hatch "
742 "(LDR_POLICY_ALLOW_UNPROTECTED_EGRESS), then select the "
743 "'Unprotected' scope yourself."
744 )
745 else:
746 message = (
747 "Egress policy refused this run: a sensitive source would "
748 f"reach an exposing destination ({two_axis.reason}). Use "
749 "local inference, mark the collection public, or — to "
750 "override — ask the server operator to enable the "
751 "Unprotected escape hatch "
752 "(LDR_POLICY_ALLOW_UNPROTECTED_EGRESS), then set Egress "
753 "Scope to 'Unprotected' yourself."
754 )
755 return JSONResponse(
756 {
757 "status": "error",
758 "message": message,
759 "reason": two_axis.reason,
760 },
761 status_code=400,
762 )
763 return None
764 except SettingsValueError as error:
765 # The snapshot read above validates every effective setting, so one
766 # invalid row anywhere lands here. Skipping the check would let the
767 # run start without it; refuse and name the setting instead.
768 return _settings_value_error_response(error)
769 except Exception:
770 # Policy module unavailable / internal error → log and fall
771 # through; the factory PEP will catch any actual violation.
772 logger.exception("egress policy pre-check skipped")
773 return None
776def _apply_policy_overrides(settings_snapshot, params):
777 """Overlay form-supplied egress policy values onto the snapshot.
779 Per-research overrides: the user picked these values for this
780 specific run and they do NOT persist to the settings DB. Search
781 engine / model / iteration overlays live in ``overlay_runtime_settings``.
782 """
783 if not isinstance(settings_snapshot, dict):
784 return
785 from ...settings.manager import check_env_setting
786 from ...security.egress.policy import (
787 parse_user_egress_scope,
788 )
790 scope_override = params.get("policy_egress_scope")
791 if (
792 scope_override is not None
793 and check_env_setting("policy.egress_scope") is None
794 ):
795 settings_snapshot["policy.egress_scope"] = parse_user_egress_scope(
796 scope_override
797 ).value
798 if (
799 params.get("llm_require_local_endpoint") is not None
800 and check_env_setting("llm.require_local_endpoint") is None
801 ):
802 settings_snapshot["llm.require_local_endpoint"] = to_bool(
803 params["llm_require_local_endpoint"]
804 )
805 if (
806 params.get("embeddings_require_local") is not None
807 and check_env_setting("embeddings.require_local") is None
808 ):
809 settings_snapshot["embeddings.require_local"] = to_bool(
810 params["embeddings_require_local"]
811 )
814def _research_not_found(research_id, message="Research not found"):
815 """Return a consistent 404 JSON for a missing research.
817 Emits ``status``, ``error`` and ``message`` so the body is a strict
818 superset of both historical 404 shapes — every frontend reader and
819 existing test keeps working without changes:
820 - Shape A readers read ``data.error``
821 - Shape B readers read ``data.status`` (``== "error"``) and/or
822 ``data.message``
824 ``research_id`` is used only for a debug log identifying which research
825 was missing; it is intentionally never echoed in the response body.
826 """
827 logger.opt(lazy=True).debug(
828 "404 for research {}: {}",
829 lambda: _log_value_preview(research_id),
830 lambda: message,
831 )
832 return JSONResponse(
833 {"status": "error", "error": message, "message": message},
834 status_code=404,
835 )
838def _queue_research(
839 db_session: Session,
840 username,
841 research_id,
842 query,
843 mode,
844 research_settings,
845 params,
846 session_id,
847 reason="",
848 research=None,
849):
850 """Add research to queue and notify processor. Returns a JSON response.
852 Args:
853 reason: Optional prefix explaining why the research was queued
854 (e.g. "due to concurrent limit").
855 research: Optional ResearchHistory object whose status should be set
856 to QUEUED atomically with the queue record insertion.
857 """
858 max_position = (
859 db_session.query(func.max(QueuedResearch.position))
860 .filter_by(username=username)
861 .scalar()
862 or 0
863 )
865 queued_record = QueuedResearch(
866 username=username,
867 research_id=research_id,
868 query=query,
869 mode=mode,
870 settings_snapshot=research_settings,
871 position=max_position + 1,
872 )
873 db_session.add(queued_record)
874 if research is not None:
875 research.status = ResearchStatus.QUEUED # type: ignore[assignment]
876 db_session.commit()
877 logger.info(
878 f"Queued research {research_id} at position {max_position + 1} for user {username}"
879 )
881 from ..queue.processor_v2 import queue_processor
883 queue_processor.notify_research_queued(
884 username,
885 research_id,
886 session_id=session_id,
887 query=query,
888 mode=mode,
889 settings_snapshot=research_settings,
890 model_provider=params["model_provider"],
891 model=params["model"],
892 custom_endpoint=params["custom_endpoint"],
893 search_engine=params["search_engine"],
894 max_results=params["max_results"],
895 time_period=params["time_period"],
896 iterations=params["iterations"],
897 questions_per_iteration=params["questions_per_iteration"],
898 strategy=params["strategy"],
899 )
901 position = max_position + 1
902 reason_text = f" {reason}" if reason else ""
903 message = f"Your research has been queued{reason_text}. Position in queue: {position}"
904 return {
905 "status": ResearchStatus.QUEUED,
906 "research_id": research_id,
907 "queue_position": position,
908 "message": message,
909 }
912@router.post("/api/start_research")
913# Primary research-submission endpoint — must be rate-limited (per-user,
914# via the shared api_rate_limit bucket) or a single account can flood the
915# research queue (#3135).
916@api_rate_limit
917async def start_research(
918 request: Request, username: Annotated[str, Depends(require_auth)]
919):
920 """
921 Async wrapper that does the only legitimate await (parsing the JSON
922 body) on the event loop, then offloads the synchronous body — which
923 opens four ``get_user_db_session`` blocks (each potentially driving
924 PBKDF2 key derivation), threads, and other blocking work — to a
925 threadpool. Without this, ``async def`` + sync DB blocks would
926 serialise the entire server behind the slowest /start_research call.
927 """
929 # main's @require_json_body(error_format="status") read the body with
930 # get_json(silent=True), which returns None for a non-JSON media type,
931 # so a JSON-looking text/plain body got this 400 and started nothing.
932 # request.json() ignores Content-Type, so the check is explicit here.
933 # POST /research/api/start delegates to this handler and shares it.
934 if not is_json_request(request):
935 return json_body_error("status")
937 # A malformed or non-object JSON body would otherwise reach
938 # _start_research_sync's data.get(...) calls and 500 (AttributeError).
939 # Reject at the boundary with a clean 400.
940 try:
941 data = await request.json()
942 except Exception:
943 return JSONResponse(
944 {"status": "error", "message": "Request body must be JSON"},
945 status_code=400,
946 )
947 if not isinstance(data, dict):
948 return JSONResponse(
949 {
950 "status": "error",
951 "message": "Request body must be a JSON object",
952 },
953 status_code=400,
954 )
955 query = data.get("query")
956 if query is not None and not isinstance(query, str):
957 return JSONResponse(
958 {"status": "error", "message": "query must be a string"},
959 status_code=400,
960 )
961 # The string fields (``_STRING_REQUEST_FIELDS``) are used with str
962 # methods after dispatch, so any JSON type other than a string or null
963 # is a 400 here, before any database work. Null and "" still count as
964 # unset in ``_extract_research_params``.
965 bad_field = _non_string_request_field(data)
966 if bad_field is not None:
967 return JSONResponse(
968 {"status": "error", "message": _string_field_error(bad_field)},
969 status_code=400,
970 )
971 metadata = data.get("metadata", {})
972 if not isinstance(metadata, dict):
973 return JSONResponse(
974 {"status": "error", "message": "metadata must be an object"},
975 status_code=400,
976 )
977 # A missing query outranks the override range checks below, so a body
978 # with no query and an invalid iterations value reports "Query is
979 # required" (the more useful message when both are wrong). Mirrors the
980 # early return in _start_research_sync and the api_v1 query-first
981 # order. The JSON type checks above (query, the string fields,
982 # metadata) still run first, so a wrong-typed field is reported even
983 # when the query is missing.
984 if not query:
985 return JSONResponse(
986 {"status": "error", "message": "Query is required"}, status_code=400
987 )
988 validation_error = validate_search_overrides(data)
989 if validation_error is not None:
990 return JSONResponse(
991 {"status": "error", "message": validation_error},
992 status_code=400,
993 )
994 base_url = str(request.base_url)
995 session_id = request.session.get("session_id")
996 return await run_db_sync(
997 _start_research_sync, data, username, base_url, session_id
998 )
1001def _unrunnable_subscription_response(db_session, metadata):
1002 """400 response for a run linked to a saved subscription that cannot run.
1004 ``metadata.subscription_id`` makes the finished run advance that
1005 subscription's schedule, which needs a runnable saved interval (see
1006 ``is_runnable_refresh_interval``). The manual-run endpoint and the
1007 overdue sweep check that before they call in; checking it in this shared
1008 entry point as well covers a direct ``POST /api/start_research``. An id
1009 that matches no saved subscription is left alone, as before.
1011 Any truthy id is looked up as ``str(id)``, the way the completion advance
1012 (``advance_refresh_schedule_by_id``) resolves it, so a JSON number cannot
1013 pass this check and still advance a numeric-string row at completion.
1014 """
1015 subscription_id = metadata.get("subscription_id")
1016 if not subscription_id:
1017 return None
1018 subscription_id = str(subscription_id)
1019 from ...database.models.news import NewsSubscription
1020 from ...news.constants import (
1021 SUBSCRIPTION_INTERVAL_REPAIR_MESSAGE,
1022 is_runnable_refresh_interval,
1023 )
1025 subscription = (
1026 db_session.query(NewsSubscription)
1027 .filter(NewsSubscription.id == subscription_id)
1028 .first()
1029 )
1030 if subscription is None or is_runnable_refresh_interval(
1031 subscription.refresh_interval_minutes
1032 ):
1033 return None
1034 return JSONResponse(
1035 {"status": "error", "message": SUBSCRIPTION_INTERVAL_REPAIR_MESSAGE},
1036 status_code=400,
1037 )
1040def _start_research_sync(
1041 data: dict, username: str, base_url: str, session_id: str | None
1042):
1043 # Debug logging to trace model parameter. The key list is request
1044 # input too (a body may carry any number of keys of any length), so
1045 # it goes through the same bounded preview as the values below.
1046 logger.opt(lazy=True).debug(
1047 "Request data keys: {}", lambda: _log_value_preview(list(data.keys()))
1048 )
1050 # Check if this is a news search
1051 metadata = data.get("metadata", {})
1052 if metadata.get("is_news_search"):
1053 logger.info(
1054 "News search request received: triggered_by={}",
1055 _log_value_preview(metadata.get("triggered_by", "unknown")),
1056 )
1058 query = data.get("query")
1059 mode = data.get("mode", "quick")
1061 # Missing query outranks parameter validation: a body with no query
1062 # and an invalid iterations value reports "Query is required" (the
1063 # more useful message when both are wrong). The no-query log-preview
1064 # probe only exercises the two log sites above this return ("Request
1065 # data keys", news "triggered_by"); the sites below (model/provider/
1066 # engine/iterations/strategy lines) need a query to run and are pinned
1067 # by the with-query probe variant. See #6517 review and #6756.
1068 if not query:
1069 return JSONResponse(
1070 {"status": "error", "message": "Query is required"}, status_code=400
1071 )
1073 # Replace date placeholders if they exist
1074 if query and "YYYY-MM-DD" in query:
1075 # Use local system time
1076 current_date = datetime.now(UTC).strftime("%Y-%m-%d")
1078 original_query = query
1079 query = query.replace("YYYY-MM-DD", current_date)
1080 logger.info(
1081 "Replaced date placeholder in query: {} -> {}",
1082 # (#6938: redacted before it is cut.)
1083 redact_and_bound_for_log(original_query, 100),
1084 redact_and_bound_for_log(query, 100),
1085 )
1086 logger.info(f"Using date: {current_date}")
1088 # Update metadata to track the replacement
1089 if not metadata: 1089 ↛ 1091line 1089 didn't jump to line 1091 because the condition on line 1089 was always true
1090 metadata = {}
1091 metadata["original_query"] = original_query
1092 metadata["processed_query"] = query
1093 metadata["date_replaced"] = current_date
1094 data["metadata"] = metadata
1096 # Get parameters from request or use database settings
1097 from ...settings import SettingsManager
1099 with get_user_db_session(username) as db_session:
1100 unrunnable = _unrunnable_subscription_response(db_session, metadata)
1101 if unrunnable is not None:
1102 return unrunnable
1104 settings_manager = SettingsManager(db_session=db_session)
1105 try:
1106 params = _extract_research_params(data, settings_manager)
1107 except SettingsValueError as error:
1108 # Ahead of the ValueError mapping below: its substring matching
1109 # would report an invalid saved or environment value as a bad
1110 # request field (e.g. "iterations must be an integer ...").
1111 return _settings_value_error_response(error)
1112 except ValueError as exc:
1113 # SSRF guard or parameter validation in _extract_research_params
1114 # rejected the input — return main's clean 400 contract
1115 # instead of an unhandled 500. Map to explicit error messages to
1116 # avoid exposing internal exception details (CodeQL py/stack-trace-exposure).
1117 msg = str(exc)
1118 # A string-field type error keeps its own message, rebuilt from
1119 # the constant field list rather than echoed from ``exc``. Only
1120 # in-process callers that skip start_research's identical check
1121 # (the news run-now and overdue paths) can reach it.
1122 bad_field = next(
1123 (
1124 field
1125 for field in _STRING_REQUEST_FIELDS
1126 if msg == _string_field_error(field)
1127 ),
1128 None,
1129 )
1130 if bad_field is not None:
1131 error_msg = _string_field_error(bad_field)
1132 elif "iterations" in msg:
1133 error_msg = f"iterations must be an integer between {ITERATIONS_MIN} and {ITERATIONS_MAX}"
1134 elif "questions_per_iteration" in msg: 1134 ↛ 1135line 1134 didn't jump to line 1135 because the condition on line 1134 was never true
1135 error_msg = (
1136 f"questions_per_iteration must be an integer between "
1137 f"{QUESTIONS_PER_ITERATION_MIN} and {QUESTIONS_PER_ITERATION_MAX}"
1138 )
1139 else:
1140 error_msg = "Invalid custom endpoint URL"
1141 return JSONResponse(
1142 {"status": "error", "message": error_msg},
1143 status_code=400,
1144 )
1146 model_provider = params["model_provider"]
1147 model = params["model"]
1148 custom_endpoint = params["custom_endpoint"]
1149 search_engine = params["search_engine"]
1150 max_results = params["max_results"]
1151 time_period = params["time_period"]
1152 iterations = params["iterations"]
1153 questions_per_iteration = params["questions_per_iteration"]
1154 strategy = params["strategy"]
1156 # Egress policy: server-side check on the requested search
1157 # engine BEFORE we enqueue the research. The requested engine is
1158 # the run primary (same overlay the worker applies), so ADAPTIVE
1159 # scope follows this run's engine rather than the saved DB default.
1160 # The factory PEP still enforces expanded engines at instantiation.
1161 policy_error = _precheck_engine_policy(
1162 settings_manager, params, search_engine, username
1163 )
1164 if policy_error is not None:
1165 return policy_error
1167 # Per-collection ``agent_enabled`` is exclusive to the LangGraph
1168 # research agent. When the user picks LangGraph + a collection
1169 # hidden from the agent, the run is effectively broken (the
1170 # agent won't pick it as a tool, and the primary was the only
1171 # way to surface it). The frontend already disables the
1172 # combination in the dropdown; this is the second backstop for
1173 # direct API callers and stale-form submissions.
1174 agent_enabled_error = _precheck_collection_agent_enabled(
1175 search_engine, strategy, username
1176 )
1177 if agent_enabled_error is not None: 1177 ↛ 1178line 1177 didn't jump to line 1178 because the condition on line 1177 was never true
1178 return agent_enabled_error
1180 # This thread-local session survives the ``with`` block. End its
1181 # read transaction before the admission gate below; otherwise this
1182 # request can wait for the gate while retaining a pooled connection
1183 # (and, under rollback-journal SQLite, a shared database lock).
1184 db_session.rollback()
1186 # Debug logging for model parameter specifically. ``model`` is an
1187 # unvalidated request-body value, so it must not be interpolated
1188 # eagerly: an f-string builds the full repr on every request
1189 # regardless of log level. The bounded preview keeps the cost and
1190 # the logged size capped (#6305, same shape as #6191).
1191 logger.opt(lazy=True).debug(
1192 "Extracted model value: {} (type: {})",
1193 lambda: _log_value_preview(model),
1194 lambda: type(model).__name__,
1195 )
1197 # Log the selections for troubleshooting. These two lines are emitted
1198 # at the default INFO level, so the request-body values in them go
1199 # through the same bounded preview. ``max_results`` and ``time_period``
1200 # are validated at the request boundary (``validate_search_overrides``)
1201 # and are passed as they are.
1202 logger.info(
1203 "Starting research with provider: {}, model: {}, search engine: {}",
1204 _log_value_preview(model_provider),
1205 _log_value_preview(model),
1206 _log_value_preview(search_engine),
1207 )
1208 logger.info(
1209 "Additional parameters: max_results={}, time_period={}, iterations={}, questions={}, strategy={}",
1210 max_results,
1211 time_period,
1212 _log_value_preview(iterations),
1213 _log_value_preview(questions_per_iteration),
1214 _log_value_preview(strategy),
1215 )
1217 # NOTE: the missing-query 400 lives at the top of this function
1218 # (before _extract_research_params) so it outranks parameter
1219 # validation — see the comment there.
1221 # Cap query length to defend against accidental prompt blowups that would
1222 # bloat ResearchHistory storage and run up LLM context. The queue replay
1223 # path uses the same validator before dispatching a persisted row.
1224 query_length_error = validate_research_query_length(query)
1225 if query_length_error is not None:
1226 return JSONResponse(
1227 {
1228 "status": "error",
1229 "message": query_length_error,
1230 },
1231 status_code=400,
1232 )
1234 # Validate required parameters based on provider.
1235 # model_provider is normalize_provider()'d to lowercase in
1236 # _extract_research_params, so this MUST compare to the lowercase
1237 # canonical form — the old "OPENAI_ENDPOINT" literal never matched and
1238 # left this required-field check dead.
1239 if model_provider == "openai_endpoint" and not custom_endpoint:
1240 return JSONResponse(
1241 {
1242 "status": "error",
1243 "message": "Custom endpoint URL is required for OpenAI endpoint provider",
1244 },
1245 status_code=400,
1246 )
1248 # SSRF pre-flight on the user-supplied LLM endpoint: reject metadata /
1249 # link-local targets at the request boundary, before any research thread
1250 # is spawned. This is fail-fast defense-in-depth — the OpenAI-compatible
1251 # provider's assert_base_url_safe re-validates the same URL before the
1252 # client is built. Private IPs / localhost are permitted so local LLMs
1253 # (vLLM, Ollama, LM Studio) work, including scheme-less endpoints
1254 # (the helper normalizes exactly as the provider does).
1255 if model_provider == "openai_endpoint" and not is_safe_custom_llm_endpoint( 1255 ↛ 1258line 1255 didn't jump to line 1258 because the condition on line 1255 was never true
1256 custom_endpoint
1257 ):
1258 return JSONResponse(
1259 {"status": "error", "message": "Invalid custom endpoint URL"},
1260 status_code=400,
1261 )
1263 if not model:
1264 logger.error(
1265 "No model specified or configured. Provider: {}",
1266 _log_value_preview(model_provider),
1267 )
1268 return JSONResponse(
1269 {
1270 "status": "error",
1271 "message": "Model is required. Please configure a model in the settings.",
1272 },
1273 status_code=400,
1274 )
1276 # Check if the user has too many active researches
1277 from ...settings import SettingsManager
1279 should_queue = False
1280 # Pre-bind: the post-commit race check below reads this, so if the block
1281 # under try raises before the setting is read (a DB hiccup), that check
1282 # would raise UnboundLocalError and turn a recoverable failure into a 500.
1283 max_concurrent_researches = 3
1284 try:
1285 # Share one stable per-user admission lock with queue replay. The
1286 # preliminary count and the post-commit claim below both use this gate;
1287 # the second guarded section closes any race during snapshot capture.
1288 with user_research_start_gate(username):
1289 with get_user_db_session(username) as db_session:
1290 settings_manager = SettingsManager(db_session)
1291 # Clamp to the server-wide semaphore ceiling (#5549).
1292 # app.max_concurrent_researches is a per-user, user-editable
1293 # setting, but the semaphore it is checked against is global —
1294 # so without the clamp a user can raise their own limit past
1295 # the server's and monopolise it.
1296 max_concurrent_researches = clamp_user_max_concurrent(
1297 settings_manager.get_setting(
1298 "app.max_concurrent_researches", 3
1299 )
1300 )
1302 # First, clean up stale entries where the research thread died.
1303 from ..routes.globals import reclaim_stale_user_active_research
1305 if reclaim_stale_user_active_research(
1306 db_session, username, logger=logger
1307 ):
1308 db_session.commit()
1310 active_count = (
1311 db_session.query(UserActiveResearch)
1312 .filter_by(
1313 username=username,
1314 status=ResearchStatus.IN_PROGRESS,
1315 )
1316 .count()
1317 )
1319 logger.info(
1320 f"Active research count for {username}: "
1321 f"{active_count}/{max_concurrent_researches}"
1322 )
1324 should_queue = active_count >= max_concurrent_researches
1325 logger.info(f"Should queue new research: {should_queue}")
1327 # This context reuses a thread-local Session rather than
1328 # closing it on exit. End the count query's read transaction
1329 # before releasing the admission gate; otherwise this request
1330 # can retain a pooled connection and later wait for the gate
1331 # while a queue tick holds the gate and waits for a connection.
1332 db_session.rollback()
1333 except SettingsValueError as error:
1334 # An invalid concurrency limit must not become "do not queue".
1335 return _settings_value_error_response(error)
1336 except Exception:
1337 logger.exception("Failed to check active researches")
1338 # Default to not queueing if we can't check
1339 should_queue = False
1341 # For non-queued research, verify password is available BEFORE creating DB records
1342 # (queued research gets password later via queue processor)
1343 user_password = None
1344 if not should_queue:
1345 user_password, session_expired = resolve_user_password(username)
1346 if session_expired:
1347 return JSONResponse(
1348 {
1349 "status": "error",
1350 "message": "Your session has expired. Please log out and log back in to start research.",
1351 },
1352 status_code=401,
1353 )
1355 # Create a record in the database with explicit UTC timestamp
1356 import uuid
1357 import threading
1359 created_at = datetime.now(UTC).isoformat()
1360 research_id = str(uuid.uuid4())
1362 # Create organized research metadata with settings snapshot
1363 submission = {
1364 "model_provider": model_provider,
1365 "model": model,
1366 "custom_endpoint": custom_endpoint,
1367 "search_engine": search_engine,
1368 "max_results": max_results,
1369 "time_period": time_period,
1370 "iterations": iterations,
1371 "questions_per_iteration": questions_per_iteration,
1372 "strategy": strategy,
1373 }
1374 submission_overrides = []
1375 if data.get("model_provider"):
1376 submission_overrides.append("model_provider")
1377 if data.get("model"):
1378 submission_overrides.append("model")
1379 if data.get("custom_endpoint"):
1380 submission_overrides.append("custom_endpoint")
1381 if data.get("search_engine") or data.get("search_tool"):
1382 submission_overrides.append("search_engine")
1383 if data.get("max_results") is not None:
1384 submission_overrides.append("max_results")
1385 if data.get("time_period") is not None:
1386 submission_overrides.append("time_period")
1387 if data.get("iterations") is not None:
1388 submission_overrides.append("iterations")
1389 if data.get("questions_per_iteration") is not None:
1390 submission_overrides.append("questions_per_iteration")
1391 if data.get("strategy"):
1392 submission_overrides.append("strategy")
1394 research_settings = {
1395 # Direct submission parameters
1396 "submission": submission,
1397 # Only these request-supplied values remain fixed if queued work is
1398 # dispatched after operator environment settings change.
1399 "submission_overrides": submission_overrides,
1400 # System information
1401 "system": {
1402 "timestamp": created_at,
1403 "user": username,
1404 "version": "1.0", # Track metadata version for future migrations
1405 "server_url": base_url, # Server URL for link generation
1406 },
1407 }
1409 # Add any additional metadata from request
1410 additional_metadata = data.get("metadata", {})
1411 if additional_metadata:
1412 reserved_metadata_keys = {
1413 "submission",
1414 "submission_overrides",
1415 "system",
1416 "settings_snapshot",
1417 }
1418 research_settings.update(
1419 {
1420 key: value
1421 for key, value in additional_metadata.items()
1422 if key not in reserved_metadata_keys
1423 }
1424 )
1425 # Get complete settings snapshot for this research
1426 try:
1427 from ...settings import SettingsManager
1429 with get_user_db_session(username) as db_session:
1430 # Ensure any pending changes are committed
1431 try:
1432 db_session.commit()
1433 except Exception:
1434 db_session.rollback()
1435 settings_manager = SettingsManager(db_session, owns_session=False)
1436 # Get all current settings as a snapshot (bypass cache to ensure fresh data)
1437 all_settings = settings_manager.get_all_settings(
1438 bypass_cache=True,
1439 include_environment_overrides=False,
1440 )
1441 # Apply per-research egress policy overrides (form-supplied
1442 # values override saved settings JUST FOR THIS RUN; they do
1443 # not persist to the user's settings DB).
1444 _apply_policy_overrides(all_settings, params)
1445 overlay_runtime_settings(
1446 all_settings,
1447 search_engine=search_engine,
1448 model_provider=model_provider,
1449 model=model,
1450 custom_endpoint=custom_endpoint,
1451 iterations=iterations,
1452 questions_per_iteration=questions_per_iteration,
1453 strategy=strategy,
1454 max_results=max_results,
1455 time_period=time_period,
1456 )
1458 # Add settings snapshot to metadata
1459 research_settings["settings_snapshot"] = all_settings
1460 logger.info(
1461 f"Captured {len(all_settings)} settings for research {research_id}"
1462 )
1463 # As above, release the cached session's read transaction before
1464 # taking the second admission gate for the durable slot claim.
1465 db_session.rollback()
1466 except Exception:
1467 logger.exception("Failed to capture settings snapshot")
1468 # Cannot continue without settings snapshot for thread-based research
1469 return JSONResponse(
1470 {
1471 "status": "error",
1472 "message": "Failed to capture settings for research. Please try again.",
1473 },
1474 status_code=500,
1475 )
1477 # Maintain the process-wide order used by queue replay and rekey/logout:
1478 # admission gate before database/session work. The narrower gate below is
1479 # deliberately retained and re-enters this RLock to document the exact
1480 # count -> claim section.
1481 admission_lock = get_user_research_start_lock(username)
1482 hold_admission_gate_until_spawn = False
1483 admission_lock.acquire()
1484 try:
1485 with get_user_db_session(username) as db_session:
1486 # Determine initial status based on whether we need to queue
1487 initial_status = (
1488 ResearchStatus.QUEUED
1489 if should_queue
1490 else ResearchStatus.IN_PROGRESS
1491 )
1493 research = ResearchHistory(
1494 id=research_id, # Set UUID as primary key
1495 query=query,
1496 mode=mode,
1497 status=initial_status,
1498 created_at=created_at,
1499 progress_log=[{"time": created_at, "progress": 0}],
1500 research_meta=research_settings,
1501 )
1502 db_session.add(research)
1503 db_session.commit()
1504 logger.info(
1505 f"Created research entry with UUID: {research_id}, status: {initial_status}"
1506 )
1508 if should_queue:
1509 return _queue_research(
1510 db_session,
1511 username,
1512 research_id,
1513 query,
1514 mode,
1515 research_settings,
1516 params,
1517 session_id,
1518 )
1519 # Atomically claim a user slot against queue replay. A queue tick
1520 # may have started work while this request captured its settings;
1521 # inserting and re-counting under the shared gate makes one side
1522 # demote before either can exceed the user's cap.
1523 with user_research_start_gate(username):
1524 import threading
1526 active_record = UserActiveResearch(
1527 username=username,
1528 research_id=research_id,
1529 status=ResearchStatus.IN_PROGRESS,
1530 thread_id=str(threading.current_thread().ident),
1531 settings_snapshot=research_settings,
1532 )
1533 db_session.add(active_record)
1534 db_session.commit()
1535 logger.info(
1536 f"Created active research record for user {username}"
1537 )
1539 # Double-check after committing to catch starts that landed
1540 # between the preliminary count and this guarded claim.
1541 try:
1542 final_count = (
1543 db_session.query(UserActiveResearch)
1544 .filter_by(
1545 username=username,
1546 status=ResearchStatus.IN_PROGRESS,
1547 )
1548 .count()
1549 )
1550 logger.info(
1551 f"Final active count after commit: "
1552 f"{final_count}/{max_concurrent_researches}"
1553 )
1555 if final_count > max_concurrent_researches:
1556 logger.warning(
1557 f"Race condition detected: {final_count} > "
1558 f"{max_concurrent_researches}, moving to queue"
1559 )
1560 db_session.delete(active_record)
1561 db_session.commit()
1563 return _queue_research(
1564 db_session,
1565 username,
1566 research_id,
1567 query,
1568 mode,
1569 research_settings,
1570 params,
1571 session_id,
1572 reason="due to concurrent limit",
1573 research=research,
1574 )
1575 except Exception:
1576 logger.warning("Could not recheck active count")
1578 # The durable claim is not live until start_research_process registers
1579 # its worker in the in-memory research registry. Queue replay uses that
1580 # registry to distinguish live rows from stale ones, so releasing the
1581 # shared gate here would let it reclaim this brand-new claim as FAILED
1582 # before the worker exists. Keep the gate through registration (and
1583 # through any spawn-failure cleanup) below.
1584 hold_admission_gate_until_spawn = True
1586 except Exception:
1587 logger.exception("Failed to create research entry")
1588 return JSONResponse(
1589 {"status": "error", "message": "Failed to create research entry"},
1590 status_code=500,
1591 )
1592 finally:
1593 # Queued and failed claims return from the section above and must not
1594 # retain the gate. A successful direct claim transfers responsibility
1595 # for releasing it to the prepare/spawn try-finally below.
1596 if not hold_admission_gate_until_spawn:
1597 admission_lock.release()
1599 # Only start the research if not queued
1600 if not should_queue: 1600 ↛ 1754line 1600 didn't jump to line 1754 because the condition on line 1600 was always true
1601 try:
1602 # Save the research strategy to the database before starting the
1603 # thread. This remains inside the admission gate so no queue tick
1604 # can reclaim the durable claim before registration.
1605 try:
1606 from ..services.research_service import save_research_strategy
1608 save_research_strategy(research_id, strategy, username=username)
1609 except Exception:
1610 logger.warning("Could not save research strategy")
1612 # Debug logging for settings snapshot
1613 snapshot_data = research_settings.get("settings_snapshot", {})
1614 log_settings(
1615 snapshot_data, "Settings snapshot being passed to thread"
1616 )
1617 if "search.tool" in snapshot_data: 1617 ↛ 1622line 1617 didn't jump to line 1622 because the condition on line 1617 was always true
1618 logger.debug(
1619 f"search.tool in snapshot: {snapshot_data['search.tool']}"
1620 )
1621 else:
1622 logger.debug("search.tool NOT in snapshot")
1624 # Start the research process with the selected parameters.
1625 # If the spawn raises, the UserActiveResearch + IN_PROGRESS
1626 # ResearchHistory rows persisted above would otherwise be
1627 # permanently orphaned (no thread, no cleanup path). Catch any
1628 # exception, mark the research FAILED, delete the active row,
1629 # and return 500 — same contract as the queue processor's
1630 # terminal-failure branch introduced in #3481.
1631 research_thread = start_research_process(
1632 research_id,
1633 query,
1634 mode,
1635 run_research_process,
1636 username=username, # Pass username to the thread
1637 user_password=user_password, # Pass password for database access
1638 model_provider=model_provider,
1639 model=model,
1640 custom_endpoint=custom_endpoint,
1641 search_engine=search_engine,
1642 max_results=max_results,
1643 time_period=time_period,
1644 iterations=iterations,
1645 questions_per_iteration=questions_per_iteration,
1646 strategy=strategy,
1647 settings_snapshot=snapshot_data, # Pass complete settings
1648 )
1649 except DuplicateResearchError:
1650 # A live thread already owns this research_id. Do NOT delete
1651 # the UserActiveResearch row or mark ResearchHistory FAILED —
1652 # that state belongs to the live thread, and mutating it
1653 # would terminate a running research from the user's
1654 # perspective while it keeps executing. Same contract as the
1655 # queue processor's dedicated dup branch (#3506).
1656 logger.warning(
1657 f"Duplicate live thread detected for {research_id} "
1658 "on direct submission; leaving state intact"
1659 )
1660 return JSONResponse(
1661 content={
1662 "status": "error",
1663 "message": "Research is already running.",
1664 },
1665 status_code=409,
1666 )
1667 except SystemAtCapacityError:
1668 # System at concurrent-research capacity. Roll back the rows
1669 # committed above (UserActiveResearch + IN_PROGRESS history)
1670 # and return 429 so the client can retry shortly.
1671 logger.warning(
1672 f"SystemAtCapacityError on direct submission for "
1673 f"{research_id}; rolling back orphan rows"
1674 )
1675 try:
1676 with get_user_db_session(username) as cleanup_session:
1677 stale_active = (
1678 cleanup_session.query(UserActiveResearch)
1679 .filter_by(username=username, research_id=research_id)
1680 .first()
1681 )
1682 if stale_active:
1683 cleanup_session.delete(stale_active)
1684 cleanup_session.query(ResearchHistory).filter_by(
1685 id=research_id
1686 ).delete()
1687 cleanup_session.commit()
1688 except Exception:
1689 logger.exception(
1690 "Cleanup after SystemAtCapacityError raised; "
1691 "leaving orphan rows for the reconciler"
1692 )
1693 return JSONResponse(
1694 content={
1695 "status": "error",
1696 "message": "Server is at research capacity. Please retry shortly.",
1697 },
1698 status_code=429,
1699 )
1700 except Exception:
1701 logger.exception(
1702 f"Failed to prepare or spawn research thread for {research_id}"
1703 )
1704 try:
1705 with get_user_db_session(username) as cleanup_session:
1706 stale_active = (
1707 cleanup_session.query(UserActiveResearch)
1708 .filter_by(username=username, research_id=research_id)
1709 .first()
1710 )
1711 if stale_active: 1711 ↛ 1713line 1711 didn't jump to line 1713 because the condition on line 1711 was always true
1712 cleanup_session.delete(stale_active)
1713 research_row = (
1714 cleanup_session.query(ResearchHistory)
1715 .filter_by(id=research_id)
1716 .first()
1717 )
1718 if research_row: 1718 ↛ 1720line 1718 didn't jump to line 1720 because the condition on line 1718 was always true
1719 research_row.status = ResearchStatus.FAILED
1720 cleanup_session.commit()
1721 except Exception:
1722 logger.exception(
1723 "Cleanup after spawn failure raised; leaving "
1724 "orphan rows for the reconciler to handle"
1725 )
1726 return JSONResponse(
1727 content={
1728 "status": "error",
1729 "message": "Failed to start research. Please try again.",
1730 },
1731 status_code=500,
1732 )
1733 finally:
1734 # start_research_process registers the worker before returning.
1735 # On failure, the exception handlers above finish durable cleanup
1736 # before this runs. Either way, queue replay can now safely inspect
1737 # the claim after acquiring the shared gate.
1738 admission_lock.release()
1740 # Update the active research record with the actual thread ID.
1741 try:
1742 with get_user_db_session(username) as thread_session:
1743 active_record = (
1744 thread_session.query(UserActiveResearch)
1745 .filter_by(username=username, research_id=research_id)
1746 .first()
1747 )
1748 if active_record: 1748 ↛ 1754line 1748 didn't jump to line 1754
1749 active_record.thread_id = str(research_thread.ident)
1750 thread_session.commit()
1751 except Exception:
1752 logger.warning("Could not update thread ID")
1754 return {"status": "success", "research_id": research_id}
1757@router.post("/api/terminate/{research_id}")
1758def terminate_research(
1759 request: Request,
1760 research_id,
1761 username: Annotated[str, Depends(require_auth)],
1762):
1763 """Terminate an in-progress research process"""
1765 # Check if the research exists and is in progress
1766 try:
1767 with get_user_db_session(username) as db_session:
1768 research = (
1769 db_session.query(ResearchHistory)
1770 .filter_by(id=research_id)
1771 .first()
1772 )
1774 if not research:
1775 return _research_not_found(research_id)
1777 status = research.status
1779 # If it's already in a terminal state, return success
1780 if status in (
1781 ResearchStatus.COMPLETED,
1782 ResearchStatus.SUSPENDED,
1783 ResearchStatus.FAILED,
1784 ResearchStatus.ERROR,
1785 ):
1786 return {
1787 "status": "success",
1788 "message": f"Research already {status}",
1789 "research_status": status,
1790 }
1792 # Check if it's in the active_research dict
1793 if not is_research_active(research_id):
1794 # The worker may not be registered in _active_research yet: a
1795 # just-submitted research commits its IN_PROGRESS row before the
1796 # worker thread registers itself (spawn-grace window). Set the
1797 # termination flag anyway so a worker that starts right after
1798 # this still sees it and aborts at its first checkpoint —
1799 # otherwise the user's Stop is silently ignored and the research
1800 # runs to completion (overwriting this SUSPENDED status). The
1801 # flag is harmless if no worker ever starts.
1802 set_termination_flag(research_id)
1803 if status == ResearchStatus.QUEUED:
1804 claimed_queue_row = (
1805 db_session.query(QueuedResearch.id)
1806 .filter(
1807 QueuedResearch.research_id == research_id,
1808 QueuedResearch.is_processing.is_(True),
1809 )
1810 .exists()
1811 )
1812 suspended = (
1813 db_session.query(ResearchHistory)
1814 .filter(
1815 ResearchHistory.id == research_id,
1816 ResearchHistory.status == ResearchStatus.QUEUED,
1817 ~claimed_queue_row,
1818 )
1819 .update(
1820 {ResearchHistory.status: ResearchStatus.SUSPENDED},
1821 synchronize_session=False,
1822 )
1823 )
1824 if suspended: 1824 ↛ 1834line 1824 didn't jump to line 1834 because the condition on line 1824 was always true
1825 from ..queue.lifecycle_cleanup import (
1826 cleanup_queued_research_state,
1827 )
1829 cleanup_queued_research_state(
1830 db_session, [research_id], acting_user=username
1831 )
1832 else:
1833 research.status = ResearchStatus.SUSPENDED
1834 db_session.commit()
1835 return {"status": "success", "message": "Research terminated"}
1837 # Set the termination flag
1838 set_termination_flag(research_id)
1840 # Log the termination request - using UTC timestamp
1841 timestamp = datetime.now(UTC).isoformat()
1842 termination_message = "Research termination requested by user"
1843 current_progress = get_research_field(research_id, "progress", 0)
1845 # Create log entry
1846 log_entry = {
1847 "time": timestamp,
1848 "message": termination_message,
1849 "progress": current_progress,
1850 "metadata": {"phase": "termination"},
1851 }
1853 # Add to in-memory log
1854 append_research_log(research_id, log_entry)
1856 # Add to database log
1857 logger.log("MILESTONE", f"Research ended: {termination_message}")
1859 # Update the log in the database
1860 if research.progress_log: 1860 ↛ 1869line 1860 didn't jump to line 1869 because the condition on line 1860 was always true
1861 try:
1862 if isinstance(research.progress_log, str):
1863 current_log = json.loads(research.progress_log)
1864 else:
1865 current_log = research.progress_log
1866 except Exception:
1867 current_log = []
1868 else:
1869 current_log = []
1871 # The column is a log LIST, but a JSON column accepts any shape
1872 # and the reader hands one back verbatim: a legacy TEXT scalar
1873 # ('true') parses to a bool here, and a value the driver already
1874 # decoded arrives as an int. Appending to either raises
1875 # AttributeError outside the try above — one malformed scalar
1876 # would turn a Stop request into a 500. A non-list is not a log;
1877 # start a fresh one instead of failing the request.
1878 if not isinstance(current_log, list):
1879 logger.warning(
1880 "Ignoring a non-list progress_log for research {}",
1881 research_id,
1882 )
1883 current_log = []
1885 # Assign a new value so SQLAlchemy sees the JSON column change.
1886 research.progress_log = [*current_log, log_entry]
1887 research.status = ResearchStatus.SUSPENDED
1888 db_session.commit()
1890 # Emit a socket event for the termination request. Must be
1891 # scoped to subscribers of this research_id — a roomless emit
1892 # broadcasts the user's SUSPENDED status to every connected
1893 # client on the default namespace.
1894 try:
1895 event_data = {
1896 "status": ResearchStatus.SUSPENDED,
1897 "message": "Research was suspended by user request",
1898 }
1900 from ..services.socketio_asgi import emit_to_subscribers
1902 emit_to_subscribers(
1903 "research_progress",
1904 research_id,
1905 event_data,
1906 owner=username,
1907 )
1909 except Exception:
1910 logger.exception("Socket emit error (non-critical)")
1912 return {
1913 "status": "success",
1914 "message": "Research termination requested",
1915 }
1917 except Exception:
1918 logger.exception("Error terminating research")
1919 return JSONResponse(
1920 {"status": "error", "message": "Failed to terminate research"},
1921 status_code=500,
1922 )
1925def _unlink_report_file_if_inside_root(report_path: Optional[str]) -> bool:
1926 """Delete a stored report file, but only when it is a regular file
1927 inside the research reports root.
1929 The path arrives from the ``research_history.report_path`` column, so it
1930 is treated as untrusted: a corrupted row, or a future buggy writer, could
1931 point it anywhere on the filesystem.
1933 Three properties matter:
1935 * Symlinks are refused outright. Resolving first and unlinking the
1936 resolved path deletes the *target*, so an in-root symlink would delete
1937 a file elsewhere and leave a dangling link behind; the library deletion
1938 path refuses symlinks for the same reason.
1939 * Containment is decided on the resolved location, which also catches an
1940 escape through a symlinked *ancestor* directory -- one already swapped in
1941 before the check. Both sides are resolved with resolve_allow_missing(),
1942 so a path that cannot be fully resolved (a symlink chain whose resolved
1943 form outgrows PATH_MAX, a loop) is refused rather than compared
1944 half-resolved. An ancestor swapped in between the resolve and the
1945 unlink() can still redirect the unlink, since the unlink re-traverses
1946 the path; that residual race needs local write access to the reports
1947 tree mid-request and is outside this guard's threat model.
1948 * The unlink targets the path as given -- never a resolved copy -- and
1949 tolerates a missing file, so there is no separate
1950 exists()-before-unlink() window.
1952 The refused path is not echoed, mirroring config/paths.py.
1954 Returns True when a file was removed.
1955 """
1956 if not report_path:
1957 return False
1958 try:
1959 from ...config.paths import get_research_outputs_directory
1960 from ...security.path_validator import resolve_allow_missing
1962 reports_root = resolve_allow_missing(get_research_outputs_directory())
1963 path = Path(report_path)
1965 if path.is_symlink():
1966 logger.warning(
1967 "Refusing to unlink a symlinked report path; "
1968 "the link target may be outside the reports root"
1969 )
1970 return False
1972 try:
1973 contained = resolve_allow_missing(path).is_relative_to(reports_root)
1974 except OSError:
1975 contained = False
1976 if not contained:
1977 logger.warning(
1978 "Refusing to unlink a report path outside the reports root"
1979 )
1980 return False
1982 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Leaf symlinks are refused and the fully resolved path must be inside the reports root before unlink
1983 path.unlink()
1984 return True
1985 except FileNotFoundError:
1986 # Already gone; nothing to remove.
1987 return False
1988 except Exception:
1989 logger.exception("Error removing report file")
1990 return False
1993@router.delete("/api/delete/{research_id}")
1994def delete_research(
1995 request: Request,
1996 research_id,
1997 username: Annotated[str, Depends(require_auth)],
1998):
1999 """Delete a research record"""
2001 def delete_once():
2002 with get_user_db_session(username) as db_session:
2003 research = (
2004 db_session.query(ResearchHistory)
2005 .filter_by(id=research_id)
2006 .first()
2007 )
2009 if not research:
2010 return _research_not_found(research_id)
2012 status = research.status
2013 report_path = research.report_path
2015 # Atomic guard, part 1: refuse the obvious "row is
2016 # IN_PROGRESS" case before opening the helper. The helper
2017 # repeats this check *inside* the parent DELETE so a row
2018 # that becomes IN_PROGRESS between here and the DELETE
2019 # wins the race (the same #5074 invariant the old
2020 # inline-WHERE shape preserved). The two checks together
2021 # guarantee we never silently delete a row a worker owns.
2022 if status == ResearchStatus.IN_PROGRESS:
2023 return JSONResponse(
2024 {
2025 "status": "error",
2026 "message": "Cannot delete research that is in progress",
2027 },
2028 status_code=400,
2029 )
2031 # Atomic guard, part 2: the claimed-queue predicate. Again,
2032 # the helper repeats this inside the parent DELETE; the
2033 # pre-check exists only to fast-path the 400 so we don't
2034 # pay for the cascade on a row we already know is claimed.
2035 is_claimed = db_session.query(
2036 db_session.query(QueuedResearch.id)
2037 .filter(
2038 QueuedResearch.research_id == research_id,
2039 QueuedResearch.is_processing.is_(True),
2040 )
2041 .exists()
2042 ).scalar()
2043 if is_claimed:
2044 return JSONResponse(
2045 {
2046 "status": "error",
2047 "message": "Cannot delete research that is in progress",
2048 },
2049 status_code=400,
2050 )
2052 # Application-level cascade with the #5074 predicates
2053 # baked into the parent DELETE (see
2054 # ``_delete_research_with_dependents``).
2055 rag_targets = collect_research_document_rag_targets(
2056 db_session, research_id
2057 )
2058 deleted = _delete_research_with_dependents(db_session, research_id)
2059 if not deleted:
2060 # A concurrent request claimed the row between our
2061 # pre-check and the helper's parent DELETE — the
2062 # #5074 invariant holds (the row still exists, the
2063 # worker is in charge). Surface the same 400 the
2064 # pre-check would have emitted.
2065 if db_session.get(ResearchHistory, research_id) is not None: 2065 ↛ 2077line 2065 didn't jump to line 2077 because the condition on line 2065 was always true
2066 return JSONResponse(
2067 {
2068 "status": "error",
2069 "message": (
2070 "Cannot delete research that is in progress"
2071 ),
2072 },
2073 status_code=400,
2074 )
2075 # Concurrent delete instead: the row is already gone,
2076 # honour the user's intent.
2077 db_session.commit()
2078 return {"status": "success"}
2080 # ``user_active_researches`` is not a CASCADE child of
2081 # ``research_history.id`` (see ``chat/service.py``'s parallel
2082 # cleanup), so the helper above does not touch it. Drop any
2083 # leftover per-user counter row here so deleting a research
2084 # doesn't leave the slot counted as occupied forever.
2085 db_session.query(UserActiveResearch).filter(
2086 UserActiveResearch.research_id == research_id,
2087 ).delete(synchronize_session=False)
2089 from ..queue.lifecycle_cleanup import cleanup_queued_research_state
2091 cleanup_queued_research_state(
2092 db_session, [research_id], acting_user=username
2093 )
2094 db_session.commit()
2096 # Files cannot roll back with the transaction. Remove the report only
2097 # after its database deletion commits, retaining the containment guard.
2098 _unlink_report_file_if_inside_root(report_path)
2099 # The chunk rows must outlive the parent transaction: vector deletion
2100 # resolves FAISS IDs from them. Purge only after the guarded delete
2101 # committed, never after a refusal or rollback.
2102 if rag_targets:
2103 from ...research_library.deletion.services.document_deletion import (
2104 DocumentDeletionService,
2105 )
2107 DocumentDeletionService(username).purge_deleted_research_documents(
2108 rag_targets
2109 )
2110 return {"status": "success"}
2112 try:
2113 # A concurrent writer can invalidate the cascade's read snapshot.
2114 # The failed context rolls back; each bounded retry re-reads the row,
2115 # queue claim and shared references in a new transaction.
2116 return retry_on_db_lock(delete_once)
2117 except Exception:
2118 logger.exception("Error deleting research")
2119 return JSONResponse(
2120 {"status": "error", "message": "Failed to delete research"},
2121 status_code=500,
2122 )
2125@router.post("/api/clear_history")
2126def clear_history(
2127 request: Request, username: Annotated[str, Depends(require_auth)]
2128):
2129 """Clear all research history"""
2131 def clear_once():
2132 with get_user_db_session(username) as db_session:
2133 # Get all research records first to clean up files. Select
2134 # only id + report_path (the columns the cleanup loop uses)
2135 # so clearing history doesn't load every report body into
2136 # memory (#4560).
2137 research_records = db_session.query(
2138 ResearchHistory.id,
2139 ResearchHistory.report_path,
2140 ResearchHistory.status,
2141 ).all()
2143 # Get IDs of currently active research (snapshot)
2144 protected_ids = set(get_active_research_ids())
2145 protected_ids.update(
2146 research.id
2147 for research in research_records
2148 if research.status == ResearchStatus.IN_PROGRESS
2149 )
2150 deletable_ids = [
2151 research.id
2152 for research in research_records
2153 if research.id not in protected_ids
2154 ]
2156 # One scan of research_resources for the whole set instead of
2157 # one per tracker-owning research. Building the index starts
2158 # a real DB transaction shared with all deletes. A concurrent
2159 # WAL write makes a stale snapshot fail closed on mutation;
2160 # retrying the request builds a fresh index.
2161 shared_downloads = SharedDownloadIndex.build(
2162 db_session, deletable_ids
2163 )
2164 deleted_ids = []
2165 rag_targets = []
2166 for research_id in deletable_ids:
2167 # The #5074 atomic guard is applied INSIDE the helper's
2168 # parent DELETE; a row whose queue entry was claimed
2169 # after this snapshot was taken is observed by SQL and
2170 # preserved. The pre-check below is purely a fast-path
2171 # so we don't pay for the cascade on a row we already
2172 # know is claimed.
2173 is_claimed = db_session.query(
2174 db_session.query(QueuedResearch.id)
2175 .filter(
2176 QueuedResearch.research_id == research_id,
2177 QueuedResearch.is_processing.is_(True),
2178 )
2179 .exists()
2180 ).scalar()
2181 if is_claimed:
2182 continue
2183 # Application-level cascade with the IN_PROGRESS and
2184 # claimed-queue predicates baked into the parent
2185 # DELETE; see ``_delete_research_with_dependents``.
2186 document_targets = collect_research_document_rag_targets(
2187 db_session, research_id
2188 )
2189 if _delete_research_with_dependents(
2190 db_session, research_id, shared_downloads=shared_downloads
2191 ):
2192 deleted_ids.append(research_id)
2193 rag_targets.extend(document_targets)
2195 if deleted_ids:
2196 # The cascade has no FK path to this per-user counter table.
2197 # Remove only rows whose guarded parent DELETE succeeded;
2198 # protected or raced-to-active research keeps its slot.
2199 db_session.query(UserActiveResearch).filter(
2200 UserActiveResearch.research_id.in_(deleted_ids)
2201 ).delete(synchronize_session=False)
2203 from ..queue.lifecycle_cleanup import (
2204 cleanup_queued_research_state,
2205 )
2207 cleanup_queued_research_state(
2208 db_session, deleted_ids, acting_user=username
2209 )
2211 db_session.commit()
2213 for research in research_records:
2214 if research.id in deleted_ids:
2215 _unlink_report_file_if_inside_root(research.report_path)
2216 if rag_targets:
2217 from ...research_library.deletion.services.document_deletion import (
2218 DocumentDeletionService,
2219 )
2221 DocumentDeletionService(username).purge_deleted_research_documents(
2222 rag_targets
2223 )
2224 return {"status": "success"}
2226 try:
2227 # Retry the whole batch so its protection snapshot and shared-download
2228 # index are rebuilt after rollback; never reuse a partially applied plan.
2229 return retry_on_db_lock(clear_once)
2230 except Exception:
2231 logger.exception("Error clearing history")
2232 return JSONResponse(
2233 {"status": "error", "message": "Failed to process request"},
2234 status_code=500,
2235 )
2238@router.post("/open_file_location")
2239def open_file_location(
2240 request: Request, username: Annotated[str, Depends(require_auth)]
2241):
2242 """Open a file location in the system file explorer.
2244 Security: This endpoint is disabled for server deployments.
2245 It only makes sense for desktop usage where the server and client are on the same machine.
2246 """
2247 return JSONResponse(
2248 {
2249 "status": "error",
2250 "message": "This feature is disabled. It is only available in desktop mode.",
2251 },
2252 status_code=403,
2253 )
2256# save_raw_config refuses a configuration with any key whose lowercased name
2257# contains one of these: such keys name a Python module or class to import.
2258_BLOCKED_KEY_PATTERNS = ("module_path", "class_name", "module", "class")
2260# Its refusal lists at most this many protected keys, each path cut to at
2261# most this many characters, so the response and its log line stay small
2262# however many protected keys a configuration has and however long the names
2263# above them are. One protected key is enough to refuse the save, so a cut
2264# list changes only what is listed.
2265_MAX_REPORTED_BLOCKED_KEYS = 100
2266_MAX_BLOCKED_KEY_PATH_CHARS = 200
2269def _blocked_key_path(parts: list[str | int]) -> str:
2270 """Render *parts* as ``a.b[0].c``, cut to _MAX_BLOCKED_KEY_PATH_CHARS.
2272 Names join with dots and array indexes go in brackets, as
2273 save_raw_config has always reported them. Rendering stops once the
2274 path passes the limit, so a long name is never copied whole, and a cut
2275 path ends with an ellipsis.
2276 """
2277 limit = _MAX_BLOCKED_KEY_PATH_CHARS
2278 pieces: list[str] = []
2279 size = 0
2280 for part in parts:
2281 if isinstance(part, int):
2282 # Not f"[{part}]": Semgrep's sql-string-concatenation rule
2283 # reports that assignment as SQL built from input.
2284 piece = "".join(("[", str(part), "]"))
2285 elif size:
2286 piece = "." + part[: limit + 1]
2287 else:
2288 piece = part[: limit + 1]
2289 pieces.append(piece)
2290 size += len(piece)
2291 if size > limit:
2292 break
2293 path = "".join(pieces)
2294 if len(path) > limit:
2295 path = path[: limit - 1] + "\N{HORIZONTAL ELLIPSIS}"
2296 return path
2299def _find_blocked_keys(config: dict) -> tuple[list[str], bool]:
2300 """Return the paths of the protected keys in a parsed configuration.
2302 Every table, inline table and array is searched, depth first in the
2303 parsed configuration's key order, for a key whose lowercased name
2304 contains one of _BLOCKED_KEY_PATTERNS. The second value is True when
2305 there are more than _MAX_REPORTED_BLOCKED_KEYS of them: the walk stops
2306 at the first one past that limit, so the list holds the first
2307 _MAX_REPORTED_BLOCKED_KEYS it found.
2309 A path is rendered only for a protected key, by _blocked_key_path, so
2310 the walk's work does not grow with the length of the names above one.
2311 The walk keeps an explicit stack instead of recursing: a dotted key in
2312 an inline table nests up to 32 tables for a few frames of tomllib's own
2313 recursion, so a configuration within the parsing limits can nest deeper
2314 than Python's recursion limit.
2315 """
2316 found: list[str] = []
2317 parts: list[str | int] = []
2318 stack = [iter(config.items())]
2319 while stack:
2320 entry = next(stack[-1], None)
2321 if entry is None:
2322 stack.pop()
2323 if stack:
2324 parts.pop()
2325 continue
2326 name, value = entry
2327 parts.append(name)
2328 if isinstance(name, str):
2329 lowered = name.lower()
2330 if any(pattern in lowered for pattern in _BLOCKED_KEY_PATTERNS):
2331 if len(found) == _MAX_REPORTED_BLOCKED_KEYS:
2332 return found, True
2333 found.append(_blocked_key_path(parts))
2334 if isinstance(value, dict):
2335 stack.append(iter(value.items()))
2336 elif isinstance(value, list):
2337 stack.append(enumerate(value))
2338 else:
2339 parts.pop()
2340 return found, False
2343@router.post("/api/save_raw_config")
2344async def save_raw_config(
2345 request: Request, username: Annotated[str, Depends(require_auth)]
2346):
2347 """Save raw configuration"""
2348 # Same media-type check as start_research: main's
2349 # @require_json_body(error_format="success") refused a non-JSON media
2350 # type with this 400 before the config file could be written.
2351 if not is_json_request(request):
2352 return json_body_error("success", "Request body must be valid JSON")
2353 data, body_error = await read_json_dict(
2354 request, "success", "Request body must be valid JSON"
2355 )
2356 if body_error is not None:
2357 return body_error
2358 raw_config = data.get("raw_config")
2360 if not raw_config:
2361 return JSONResponse(
2362 {"success": False, "error": "Raw configuration is required"},
2363 status_code=400,
2364 )
2366 # Security: Parse and validate the TOML to block dangerous keys
2367 from ...security import toml_limits
2369 try:
2370 # Bounded: the body may be 16 MB (fastapi_app.py's default JSON body
2371 # limit), and tomllib's cost for one long dotted key grows with the
2372 # square of its parts. loads_toml_within_limits refuses text over
2373 # 512 KiB, or with a key or table header of more than 32 dotted parts,
2374 # before tomllib runs (security/toml_limits.py).
2375 #
2376 # Offloaded: within those limits the scan and the parse are still
2377 # pure-Python CPU work. On the event loop, with the single-worker
2378 # deployment this ships as, that would stall EVERY request for the
2379 # duration, for any authenticated user. asyncio.to_thread rather than
2380 # run_db_sync: this opens no DB session, and run_db_sync's docstring
2381 # says to prefer to_thread for purely CPU-bound work. The GIL is
2382 # released periodically during the parse, so the loop interleaves
2383 # instead of stopping dead.
2384 #
2385 # Note this, and the protected-key walk below, run BEFORE the
2386 # system.allow_config_write gate (enforced inside write_file_verified
2387 # below), so they are reachable even on deployments that have config
2388 # writing turned off. Offloading and the limits are what make that
2389 # ordering safe; they are not an argument for keeping it.
2390 parsed_config = await asyncio.to_thread(
2391 toml_limits.loads_toml_within_limits, raw_config
2392 )
2393 except toml_limits.TOMLLimitError:
2394 logger.warning("Refused TOML configuration over the parsing limits")
2395 return JSONResponse(
2396 {
2397 "success": False,
2398 "error": (
2399 "Configuration exceeds the parsing limits: at most "
2400 f"{toml_limits.MAX_TOML_BYTES // 1024} KiB, and at most "
2401 f"{toml_limits.MAX_TOML_KEY_PARTS} dotted parts in any "
2402 "key or table header."
2403 ),
2404 },
2405 status_code=400,
2406 )
2407 except Exception:
2408 logger.warning("Invalid TOML configuration")
2409 # Don't expose internal exception details to users (CWE-209)
2410 return JSONResponse(
2411 {
2412 "success": False,
2413 "error": "Invalid TOML syntax. Please check your configuration format.",
2414 },
2415 status_code=400,
2416 )
2418 # Security: Check for dangerous keys that could enable code execution.
2419 # Offloaded like the parse: the walk is pure-Python work that grows with
2420 # the size of the parsed configuration.
2421 blocked_keys, more_blocked_keys = await asyncio.to_thread(
2422 _find_blocked_keys, parsed_config
2423 )
2424 if blocked_keys or more_blocked_keys:
2425 logger.warning(
2426 "Security: Blocked attempt to write config with dangerous keys: {}",
2427 _log_value_preview(blocked_keys),
2428 )
2429 error = "Configuration contains protected keys that cannot be modified"
2430 if more_blocked_keys:
2431 error += f"; only the first {len(blocked_keys)} are listed"
2432 return JSONResponse(
2433 {
2434 "success": False,
2435 "error": error,
2436 "blocked_keys": blocked_keys,
2437 "blocked_keys_truncated": more_blocked_keys,
2438 },
2439 status_code=403,
2440 )
2442 try:
2443 from ...security.file_write_verifier import write_file_verified
2445 # Get the config file path (uses centralized path config, respects LDR_DATA_DIR)
2446 config_dir = get_config_directory()
2447 config_path = config_dir / "config.toml"
2449 # Write the configuration to file. Threadpooled: the verifier
2450 # resolves the allow-flag setting (may open the user's DB session)
2451 # and does sync disk IO — neither belongs on the event loop.
2452 await run_db_sync(
2453 write_file_verified,
2454 config_path,
2455 raw_config,
2456 "system.allow_config_write",
2457 context="system configuration file",
2458 )
2460 return {"success": True}
2461 except Exception:
2462 logger.exception("Error saving configuration file")
2463 return JSONResponse(
2464 {"success": False, "error": "Failed to process request"},
2465 status_code=500,
2466 )
2469@router.get("/api/history")
2470def get_history(
2471 request: Request, username: Annotated[str, Depends(require_auth)]
2472):
2473 """Get research history.
2475 NOTE: Returns a raw list of items. The newer canonical endpoint
2476 ``GET /history/api`` (in routers/history.py) returns the wrapped
2477 shape ``{"status": "success", "items": [...]}`` — JS callers
2478 should prefer that one. Kept here because Puppeteer tests still
2479 target this path; consolidating to a single endpoint with a
2480 consistent shape is tracked as a follow-up cleanup.
2481 """
2483 # Bound the result set. Without a limit this endpoint loaded every
2484 # research row (and its research_meta JSON, which can hold a settings
2485 # snapshot) into memory at once (#4560). Mirrors the clamp used by the
2486 # symmetric /history/api endpoint.
2487 try:
2488 limit = int(request.query_params.get("limit", 200))
2489 except (TypeError, ValueError):
2490 limit = 200
2491 limit = max(1, min(limit, 500))
2492 try:
2493 offset = parse_bounded_offset(request.query_params.get("offset"))
2494 except ValueError:
2495 return JSONResponse({"error": OFFSET_LIMIT_ERROR}, status_code=400)
2497 try:
2498 with get_user_db_session(username) as db_session:
2499 # Query research history ordered by created_at. Project
2500 # only the metadata columns the loop below consumes — never
2501 # the large ``report_content`` Text body — so this listing
2502 # doesn't pull every report into memory (#4560). This mirrors
2503 # the projection used by the symmetric /history/api endpoint.
2504 research_records = (
2505 db_session.query(
2506 ResearchHistory.id,
2507 ResearchHistory.title,
2508 ResearchHistory.query,
2509 ResearchHistory.mode,
2510 ResearchHistory.status,
2511 ResearchHistory.created_at,
2512 ResearchHistory.completed_at,
2513 ResearchHistory.research_meta,
2514 ResearchHistory.chat_session_id,
2515 )
2516 .order_by(ResearchHistory.created_at.desc())
2517 .limit(limit)
2518 .offset(offset)
2519 .all()
2520 )
2522 # Pre-compute Document counts in a single GROUP BY query
2523 # to avoid an N+1 SELECT-COUNT-per-row inside the loop. The
2524 # symmetric /history/api endpoint in web/routers/history.py
2525 # already uses an outerjoin + group_by — this brings
2526 # /api/history to parity for users with deep history.
2527 research_ids = [r.id for r in research_records]
2528 if research_ids:
2529 doc_count_rows = (
2530 db_session.query(
2531 Document.research_id, func.count(Document.id)
2532 )
2533 .filter(Document.research_id.in_(research_ids))
2534 .group_by(Document.research_id)
2535 .all()
2536 )
2537 doc_counts = dict(doc_count_rows)
2538 else:
2539 doc_counts = {}
2541 # Build history items while session is active to avoid
2542 # DetachedInstanceError on ORM attribute access
2543 history_items = []
2544 for research in research_records:
2545 # Calculate duration if completed
2546 duration_seconds = None
2547 if research.completed_at and research.created_at:
2548 try:
2549 duration_seconds = calculate_duration(
2550 research.created_at, research.completed_at
2551 )
2552 except Exception:
2553 logger.exception("Error calculating duration")
2555 # Look up the pre-computed document count.
2556 doc_count = doc_counts.get(research.id, 0)
2558 # Create a history item
2559 item = {
2560 "id": research.id,
2561 "query": research.query,
2562 "mode": research.mode,
2563 "status": research.status,
2564 "created_at": research.created_at,
2565 "completed_at": research.completed_at,
2566 "duration_seconds": duration_seconds,
2567 "metadata": filter_research_metadata(
2568 research.research_meta
2569 ),
2570 "document_count": doc_count,
2571 }
2572 if research.chat_session_id is not None: 2572 ↛ 2573line 2572 didn't jump to line 2573 because the condition on line 2572 was never true
2573 item["metadata"]["chat_session_id"] = (
2574 research.chat_session_id
2575 )
2577 # Add title if it exists
2578 if hasattr(research, "title") and research.title is not None:
2579 item["title"] = research.title
2581 history_items.append(item)
2583 return {"status": "success", "items": history_items}
2584 except Exception:
2585 logger.exception("Error getting history")
2586 return JSONResponse(
2587 {"status": "error", "message": "Failed to process request"},
2588 status_code=500,
2589 )
2592@router.get("/api/research/{research_id}")
2593def get_research_details(
2594 request: Request,
2595 research_id,
2596 username: Annotated[str, Depends(require_auth)],
2597):
2598 """Get full details of a research using ORM"""
2600 try:
2601 with get_user_db_session(username) as db_session:
2602 research = (
2603 db_session.query(ResearchHistory)
2604 .filter(ResearchHistory.id == research_id)
2605 .first()
2606 )
2608 if not research:
2609 return _research_not_found(research_id)
2611 return {
2612 "id": research.id,
2613 "query": research.query,
2614 "status": research.status,
2615 "progress": research.progress,
2616 "progress_percentage": research.progress or 0,
2617 "mode": research.mode,
2618 "created_at": research.created_at,
2619 "completed_at": research.completed_at,
2620 "metadata": strip_settings_snapshot(research.research_meta),
2621 }
2622 except Exception:
2623 logger.exception("Error getting research details")
2624 return JSONResponse(
2625 {"error": "An internal error has occurred"}, status_code=500
2626 )
2629@router.get("/api/research/{research_id}/logs")
2630def get_research_logs(
2631 request: Request,
2632 research_id,
2633 username: Annotated[str, Depends(require_auth)],
2634):
2635 """Get logs for a specific research.
2637 Accepts an optional ``?limit=N`` that bounds the response to ``N`` rows
2638 (returned oldest-first, matching the default ordering) and is
2639 clamped to ``[1, HISTORY_LOGS_HARD_CAP]`` so a client cannot force an
2640 unbounded load — a long langgraph run can persist thousands of rows. When
2641 ``?limit`` is absent or not a valid integer (``type=int`` yields
2642 ``None``) the historical contract is preserved: every row is returned. The
2643 frontend log panel always sends a valid limit and
2644 ``priority=diagnostic``. That mode does a best-effort newest-first
2645 selection prioritizing errors/CRITICAL/FATAL, then warnings, then
2646 milestones/SUCCESS, before filling the remaining window with routine rows.
2647 If the number of higher-priority rows exceeds the limit, the oldest
2648 diagnostics in those categories and all lower-priority/routine rows are
2649 dropped.
2650 Direct API callers retain newest-N behavior unless they opt in.
2652 Each ``timestamp`` is a raw ``datetime`` that FastAPI's encoder
2653 serializes as ISO 8601, matching the streaming ``/logs/export``
2654 endpoint. (Under Flask this endpoint emitted RFC 822 HTTP-dates via
2655 ``jsonify`` and the two wire formats disagreed; the migration to
2656 FastAPI aligned them.)
2657 """
2659 _limit_raw = request.query_params.get("limit")
2660 try:
2661 limit = int(_limit_raw) if _limit_raw is not None else None
2662 except (TypeError, ValueError):
2663 limit = None
2664 if limit is not None:
2665 limit = max(1, min(limit, HISTORY_LOGS_HARD_CAP))
2666 prioritize_diagnostics = (
2667 request.query_params.get("priority") == "diagnostic"
2668 )
2670 try:
2671 # First check if the research exists
2672 with get_user_db_session(username) as db_session:
2673 research = (
2674 db_session.query(ResearchHistory)
2675 .filter_by(id=research_id)
2676 .first()
2677 )
2678 if not research:
2679 return _research_not_found(research_id)
2681 # Get logs from research_logs table
2682 log_query = db_session.query(ResearchLog).filter_by(
2683 research_id=research_id
2684 )
2685 if limit is None:
2686 log_results = log_query.order_by(
2687 ResearchLog.timestamp, ResearchLog.id
2688 ).all()
2689 elif prioritize_diagnostics:
2690 normalized_level = func.lower(ResearchLog.level)
2691 diagnostic_priority = case(
2692 (
2693 normalized_level.in_(("error", "critical", "fatal")),
2694 0,
2695 ),
2696 (normalized_level == "warning", 1),
2697 (
2698 normalized_level.in_(("milestone", "success")),
2699 2,
2700 ),
2701 else_=3,
2702 )
2703 log_results = (
2704 log_query.order_by(
2705 diagnostic_priority,
2706 ResearchLog.timestamp.desc(),
2707 ResearchLog.id.desc(),
2708 )
2709 .limit(limit)
2710 .all()
2711 )
2712 log_results.sort(key=lambda row: (row.timestamp, row.id))
2713 else:
2714 # Take the newest ``limit`` rows at the SQL layer, then flip
2715 # back to oldest-first so the response ordering is unchanged.
2716 # ``id`` is the tie-break: timestamps are not unique, so without
2717 # it the rows that survive ``.limit()`` at a shared-timestamp
2718 # boundary would be SQL-undefined.
2719 log_results = list(
2720 reversed(
2721 log_query.order_by(
2722 ResearchLog.timestamp.desc(),
2723 ResearchLog.id.desc(),
2724 )
2725 .limit(limit)
2726 .all()
2727 )
2728 )
2730 # Extract log attributes while session is active
2731 # to avoid DetachedInstanceError on ORM attribute access
2732 logs = []
2733 for row in log_results:
2734 logs.append(
2735 {
2736 "id": row.id,
2737 "message": row.message,
2738 "timestamp": row.timestamp,
2739 "log_type": row.level,
2740 }
2741 )
2743 return logs
2745 except Exception:
2746 logger.exception("Error getting research logs")
2747 return JSONResponse(
2748 {"error": "An internal error has occurred"}, status_code=500
2749 )
2752def _log_export_exempt(request: Request) -> bool:
2753 """Keep HEAD pre-flights outside the GET export quota.
2755 Port of main's ``_is_log_export_rate_limit_exempt`` (#5369). Main's
2756 version read Flask's ``request`` global, which does not exist here;
2757 slowapi inspects the callable's signature and passes the Starlette
2758 ``Request`` when it declares exactly one parameter (see
2759 ``slowapi/wrappers.py``'s ``_exempt_when_takes_request``), so the
2760 method is read off the live request instead.
2762 Still needed after the migration: Starlette answers HEAD on a route
2763 registered with ``@router.get``, so a browser or proxy pre-flight
2764 would otherwise burn a slot of the caller's 10/min export budget
2765 without transferring a byte.
2766 """
2767 return request.method == "HEAD" or _api_exempt()
2770_log_export_limit = limiter.shared_limit(
2771 "10 per minute",
2772 scope="log_export",
2773 key_func=_api_user_key,
2774 exempt_when=_log_export_exempt,
2775)
2778# GET *and* HEAD. The log panel's "Download Logs" button issues a HEAD
2779# pre-flight (static/js/components/logpanel.js:1854) so the browser never
2780# saves a JSON error body as a .jsonl file, and it bails out on a non-ok
2781# response before creating the download anchor. Under Flask that worked
2782# for free: werkzeug adds HEAD to any rule that serves GET
2783# (routing/rules.py: `if "HEAD" not in methods and "GET" in methods`).
2784# FastAPI takes `methods` literally, so a bare @router.get answers HEAD
2785# with 405 -- the pre-flight fails, the function returns early, and the
2786# export silently never starts. Serving both verbs restores the Flask
2787# behaviour; Starlette runs the handler and discards the body for HEAD,
2788# exactly as werkzeug did.
2789# Registered as two routes rather than one api_route(methods=["GET","HEAD"]):
2790# a single route with both verbs makes FastAPI emit two OpenAPI operations
2791# that share one generated operationId, which
2792# test_route_contracts.py::test_in_process_openapi_covers_schema_included_routes
2793# correctly rejects as a duplicate. GET stays the documented operation; the
2794# HEAD twin is include_in_schema=False because it is an implementation detail
2795# of the pre-flight, exactly as werkzeug's auto-added HEAD never appeared in
2796# any schema either.
2797@router.head("/api/research/{research_id}/logs/export", include_in_schema=False)
2798@router.get("/api/research/{research_id}/logs/export")
2799@_log_export_limit
2800def export_research_logs(
2801 request: Request,
2802 research_id: str,
2803 username: Annotated[str, Depends(require_auth)],
2804):
2805 """Stream every persisted log row for a research as newline-delimited JSON.
2807 This endpoint deliberately bypasses ``HISTORY_LOGS_HARD_CAP``: a long
2808 langgraph run can persist tens of thousands of rows and the on-screen
2809 log panel rightly caps at ``?limit=5000`` to bound DOM and parsing
2810 memory. A *download*, by contrast, is short-lived — the browser's
2811 download manager streams the response body straight to disk — so the
2812 cap does not protect anything here and only hides data.
2814 Memory layout for very large exports:
2816 * **Server**: ``get_user_db_session`` opens a fresh session inside
2817 the generator so it survives until the response finishes flushing;
2818 ``.yield_per(500)`` pulls rows in 500-row batches from the SQLite
2819 cursor so the full set is never resident in process memory.
2820 * **Wire**: each yielded row is a single ``\\n``-terminated JSON
2821 line, so the server flushes incrementally and the browser receives
2822 chunks as they arrive.
2823 * **Client**: ``Content-Disposition: attachment`` plus the
2824 ``log-download-button`` JS trigger a native browser download, so
2825 the response body streams to disk without first being buffered
2826 into a JS Blob.
2828 The NDJSON format is the standard streaming interchange for
2829 line-oriented records. Each line emits ``id``, ``timestamp``, ``message``,
2830 ``level`` (log level), ``log_type`` (aligned with ``/logs`` endpoint),
2831 ``module``, and ``line_no``. Consumers can ``for line in file: json.loads(line)``
2832 without parsing a wrapper array. ``timestamp`` is serialized explicitly
2833 with ``datetime.isoformat()`` (ISO 8601), which matches how FastAPI's
2834 encoder renders the datetimes returned by the sibling ``/logs`` endpoint.
2835 """
2836 # Verify the research exists (and belongs to this user, since
2837 # ``get_user_db_session(username)`` scopes queries to the user's
2838 # encrypted DB). Failing fast here avoids opening a streaming session
2839 # for a 404.
2840 try:
2841 with get_user_db_session(username) as db_session:
2842 research = (
2843 db_session.query(ResearchHistory)
2844 .filter_by(id=research_id)
2845 .first()
2846 )
2847 if not research:
2848 return _research_not_found(research_id)
2849 except Exception:
2850 logger.exception("Error verifying research for log export")
2851 return JSONResponse(
2852 {"error": "An internal error has occurred"}, status_code=500
2853 )
2855 def generate():
2856 # NEVER hold a DB session across a yield: Starlette iterates sync
2857 # generators via anyio's threadpool, so each next() can land on a
2858 # different OS thread and a session entered on one thread would be
2859 # exited on another (see download_bulk for the thread-affinity
2860 # rationale). Snapshot the ordered id list in one short-lived
2861 # session, then hydrate rows in per-batch sessions, serializing
2862 # each batch fully BEFORE yielding it.
2863 yielded_count = 0
2864 try:
2865 with get_user_db_session(username) as db_session:
2866 ordered_ids = [
2867 row_id
2868 for (row_id,) in db_session.query(ResearchLog.id)
2869 .filter_by(research_id=research_id)
2870 .order_by(ResearchLog.timestamp.asc(), ResearchLog.id.asc())
2871 .all()
2872 ]
2874 batch_size = 500
2875 for start in range(0, len(ordered_ids), batch_size):
2876 batch_ids = ordered_ids[start : start + batch_size]
2877 lines = []
2878 with get_user_db_session(username) as db_session:
2879 rows = (
2880 db_session.query(ResearchLog)
2881 .filter(ResearchLog.id.in_(batch_ids))
2882 .all()
2883 )
2884 by_id = {row.id: row for row in rows}
2885 # Serialize inside the session (ORM attribute access
2886 # may lazy-load); emit in snapshot order so the
2887 # timestamp-then-id ordering is preserved exactly.
2888 for row_id in batch_ids:
2889 row = by_id.get(row_id)
2890 if row is None:
2891 continue
2892 yielded_count += 1
2893 lines.append(
2894 json.dumps(
2895 {
2896 "id": row.id,
2897 "timestamp": row.timestamp.isoformat()
2898 if row.timestamp is not None
2899 else None,
2900 "message": row.message,
2901 "level": row.level,
2902 "log_type": row.level,
2903 "module": row.module,
2904 "line_no": row.line_no,
2905 },
2906 default=str,
2907 )
2908 + "\n"
2909 )
2910 yield "".join(lines)
2911 except Exception:
2912 # If the DB blows up mid-stream we have already sent a 200 +
2913 # partial body, so we can't recover with a JSON error here.
2914 # Log and let the iterator end; the client will see a
2915 # truncated file, which is the best we can do without
2916 # buffering the whole result.
2917 logger.exception(
2918 f"Error streaming logs for export (research_id={research_id}, yielded={yielded_count})"
2919 )
2921 safe_id = re.sub(r"[^a-zA-Z0-9_-]", "", research_id)
2922 filename = f"research_logs_{safe_id or 'export'}.jsonl"
2923 return WorkerCleanupStreamingResponse(
2924 generate(),
2925 media_type="application/x-ndjson",
2926 headers={
2927 "Content-Disposition": f'attachment; filename="{filename}"',
2928 # Disable caching: this is a one-shot download, and a stale
2929 # partial body from a previous run could otherwise confuse a
2930 # retry after a mid-stream failure.
2931 "Cache-Control": "no-store",
2932 "X-Content-Type-Options": "nosniff",
2933 },
2934 )
2937@router.get("/api/report/{research_id}")
2938def get_research_report(
2939 request: Request,
2940 research_id,
2941 username: Annotated[str, Depends(require_auth)],
2942):
2943 """Get the research report content"""
2945 try:
2946 with get_user_db_session(username) as db_session:
2947 # Query using ORM
2948 research = (
2949 db_session.query(ResearchHistory)
2950 .filter_by(id=research_id)
2951 .first()
2952 )
2954 if research is None:
2955 return _research_not_found(research_id)
2957 # Parse metadata if it exists
2958 metadata = research.research_meta
2960 # research.report_content holds the answer-only string;
2961 # rebuild the legacy display shape (answer + Sources from
2962 # research_resources + Metrics from research_meta) on demand.
2963 from ..services.report_assembly_service import (
2964 assemble_full_report,
2965 get_research_source_links_batch,
2966 )
2968 content = assemble_full_report(research, db_session)
2969 # Only None means "research not found" — guarded above.
2970 # Empty-but-found rows return "" and are valid responses.
2971 if content is None:
2972 return _research_not_found(
2973 research_id, message="Report not found"
2974 )
2976 # Sources live in the research_resources table, not research_meta.
2977 # The post-refactor save path never writes the legacy
2978 # `all_links_of_system` metadata key, so reading it here returned
2979 # [] for every research created since chat-mode-v2 (#3665). Read
2980 # the structured table instead — the same source of truth the
2981 # assembled `content` and the news feed already use. limit=None
2982 # returns every source (this field was never top-N), matching the
2983 # full list the assembled `content` renders for the same research.
2984 sources = get_research_source_links_batch(
2985 [research.id], db_session, limit=None
2986 ).get(research.id, [])
2988 # Return the report data with backwards-compatible fields
2989 # Examples expect 'summary', 'sources', 'findings' at top level
2990 safe_metadata = strip_settings_snapshot(metadata)
2991 return {
2992 "content": content,
2993 # Backwards-compatible fields for examples
2994 "summary": content, # The markdown report is the summary
2995 "sources": sources,
2996 "findings": safe_metadata.get("findings", []),
2997 "metadata": {
2998 "title": research.title if research.title else None,
2999 "query": research.query,
3000 "mode": research.mode if research.mode else None,
3001 "created_at": research.created_at
3002 if research.created_at
3003 else None,
3004 "completed_at": research.completed_at
3005 if research.completed_at
3006 else None,
3007 **safe_metadata,
3008 },
3009 }
3011 except Exception:
3012 logger.exception("Error getting research report")
3013 return JSONResponse(
3014 {"error": "An internal error has occurred"}, status_code=500
3015 )
3018@router.post("/api/v1/research/{research_id}/export/{format}")
3019def export_research_report(
3020 request: Request,
3021 research_id,
3022 format,
3023 username: Annotated[str, Depends(require_auth)],
3024):
3025 """Export research report to different formats (LaTeX, Quarto, RIS, PDF, ODT, etc.)"""
3026 try:
3027 # Use the exporter registry to validate format
3028 from ...exporters import ExporterRegistry
3030 if not ExporterRegistry.is_format_supported(format):
3031 available = ExporterRegistry.get_available_formats()
3032 return JSONResponse(
3033 {
3034 "error": f"Invalid format. Available formats: {', '.join(available)}"
3035 },
3036 status_code=400,
3037 )
3039 # Get research from database
3041 try:
3042 with get_user_db_session(username) as db_session:
3043 research = (
3044 db_session.query(ResearchHistory)
3045 .filter_by(id=research_id)
3046 .first()
3047 )
3048 if not research:
3049 return _research_not_found(research_id)
3051 # Build the full assembled report (answer + Sources +
3052 # Metrics) so exporters get the same shape they did
3053 # before the report_content refactor.
3054 from ..services.report_assembly_service import (
3055 assemble_full_report,
3056 )
3058 report_content = assemble_full_report(research, db_session)
3059 if report_content is None: 3059 ↛ 3060line 3059 didn't jump to line 3060 because the condition on line 3059 was never true
3060 return _research_not_found(
3061 research_id, message="Report content not found"
3062 )
3064 # Export to requested format (all in memory)
3065 try:
3066 # Use title or query for the PDF title
3067 pdf_title = research.title or research.query
3069 # Generate export content in memory
3070 export_content, filename, mimetype = (
3071 export_report_to_memory(
3072 report_content, format, title=pdf_title
3073 )
3074 )
3076 # Send the file directly from memory.
3077 #
3078 # A plain ``Response``, not ``StreamingResponse``:
3079 # ``export_content`` is an in-memory bytes object that is
3080 # already fully materialised, and iterating a ``BytesIO``
3081 # yields one chunk per ``0x0A`` byte — for binary PDF/ODT
3082 # payloads that is thousands of tiny ASGI sends, each a
3083 # threadpool dispatch, and it also suppresses
3084 # ``Content-Length`` so the browser gets no download
3085 # progress. Same fix already applied to library.py's PDF
3086 # route; this is its sibling call site.
3087 from fastapi.responses import Response
3089 return Response(
3090 content=export_content,
3091 media_type=mimetype,
3092 headers={
3093 "Content-Disposition": build_content_disposition(
3094 filename
3095 ),
3096 },
3097 )
3098 except MissingPDFDependencyError:
3099 logger.exception(
3100 "PDF export failed: WeasyPrint unavailable"
3101 )
3102 return JSONResponse(
3103 {"error": get_weasyprint_install_instructions()},
3104 status_code=500,
3105 )
3106 except Exception:
3107 logger.exception("Error exporting report")
3108 return JSONResponse(
3109 {
3110 "error": f"Failed to export to {format}. Please try again later."
3111 },
3112 status_code=500,
3113 )
3115 except Exception:
3116 logger.exception("Error in export endpoint")
3117 return JSONResponse(
3118 {"error": "An internal error has occurred"}, status_code=500
3119 )
3121 except Exception:
3122 logger.exception("Unexpected error in export endpoint")
3123 return JSONResponse(
3124 {"error": "An internal error has occurred"}, status_code=500
3125 )
3128@router.get("/api/research/{research_id}/status")
3129@limiter.exempt
3130def get_research_status(
3131 request: Request,
3132 research_id,
3133 username: Annotated[str, Depends(require_auth)],
3134):
3135 """Get the status of a research process"""
3137 try:
3138 with get_user_db_session(username) as db_session:
3139 research = (
3140 db_session.query(ResearchHistory)
3141 .filter_by(id=research_id)
3142 .first()
3143 )
3145 if research is None:
3146 return _research_not_found(research_id)
3148 status = research.status
3149 progress = research.progress
3150 completed_at = research.completed_at
3151 metadata = _parse_research_metadata(research.research_meta)
3152 filtered_metadata = strip_settings_snapshot(metadata)
3154 # Extract and format error information for better UI display
3155 error_info = {}
3156 if metadata.get("error"):
3157 error_msg = str(metadata["error"])
3158 known_message = get_known_research_error_message(error_msg)
3159 error_type = "unknown"
3161 # Detect specific error types
3162 if "timeout" in error_msg.lower():
3163 error_type = "timeout"
3164 error_info = {
3165 "type": "timeout",
3166 "message": "LLM service timed out during synthesis. This may be due to high server load or connectivity issues.",
3167 "suggestion": "Try again later or use a smaller query scope.",
3168 }
3169 elif (
3170 "token limit" in error_msg.lower()
3171 or "context length" in error_msg.lower()
3172 ):
3173 error_type = "token_limit"
3174 error_info = {
3175 "type": "token_limit",
3176 "message": "The research query exceeded the AI model's token limit during synthesis.",
3177 "suggestion": "Try using a more specific query or reduce the research scope.",
3178 }
3179 elif (
3180 "final answer synthesis fail" in error_msg.lower()
3181 or "llm error" in error_msg.lower()
3182 ):
3183 error_type = "llm_error"
3184 error_info = {
3185 "type": "llm_error",
3186 "message": "The AI model encountered an error during final answer synthesis.",
3187 "suggestion": "Check that your LLM service is running correctly or try a different model.",
3188 }
3189 elif "ollama" in error_msg.lower():
3190 error_type = "ollama_error"
3191 error_info = {
3192 "type": "ollama_error",
3193 "message": "The Ollama service is not responding properly.",
3194 "suggestion": "Make sure Ollama is running with 'ollama serve' and the model is downloaded.",
3195 }
3196 elif "connection" in error_msg.lower():
3197 error_type = "connection"
3198 error_info = {
3199 "type": "connection",
3200 "message": "Connection error with the AI service.",
3201 "suggestion": "Check your internet connection and AI service status.",
3202 }
3203 elif metadata.get("solution"):
3204 # Use the solution provided in metadata if available
3205 error_info = {
3206 "type": error_type,
3207 "message": filtered_metadata["error"],
3208 "suggestion": str(metadata.get("solution")),
3209 }
3210 else:
3211 # Legacy rows can contain str(exception), including
3212 # absolute report paths. The stored detail stays private.
3213 error_info = {
3214 "type": error_type,
3215 "message": filtered_metadata["error"],
3216 "suggestion": "Try again with a different query or check the application logs.",
3217 }
3219 # The polling UI displays message alone. Preserve the worker's
3220 # exact authored classification instead of replacing e.g. a
3221 # missing-model message with a generic Ollama-service error.
3222 if known_message is not None:
3223 error_info["message"] = known_message
3224 if metadata.get("solution"):
3225 error_info["suggestion"] = str(metadata["solution"])
3227 # Get the latest milestone log for this research
3228 latest_milestone = None
3229 try:
3230 milestone_log = (
3231 db_session.query(ResearchLog)
3232 .filter_by(research_id=research_id, level="MILESTONE")
3233 # id tie-breaks equal timestamps so "latest" is
3234 # deterministic (the most recently inserted milestone).
3235 .order_by(
3236 ResearchLog.timestamp.desc(), ResearchLog.id.desc()
3237 )
3238 .first()
3239 )
3240 if milestone_log:
3241 latest_milestone = {
3242 "message": milestone_log.message,
3243 "time": milestone_log.timestamp.isoformat()
3244 if milestone_log.timestamp
3245 else None,
3246 "type": "MILESTONE",
3247 }
3248 logger.debug(
3249 f"Found latest milestone for research {research_id}: {milestone_log.message}"
3250 )
3251 else:
3252 logger.debug(
3253 f"No milestone logs found for research {research_id}"
3254 )
3255 except Exception:
3256 logger.warning("Error fetching latest milestone")
3258 if error_info:
3259 filtered_metadata["error_info"] = error_info
3261 response_data = {
3262 "status": status,
3263 "progress": progress,
3264 "completed_at": completed_at,
3265 "metadata": filtered_metadata,
3266 }
3268 # Include latest milestone as a log_entry for frontend compatibility
3269 if latest_milestone:
3270 response_data["log_entry"] = latest_milestone
3272 return response_data
3273 except Exception:
3274 logger.exception("Error getting research status")
3275 return JSONResponse(
3276 {"error": "Error checking research status"}, status_code=500
3277 )
3280@router.get("/api/queue/status")
3281def get_queue_status(
3282 request: Request, username: Annotated[str, Depends(require_auth)]
3283):
3284 """Get the current queue status for the user"""
3286 from ..queue import QueueManager
3288 try:
3289 queue_items = QueueManager.get_user_queue(username)
3291 return {
3292 "status": "success",
3293 "queue": queue_items,
3294 "total": len(queue_items),
3295 }
3296 except Exception:
3297 logger.exception("Error getting queue status")
3298 return JSONResponse(
3299 {"status": "error", "message": "Failed to process request"},
3300 status_code=500,
3301 )
3304@router.get("/api/queue/{research_id}/position")
3305def get_queue_position(
3306 request: Request,
3307 research_id,
3308 username: Annotated[str, Depends(require_auth)],
3309):
3310 """Get the queue position for a specific research"""
3312 from ..queue import QueueManager
3314 try:
3315 position = QueueManager.get_queue_position(username, research_id)
3317 if position is None:
3318 return _research_not_found(
3319 research_id, message="Research not found in queue"
3320 )
3322 return {"status": "success", "position": position}
3323 except Exception:
3324 logger.exception("Error getting queue position")
3325 return JSONResponse(
3326 {"status": "error", "message": "Failed to process request"},
3327 status_code=500,
3328 )
3331@router.get("/api/config/limits")
3332def get_upload_limits(
3333 request: Request, username: Annotated[str, Depends(require_auth)]
3334):
3335 """
3336 Get file upload configuration limits.
3338 Returns the backend's authoritative limits for file uploads,
3339 allowing the frontend to stay in sync without hardcoding values.
3341 Static constants only, but auth-gated to match the pre-migration
3342 Flask route (@login_required) — no endpoint enumeration for free.
3343 """
3344 return {
3345 "max_file_size": FileUploadValidator.MAX_FILE_SIZE,
3346 "max_files": FileUploadValidator.MAX_FILES_PER_REQUEST,
3347 "allowed_mime_types": list(FileUploadValidator.ALLOWED_MIME_TYPES),
3348 }
3351@router.post("/api/upload/pdf")
3352@upload_rate_limit_user
3353@upload_rate_limit_ip
3354async def upload_pdf(
3355 request: Request,
3356 username: Annotated[str, Depends(require_auth)],
3357):
3358 """
3359 Upload and extract text from PDF files with comprehensive security validation.
3361 Security features:
3362 - Rate limiting per user and per IP (``security.rate_limit_upload_user``
3363 and ``security.rate_limit_upload_ip``)
3364 - File size validation (``FileUploadValidator.MAX_FILE_SIZE``,
3365 configurable via ``security.upload_max_file_size_mb`` /
3366 ``LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB``)
3367 - File count validation (``FileUploadValidator.MAX_FILES_PER_REQUEST``)
3368 - PDF structure validation
3369 - MIME type validation
3371 Performance improvements:
3372 - Single-pass PDF processing (text + metadata)
3373 - Optimized extraction service
3374 """
3375 # Starlette's form parser yields STARLETTE UploadFile instances;
3376 # fastapi.UploadFile is a SUBCLASS of it on fastapi>=0.113, so
3377 # `isinstance(f, fastapi.UploadFile)` is False for every real
3378 # upload and the filter below silently dropped all files
3379 # (every upload 400'd with "No files provided").
3380 from starlette.datastructures import UploadFile
3382 try:
3383 # Early request size validation (before reading any files)
3384 # This prevents memory exhaustion from chunked encoding attacks
3385 max_request_size = (
3386 FileUploadValidator.MAX_FILES_PER_REQUEST
3387 * FileUploadValidator.MAX_FILE_SIZE
3388 )
3389 content_length = request.headers.get("content-length")
3390 if content_length and int(content_length) > max_request_size: 3390 ↛ 3391line 3390 didn't jump to line 3391 because the condition on line 3390 was never true
3391 return JSONResponse(
3392 {
3393 "error": f"Request too large. Maximum size is {max_request_size // (1024 * 1024)}MB"
3394 },
3395 status_code=413,
3396 )
3398 # Parse multipart form data — FastAPI's UploadFile parameter binding
3399 # doesn't work cleanly with the `files: list` shape we want, so we
3400 # parse the form manually and pull out everything named "files".
3401 #
3402 # Nothing in the signature asks FastAPI for the body, so this is
3403 # the first parse of the request, and leaving the block closes
3404 # every spooled upload whether the loop below reached it or not
3405 # (early returns, rejected names, parts under other field names).
3406 # FastAPI's form extraction used to register that cleanup.
3407 async with request.form() as form:
3408 files: list[UploadFile] = [
3409 f for f in form.getlist("files") if isinstance(f, UploadFile)
3410 ]
3412 if not files:
3413 return JSONResponse(
3414 {"error": "No files provided"}, status_code=400
3415 )
3416 if all(not f.filename for f in files):
3417 return JSONResponse(
3418 {"error": "No files selected"}, status_code=400
3419 )
3421 # Validate file count
3422 is_valid, error_msg = FileUploadValidator.validate_file_count(
3423 len(files)
3424 )
3425 if not is_valid:
3426 return JSONResponse({"error": error_msg}, status_code=400)
3428 # Get PDF extraction service
3429 pdf_service = get_pdf_extraction_service()
3431 extracted_texts = []
3432 total_files = len(files)
3433 processed_files = 0
3434 errors = []
3436 for file in files:
3437 if not file or not file.filename: 3437 ↛ 3438line 3437 didn't jump to line 3438 because the condition on line 3437 was never true
3438 errors.append("Unnamed file: Skipped")
3439 continue
3441 try:
3442 filename = sanitize_filename(
3443 file.filename, allowed_extensions={".pdf"}
3444 )
3445 except UnsafeFilenameError:
3446 errors.append(
3447 "Rejected file: invalid or disallowed filename"
3448 )
3449 continue
3451 try:
3452 # Per-file size guard BEFORE reading the spool into
3453 # RAM (GHSA-38fh finding 2): the request-level
3454 # precheck keys off Content-Length, which chunked
3455 # multipart bodies do not carry, so waiting for the
3456 # post-read validator would read the whole part into
3457 # RAM before rejecting it. Mirrors rag.py's pre-read
3458 # guard. On the pinned Starlette range (1.3.x),
3459 # file.size is the byte count request.form() actually
3460 # wrote to the part's disk-backed SpooledTemporaryFile
3461 # (it starts at 0 and grows per chunk), not a
3462 # client-declared value, so it is never None here.
3463 # The post-read validator's size check below is
3464 # therefore only defence in depth for a size of None,
3465 # which this Starlette range never produces.
3466 # request.form() has already spooled the whole part
3467 # to disk by this point; the guard prevents only the
3468 # read into RAM. A part under the cap is still read
3469 # fully into RAM, by the cap's design (unchanged by
3470 # this guard).
3471 size_ok, size_err = FileUploadValidator.validate_file_size(
3472 file.size, None
3473 )
3474 if not size_ok:
3475 errors.append(f"{filename}: {size_err}")
3476 continue
3478 # Read file content (UploadFile spools large bodies to disk)
3479 pdf_content = await file.read()
3481 # Validation + PDF text extraction are CPU-bound (up to
3482 # FileUploadValidator.MAX_FILE_SIZE per file ×
3483 # MAX_FILES_PER_REQUEST files) — run them in the
3484 # threadpool so they don't freeze the event loop (and
3485 # with it every other request and WebSocket) for the
3486 # duration of the parse. No DB session is opened here,
3487 # so a plain to_thread (not run_db_sync) is correct.
3488 def _validate_and_extract(
3489 content=pdf_content, name=filename
3490 ):
3491 ok, err = FileUploadValidator.validate_upload(
3492 filename=name,
3493 file_content=content,
3494 content_length=file.size,
3495 )
3496 if not ok:
3497 return None, err
3498 return (
3499 pdf_service.extract_text_and_metadata(
3500 content, name
3501 ),
3502 None,
3503 )
3505 result, error_msg = await asyncio.to_thread(
3506 _validate_and_extract
3507 )
3509 if result is None:
3510 errors.append(f"{filename}: {error_msg}")
3511 continue
3513 if result["success"]:
3514 extracted_texts.append(
3515 {
3516 "filename": result["filename"],
3517 "text": result["text"],
3518 "size": result["size"],
3519 "pages": result["pages"],
3520 "truncated": result.get("truncated", False),
3521 }
3522 )
3523 processed_files += 1
3524 else:
3525 errors.append(f"{filename}: {result['error']}")
3527 except Exception:
3528 logger.exception(f"Error processing {filename}")
3529 errors.append(f"{filename}: Error processing file")
3530 finally:
3531 # Close the file stream to release resources
3532 try:
3533 await file.close()
3534 except Exception:
3535 logger.debug(
3536 "best-effort file stream close", exc_info=True
3537 )
3539 # Prepare response
3540 response_data = {
3541 "status": "success",
3542 "processed_files": processed_files,
3543 "total_files": total_files,
3544 "extracted_texts": extracted_texts,
3545 "combined_text": "\n\n".join(
3546 [
3547 f"--- From {item['filename']} ---\n"
3548 + (
3549 "[Partial PDF extraction: only part of this "
3550 "document's text is included.]\n"
3551 if item["truncated"]
3552 else ""
3553 )
3554 + item["text"]
3555 for item in extracted_texts
3556 ]
3557 ),
3558 "errors": errors,
3559 }
3561 if processed_files == 0:
3562 return JSONResponse(
3563 {
3564 "status": "error",
3565 "message": "No files were processed successfully",
3566 "errors": errors,
3567 },
3568 status_code=400,
3569 )
3571 return response_data
3573 except FormParserError as e:
3574 # python-multipart could not parse the body (garbage where the
3575 # boundary should be, a malformed part header, an over-long
3576 # boundary): the client's fault, and the 400 FastAPI's form
3577 # handling answered before this handler parsed the form itself.
3578 # Only the parser's own errors are relabelled; anything else that
3579 # fails while reading the form (an OSError spooling an upload,
3580 # say) is the server's and goes to the generic handler below,
3581 # which logs it.
3582 raise HTTPException(
3583 status_code=400, detail="There was an error parsing the body"
3584 ) from e
3585 except StarletteHTTPException:
3586 # Starlette's own multipart limits (missing boundary parameter,
3587 # oversized field) already arrive as a 400: let it through instead
3588 # of turning it into the generic 500 below.
3589 raise
3590 except Exception:
3591 logger.exception("Error processing PDF upload")
3592 return JSONResponse(
3593 {"error": "Failed to process PDF files"}, status_code=500
3594 )