Coverage for src/local_deep_research/web/api.py: 98%
204 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2REST API for Local Deep Research.
3Provides HTTP access to programmatic search and research capabilities.
4"""
6import inspect
7import os
8import threading
9import time
10from functools import wraps
11from typing import Dict, Any, Optional, Tuple
13try:
14 import resource as _resource_mod
15except ImportError:
16 _resource_mod = None # Windows: no POSIX resource limits
18from flask import Blueprint, jsonify, request, Response
19from loguru import logger
21from ..api.research_functions import analyze_documents
22from ..database.session_context import get_user_db_session
23from ..security.decorators import require_json_body
24from ..security.log_sanitizer import sanitize_error_for_client
25from ..utilities.db_utils import get_settings_manager
26from ..security.rate_limiter import (
27 API_RATE_LIMIT_DEFAULT,
28 api_rate_limit,
29 get_current_username,
30)
32# Match the largest strategy-layer cap (_TOOL_ERROR_MAX_LEN = 500 in
33# langgraph_agent_strategy.py) at the HTTP boundary so a message already
34# scrubbed at the strategy layer is never re-truncated here, and
35# categorizable exception tokens (e.g. "Connection refused" sitting deep
36# in a long error) survive to the API client. The 200-char default of
37# sanitize_error_for_client would truncate that signal prematurely.
38_ERROR_BOUNDARY_MAX_LEN = 500
41def _scrub_error_fields(results: Dict[str, Any]) -> None:
42 """In-place defense-in-depth scrub for exception-derived fields about
43 to leave via jsonify (CWE-209, CodeQL #8019).
45 Strategy-layer `_scrub_tool_error`/`sanitize_error_for_client` already
46 wraps exception text at the source; this is the final HTTP boundary.
47 Only fires on fields that start with the literal ``"Error:"`` marker
48 so legitimate research prose is never touched, and truncation is from
49 the tail so the marker survives the scrub.
51 Field names: strategies emit error text into ``current_knowledge``,
52 but ``quick_summary()`` (research_functions.py) returns that value
53 under the key ``summary`` — and ``analyze_documents()`` uses
54 ``summary`` as well — so both spellings are scrubbed here.
55 """
56 for field in ("current_knowledge", "summary", "formatted_findings"):
57 value = results.get(field)
58 if isinstance(value, str) and value.startswith("Error:"):
59 results[field] = sanitize_error_for_client(
60 value, max_length=_ERROR_BOUNDARY_MAX_LEN
61 )
62 for finding in results.get("findings", []):
63 content = finding.get("content")
64 if isinstance(content, str) and content.startswith("Error:"):
65 finding["content"] = sanitize_error_for_client(
66 content, max_length=_ERROR_BOUNDARY_MAX_LEN
67 )
70# Create a blueprint for the API
71api_blueprint = Blueprint("api_v1", __name__, url_prefix="/api/v1")
73# Body params /analyze_documents accepts beyond the positional
74# query/collection_name. Derived from the real signature so the two can't
75# drift: analyze_documents (unlike quick_summary/generate_report) has no
76# **kwargs, so an unknown key would TypeError at call time — surfacing as
77# an opaque 500. Validating up front turns that into a clear 400.
78# username/settings_snapshot are excluded because they are server-set by
79# _load_user_context_into_params (and overwritten if a body supplied
80# them); programmatic_mode stays accepted as the documented body
81# override.
82_ANALYZE_DOCUMENTS_PARAMS = frozenset(
83 inspect.signature(analyze_documents).parameters
84) - {"query", "collection_name", "username", "settings_snapshot"}
87# Contract enforced by the test_user_context_loaded /
88# test_settings_snapshot_loaded tests in tests/web/test_api_coverage.py.
89# Each authed REST endpoint that calls a research function must invoke
90# this helper. Each endpoint's contract test must assert that username,
91# settings_snapshot (with the tracer key from _mock_access_control), and
92# programmatic_mode=False reach the underlying research function.
93def _load_user_context_into_params(
94 params: Dict[str, Any],
95 username: str | None,
96 allow_default_settings: bool = False,
97) -> Optional[Tuple[Response, int]]:
98 """Mutate ``params`` in place to thread the authenticated user's context
99 down to the research-function call.
101 All authenticated REST endpoints share the same shape: the user has an
102 encrypted DB whose settings snapshot must be loaded and passed through,
103 so calls honor the user's stored API keys, model preference, search
104 tool, and other config — not just the application defaults plus
105 ``LDR_*`` env vars that the programmatic-API fallback would produce.
107 Sets ``username``, ``settings_snapshot``, and (for authenticated
108 requests) ``programmatic_mode=False`` so DB-backed rate-limit
109 estimates persist across requests. Uses ``setdefault`` for
110 ``programmatic_mode`` so an explicit override in the request body
111 is respected.
113 Returns ``None`` on success. If the settings snapshot cannot be loaded,
114 fails CLOSED: returns a ``(response, 503)`` tuple the endpoint must
115 return to the caller. Continuing with an empty snapshot would resolve
116 to the permissive default egress scope, silently downgrading a
117 configured PRIVATE_ONLY / require-local user — bypassing the very
118 boundary they configured. ``allow_default_settings=True`` is the
119 caller's CONSCIOUS opt-in to proceed with defaults (empty snapshot,
120 no egress policy) instead; it is logged loudly so it is never silent.
121 """
122 if not username: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 logger.debug("No username in session, skipping settings snapshot")
124 params["settings_snapshot"] = {}
125 return None
127 params["username"] = username
128 params.setdefault("programmatic_mode", False)
129 try:
130 with get_user_db_session(username) as db_session:
131 if db_session is None:
132 logger.warning(f"No database session for user: {username}")
133 params["settings_snapshot"] = {}
134 return None
135 settings_manager = get_settings_manager(db_session, username)
136 snapshot = settings_manager.get_settings_snapshot()
137 params["settings_snapshot"] = snapshot
138 logger.debug(
139 f"Loaded settings snapshot for user '{username}' "
140 f"with {len(snapshot)} settings"
141 )
142 return None
143 except Exception:
144 # logger.exception captures the traceback so the root cause
145 # (e.g. SQLCipher decrypt failure, settings table corruption,
146 # missing column after a migration) is visible. Without this
147 # the downstream error misleads — looks like "no provider",
148 # really was "couldn't read user settings".
149 logger.exception("Failed to load user settings snapshot")
150 if allow_default_settings:
151 # Caller explicitly opted in to run without their settings.
152 # Proceed with defaults (empty snapshot → permissive scope).
153 # Logged loudly so it's never a silent downgrade.
154 logger.bind(policy_audit=True).warning(
155 "Settings snapshot failed to load; proceeding with "
156 "DEFAULT settings because allow_default_settings=true "
157 "— this run is NOT bound by the user's egress policy",
158 user=username,
159 )
160 params["settings_snapshot"] = {}
161 return None
162 return (
163 jsonify(
164 {
165 "error": (
166 "Your settings could not be loaded, so the "
167 "research was REFUSED to avoid silently "
168 "running without your privacy/egress policy "
169 "(which could send your data to the cloud "
170 "when you meant to keep it local)."
171 ),
172 "how_to_fix": (
173 "This is usually transient — try again. If "
174 "it persists, your encrypted settings "
175 "database may be unavailable (e.g. a session "
176 "/ password issue), so re-authenticate. To "
177 "deliberately run with default settings and "
178 "NO egress policy, resend the request with "
179 '"allow_default_settings": true.'
180 ),
181 "reason": "settings_unavailable",
182 }
183 ),
184 503,
185 )
188def api_access_control(f):
189 """
190 Decorator to enforce API access control:
191 - Check if user is authenticated
192 - Check if API is enabled for the user
193 - Pre-cache api_rate_limit on g so the rate limiter avoids a second DB read
194 """
196 @wraps(f)
197 def decorated_function(*args, **kwargs):
198 from flask import g
200 username = get_current_username()
202 if not username:
203 return jsonify({"error": "Authentication required"}), 401
205 # Read both settings in a single DB session
206 api_enabled = True
207 with get_user_db_session(username) as db_session:
208 if db_session:
209 settings_manager = get_settings_manager(db_session, username)
210 api_enabled = settings_manager.get_setting(
211 "app.enable_api", True
212 )
213 # Pre-cache for _get_user_api_rate_limit() to avoid a second DB read
214 g._api_rate_limit = settings_manager.get_setting(
215 "app.api_rate_limit", API_RATE_LIMIT_DEFAULT
216 )
218 if not api_enabled:
219 return jsonify({"error": "API access is disabled"}), 403
221 return f(*args, **kwargs)
223 return decorated_function
226@api_blueprint.route("/", methods=["GET"])
227@api_access_control
228@api_rate_limit
229def api_documentation():
230 """
231 Provide documentation on the available API endpoints.
232 """
233 api_docs = {
234 "api_version": "v1",
235 "description": "REST API for Local Deep Research",
236 "endpoints": [
237 {
238 "path": "/api/v1/quick_summary",
239 "method": "POST",
240 "description": "Generate a quick research summary",
241 "parameters": {
242 "query": "Research query (required)",
243 "search_tool": "Search engine to use (optional)",
244 "iterations": "Number of search iterations (optional)",
245 "temperature": "LLM temperature (optional)",
246 "allow_default_settings": "Set to true to proceed with default settings (and NO egress policy) when your stored settings cannot be loaded; default is to refuse with 503 (optional)",
247 },
248 },
249 {
250 "path": "/api/v1/generate_report",
251 "method": "POST",
252 "description": "Generate a comprehensive research report",
253 "parameters": {
254 "query": "Research query (required)",
255 "output_file": "Path to save report (optional)",
256 "searches_per_section": "Searches per report section (optional)",
257 "model_name": "LLM model to use (optional)",
258 "temperature": "LLM temperature (optional)",
259 "allow_default_settings": "Set to true to proceed with default settings (and NO egress policy) when your stored settings cannot be loaded; default is to refuse with 503 (optional)",
260 },
261 },
262 {
263 "path": "/api/v1/analyze_documents",
264 "method": "POST",
265 "description": "Search and analyze documents in a local collection",
266 "parameters": {
267 "query": "Search query (required)",
268 "collection_name": "Local collection name (required)",
269 "max_results": "Maximum results to return (optional)",
270 "temperature": "LLM temperature (optional)",
271 "force_reindex": "Force collection reindexing (optional)",
272 "allow_default_settings": "Set to true to proceed with default settings (and NO egress policy) when your stored settings cannot be loaded; default is to refuse with 503 (optional)",
273 },
274 },
275 ],
276 }
278 return jsonify(api_docs)
281@api_blueprint.route("/health", methods=["GET"])
282def health_check():
283 """Health check endpoint with resource usage diagnostics.
285 The basic ``status``/``message``/``timestamp`` fields are public so the
286 Docker healthcheck (which only inspects the HTTP status code) keeps
287 working. File-descriptor and thread diagnostics are only included for
288 authenticated users, to avoid leaking process internals to anonymous
289 callers.
290 """
291 diagnostics = {
292 "status": "ok",
293 "message": "API is running",
294 "timestamp": time.time(),
295 }
297 # Only expose resource diagnostics to authenticated users
298 username = get_current_username()
299 if username:
300 # File descriptor count (Linux only; /proc not available on macOS)
301 try:
302 fd_count = len(os.listdir("/proc/self/fd"))
303 except OSError:
304 fd_count = None
306 # FD soft/hard limits (POSIX)
307 soft_limit = hard_limit = None
308 if _resource_mod is not None:
309 try:
310 soft_limit, hard_limit = _resource_mod.getrlimit(
311 _resource_mod.RLIMIT_NOFILE
312 )
313 if soft_limit == _resource_mod.RLIM_INFINITY:
314 soft_limit = None
315 if hard_limit == _resource_mod.RLIM_INFINITY:
316 hard_limit = None
317 except (AttributeError, ValueError, OSError):
318 pass
320 thread_count = threading.active_count()
322 fd_usage_percent = (
323 round(fd_count / soft_limit * 100, 1)
324 if fd_count is not None
325 and soft_limit is not None
326 and soft_limit > 0
327 else None
328 )
330 diagnostics["resources"] = {
331 "fd_count": fd_count,
332 "fd_soft_limit": soft_limit,
333 "fd_hard_limit": hard_limit,
334 "fd_usage_percent": fd_usage_percent,
335 "thread_count": thread_count,
336 }
338 if fd_usage_percent is not None and fd_usage_percent > 70:
339 diagnostics["status"] = "warning"
340 diagnostics["message"] = (
341 f"High FD usage: {fd_count}/{soft_limit} ({fd_usage_percent}%)"
342 )
344 return jsonify(diagnostics)
347def _serialize_results(results: Dict[str, Any]) -> Response:
348 """
349 Converts the results dictionary into a JSON string.
351 Args:
352 results: The results dictionary.
354 Returns:
355 The JSON string.
357 """
358 # The main thing that needs to be handled here is the `Document` instances.
359 converted_results = results.copy()
360 for finding in converted_results.get("findings", []):
361 for i, document in enumerate(finding.get("documents", [])):
362 finding["documents"][i] = {
363 "metadata": document.metadata,
364 "content": document.page_content,
365 }
367 # CWE-209 / CodeQL #8019: scrub exception-derived fields before
368 # jsonify. See _scrub_error_fields for rationale.
369 _scrub_error_fields(converted_results)
371 return jsonify(converted_results)
374@api_blueprint.route("/quick_summary", methods=["POST"])
375@api_access_control
376@api_rate_limit
377@require_json_body(error_message="Query parameter is required")
378def api_quick_summary():
379 """
380 Generate a quick research summary via REST API.
382 POST /api/v1/quick_summary
383 {
384 "query": "Advances in fusion energy research",
385 "search_tool": "searxng", # Optional: search engine to use (defaults to your configured search.tool setting)
386 "iterations": 2, # Optional: number of search iterations
387 "temperature": 0.7 # Optional: LLM temperature
388 }
389 """
390 logger.debug("API quick_summary endpoint called")
391 data = request.json
392 logger.debug(f"Request data keys: {list(data.keys())}")
394 if "query" not in data:
395 logger.debug("Missing query parameter")
396 return jsonify({"error": "Query parameter is required"}), 400
398 # Extract query and validate type
399 query = data.get("query")
400 if not isinstance(query, str):
401 return jsonify({"error": "Query must be a string"}), 400
402 # Opt-in escape hatch for programmatic callers: when settings can't be
403 # loaded, proceed with defaults (empty snapshot → permissive scope) instead
404 # of failing closed (503). Default False (fail closed) so a configured
405 # PRIVATE_ONLY user is never silently downgraded; setting it true is a
406 # CONSCIOUS "I'm fine running without my settings/egress policy" choice.
407 # Excluded from ``params`` so it isn't forwarded to quick_summary().
408 # Strict ``is True`` (not bool()): for a security-boundary flag we only opt
409 # in on a real JSON ``true`` — not on a truthy string like "false"/"0".
410 allow_default_settings = data.get("allow_default_settings") is True
411 params = {
412 k: v
413 for k, v in data.items()
414 if k not in ("query", "allow_default_settings")
415 }
416 logger.debug(
417 f"Query length: {len(query) if query else 0}, params keys: {list(params.keys()) if params else 'None'}"
418 )
420 username = get_current_username()
422 try:
423 # Import here to avoid circular imports. NOTE: get_user_db_session and
424 # get_settings_manager are NOT re-imported here — both are bound at
425 # module level (top of file). A local re-import would shadow the
426 # module-level name and silently defeat ``patch("...web.api.<name>")``
427 # in tests (the function would fetch the real, encrypted-DB-requiring
428 # implementation instead of the mock), so keep them module-level for one
429 # consistent patch target. Only quick_summary genuinely needs the local
430 # import (it pulls in the research stack, which would cycle).
431 from ..api.research_functions import quick_summary
433 logger.info(
434 f"Processing quick_summary request: query='{query}' for user='{username}'"
435 )
437 # Set reasonable defaults for API use. search_tool deliberately has
438 # no default here: when omitted, quick_summary reads the user's
439 # configured search.tool from the settings snapshot.
440 params.setdefault("temperature", 0.7)
441 params.setdefault("iterations", 1)
443 error = _load_user_context_into_params(
444 params, username, allow_default_settings
445 )
446 if error is not None:
447 return error
449 # Call the actual research function
450 result = quick_summary(query, **params)
452 return _serialize_results(result)
453 except TimeoutError:
454 logger.exception("Request timed out")
455 return (
456 jsonify(
457 {
458 "error": "Request timed out. Please try with a simpler query or fewer iterations."
459 }
460 ),
461 504,
462 )
463 except Exception:
464 logger.exception("Error in quick_summary API")
465 return (
466 jsonify(
467 {
468 "error": "An internal error has occurred. Please try again later."
469 }
470 ),
471 500,
472 )
475@api_blueprint.route("/generate_report", methods=["POST"])
476@api_access_control
477@api_rate_limit
478@require_json_body(error_message="Query parameter is required")
479def api_generate_report():
480 """
481 Generate a comprehensive research report via REST API.
483 POST /api/v1/generate_report
484 {
485 "query": "Impact of climate change on agriculture",
486 "output_file": "/path/to/save/report.md", # Optional
487 "searches_per_section": 2, # Optional
488 "model_name": "gpt-4", # Optional
489 "temperature": 0.5 # Optional
490 }
491 """
492 data = request.json
493 if "query" not in data:
494 return jsonify({"error": "Query parameter is required"}), 400
496 query = data.get("query")
497 # See api_quick_summary for the allow_default_settings semantics
498 # (opt-in escape hatch, strict ``is True``, excluded from params).
499 allow_default_settings = data.get("allow_default_settings") is True
500 params = {
501 k: v
502 for k, v in data.items()
503 if k not in ("query", "allow_default_settings")
504 }
506 username = get_current_username()
508 try:
509 # Import here to avoid circular imports
510 from ..api.research_functions import generate_report
512 # Set reasonable defaults for API use
513 params.setdefault("searches_per_section", 1)
514 params.setdefault("temperature", 0.7)
516 error = _load_user_context_into_params(
517 params, username, allow_default_settings
518 )
519 if error is not None:
520 return error
522 logger.info(
523 f"Processing generate_report request: query='{query}' for user='{username}'"
524 )
526 result = generate_report(query, **params)
528 # Don't return the full content for large reports
529 if (
530 result
531 and "content" in result
532 and isinstance(result["content"], str)
533 and len(result["content"]) > 10000
534 ):
535 # Include a summary of the report content
536 content_preview = (
537 result["content"][:2000] + "... [Content truncated]"
538 )
539 result["content"] = content_preview
540 result["content_truncated"] = True
542 # CWE-209 / CodeQL #8019: same boundary scrub as _serialize_results.
543 # generate_report bypasses that helper, so apply _scrub_error_fields
544 # directly. NOTE: today's payload ({content, metadata, file_path})
545 # contains none of the scrubbed field names — exceptions propagate
546 # to the generic handlers below instead — so this is purely
547 # precautionary: it keeps the every-jsonify-sink boundary policy
548 # intact if the payload shape ever grows error-carrying fields.
549 _scrub_error_fields(result)
551 return jsonify(result)
552 except TimeoutError:
553 logger.exception("Request timed out")
554 return (
555 jsonify(
556 {"error": "Request timed out. Please try with a simpler query."}
557 ),
558 504,
559 )
560 except Exception:
561 logger.exception("Error in generate_report API")
562 return (
563 jsonify(
564 {
565 "error": "An internal error has occurred. Please try again later."
566 }
567 ),
568 500,
569 )
572@api_blueprint.route("/analyze_documents", methods=["POST"])
573@api_access_control
574@api_rate_limit
575@require_json_body(
576 error_message="Both query and collection_name parameters are required"
577)
578def api_analyze_documents():
579 """
580 Search and analyze documents in a local collection via REST API.
582 POST /api/v1/analyze_documents
583 {
584 "query": "neural networks in medicine",
585 "collection_name": "my_collection", # Required: local collection name
586 "max_results": 20, # Optional: max results to return
587 "temperature": 0.7, # Optional: LLM temperature
588 "force_reindex": false # Optional: force reindexing
589 }
590 """
591 data = request.json
592 if "query" not in data or "collection_name" not in data:
593 return (
594 jsonify(
595 {
596 "error": "Both query and collection_name parameters are required"
597 }
598 ),
599 400,
600 )
602 query = data.get("query")
603 collection_name = data.get("collection_name")
604 # See api_quick_summary for the allow_default_settings semantics
605 # (opt-in escape hatch, strict ``is True``, excluded from params).
606 allow_default_settings = data.get("allow_default_settings") is True
607 params = {
608 k: v
609 for k, v in data.items()
610 if k not in ("query", "collection_name", "allow_default_settings")
611 }
613 unknown_params = sorted(set(params) - _ANALYZE_DOCUMENTS_PARAMS)
614 if unknown_params:
615 return (
616 jsonify(
617 {
618 "error": (
619 f"Unknown parameter(s) for analyze_documents: "
620 f"{', '.join(unknown_params)}"
621 ),
622 "allowed_parameters": sorted(_ANALYZE_DOCUMENTS_PARAMS),
623 }
624 ),
625 400,
626 )
628 username = get_current_username()
630 try:
631 error = _load_user_context_into_params(
632 params, username, allow_default_settings
633 )
634 if error is not None:
635 return error
636 result = analyze_documents(query, collection_name, **params)
637 # CWE-209 / CodeQL #8019: same boundary scrub as _serialize_results.
638 # The live field here is `summary` (analyze_documents returns
639 # {summary, documents, ...} and puts "Error: ..." text in summary,
640 # e.g. the unknown-collection message).
641 _scrub_error_fields(result)
642 return jsonify(result)
643 except Exception:
644 logger.exception("Error in analyze_documents API")
645 return (
646 jsonify(
647 {
648 "error": "An internal error has occurred. Please try again later."
649 }
650 ),
651 500,
652 )