Coverage for src/local_deep_research/web/routes/research_routes.py: 91%
780 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1import io
2import json
3from datetime import datetime, UTC
4from pathlib import Path
6from flask import (
7 Blueprint,
8 jsonify,
9 redirect,
10 request,
11 send_file,
12 session,
13 url_for,
14)
15from loguru import logger
16from ...settings.logger import log_settings
17from sqlalchemy import func
18from sqlalchemy.orm import Session
20# Security imports
21from ...config.constants import DEFAULT_OLLAMA_URL
22from ...exceptions import DuplicateResearchError, SystemAtCapacityError
23from ...llm.providers.base import normalize_provider
24from ...constants import HISTORY_LOGS_HARD_CAP, ResearchStatus
25from ...security import (
26 FileUploadValidator,
27 UnsafeFilenameError,
28 filter_research_metadata,
29 sanitize_filename,
30 strip_settings_snapshot,
31)
32from ...utilities.url_utils import is_safe_custom_llm_endpoint
33from ...security.rate_limiter import (
34 api_rate_limit,
35 upload_rate_limit_ip,
36 upload_rate_limit_user,
37)
38from ...security.decorators import require_json_body
39from ...config.paths import get_config_directory
41# Services imports
42from ..services.pdf_extraction_service import get_pdf_extraction_service
43from ..services.pdf_service import (
44 MissingPDFDependencyError,
45 get_weasyprint_install_instructions,
46)
48from ...database.models import (
49 QueuedResearch,
50 ResearchHistory,
51 ResearchLog,
52 UserActiveResearch,
53)
54from ...database.models.library import Document as Document
55from ...database.session_context import get_g_db_session, get_user_db_session
56from ..auth.decorators import login_required
57from ..auth.password_utils import get_user_password, resolve_user_password
58from ..models.database import calculate_duration
59from ..services.research_service import (
60 export_report_to_memory,
61 run_research_process,
62 start_research_process,
63)
64from ...security.rate_limiter import limiter
65from ..utils.templates import render_template_with_defaults
66from .globals import (
67 append_research_log,
68 get_active_research_ids,
69 get_research_field,
70 is_research_active,
71 set_termination_flag,
72)
73from ...constants import DEFAULT_SEARCH_TOOL
75# Create a Blueprint for the research application
76research_bp = Blueprint("research", __name__)
79# NOTE: Routes use session["username"] (not .get()) intentionally.
80# @login_required guarantees the key exists; direct access fails fast
81# if the decorator is ever removed.
84# Add static route at the root level
85@research_bp.route("/redirect-static/<path:path>")
86def redirect_static(path):
87 """Redirect old static URLs to new static URLs"""
88 return redirect(url_for("static", filename=path))
91@research_bp.route("/progress/<string:research_id>")
92@login_required
93def progress_page(research_id):
94 """Render the research progress page"""
95 return render_template_with_defaults("pages/progress.html")
98@research_bp.route("/details/<string:research_id>")
99@login_required
100def research_details_page(research_id):
101 """Render the research details page"""
102 return render_template_with_defaults("pages/details.html")
105@research_bp.route("/results/<string:research_id>")
106@login_required
107def results_page(research_id):
108 """Render the research results page"""
109 return render_template_with_defaults("pages/results.html")
112@research_bp.route("/history")
113@login_required
114def history_page():
115 """Render the history page"""
116 return render_template_with_defaults("pages/history.html")
119# Add missing settings routes
120@research_bp.route("/settings", methods=["GET"])
121@login_required
122def settings_page():
123 """Render the settings page"""
124 return render_template_with_defaults("settings_dashboard.html")
127def _extract_research_params(data, settings_manager):
128 """Extract and resolve research parameters from request data and settings.
130 Returns a dict with keys: model_provider, model, custom_endpoint,
131 ollama_url, search_engine, max_results, time_period, iterations,
132 questions_per_iteration, strategy.
133 """
134 model_provider = data.get("model_provider")
135 if not model_provider:
136 model_provider = settings_manager.get_setting("llm.provider", "ollama")
137 logger.debug(
138 f"No model_provider in request, using database setting: {model_provider}"
139 )
140 else:
141 logger.debug(f"Using model_provider from request: {model_provider}")
142 # Normalize provider to lowercase canonical form
143 model_provider = normalize_provider(model_provider)
145 model = data.get("model")
146 if not model:
147 model = settings_manager.get_setting("llm.model", None)
148 logger.debug(f"No model in request, using database setting: {model}")
149 else:
150 logger.debug(f"Using model from request: {model}")
152 custom_endpoint = data.get("custom_endpoint")
153 if not custom_endpoint and model_provider == "openai_endpoint":
154 custom_endpoint = settings_manager.get_setting(
155 "llm.openai_endpoint.url", None
156 )
157 logger.debug(
158 f"No custom_endpoint in request, using database setting: {custom_endpoint}"
159 )
161 ollama_url = data.get("ollama_url")
162 if not ollama_url and model_provider == "ollama":
163 ollama_url = settings_manager.get_setting(
164 "llm.ollama.url", DEFAULT_OLLAMA_URL
165 )
166 logger.debug(
167 f"No ollama_url in request, using database setting: {ollama_url}"
168 )
170 search_engine = data.get("search_engine") or data.get("search_tool")
171 if not search_engine:
172 search_engine = settings_manager.get_setting(
173 "search.tool", DEFAULT_SEARCH_TOOL
174 )
176 max_results = data.get("max_results")
177 time_period = data.get("time_period")
179 iterations = data.get("iterations")
180 if iterations is None:
181 iterations = settings_manager.get_setting("search.iterations", 5)
183 questions_per_iteration = data.get("questions_per_iteration")
184 if questions_per_iteration is None:
185 questions_per_iteration = settings_manager.get_setting(
186 "search.questions_per_iteration", 5
187 )
189 strategy = data.get("strategy")
190 if not strategy:
191 strategy = settings_manager.get_setting(
192 "search.search_strategy", "source-based"
193 )
195 # Egress policy per-research overrides. Mirror the
196 # model/search_engine pattern: missing values fall back to saved
197 # settings; supplied values override JUST FOR THIS RUN. They do
198 # NOT persist to the user's settings DB.
199 policy_egress_scope = data.get("policy_egress_scope")
200 llm_require_local_endpoint = data.get("llm_require_local_endpoint")
201 embeddings_require_local = data.get("embeddings_require_local")
203 return {
204 "model_provider": model_provider,
205 "model": model,
206 "custom_endpoint": custom_endpoint,
207 "ollama_url": ollama_url,
208 "search_engine": search_engine,
209 "max_results": max_results,
210 "time_period": time_period,
211 "iterations": iterations,
212 "questions_per_iteration": questions_per_iteration,
213 "strategy": strategy,
214 "policy_egress_scope": policy_egress_scope,
215 "llm_require_local_endpoint": llm_require_local_endpoint,
216 "embeddings_require_local": embeddings_require_local,
217 }
220def _precheck_engine_policy(settings_manager, params, search_engine, username):
221 """Validate the requested search engine against the saved egress
222 policy at the request boundary.
224 Returns a Flask ``(response, status)`` tuple when the request should
225 be rejected, or ``None`` to continue. Falls through (returns None)
226 when there's no real dict snapshot to validate or the policy module
227 errors — the factory PEP still enforces at engine-instantiation time.
228 """
229 try:
230 from ...security.egress.policy import (
231 PolicyDeniedError,
232 context_from_snapshot,
233 evaluate_engine,
234 resolve_run_primary_engine,
235 )
237 policy_snapshot = settings_manager.get_settings_snapshot()
238 # Only validate against a real dict snapshot. A test double or
239 # an unavailable settings backend hands back something else;
240 # skip rather than misfire.
241 if not isinstance(policy_snapshot, dict):
242 return None
244 # Overlay per-research form overrides so the snapshot reflects
245 # what the user picked for THIS run (per-research overrides, not
246 # a global settings save).
247 _apply_policy_overrides(policy_snapshot, params)
249 # Resolve the primary the SAME way the worker does (single source of
250 # truth) so the precheck and the background worker agree on accept vs.
251 # refuse — including the fail-closed missing-primary case, which the
252 # ValueError handler below maps to a 400. (Previously this substituted
253 # searxng and accepted runs the worker then refused.)
254 try:
255 primary = resolve_run_primary_engine(policy_snapshot)
256 policy_ctx = context_from_snapshot(
257 policy_snapshot,
258 primary,
259 username=username,
260 )
261 except PolicyDeniedError as exc:
262 return (
263 jsonify(
264 {
265 "status": "error",
266 "message": (
267 f"Egress policy refused this run: "
268 f"{exc.decision.reason}"
269 ),
270 }
271 ),
272 400,
273 )
274 except ValueError as exc:
275 # An invalid policy config is unrecoverable. Previously this
276 # raised, fell through to the outer ``except Exception``
277 # below, and silently returned None — so the run started
278 # successfully at the precheck and only failed at a
279 # downstream PEP. Surface it here as a 400 instead.
280 logger.bind(policy_audit=True).warning(
281 "POST /api/start_research policy precheck rejected",
282 reason=str(exc),
283 )
284 # Return a generic reason to the client — the raw ValueError text
285 # can carry policy-config internals and is already captured in the
286 # policy_audit warning above (CWE-209). Unlike the PolicyDeniedError
287 # branch, this path has no curated, user-safe decision reason.
288 return (
289 jsonify(
290 {
291 "status": "error",
292 "message": (
293 "Egress policy refused this run due to an "
294 "invalid policy configuration."
295 ),
296 }
297 ),
298 400,
299 )
301 decision = evaluate_engine(
302 search_engine,
303 policy_ctx,
304 settings_snapshot=policy_snapshot,
305 )
306 if not decision.allowed:
307 logger.bind(policy_audit=True).warning(
308 "POST /api/start_research search_engine refused",
309 engine=search_engine,
310 reason=decision.reason,
311 )
312 # Local import: this module imports the egress policy lazily inside
313 # this function to avoid a circular import at load time; keep the
314 # guidance import with it.
315 from ...security.egress.guidance import denial_guidance
317 return (
318 jsonify(
319 {
320 "status": "error",
321 # Clear, actionable message (what + why + how to allow),
322 # plus the raw reason code for support/logs.
323 "message": denial_guidance(
324 decision.reason,
325 target=f"Search engine '{search_engine}'",
326 ),
327 "reason": decision.reason,
328 }
329 ),
330 400,
331 )
333 # Stage C (ADR-0007): enforce the two-axis (sensitivity x exposure)
334 # admissibility as defense-in-depth on top of the scope check. This is
335 # a fast pre-start 400 for the API path; the same rule is also enforced
336 # at the shared run chokepoint in run_research_process, so follow-up /
337 # chat / queue runs are covered too. Under the UNPROTECTED escape hatch
338 # audit_run evaluates in permissive mode (always allowed), so an
339 # explicitly-unprotected run is never blocked. On an internal error
340 # audit_run fails CLOSED (reason "audit_error") so the run is refused
341 # visibly rather than silently degrading to the scope PEPs.
342 #
343 # NOTE: this sees the run's PRIMARY engine only. The per-engine scope
344 # PEP still runs for expanded engines, but it enforces SCOPE, not the
345 # two-axis rule — so a non-primary sensitive engine reaching a cloud
346 # sink under a permissive scope (legacy `both`) is not caught here.
347 # Widening to the full resolved engine set is a follow-up.
348 from ...security.egress.run_classification import (
349 audit_run_from_snapshot,
350 )
352 # Same shared entry point the worker chokepoint uses, so both resolve
353 # providers identically (LLM override wins over the snapshot; embeddings
354 # only for RAG primaries) and fail closed the same way.
355 two_axis = audit_run_from_snapshot(
356 policy_snapshot,
357 policy_ctx,
358 search_engine,
359 llm_provider=params.get("model_provider"),
360 )
361 if not two_axis.allowed:
362 logger.bind(policy_audit=True).warning(
363 "POST /api/start_research two-axis egress refused",
364 reason=two_axis.reason,
365 offending=two_axis.offending,
366 )
367 if two_axis.reason == "audit_error": 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true
368 message = (
369 "Egress policy could not verify this run, so it was "
370 "refused (fail-closed). Set Egress Scope to 'Unprotected' "
371 "to override."
372 )
373 else:
374 message = (
375 "Egress policy refused this run: a sensitive source would "
376 f"reach an exposing destination ({two_axis.reason}). Use "
377 "local inference, mark the collection public, or set Egress "
378 "Scope to 'Unprotected' to override."
379 )
380 return (
381 jsonify(
382 {
383 "status": "error",
384 "message": message,
385 "reason": two_axis.reason,
386 }
387 ),
388 400,
389 )
390 return None
391 except Exception:
392 # Policy module unavailable / internal error → log and fall
393 # through; the factory PEP will catch any actual violation.
394 logger.exception("egress policy pre-check skipped")
395 return None
398def _apply_policy_overrides(settings_snapshot, params):
399 """Overlay form-supplied egress policy values onto the snapshot.
401 Per-research overrides: the user picked these values for this
402 specific run and they do NOT persist to the settings DB. Mirrors
403 how model / search_engine overrides work today.
404 """
405 if not isinstance(settings_snapshot, dict):
406 return
407 if params.get("policy_egress_scope") is not None:
408 settings_snapshot["policy.egress_scope"] = params["policy_egress_scope"]
409 if params.get("llm_require_local_endpoint") is not None:
410 settings_snapshot["llm.require_local_endpoint"] = bool(
411 params["llm_require_local_endpoint"]
412 )
413 if params.get("embeddings_require_local") is not None:
414 settings_snapshot["embeddings.require_local"] = bool(
415 params["embeddings_require_local"]
416 )
417 # NOTE: the per-run `custom_endpoint` form value is deliberately NOT
418 # overlaid onto llm.openai_endpoint.url — that kwarg is non-functional
419 # (config/llm_config.py) and the run reads the URL from the DB setting, so
420 # the snapshot already carries the endpoint the run actually uses.
423def _research_not_found(research_id, message="Research not found"):
424 """Return a consistent 404 JSON for a missing research.
426 Emits ``status``, ``error`` and ``message`` so the body is a strict
427 superset of both historical 404 shapes — every frontend reader and
428 existing test keeps working without changes:
429 - Shape A readers read ``data.error``
430 - Shape B readers read ``data.status`` (``== "error"``) and/or
431 ``data.message``
433 ``research_id`` is used only for a debug log identifying which research
434 was missing; it is intentionally never echoed in the response body.
435 """
436 logger.debug(f"404 for research {research_id}: {message}")
437 return (
438 jsonify({"status": "error", "error": message, "message": message}),
439 404,
440 )
443def _queue_research(
444 db_session: Session,
445 username,
446 research_id,
447 query,
448 mode,
449 research_settings,
450 params,
451 session_id,
452 reason="",
453 research=None,
454):
455 """Add research to queue and notify processor. Returns a JSON response.
457 Args:
458 reason: Optional prefix explaining why the research was queued
459 (e.g. "due to concurrent limit").
460 research: Optional ResearchHistory object whose status should be set
461 to QUEUED atomically with the queue record insertion.
462 """
463 max_position = (
464 db_session.query(func.max(QueuedResearch.position))
465 .filter_by(username=username)
466 .scalar()
467 or 0
468 )
470 queued_record = QueuedResearch(
471 username=username,
472 research_id=research_id,
473 query=query,
474 mode=mode,
475 settings_snapshot=research_settings,
476 position=max_position + 1,
477 )
478 db_session.add(queued_record)
479 if research is not None:
480 research.status = ResearchStatus.QUEUED # type: ignore[assignment]
481 db_session.commit()
482 logger.info(
483 f"Queued research {research_id} at position {max_position + 1} for user {username}"
484 )
486 from ..queue.processor_v2 import queue_processor
488 queue_processor.notify_research_queued(
489 username,
490 research_id,
491 session_id=session_id,
492 query=query,
493 mode=mode,
494 settings_snapshot=research_settings,
495 model_provider=params["model_provider"],
496 model=params["model"],
497 custom_endpoint=params["custom_endpoint"],
498 search_engine=params["search_engine"],
499 max_results=params["max_results"],
500 time_period=params["time_period"],
501 iterations=params["iterations"],
502 questions_per_iteration=params["questions_per_iteration"],
503 strategy=params["strategy"],
504 )
506 position = max_position + 1
507 reason_text = f" {reason}" if reason else ""
508 message = f"Your research has been queued{reason_text}. Position in queue: {position}"
509 return jsonify(
510 {
511 "status": ResearchStatus.QUEUED,
512 "research_id": research_id,
513 "queue_position": position,
514 "message": message,
515 }
516 )
519@research_bp.route("/api/start_research", methods=["POST"])
520@login_required
521@api_rate_limit
522@require_json_body(error_format="status")
523def start_research():
524 data = request.json
525 # Debug logging to trace model parameter
526 logger.debug(f"Request data keys: {list(data.keys())}")
528 # Check if this is a news search
529 metadata = data.get("metadata", {})
530 if metadata.get("is_news_search"): 530 ↛ 531line 530 didn't jump to line 531 because the condition on line 530 was never true
531 logger.info(
532 f"News search request received: triggered_by={metadata.get('triggered_by', 'unknown')}"
533 )
535 query = data.get("query")
536 mode = data.get("mode", "quick")
538 # Replace date placeholders if they exist
539 if query and "YYYY-MM-DD" in query:
540 # Use local system time
541 current_date = datetime.now(UTC).strftime("%Y-%m-%d")
543 original_query = query
544 query = query.replace("YYYY-MM-DD", current_date)
545 logger.info(
546 f"Replaced date placeholder in query: {original_query[:100]}... -> {query[:100]}..."
547 )
548 logger.info(f"Using date: {current_date}")
550 # Update metadata to track the replacement
551 if not metadata: 551 ↛ 553line 551 didn't jump to line 553 because the condition on line 551 was always true
552 metadata = {}
553 metadata["original_query"] = original_query
554 metadata["processed_query"] = query
555 metadata["date_replaced"] = current_date
556 data["metadata"] = metadata
558 # Get parameters from request or use database settings
559 from ...settings import SettingsManager
561 username = session["username"]
563 with get_user_db_session(username) as db_session:
564 settings_manager = SettingsManager(db_session=db_session)
565 params = _extract_research_params(data, settings_manager)
567 model_provider = params["model_provider"]
568 model = params["model"]
569 custom_endpoint = params["custom_endpoint"]
570 search_engine = params["search_engine"]
571 max_results = params["max_results"]
572 time_period = params["time_period"]
573 iterations = params["iterations"]
574 questions_per_iteration = params["questions_per_iteration"]
575 strategy = params["strategy"]
577 # Egress policy: server-side check on the requested search
578 # engine BEFORE we enqueue the research. Defends against an API
579 # client posting an engine name that violates the saved policy
580 # (e.g. ``{"search_engine": "pubmed"}`` under STRICT+primary=arxiv).
581 # The factory PEP catches the same case at instantiation time,
582 # but rejecting at the request boundary lets us return a clean
583 # 4xx instead of an opaque background failure.
584 policy_error = _precheck_engine_policy(
585 settings_manager, params, search_engine, username
586 )
587 if policy_error is not None: 587 ↛ 588line 587 didn't jump to line 588 because the condition on line 587 was never true
588 return policy_error
590 # Debug logging for model parameter specifically
591 logger.debug(
592 f"Extracted model value: '{model}' (type: {type(model).__name__})"
593 )
595 # Log the selections for troubleshooting
596 logger.info(
597 f"Starting research with provider: {model_provider}, model: {model}, search engine: {search_engine}"
598 )
599 logger.info(
600 f"Additional parameters: max_results={max_results}, time_period={time_period}, iterations={iterations}, questions={questions_per_iteration}, strategy={strategy}"
601 )
603 if not query:
604 return jsonify({"status": "error", "message": "Query is required"}), 400
606 # Validate required parameters based on provider
607 if model_provider == "openai_endpoint" and not custom_endpoint:
608 return (
609 jsonify(
610 {
611 "status": "error",
612 "message": "Custom endpoint URL is required for OpenAI endpoint provider",
613 }
614 ),
615 400,
616 )
618 # SSRF pre-flight on the user-supplied LLM endpoint: reject metadata /
619 # link-local targets at the request boundary, before any research thread
620 # is spawned. This is fail-fast defense-in-depth — the OpenAI-compatible
621 # provider's assert_base_url_safe re-validates the same URL before the
622 # client is built. Private IPs / localhost are permitted so local LLMs
623 # (vLLM, Ollama, LM Studio) work, including scheme-less endpoints
624 # (the helper normalizes exactly as the provider does).
625 if not is_safe_custom_llm_endpoint(custom_endpoint):
626 return (
627 jsonify(
628 {"status": "error", "message": "Invalid custom endpoint URL"}
629 ),
630 400,
631 )
633 if not model:
634 logger.error(
635 f"No model specified or configured. Provider: {model_provider}"
636 )
637 return jsonify(
638 {
639 "status": "error",
640 "message": "Model is required. Please configure a model in the settings.",
641 }
642 ), 400
644 # Check if the user has too many active researches
645 username = session["username"]
647 # Get max concurrent researches from settings
648 from ...settings import SettingsManager
650 with get_user_db_session() as db_session:
651 settings_manager = SettingsManager(db_session)
652 max_concurrent_researches = settings_manager.get_setting(
653 "app.max_concurrent_researches", 3
654 )
656 # Use existing session from g to check active researches
657 try:
658 db_session = get_g_db_session()
659 if db_session:
660 # First, clean up stale entries where the research thread has died
661 # (e.g. crashed with an unhandled exception before cleanup ran).
662 # Without this, dead researches permanently block the queue.
663 from ..routes.globals import reclaim_stale_user_active_research
665 # No grace cutoff here — research_routes.start_research has
666 # always reclaimed all dead-thread rows immediately. The chat
667 # send-message path uses a 30s grace via grace_cutoff_dt.
668 if reclaim_stale_user_active_research( 668 ↛ 671line 668 didn't jump to line 671 because the condition on line 668 was never true
669 db_session, username, logger=logger
670 ):
671 db_session.commit()
673 # Now count truly active researches
674 active_count = (
675 db_session.query(UserActiveResearch)
676 .filter_by(username=username, status=ResearchStatus.IN_PROGRESS)
677 .count()
678 )
680 # Debug logging
681 logger.info(
682 f"Active research count for {username}: {active_count}/{max_concurrent_researches}"
683 )
685 should_queue = active_count >= max_concurrent_researches
686 logger.info(f"Should queue new research: {should_queue}")
687 else:
688 logger.warning(
689 "No database session available to check active researches"
690 )
691 should_queue = False
692 except Exception:
693 logger.exception("Failed to check active researches")
694 # Default to not queueing if we can't check
695 should_queue = False
697 # For non-queued research, verify password is available BEFORE creating DB records
698 # (queued research gets password later via queue processor)
699 user_password = None
700 if not should_queue:
701 user_password, session_expired = resolve_user_password(username)
702 if session_expired:
703 # Use status/message keys to match the research API convention
704 # (the research frontend checks data.status and data.message)
705 return jsonify(
706 {
707 "status": "error",
708 "message": "Your session has expired. Please log out and log back in to start research.",
709 }
710 ), 401
712 # Create a record in the database with explicit UTC timestamp
713 import uuid
714 import threading
716 created_at = datetime.now(UTC).isoformat()
717 research_id = str(uuid.uuid4())
719 # Create organized research metadata with settings snapshot
720 research_settings = {
721 # Direct submission parameters
722 "submission": {
723 "model_provider": model_provider,
724 "model": model,
725 "custom_endpoint": custom_endpoint,
726 "search_engine": search_engine,
727 "max_results": max_results,
728 "time_period": time_period,
729 "iterations": iterations,
730 "questions_per_iteration": questions_per_iteration,
731 "strategy": strategy,
732 },
733 # System information
734 "system": {
735 "timestamp": created_at,
736 "user": username,
737 "version": "1.0", # Track metadata version for future migrations
738 "server_url": request.host_url, # Add server URL for link generation
739 },
740 }
742 # Add any additional metadata from request
743 additional_metadata = data.get("metadata", {})
744 if additional_metadata:
745 research_settings.update(additional_metadata)
746 # Get complete settings snapshot for this research
747 try:
748 from ...settings import SettingsManager
750 # Get or lazily create a session for settings snapshot
751 db_session_for_settings = get_g_db_session()
752 if db_session_for_settings:
753 # Create SettingsManager with the existing session
754 username = session["username"]
755 # Ensure any pending changes are committed
756 try:
757 db_session_for_settings.commit()
758 except Exception:
759 db_session_for_settings.rollback()
760 settings_manager = SettingsManager(
761 db_session_for_settings, owns_session=False
762 )
763 # Get all current settings as a snapshot (bypass cache to ensure fresh data)
764 all_settings = settings_manager.get_all_settings(bypass_cache=True)
765 # Apply per-research egress policy overrides (form-supplied
766 # values override saved settings JUST FOR THIS RUN; they do
767 # not persist to the user's settings DB).
768 _apply_policy_overrides(all_settings, params)
770 # Add settings snapshot to metadata
771 research_settings["settings_snapshot"] = all_settings
772 logger.info(
773 f"Captured {len(all_settings)} settings for research {research_id}"
774 )
775 else:
776 # If no session in g, create a new one temporarily to get settings
777 logger.warning(
778 "No database session in g, creating temporary session for settings snapshot"
779 )
780 from ...database.thread_local_session import get_metrics_session
782 password = get_user_password(username)
784 if password:
785 temp_session = get_metrics_session(username, password)
786 if temp_session:
787 username = session["username"]
788 settings_manager = SettingsManager(
789 temp_session, owns_session=False
790 )
791 all_settings = settings_manager.get_all_settings(
792 bypass_cache=True
793 )
794 _apply_policy_overrides(all_settings, params)
795 research_settings["settings_snapshot"] = all_settings
796 logger.info(
797 f"Captured {len(all_settings)} settings using temporary session for research {research_id}"
798 )
799 else:
800 logger.error(
801 "Failed to create temporary session for settings snapshot"
802 )
803 return jsonify(
804 {
805 "status": "error",
806 "message": "Cannot create research without settings snapshot.",
807 }
808 ), 500
809 else:
810 logger.error(
811 "No password available to create session for settings snapshot"
812 )
813 return jsonify(
814 {
815 "status": "error",
816 "message": "Cannot create research without settings snapshot.",
817 }
818 ), 500
819 except Exception:
820 logger.exception("Failed to capture settings snapshot")
821 # Cannot continue without settings snapshot for thread-based research
822 return jsonify(
823 {
824 "status": "error",
825 "message": "Failed to capture settings for research. Please try again.",
826 }
827 ), 500
829 # Use existing session from g
830 username = session["username"]
832 try:
833 # Get or lazily create a session
834 db_session = get_g_db_session()
835 if db_session:
836 # Determine initial status based on whether we need to queue
837 initial_status = (
838 ResearchStatus.QUEUED
839 if should_queue
840 else ResearchStatus.IN_PROGRESS
841 )
843 research = ResearchHistory(
844 id=research_id, # Set UUID as primary key
845 query=query,
846 mode=mode,
847 status=initial_status,
848 created_at=created_at,
849 progress_log=[{"time": created_at, "progress": 0}],
850 research_meta=research_settings,
851 )
852 db_session.add(research)
853 db_session.commit()
854 logger.info(
855 f"Created research entry with UUID: {research_id}, status: {initial_status}"
856 )
858 if should_queue:
859 session_id = session.get("session_id")
860 return _queue_research(
861 db_session,
862 username,
863 research_id,
864 query,
865 mode,
866 research_settings,
867 params,
868 session_id,
869 )
870 # Start immediately
871 # Create active research tracking record
872 import threading
874 active_record = UserActiveResearch(
875 username=username,
876 research_id=research_id,
877 status=ResearchStatus.IN_PROGRESS,
878 thread_id=str(threading.current_thread().ident),
879 settings_snapshot=research_settings,
880 )
881 db_session.add(active_record)
882 db_session.commit()
883 logger.info(f"Created active research record for user {username}")
885 # Double-check the count after committing to handle race conditions
886 # Use the existing session for the recheck
887 try:
888 # Use the same session we already have
889 recheck_session = db_session
890 final_count = (
891 recheck_session.query(UserActiveResearch)
892 .filter_by(
893 username=username, status=ResearchStatus.IN_PROGRESS
894 )
895 .count()
896 )
897 logger.info(
898 f"Final active count after commit: {final_count}/{max_concurrent_researches}"
899 )
901 if final_count > max_concurrent_researches:
902 # We exceeded the limit due to a race condition
903 # Remove this record and queue instead
904 logger.warning(
905 f"Race condition detected: {final_count} > {max_concurrent_researches}, moving to queue"
906 )
907 db_session.delete(active_record)
908 db_session.commit()
910 session_id = session.get("session_id")
911 return _queue_research(
912 db_session,
913 username,
914 research_id,
915 query,
916 mode,
917 research_settings,
918 params,
919 session_id,
920 reason="due to concurrent limit",
921 research=research,
922 )
923 except Exception:
924 logger.warning("Could not recheck active count")
926 except Exception:
927 logger.exception("Failed to create research entry")
928 return jsonify(
929 {"status": "error", "message": "Failed to create research entry"}
930 ), 500
932 # Only start the research if not queued
933 if not should_queue: 933 ↛ 1074line 933 didn't jump to line 1074 because the condition on line 933 was always true
934 # Save the research strategy to the database before starting the thread
935 try:
936 from ..services.research_service import save_research_strategy
938 save_research_strategy(research_id, strategy, username=username)
939 except Exception:
940 logger.warning("Could not save research strategy")
942 # Debug logging for settings snapshot
943 snapshot_data = research_settings.get("settings_snapshot", {})
944 log_settings(snapshot_data, "Settings snapshot being passed to thread")
945 if "search.tool" in snapshot_data:
946 logger.debug(
947 f"search.tool in snapshot: {snapshot_data['search.tool']}"
948 )
949 else:
950 logger.debug("search.tool NOT in snapshot")
952 # Start the research process with the selected parameters.
953 # If the spawn raises, the UserActiveResearch + IN_PROGRESS
954 # ResearchHistory rows persisted above would otherwise be
955 # permanently orphaned (no thread, no cleanup path). Catch any
956 # exception, mark the research FAILED, delete the active row,
957 # and return 500 — same contract as the queue processor's
958 # terminal-failure branch introduced in #3481.
959 try:
960 research_thread = start_research_process(
961 research_id,
962 query,
963 mode,
964 run_research_process,
965 username=username, # Pass username to the thread
966 user_password=user_password, # Pass password for database access
967 model_provider=model_provider,
968 model=model,
969 custom_endpoint=custom_endpoint,
970 search_engine=search_engine,
971 max_results=max_results,
972 time_period=time_period,
973 iterations=iterations,
974 questions_per_iteration=questions_per_iteration,
975 strategy=strategy,
976 settings_snapshot=snapshot_data, # Pass complete settings
977 )
978 except DuplicateResearchError:
979 # A live thread already owns this research_id. Do NOT delete
980 # the UserActiveResearch row or mark ResearchHistory FAILED —
981 # that state belongs to the live thread, and mutating it
982 # would terminate a running research from the user's
983 # perspective while it keeps executing. Same contract as the
984 # queue processor's dedicated dup branch (#3506).
985 logger.warning(
986 f"Duplicate live thread detected for {research_id} "
987 "on direct submission; leaving state intact"
988 )
989 return jsonify(
990 {
991 "status": "error",
992 "message": "Research is already running.",
993 }
994 ), 409
995 except SystemAtCapacityError:
996 # System at concurrent-research capacity. Roll back the rows
997 # committed above (UserActiveResearch + IN_PROGRESS history)
998 # and return 429 so the client can retry shortly.
999 logger.warning(
1000 f"SystemAtCapacityError on direct submission for "
1001 f"{research_id}; rolling back orphan rows"
1002 )
1003 try:
1004 with get_user_db_session(username) as cleanup_session:
1005 stale_active = (
1006 cleanup_session.query(UserActiveResearch)
1007 .filter_by(username=username, research_id=research_id)
1008 .first()
1009 )
1010 if stale_active:
1011 cleanup_session.delete(stale_active)
1012 cleanup_session.query(ResearchHistory).filter_by(
1013 id=research_id
1014 ).delete()
1015 cleanup_session.commit()
1016 except Exception:
1017 logger.exception(
1018 "Cleanup after SystemAtCapacityError raised; "
1019 "leaving orphan rows for the reconciler"
1020 )
1021 return jsonify(
1022 {
1023 "status": "error",
1024 "message": "Server is at research capacity. Please retry shortly.",
1025 }
1026 ), 429
1027 except Exception:
1028 logger.exception(
1029 f"Failed to spawn research thread for {research_id}"
1030 )
1031 try:
1032 with get_user_db_session(username) as cleanup_session:
1033 stale_active = (
1034 cleanup_session.query(UserActiveResearch)
1035 .filter_by(username=username, research_id=research_id)
1036 .first()
1037 )
1038 if stale_active: 1038 ↛ 1040line 1038 didn't jump to line 1040 because the condition on line 1038 was always true
1039 cleanup_session.delete(stale_active)
1040 research_row = (
1041 cleanup_session.query(ResearchHistory)
1042 .filter_by(id=research_id)
1043 .first()
1044 )
1045 if research_row: 1045 ↛ 1047line 1045 didn't jump to line 1047 because the condition on line 1045 was always true
1046 research_row.status = ResearchStatus.FAILED
1047 cleanup_session.commit()
1048 except Exception:
1049 logger.exception(
1050 "Cleanup after spawn failure raised; leaving "
1051 "orphan rows for the reconciler to handle"
1052 )
1053 return jsonify(
1054 {
1055 "status": "error",
1056 "message": "Failed to start research. Please try again.",
1057 }
1058 ), 500
1060 # Update the active research record with the actual thread ID.
1061 try:
1062 with get_user_db_session(username) as thread_session:
1063 active_record = (
1064 thread_session.query(UserActiveResearch)
1065 .filter_by(username=username, research_id=research_id)
1066 .first()
1067 )
1068 if active_record: 1068 ↛ 1074line 1068 didn't jump to line 1074
1069 active_record.thread_id = str(research_thread.ident)
1070 thread_session.commit()
1071 except Exception:
1072 logger.warning("Could not update thread ID")
1074 return jsonify({"status": "success", "research_id": research_id})
1077@research_bp.route("/api/terminate/<string:research_id>", methods=["POST"])
1078@login_required
1079def terminate_research(research_id):
1080 """Terminate an in-progress research process"""
1081 username = session["username"]
1083 # Check if the research exists and is in progress
1084 try:
1085 with get_user_db_session(username) as db_session:
1086 research = (
1087 db_session.query(ResearchHistory)
1088 .filter_by(id=research_id)
1089 .first()
1090 )
1092 if not research:
1093 return _research_not_found(research_id)
1095 status = research.status
1097 # If it's already in a terminal state, return success
1098 if status in (
1099 ResearchStatus.COMPLETED,
1100 ResearchStatus.SUSPENDED,
1101 ResearchStatus.FAILED,
1102 ResearchStatus.ERROR,
1103 ):
1104 return jsonify(
1105 {
1106 "status": "success",
1107 "message": f"Research already {status}",
1108 }
1109 )
1111 # Check if it's in the active_research dict
1112 if not is_research_active(research_id):
1113 # The worker may not be registered in _active_research yet: a
1114 # just-submitted research commits its IN_PROGRESS row before the
1115 # worker thread registers itself (spawn-grace window). Set the
1116 # termination flag anyway so a worker that starts right after
1117 # this still sees it and aborts at its first checkpoint —
1118 # otherwise the user's Stop is silently ignored and the research
1119 # runs to completion (overwriting this SUSPENDED status). The
1120 # flag is harmless if no worker ever starts.
1121 set_termination_flag(research_id)
1122 if status == ResearchStatus.QUEUED:
1123 claimed_queue_row = (
1124 db_session.query(QueuedResearch.id)
1125 .filter(
1126 QueuedResearch.research_id == research_id,
1127 QueuedResearch.is_processing.is_(True),
1128 )
1129 .exists()
1130 )
1131 suspended = (
1132 db_session.query(ResearchHistory)
1133 .filter(
1134 ResearchHistory.id == research_id,
1135 ResearchHistory.status == ResearchStatus.QUEUED,
1136 ~claimed_queue_row,
1137 )
1138 .update(
1139 {ResearchHistory.status: ResearchStatus.SUSPENDED},
1140 synchronize_session=False,
1141 )
1142 )
1143 if suspended: 1143 ↛ 1151line 1143 didn't jump to line 1151 because the condition on line 1143 was always true
1144 from ..queue.lifecycle_cleanup import (
1145 cleanup_queued_research_state,
1146 )
1148 cleanup_queued_research_state(db_session, [research_id])
1149 else:
1150 research.status = ResearchStatus.SUSPENDED
1151 db_session.commit()
1152 return jsonify(
1153 {"status": "success", "message": "Research terminated"}
1154 )
1156 # Set the termination flag
1157 set_termination_flag(research_id)
1159 # Log the termination request - using UTC timestamp
1160 timestamp = datetime.now(UTC).isoformat()
1161 termination_message = "Research termination requested by user"
1162 current_progress = get_research_field(research_id, "progress", 0)
1164 # Create log entry
1165 log_entry = {
1166 "time": timestamp,
1167 "message": termination_message,
1168 "progress": current_progress,
1169 "metadata": {"phase": "termination"},
1170 }
1172 # Add to in-memory log
1173 append_research_log(research_id, log_entry)
1175 # Add to database log
1176 logger.log("MILESTONE", f"Research ended: {termination_message}")
1178 # Update the log in the database
1179 if research.progress_log: 1179 ↛ 1188line 1179 didn't jump to line 1188 because the condition on line 1179 was always true
1180 try:
1181 if isinstance(research.progress_log, str): 1181 ↛ 1184line 1181 didn't jump to line 1184 because the condition on line 1181 was always true
1182 current_log = json.loads(research.progress_log)
1183 else:
1184 current_log = research.progress_log
1185 except Exception:
1186 current_log = []
1187 else:
1188 current_log = []
1190 current_log.append(log_entry)
1191 research.progress_log = current_log
1192 research.status = ResearchStatus.SUSPENDED
1193 db_session.commit()
1195 # Emit a socket event for the termination request
1196 try:
1197 event_data = {
1198 "status": ResearchStatus.SUSPENDED,
1199 "message": "Research was suspended by user request",
1200 }
1202 from ..services.socket_service import SocketIOService
1204 SocketIOService().emit_to_subscribers(
1205 "progress", research_id, event_data
1206 )
1208 except Exception:
1209 logger.exception("Socket emit error (non-critical)")
1211 return jsonify(
1212 {
1213 "status": "success",
1214 "message": "Research termination requested",
1215 }
1216 )
1217 except Exception:
1218 logger.exception("Error terminating research")
1219 return jsonify(
1220 {"status": "error", "message": "Failed to terminate research"}
1221 ), 500
1224@research_bp.route("/api/delete/<string:research_id>", methods=["DELETE"])
1225@login_required
1226def delete_research(research_id):
1227 """Delete a research record"""
1228 username = session["username"]
1230 try:
1231 with get_user_db_session(username) as db_session:
1232 research = (
1233 db_session.query(ResearchHistory)
1234 .filter_by(id=research_id)
1235 .first()
1236 )
1238 if not research:
1239 return _research_not_found(research_id)
1241 status = research.status
1242 report_path = research.report_path
1244 # Don't allow deleting research in progress
1245 if status == ResearchStatus.IN_PROGRESS:
1246 return (
1247 jsonify(
1248 {
1249 "status": "error",
1250 "message": "Cannot delete research that is in progress",
1251 }
1252 ),
1253 400,
1254 )
1256 # Delete the database record
1257 claimed_queue_row = (
1258 db_session.query(QueuedResearch.id)
1259 .filter(
1260 QueuedResearch.research_id == research_id,
1261 QueuedResearch.is_processing.is_(True),
1262 )
1263 .exists()
1264 )
1265 deleted = (
1266 db_session.query(ResearchHistory)
1267 .filter(
1268 ResearchHistory.id == research_id,
1269 ResearchHistory.status != ResearchStatus.IN_PROGRESS,
1270 ~claimed_queue_row,
1271 )
1272 .delete(synchronize_session=False)
1273 )
1274 if not deleted:
1275 return (
1276 jsonify(
1277 {
1278 "status": "error",
1279 "message": "Cannot delete research that is in progress",
1280 }
1281 ),
1282 400,
1283 )
1285 # Delete report file if it exists
1286 if report_path and Path(report_path).exists(): 1286 ↛ 1287line 1286 didn't jump to line 1287 because the condition on line 1286 was never true
1287 try:
1288 Path(report_path).unlink()
1289 except Exception:
1290 logger.exception("Error removing report file")
1292 from ..queue.lifecycle_cleanup import cleanup_queued_research_state
1294 cleanup_queued_research_state(db_session, [research_id])
1295 db_session.commit()
1297 return jsonify({"status": "success"})
1298 except Exception:
1299 logger.exception("Error deleting research")
1300 return jsonify(
1301 {"status": "error", "message": "Failed to delete research"}
1302 ), 500
1305@research_bp.route("/api/clear_history", methods=["POST"])
1306@login_required
1307def clear_history():
1308 """Clear all research history"""
1309 username = session["username"]
1311 try:
1312 with get_user_db_session(username) as db_session:
1313 # Get all research records first to clean up files. Select
1314 # only id + report_path (the columns the cleanup loop uses)
1315 # so clearing history doesn't load every report body into
1316 # memory (#4560).
1317 research_records = db_session.query(
1318 ResearchHistory.id,
1319 ResearchHistory.report_path,
1320 ResearchHistory.status,
1321 ).all()
1323 # Get IDs of currently active research (snapshot)
1324 protected_ids = set(get_active_research_ids())
1325 protected_ids.update(
1326 research.id
1327 for research in research_records
1328 if research.status == ResearchStatus.IN_PROGRESS
1329 )
1330 deletable_ids = [
1331 research.id
1332 for research in research_records
1333 if research.id not in protected_ids
1334 ]
1336 deleted_ids = []
1337 for research_id in deletable_ids:
1338 claimed_queue_row = (
1339 db_session.query(QueuedResearch.id)
1340 .filter(
1341 QueuedResearch.research_id == research_id,
1342 QueuedResearch.is_processing.is_(True),
1343 )
1344 .exists()
1345 )
1346 deleted = (
1347 db_session.query(ResearchHistory)
1348 .filter(
1349 ResearchHistory.id == research_id,
1350 ResearchHistory.status != ResearchStatus.IN_PROGRESS,
1351 ~claimed_queue_row,
1352 )
1353 .delete(synchronize_session=False)
1354 )
1355 if deleted:
1356 deleted_ids.append(research_id)
1358 if deleted_ids:
1359 for research in research_records:
1360 if research.id not in deleted_ids:
1361 continue
1363 if ( 1363 ↛ 1367line 1363 didn't jump to line 1367 because the condition on line 1363 was never true
1364 research.report_path
1365 and Path(research.report_path).exists()
1366 ):
1367 try:
1368 Path(research.report_path).unlink()
1369 except Exception:
1370 logger.exception("Error removing report file")
1372 from ..queue.lifecycle_cleanup import (
1373 cleanup_queued_research_state,
1374 )
1376 cleanup_queued_research_state(db_session, deleted_ids)
1378 db_session.commit()
1380 return jsonify({"status": "success"})
1381 except Exception:
1382 logger.exception("Error clearing history")
1383 return jsonify(
1384 {"status": "error", "message": "Failed to process request"}
1385 ), 500
1388@research_bp.route("/open_file_location", methods=["POST"])
1389@login_required
1390def open_file_location():
1391 """Open a file location in the system file explorer.
1393 Security: This endpoint is disabled for server deployments.
1394 It only makes sense for desktop usage where the server and client are on the same machine.
1395 """
1396 return jsonify(
1397 {
1398 "status": "error",
1399 "message": "This feature is disabled. It is only available in desktop mode.",
1400 }
1401 ), 403
1404@research_bp.route("/api/save_raw_config", methods=["POST"])
1405@login_required
1406@require_json_body(error_format="success")
1407def save_raw_config():
1408 """Save raw configuration"""
1409 data = request.json
1410 raw_config = data.get("raw_config")
1412 if not raw_config:
1413 return (
1414 jsonify(
1415 {"success": False, "error": "Raw configuration is required"}
1416 ),
1417 400,
1418 )
1420 # Security: Parse and validate the TOML to block dangerous keys
1421 try:
1422 import tomllib
1423 except ImportError:
1424 import tomli as tomllib # type: ignore[no-redef]
1426 try:
1427 parsed_config = tomllib.loads(raw_config)
1428 except Exception:
1429 logger.warning("Invalid TOML configuration")
1430 # Don't expose internal exception details to users (CWE-209)
1431 return jsonify(
1432 {
1433 "success": False,
1434 "error": "Invalid TOML syntax. Please check your configuration format.",
1435 }
1436 ), 400
1438 # Security: Check for dangerous keys that could enable code execution
1439 # These patterns match keys used for dynamic module imports
1440 BLOCKED_KEY_PATTERNS = ["module_path", "class_name", "module", "class"]
1442 def find_blocked_keys(obj, path=""):
1443 """Recursively find any blocked keys in the config."""
1444 blocked = []
1445 if isinstance(obj, dict):
1446 for key, value in obj.items():
1447 current_path = f"{path}.{key}" if path else key
1448 key_lower = key.lower()
1449 for pattern in BLOCKED_KEY_PATTERNS:
1450 if pattern in key_lower:
1451 blocked.append(current_path)
1452 break
1453 # Recurse into nested dicts
1454 blocked.extend(find_blocked_keys(value, current_path))
1455 elif isinstance(obj, list): 1455 ↛ 1456line 1455 didn't jump to line 1456 because the condition on line 1455 was never true
1456 for i, item in enumerate(obj):
1457 blocked.extend(find_blocked_keys(item, f"{path}[{i}]"))
1458 return blocked
1460 blocked_keys = find_blocked_keys(parsed_config)
1461 if blocked_keys:
1462 logger.warning(
1463 f"Security: Blocked attempt to write config with dangerous keys: {blocked_keys}"
1464 )
1465 return jsonify(
1466 {
1467 "success": False,
1468 "error": "Configuration contains protected keys that cannot be modified",
1469 "blocked_keys": blocked_keys,
1470 }
1471 ), 403
1473 try:
1474 from ...security.file_write_verifier import write_file_verified
1476 # Get the config file path (uses centralized path config, respects LDR_DATA_DIR)
1477 config_dir = get_config_directory()
1478 config_path = config_dir / "config.toml"
1480 # Write the configuration to file
1481 write_file_verified(
1482 config_path,
1483 raw_config,
1484 "system.allow_config_write",
1485 context="system configuration file",
1486 )
1488 return jsonify({"success": True})
1489 except Exception:
1490 logger.exception("Error saving configuration file")
1491 return jsonify(
1492 {"success": False, "error": "Failed to process request"}
1493 ), 500
1496@research_bp.route("/api/history", methods=["GET"])
1497@login_required
1498def get_history():
1499 """Get research history"""
1500 username = session["username"]
1502 # Bound the result set. Without a limit this endpoint loaded every
1503 # research row (and its research_meta JSON, which can hold a settings
1504 # snapshot) into memory at once (#4560). Mirrors the clamp used by the
1505 # symmetric /history/api endpoint.
1506 limit = request.args.get("limit", 200, type=int)
1507 limit = max(1, min(limit, 500))
1508 offset = max(0, request.args.get("offset", 0, type=int))
1510 try:
1511 with get_user_db_session(username) as db_session:
1512 # Query research history ordered by created_at. Project
1513 # only the metadata columns the loop below consumes — never
1514 # the large ``report_content`` Text body — so this listing
1515 # doesn't pull every report into memory (#4560). This mirrors
1516 # the projection used by the symmetric /history/api endpoint.
1517 research_records = (
1518 db_session.query(
1519 ResearchHistory.id,
1520 ResearchHistory.title,
1521 ResearchHistory.query,
1522 ResearchHistory.mode,
1523 ResearchHistory.status,
1524 ResearchHistory.created_at,
1525 ResearchHistory.completed_at,
1526 ResearchHistory.research_meta,
1527 ResearchHistory.chat_session_id,
1528 )
1529 .order_by(ResearchHistory.created_at.desc())
1530 .limit(limit)
1531 .offset(offset)
1532 .all()
1533 )
1535 # Pre-compute Document counts in a single GROUP BY query
1536 # to avoid an N+1 SELECT-COUNT-per-row inside the loop. The
1537 # symmetric /history/api endpoint in history_routes.py
1538 # already uses an outerjoin + group_by — this brings
1539 # /api/history to parity for users with deep history.
1540 research_ids = [r.id for r in research_records]
1541 if research_ids:
1542 doc_count_rows = (
1543 db_session.query(
1544 Document.research_id, func.count(Document.id)
1545 )
1546 .filter(Document.research_id.in_(research_ids))
1547 .group_by(Document.research_id)
1548 .all()
1549 )
1550 doc_counts = dict(doc_count_rows)
1551 else:
1552 doc_counts = {}
1554 # Build history items while session is active to avoid
1555 # DetachedInstanceError on ORM attribute access
1556 history_items = []
1557 for research in research_records:
1558 # Calculate duration if completed
1559 duration_seconds = None
1560 if research.completed_at and research.created_at:
1561 try:
1562 duration_seconds = calculate_duration(
1563 research.created_at, research.completed_at
1564 )
1565 except Exception:
1566 logger.exception("Error calculating duration")
1568 # Look up the pre-computed document count.
1569 doc_count = doc_counts.get(research.id, 0)
1571 # Create a history item
1572 item = {
1573 "id": research.id,
1574 "query": research.query,
1575 "mode": research.mode,
1576 "status": research.status,
1577 "created_at": research.created_at,
1578 "completed_at": research.completed_at,
1579 "duration_seconds": duration_seconds,
1580 "metadata": filter_research_metadata(
1581 research.research_meta
1582 ),
1583 "document_count": doc_count,
1584 }
1585 if research.chat_session_id is not None: 1585 ↛ 1586line 1585 didn't jump to line 1586 because the condition on line 1585 was never true
1586 item["metadata"]["chat_session_id"] = (
1587 research.chat_session_id
1588 )
1590 # Add title if it exists
1591 if hasattr(research, "title") and research.title is not None:
1592 item["title"] = research.title
1594 history_items.append(item)
1596 return jsonify({"status": "success", "items": history_items})
1597 except Exception:
1598 logger.exception("Error getting history")
1599 return jsonify(
1600 {"status": "error", "message": "Failed to process request"}
1601 ), 500
1604@research_bp.route("/api/research/<string:research_id>")
1605@login_required
1606def get_research_details(research_id):
1607 """Get full details of a research using ORM"""
1608 username = session["username"]
1610 try:
1611 with get_user_db_session(username) as db_session:
1612 research = (
1613 db_session.query(ResearchHistory)
1614 .filter(ResearchHistory.id == research_id)
1615 .first()
1616 )
1618 if not research:
1619 return _research_not_found(research_id)
1621 return jsonify(
1622 {
1623 "id": research.id,
1624 "query": research.query,
1625 "status": research.status,
1626 "progress": research.progress,
1627 "progress_percentage": research.progress or 0,
1628 "mode": research.mode,
1629 "created_at": research.created_at,
1630 "completed_at": research.completed_at,
1631 "report_path": research.report_path,
1632 "metadata": strip_settings_snapshot(research.research_meta),
1633 }
1634 )
1635 except Exception:
1636 logger.exception("Error getting research details")
1637 return jsonify({"error": "An internal error has occurred"}), 500
1640@research_bp.route("/api/research/<string:research_id>/logs")
1641@login_required
1642def get_research_logs(research_id):
1643 """Get logs for a specific research.
1645 Accepts an optional ``?limit=N`` that bounds the response to the newest
1646 ``N`` rows (returned oldest-first, matching the default ordering) and is
1647 clamped to ``[1, HISTORY_LOGS_HARD_CAP]`` so a client cannot force an
1648 unbounded load — a long langgraph run can persist thousands of rows. When
1649 ``?limit`` is absent or not a valid integer (Flask ``type=int`` yields
1650 ``None``) the historical contract is preserved: every row is returned. The
1651 frontend log panel always sends a valid limit; this only affects direct
1652 API callers that omit or malform one.
1653 """
1654 username = session["username"]
1656 limit = request.args.get("limit", type=int)
1657 if limit is not None:
1658 limit = max(1, min(limit, HISTORY_LOGS_HARD_CAP))
1660 try:
1661 # First check if the research exists
1662 with get_user_db_session(username) as db_session:
1663 research = (
1664 db_session.query(ResearchHistory)
1665 .filter_by(id=research_id)
1666 .first()
1667 )
1668 if not research:
1669 return _research_not_found(research_id)
1671 # Get logs from research_logs table
1672 log_query = db_session.query(ResearchLog).filter_by(
1673 research_id=research_id
1674 )
1675 if limit is None:
1676 log_results = log_query.order_by(
1677 ResearchLog.timestamp, ResearchLog.id
1678 ).all()
1679 else:
1680 # Take the newest ``limit`` rows at the SQL layer, then flip
1681 # back to oldest-first so the response ordering is unchanged.
1682 # ``id`` is the tie-break: timestamps are not unique, so without
1683 # it the rows that survive ``.limit()`` at a shared-timestamp
1684 # boundary would be SQL-undefined.
1685 log_results = list(
1686 reversed(
1687 log_query.order_by(
1688 ResearchLog.timestamp.desc(),
1689 ResearchLog.id.desc(),
1690 )
1691 .limit(limit)
1692 .all()
1693 )
1694 )
1696 # Extract log attributes while session is active
1697 # to avoid DetachedInstanceError on ORM attribute access
1698 logs = []
1699 for row in log_results:
1700 logs.append(
1701 {
1702 "id": row.id,
1703 "message": row.message,
1704 "timestamp": row.timestamp,
1705 "log_type": row.level,
1706 }
1707 )
1709 return jsonify(logs)
1711 except Exception:
1712 logger.exception("Error getting research logs")
1713 return jsonify({"error": "An internal error has occurred"}), 500
1716@research_bp.route("/api/report/<string:research_id>")
1717@login_required
1718def get_research_report(research_id):
1719 """Get the research report content"""
1720 username = session["username"]
1722 try:
1723 with get_user_db_session(username) as db_session:
1724 # Query using ORM
1725 research = (
1726 db_session.query(ResearchHistory)
1727 .filter_by(id=research_id)
1728 .first()
1729 )
1731 if research is None:
1732 return _research_not_found(research_id)
1734 # Parse metadata if it exists
1735 metadata = research.research_meta
1737 # research.report_content holds the answer-only string;
1738 # rebuild the legacy display shape (answer + Sources from
1739 # research_resources + Metrics from research_meta) on demand.
1740 from ..services.report_assembly_service import (
1741 assemble_full_report,
1742 get_research_source_links_batch,
1743 )
1745 content = assemble_full_report(research, db_session)
1746 # Only None means "research not found" — guarded above.
1747 # Empty-but-found rows return "" and are valid responses.
1748 if content is None:
1749 return _research_not_found(
1750 research_id, message="Report not found"
1751 )
1753 # Sources live in the research_resources table, not research_meta.
1754 # The post-refactor save path never writes the legacy
1755 # `all_links_of_system` metadata key, so reading it here returned
1756 # [] for every research created since chat-mode-v2 (#3665). Read
1757 # the structured table instead — the same source of truth the
1758 # assembled `content` and the news feed already use. limit=None
1759 # returns every source (this field was never top-N), matching the
1760 # full list the assembled `content` renders for the same research.
1761 sources = get_research_source_links_batch(
1762 [research.id], db_session, limit=None
1763 ).get(research.id, [])
1765 # Return the report data with backwards-compatible fields
1766 # Examples expect 'summary', 'sources', 'findings' at top level
1767 safe_metadata = strip_settings_snapshot(metadata)
1768 return jsonify(
1769 {
1770 "content": content,
1771 # Backwards-compatible fields for examples
1772 "summary": content, # The markdown report is the summary
1773 "sources": sources,
1774 "findings": safe_metadata.get("findings", []),
1775 "metadata": {
1776 "title": research.title if research.title else None,
1777 "query": research.query,
1778 "mode": research.mode if research.mode else None,
1779 "created_at": research.created_at
1780 if research.created_at
1781 else None,
1782 "completed_at": research.completed_at
1783 if research.completed_at
1784 else None,
1785 "report_path": research.report_path,
1786 **safe_metadata,
1787 },
1788 }
1789 )
1791 except Exception:
1792 logger.exception("Error getting research report")
1793 return jsonify({"error": "An internal error has occurred"}), 500
1796@research_bp.route(
1797 "/api/v1/research/<research_id>/export/<format>", methods=["POST"]
1798)
1799@login_required
1800def export_research_report(research_id, format):
1801 """Export research report to different formats (LaTeX, Quarto, RIS, PDF, ODT, etc.)"""
1802 try:
1803 # Use the exporter registry to validate format
1804 from ...exporters import ExporterRegistry
1806 if not ExporterRegistry.is_format_supported(format):
1807 available = ExporterRegistry.get_available_formats()
1808 return jsonify(
1809 {
1810 "error": f"Invalid format. Available formats: {', '.join(available)}"
1811 }
1812 ), 400
1814 # Get research from database
1815 username = session["username"]
1817 try:
1818 with get_user_db_session(username) as db_session:
1819 research = (
1820 db_session.query(ResearchHistory)
1821 .filter_by(id=research_id)
1822 .first()
1823 )
1824 if not research:
1825 return _research_not_found(research_id)
1827 # Build the full assembled report (answer + Sources +
1828 # Metrics) so exporters get the same shape they did
1829 # before the report_content refactor.
1830 from ..services.report_assembly_service import (
1831 assemble_full_report,
1832 )
1834 report_content = assemble_full_report(research, db_session)
1835 if report_content is None:
1836 return _research_not_found(
1837 research_id, message="Report content not found"
1838 )
1840 # Export to requested format (all in memory)
1841 try:
1842 # Use title or query for the PDF title
1843 pdf_title = research.title or research.query
1845 # Generate export content in memory
1846 export_content, filename, mimetype = (
1847 export_report_to_memory(
1848 report_content, format, title=pdf_title
1849 )
1850 )
1852 # Send the file directly from memory
1853 return send_file(
1854 io.BytesIO(export_content),
1855 as_attachment=True,
1856 download_name=filename,
1857 mimetype=mimetype,
1858 )
1859 except MissingPDFDependencyError:
1860 logger.exception(
1861 "PDF export failed: WeasyPrint unavailable"
1862 )
1863 return jsonify(
1864 {"error": get_weasyprint_install_instructions()}
1865 ), 500
1866 except Exception:
1867 logger.exception("Error exporting report")
1868 return jsonify(
1869 {
1870 "error": f"Failed to export to {format}. Please try again later."
1871 }
1872 ), 500
1874 except Exception:
1875 logger.exception("Error in export endpoint")
1876 return jsonify({"error": "An internal error has occurred"}), 500
1878 except Exception:
1879 logger.exception("Unexpected error in export endpoint")
1880 return jsonify({"error": "An internal error has occurred"}), 500
1883@research_bp.route("/api/research/<string:research_id>/status")
1884@limiter.exempt
1885@login_required
1886def get_research_status(research_id):
1887 """Get the status of a research process"""
1888 username = session["username"]
1890 try:
1891 with get_user_db_session(username) as db_session:
1892 research = (
1893 db_session.query(ResearchHistory)
1894 .filter_by(id=research_id)
1895 .first()
1896 )
1898 if research is None:
1899 return _research_not_found(research_id)
1901 status = research.status
1902 progress = research.progress
1903 completed_at = research.completed_at
1904 report_path = research.report_path
1905 metadata = research.research_meta or {}
1907 # Extract and format error information for better UI display
1908 error_info = {}
1909 if metadata and "error" in metadata:
1910 error_msg = metadata["error"]
1911 error_type = "unknown"
1913 # Detect specific error types
1914 if "timeout" in error_msg.lower():
1915 error_type = "timeout"
1916 error_info = {
1917 "type": "timeout",
1918 "message": "LLM service timed out during synthesis. This may be due to high server load or connectivity issues.",
1919 "suggestion": "Try again later or use a smaller query scope.",
1920 }
1921 elif (
1922 "token limit" in error_msg.lower()
1923 or "context length" in error_msg.lower()
1924 ):
1925 error_type = "token_limit"
1926 error_info = {
1927 "type": "token_limit",
1928 "message": "The research query exceeded the AI model's token limit during synthesis.",
1929 "suggestion": "Try using a more specific query or reduce the research scope.",
1930 }
1931 elif (
1932 "final answer synthesis fail" in error_msg.lower()
1933 or "llm error" in error_msg.lower()
1934 ):
1935 error_type = "llm_error"
1936 error_info = {
1937 "type": "llm_error",
1938 "message": "The AI model encountered an error during final answer synthesis.",
1939 "suggestion": "Check that your LLM service is running correctly or try a different model.",
1940 }
1941 elif "ollama" in error_msg.lower():
1942 error_type = "ollama_error"
1943 error_info = {
1944 "type": "ollama_error",
1945 "message": "The Ollama service is not responding properly.",
1946 "suggestion": "Make sure Ollama is running with 'ollama serve' and the model is downloaded.",
1947 }
1948 elif "connection" in error_msg.lower():
1949 error_type = "connection"
1950 error_info = {
1951 "type": "connection",
1952 "message": "Connection error with the AI service.",
1953 "suggestion": "Check your internet connection and AI service status.",
1954 }
1955 elif metadata.get("solution"):
1956 # Use the solution provided in metadata if available
1957 error_info = {
1958 "type": error_type,
1959 "message": error_msg,
1960 "suggestion": str(metadata.get("solution")),
1961 }
1962 else:
1963 # Generic error with the original message
1964 error_info = {
1965 "type": error_type,
1966 "message": error_msg,
1967 "suggestion": "Try again with a different query or check the application logs.",
1968 }
1970 # Get the latest milestone log for this research
1971 latest_milestone = None
1972 try:
1973 milestone_log = (
1974 db_session.query(ResearchLog)
1975 .filter_by(research_id=research_id, level="MILESTONE")
1976 # id tie-breaks equal timestamps so "latest" is
1977 # deterministic (the most recently inserted milestone).
1978 .order_by(
1979 ResearchLog.timestamp.desc(), ResearchLog.id.desc()
1980 )
1981 .first()
1982 )
1983 if milestone_log:
1984 latest_milestone = {
1985 "message": milestone_log.message,
1986 "time": milestone_log.timestamp.isoformat()
1987 if milestone_log.timestamp
1988 else None,
1989 "type": "MILESTONE",
1990 }
1991 logger.debug(
1992 f"Found latest milestone for research {research_id}: {milestone_log.message}"
1993 )
1994 else:
1995 logger.debug(
1996 f"No milestone logs found for research {research_id}"
1997 )
1998 except Exception:
1999 logger.warning("Error fetching latest milestone")
2001 filtered_metadata = strip_settings_snapshot(metadata)
2002 if error_info:
2003 filtered_metadata["error_info"] = error_info
2005 response_data = {
2006 "status": status,
2007 "progress": progress,
2008 "completed_at": completed_at,
2009 "report_path": report_path,
2010 "metadata": filtered_metadata,
2011 }
2013 # Include latest milestone as a log_entry for frontend compatibility
2014 if latest_milestone:
2015 response_data["log_entry"] = latest_milestone
2017 return jsonify(response_data)
2018 except Exception:
2019 logger.exception("Error getting research status")
2020 return jsonify({"error": "Error checking research status"}), 500
2023@research_bp.route("/api/queue/status", methods=["GET"])
2024@login_required
2025def get_queue_status():
2026 """Get the current queue status for the user"""
2027 username = session["username"]
2029 from ..queue import QueueManager
2031 try:
2032 queue_items = QueueManager.get_user_queue(username)
2034 return jsonify(
2035 {
2036 "status": "success",
2037 "queue": queue_items,
2038 "total": len(queue_items),
2039 }
2040 )
2041 except Exception:
2042 logger.exception("Error getting queue status")
2043 return jsonify(
2044 {"status": "error", "message": "Failed to process request"}
2045 ), 500
2048@research_bp.route("/api/queue/<string:research_id>/position", methods=["GET"])
2049@login_required
2050def get_queue_position(research_id):
2051 """Get the queue position for a specific research"""
2052 username = session["username"]
2054 from ..queue import QueueManager
2056 try:
2057 position = QueueManager.get_queue_position(username, research_id)
2059 if position is None: 2059 ↛ 2060line 2059 didn't jump to line 2060 because the condition on line 2059 was never true
2060 return _research_not_found(
2061 research_id, message="Research not found in queue"
2062 )
2064 return jsonify({"status": "success", "position": position})
2065 except Exception:
2066 logger.exception("Error getting queue position")
2067 return jsonify(
2068 {"status": "error", "message": "Failed to process request"}
2069 ), 500
2072@research_bp.route("/api/config/limits", methods=["GET"])
2073@login_required
2074def get_upload_limits():
2075 """
2076 Get file upload configuration limits.
2078 Returns the backend's authoritative limits for file uploads,
2079 allowing the frontend to stay in sync without hardcoding values.
2080 """
2081 return jsonify(
2082 {
2083 "max_file_size": FileUploadValidator.MAX_FILE_SIZE,
2084 "max_files": FileUploadValidator.MAX_FILES_PER_REQUEST,
2085 "allowed_mime_types": list(FileUploadValidator.ALLOWED_MIME_TYPES),
2086 }
2087 )
2090@research_bp.route("/api/upload/pdf", methods=["POST"])
2091@login_required
2092@upload_rate_limit_user
2093@upload_rate_limit_ip
2094def upload_pdf():
2095 """
2096 Upload and extract text from PDF files with comprehensive security validation.
2098 Security features:
2099 - Rate limiting (10 uploads/min, 100/hour per user)
2100 - File size validation (50MB max per file)
2101 - File count validation (100 files max)
2102 - PDF structure validation
2103 - MIME type validation
2105 Performance improvements:
2106 - Single-pass PDF processing (text + metadata)
2107 - Optimized extraction service
2108 """
2109 try:
2110 # Early request size validation (before reading any files)
2111 # This prevents memory exhaustion from chunked encoding attacks
2112 max_request_size = (
2113 FileUploadValidator.MAX_FILES_PER_REQUEST
2114 * FileUploadValidator.MAX_FILE_SIZE
2115 )
2116 if request.content_length and request.content_length > max_request_size: 2116 ↛ 2117line 2116 didn't jump to line 2117 because the condition on line 2116 was never true
2117 return jsonify(
2118 {
2119 "error": f"Request too large. Maximum size is {max_request_size // (1024 * 1024)}MB"
2120 }
2121 ), 413
2123 # Check if files are present in the request
2124 if "files" not in request.files:
2125 return jsonify({"error": "No files provided"}), 400
2127 files = request.files.getlist("files")
2128 if not files or files[0].filename == "":
2129 return jsonify({"error": "No files selected"}), 400
2131 # Validate file count
2132 is_valid, error_msg = FileUploadValidator.validate_file_count(
2133 len(files)
2134 )
2135 if not is_valid:
2136 return jsonify({"error": error_msg}), 400
2138 # Get PDF extraction service
2139 pdf_service = get_pdf_extraction_service()
2141 extracted_texts = []
2142 total_files = len(files)
2143 processed_files = 0
2144 errors = []
2146 for file in files:
2147 if not file or not file.filename: 2147 ↛ 2148line 2147 didn't jump to line 2148 because the condition on line 2147 was never true
2148 errors.append("Unnamed file: Skipped")
2149 continue
2151 try:
2152 filename = sanitize_filename(
2153 file.filename, allowed_extensions={".pdf"}
2154 )
2155 except UnsafeFilenameError:
2156 errors.append("Rejected file: invalid or disallowed filename")
2157 continue
2159 try:
2160 # Read file content (with disk spooling, large files are read from temp file)
2161 pdf_content = file.read()
2163 # Comprehensive validation
2164 is_valid, error_msg = FileUploadValidator.validate_upload(
2165 filename=filename,
2166 file_content=pdf_content,
2167 content_length=file.content_length,
2168 )
2170 if not is_valid:
2171 errors.append(f"{filename}: {error_msg}")
2172 continue
2174 # Extract text and metadata in single pass (performance fix)
2175 result = pdf_service.extract_text_and_metadata(
2176 pdf_content, filename
2177 )
2179 if result["success"]:
2180 extracted_texts.append(
2181 {
2182 "filename": result["filename"],
2183 "text": result["text"],
2184 "size": result["size"],
2185 "pages": result["pages"],
2186 }
2187 )
2188 processed_files += 1
2189 else:
2190 errors.append(f"{filename}: {result['error']}")
2192 except Exception:
2193 logger.exception(f"Error processing {filename}")
2194 errors.append(f"{filename}: Error processing file")
2195 finally:
2196 # Close the file stream to release resources
2197 try:
2198 file.close()
2199 except Exception:
2200 logger.debug("best-effort file stream close", exc_info=True)
2202 # Prepare response
2203 response_data = {
2204 "status": "success",
2205 "processed_files": processed_files,
2206 "total_files": total_files,
2207 "extracted_texts": extracted_texts,
2208 "combined_text": "\n\n".join(
2209 [
2210 f"--- From {item['filename']} ---\n{item['text']}"
2211 for item in extracted_texts
2212 ]
2213 ),
2214 "errors": errors,
2215 }
2217 if processed_files == 0:
2218 return jsonify(
2219 {
2220 "status": "error",
2221 "message": "No files were processed successfully",
2222 "errors": errors,
2223 }
2224 ), 400
2226 return jsonify(response_data)
2228 except Exception:
2229 logger.exception("Error processing PDF upload")
2230 return jsonify({"error": "Failed to process PDF files"}), 500