Coverage for src/local_deep_research/web/services/research_service.py: 87%
1005 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1import contextvars
2import hashlib
3import json
4import re
5import threading
6import time
7from datetime import datetime, UTC
8from pathlib import Path
10from loguru import logger
12from ...exceptions import DuplicateResearchError, ResearchTerminatedException
13from ...config.llm_config import get_llm
14from ...settings.manager import (
15 SnapshotSettingsContext,
16 apply_environment_overrides_to_snapshot,
17)
19# Output directory for research results
20from ...config.paths import get_research_outputs_directory
21from ...config.search_config import get_search
22from ...constants import ResearchStatus
23from ...database.models import ResearchHistory, ResearchStrategy
24from ...database.session_context import get_user_db_session
25from ...database.thread_local_session import thread_cleanup
26from ...error_handling.openai_compat_errors import (
27 friendly_openai_compatible_error,
28 is_openai_compat_runtime_error,
29)
30from ...error_handling.report_generator import ErrorReportGenerator
31from ...utilities.thread_context import set_search_context
32from ...report_generator import IntegratedReportGenerator
33from ...search_system import AdvancedSearchSystem
34from ...text_optimization import CitationFormatter, CitationMode
35from ...utilities.log_utils import log_for_research
36from ...utilities.search_utilities import extract_links_from_search_results
37from ..models.database import calculate_duration
38from ...settings.env_registry import get_env_setting
39from .socketio_asgi import (
40 emit_to_subscribers as _sio_emit,
41 remove_subscriptions_for_research as _sio_remove,
42)
44OUTPUT_DIR = get_research_outputs_directory()
47class _SocketEmitter:
48 """Object-style adapter over ``socketio_asgi.emit_to_subscribers``.
50 Main's Chat Mode code (merged from #2953) instantiated the old
51 Flask ``SocketIOService`` and passed it around as an object with an
52 ``.emit_to_subscribers`` method. Under FastAPI the real emit path is
53 the module-level ``socketio_asgi`` function (aliased ``_sio_emit``);
54 this thin adapter preserves the object API those call sites expect.
55 """
57 @staticmethod
58 def emit_to_subscribers(event, research_id, data, *, owner):
59 # ``owner`` is forwarded, not defaulted. Subscriptions are keyed by
60 # (owner, research_id), so a default here would send every event
61 # through the adapter to the wrong key and deliver nothing --
62 # silently, because every call site sits inside a swallowing
63 # ``except``. Keeping it required means a caller that forgets it
64 # fails loudly at the call.
65 return _sio_emit(event, research_id, data, owner=owner)
68# Module-level singleton — the adapter is stateless.
69_socket_emitter = _SocketEmitter()
72# Global concurrent research limit (server-wide, across all users)
73_MAX_GLOBAL_CONCURRENT = get_env_setting(
74 "server.max_concurrent_research", default=10
75)
76_global_research_semaphore = threading.Semaphore(_MAX_GLOBAL_CONCURRENT)
78# Fallback used by clamp_user_max_concurrent when the stored
79# `app.max_concurrent_researches` value can't be parsed as an int. Mirrors
80# that setting's own JSON-schema default (see default_settings.json).
81_DEFAULT_USER_MAX_CONCURRENT = 3
84def clamp_user_max_concurrent(raw_value) -> int:
85 """Clamp a user-supplied ``app.max_concurrent_researches`` value to the
86 server-wide semaphore ceiling.
88 ``app.max_concurrent_researches`` is a per-user, user-editable setting
89 (default 3) gating how many of a user's OWN researches may start
90 immediately instead of queueing. Its JSON schema now caps it at 20
91 (previously uncapped), but a value written to a user's DB row before
92 that cap existed keeps its raw stored value -- schema metadata is only
93 reconciled (value is preserved) on the next login after a version bump,
94 see ``SettingsManager.import_settings``. Without this clamp, a user
95 could set an arbitrarily high per-user limit and flood the global
96 research semaphore (``server.max_concurrent_research``, default 10)
97 with their own researches, starving every other user. Clamping at each
98 read site is defense-in-depth on top of the JSON-schema ``max_value``.
100 ACCEPTED RESIDUAL: this clamp fixes accounting against the global
101 ceiling, not per-user fairness. A single user whose clamped value equals
102 the full global ceiling (the default is uncapped up to 20, clamped to
103 10) can still legitimately occupy every global semaphore slot at once,
104 queueing every other user's researches until one of theirs finishes.
105 Per-user fairness / anti-monopoly scheduling (e.g. a lower default
106 per-user share, or reserving slots across users) is intentionally out of
107 scope here -- tracked as a known limitation, not a bug in this function.
109 Args:
110 raw_value: The raw setting value (any type -- callers pass whatever
111 ``SettingsManager.get_setting`` returned).
113 Returns:
114 An int between 1 and the live global semaphore ceiling
115 (``_MAX_GLOBAL_CONCURRENT``), inclusive.
116 """
117 try:
118 value = int(raw_value)
119 except (TypeError, ValueError):
120 value = _DEFAULT_USER_MAX_CONCURRENT
121 value = max(1, value)
122 return min(value, _MAX_GLOBAL_CONCURRENT)
125# Progress allocation for detailed mode — report generation is the bulk of the work
126_DETAILED_SEARCH_PROGRESS_CAP = 8 # Search/output phases capped here
127_DETAILED_REPORT_PROGRESS_START = 10 # Report generation starts here
128_DETAILED_REPORT_PROGRESS_END = 100 # Report generation ends here
129# Phases that belong to the report-generation stage. "report_generation" is
130# emitted in this file; the other four are emitted by report_generator.py. If
131# you add or rename a phase, update both this set and the emitter.
132_REPORT_PHASES = frozenset(
133 {
134 "report_generation",
135 "report_section_research",
136 "report_formatting",
137 "report_structure",
138 "report_complete",
139 }
140)
142# Phases that produce user-visible step messages in chat mode.
143# "complete" is excluded — it fires AFTER the response message is written,
144# which would create a step with a higher sequence_number than the response.
145_STEP_PHASES = frozenset(
146 {
147 "init",
148 "setup",
149 "search_planning",
150 "search",
151 "observation",
152 "output_generation",
153 "synthesis_error",
154 "synthesis_fallback",
155 "report_generation",
156 "report_complete",
157 "error",
158 }
159)
161# Socket.IO emission throttling: minimum interval between progress emissions per research
162_EMIT_THROTTLE_SECONDS = 0.2 # 200ms
163_EMIT_TTL_SECONDS = 3600 # 1 hour — evict stale entries from orphaned research
165# Cap on the partial-content buffer kept server-side so chat-mode termination
166# can persist whatever was already streamed. Bounded to keep memory predictable
167# under pathologically long answers (typical answers are a few KB).
168_MAX_PARTIAL_BUFFER_BYTES = 256 * 1024 # 256 KB
169_emit_cleanup_counter = 0
170_last_emit_times: dict[str, float] = {}
171_last_emit_lock = threading.Lock()
174def _chat_step_decision(
175 phase: str | None,
176 last_step_phase: str | None,
177 is_final: bool,
178) -> tuple[bool, bool]:
179 """Decide whether to persist + emit a chat-mode step event.
181 Encodes the symmetry invariant the progress_callback in
182 run_research_process enforces: for chat sessions, what the live UI
183 surfaces over the socket must equal what `loadSession` reconstructs
184 from chat_progress_steps on reload. The repeat-phase dedup must
185 therefore block BOTH writes, not just the DB write.
187 Returns:
188 (persist, suppress_emit)
189 - persist: True iff add_progress_step should be called for this event.
190 - suppress_emit: True iff the caller should null the socket payload
191 so the emit is dropped. Only suppressed when this is a non-final
192 repeat — final phases (complete/error/report_complete) always
193 emit so the client completion handler fires.
195 Args:
196 phase: the phase tag from this event (e.g. "search", "observation")
197 last_step_phase: phase tag of the previously persisted chat step
198 (None until the first persist of this research).
199 is_final: True if this is a "final" phase event the client must
200 see to fire its completion handler (complete | error |
201 report_complete | progress==100).
202 """
203 if phase not in _STEP_PHASES:
204 # Not a chat-step phase at all (e.g. "complete"). Persist no,
205 # and let the emit through unchanged — completion / control
206 # events still need to reach the client.
207 return False, False
208 dedup_ok = phase != last_step_phase or phase == "observation"
209 if dedup_ok:
210 return True, False
211 return False, not is_final
214def _compose_chat_step_content(
215 message: str, phase: str | None, metadata: dict
216) -> str:
217 """Build the content persisted for a chat progress step.
219 Observation events carry the (bounded) full tool output in
220 ``metadata["content"]`` alongside the one-line ``message``. Persist it
221 beneath the message so the chat step row can expand to show what the
222 tool actually returned — and so a session reload reconstructs exactly
223 what the live accordion showed (the same composition happens client-side
224 in chat.js handleProgressUpdate; keep the two in sync).
226 Only observation events compose: other phases reuse ``metadata["content"]``
227 for large payloads (e.g. full synthesized answers) that must not be
228 duplicated into step rows.
229 """
230 detail = metadata.get("content")
231 if phase == "observation" and detail:
232 return f"{message}\n\n{detail}"
233 return message
236def _parse_research_metadata(research_meta) -> dict:
237 """Parse research_meta into a dict, handling both dict and JSON string types."""
238 if isinstance(research_meta, dict):
239 return dict(research_meta)
240 if isinstance(research_meta, str):
241 try:
242 parsed = json.loads(research_meta)
243 return dict(parsed) if isinstance(parsed, dict) else {}
244 except json.JSONDecodeError:
245 logger.exception("Failed to parse research_meta as JSON")
246 return {}
247 return {}
250def _extract_synthesized_answer(results: dict) -> str:
251 """Pull the LLM-synthesized answer out of a strategy result dict.
253 ``report_content`` must store ONLY the synthesized answer (LLM
254 prose with [N] inline citations). The strategy's
255 ``formatted_findings`` is the full ``format_findings`` blob —
256 answer + ``format_links_to_markdown`` source list +
257 ``## SEARCH QUESTIONS BY ITERATION`` + ``## DETAILED FINDINGS``
258 + ``## ALL SOURCES`` — and ``format_document_split`` only knows
259 how to strip ``## Sources`` headers, not those other sections.
260 Saving the blob would leak sources/findings into ``report_content``.
262 Resolution order:
263 1. ``Final synthesis`` finding (set by source_based, parallel,
264 rapid, focused_iteration, iterdrag).
265 2. ``current_knowledge`` (other strategies expose the answer
266 there).
267 3. Empty string — caller decides whether to fall back further.
268 """
269 for finding in results.get("findings") or []:
270 if finding.get("phase") == "Final synthesis":
271 content = finding.get("content") or ""
272 if content:
273 return content
274 return results.get("current_knowledge") or ""
277def get_citation_formatter():
278 """Get citation formatter with settings from thread context."""
279 # Import here to avoid circular imports
280 from ...config.search_config import get_setting_from_snapshot
282 citation_format = get_setting_from_snapshot(
283 "report.citation_format", "number_hyperlinks"
284 )
285 mode_map = {
286 "number_hyperlinks": CitationMode.NUMBER_HYPERLINKS,
287 "domain_hyperlinks": CitationMode.DOMAIN_HYPERLINKS,
288 "domain_id_hyperlinks": CitationMode.DOMAIN_ID_HYPERLINKS,
289 "domain_id_always_hyperlinks": CitationMode.DOMAIN_ID_ALWAYS_HYPERLINKS,
290 "source_tagged_hyperlinks": CitationMode.SOURCE_TAGGED_HYPERLINKS,
291 "no_hyperlinks": CitationMode.NO_HYPERLINKS,
292 }
293 mode = mode_map.get(citation_format, CitationMode.NUMBER_HYPERLINKS)
294 return CitationFormatter(mode=mode)
297def export_report_to_memory(
298 markdown_content: str, format: str, title: str | None = None
299):
300 """
301 Export a markdown report to different formats in memory.
303 Uses the modular exporter registry to support multiple formats.
304 Available formats can be queried with ExporterRegistry.get_available_formats().
306 Args:
307 markdown_content: The markdown content to export
308 format: Export format (e.g., 'pdf', 'odt', 'latex', 'quarto', 'ris')
309 title: Optional title for the document
311 Returns:
312 Tuple of (content_bytes, filename, mimetype)
313 """
314 from ...exporters import ExporterRegistry, ExportOptions
316 # Normalize format
317 format_lower = format.lower()
319 # Get exporter from registry
320 exporter = ExporterRegistry.get_exporter(format_lower)
322 if exporter is None:
323 available = ExporterRegistry.get_available_formats()
324 raise ValueError(
325 f"Unsupported export format: {format}. "
326 f"Available formats: {', '.join(available)}"
327 )
329 # Title prepending is now handled by each exporter via _prepend_title_if_needed()
330 # PDF and ODT exporters prepend titles; RIS and other formats ignore them
332 # Create options
333 options = ExportOptions(title=title)
335 # Export
336 result = exporter.export(markdown_content, options)
338 logger.info(
339 f"Generated {format_lower} in memory, size: {len(result.content)} bytes"
340 )
342 return result.content, result.filename, result.mimetype
345def save_research_strategy(research_id, strategy_name, *, username):
346 """
347 Save the strategy used for a research to the database.
349 Args:
350 research_id: The ID of the research
351 strategy_name: The name of the strategy used
352 username: The username whose encrypted DB to write. Required —
353 without it get_user_db_session would silently fall back to
354 the Flask session user (or fail off-request-context), so
355 callers must state whose DB they mean.
356 """
357 try:
358 logger.debug(
359 f"save_research_strategy called with research_id={research_id}, strategy_name={strategy_name}"
360 )
361 with get_user_db_session(username) as session:
362 # Check if a strategy already exists for this research
363 existing_strategy = (
364 session.query(ResearchStrategy)
365 .filter_by(research_id=research_id)
366 .first()
367 )
369 if existing_strategy:
370 # Update existing strategy
371 existing_strategy.strategy_name = strategy_name
372 logger.debug(
373 f"Updating existing strategy for research {research_id}"
374 )
375 else:
376 # Create new strategy record
377 new_strategy = ResearchStrategy(
378 research_id=research_id, strategy_name=strategy_name
379 )
380 session.add(new_strategy)
381 logger.debug(
382 f"Creating new strategy record for research {research_id}"
383 )
385 session.commit()
386 logger.info(
387 f"Saved strategy '{strategy_name}' for research {research_id}"
388 )
389 except Exception:
390 logger.exception("Error saving research strategy")
393def get_research_strategy(research_id, *, username):
394 """
395 Get the strategy used for a research.
397 Args:
398 research_id: The ID of the research
399 username: The username whose encrypted DB to read. Required —
400 without it get_user_db_session would silently fall back to
401 the Flask session user (or fail off-request-context, which
402 the except below would swallow into a None return), so
403 callers must state whose DB they mean.
405 Returns:
406 str: The strategy name or None if not found
407 """
408 try:
409 with get_user_db_session(username) as session:
410 strategy = (
411 session.query(ResearchStrategy)
412 .filter_by(research_id=research_id)
413 .first()
414 )
416 return strategy.strategy_name if strategy else None
417 except Exception:
418 logger.exception("Error getting research strategy")
419 return None
422def start_research_process(
423 research_id,
424 query,
425 mode,
426 run_research_callback,
427 **kwargs,
428):
429 """
430 Start a research process in a background thread.
432 Args:
433 research_id: The ID of the research
434 query: The research query
435 mode: The research mode (quick/detailed)
436 run_research_callback: The callback function to run the research
437 **kwargs: Additional parameters to pass to the research process (model, search_engine, etc.)
439 Returns:
440 threading.Thread: The thread running the research
441 """
442 from ..research_state import check_and_start_research
443 from ...exceptions import SystemAtCapacityError
445 # Acquire the global concurrency semaphore SYNCHRONOUSLY in the caller's
446 # thread. Previously this happened inside the worker after the HTTP route
447 # had already returned 200 — at capacity, the worker parked and the user
448 # saw an infinite thinking spinner with the partial unique in-progress
449 # index blocking retries. Surfacing capacity as an exception lets the
450 # route return HTTP 429 before committing any DB state.
451 if not _global_research_semaphore.acquire(blocking=False):
452 raise SystemAtCapacityError(
453 f"At research capacity (max {_MAX_GLOBAL_CONCURRENT} concurrent)"
454 )
456 # Wrap callback so the worker releases the already-held semaphore on exit.
457 original_callback = run_research_callback
459 def _release_semaphore_on_exit(*args, **kw):
460 try:
461 return original_callback(*args, **kw)
462 finally:
463 _global_research_semaphore.release()
465 # Copy the calling request's contextvars (username, session_id) so
466 # service code inside the worker thread sees the authenticated user
467 # when it calls `get_current_username()`. Bare `threading.Thread`
468 # starts with an empty context; without this, metrics/token attribution
469 # and any code path using request_context silently fails to the
470 # Flask-compat fallback (which returns None under FastAPI).
471 _ctx = contextvars.copy_context()
473 def _ctx_wrapped(*a, **kw):
474 return _ctx.run(_release_semaphore_on_exit, *a, **kw)
476 # Prepare (but do not start) the background thread.
477 thread = threading.Thread(
478 target=_ctx_wrapped,
479 args=(
480 research_id,
481 query,
482 mode,
483 ),
484 kwargs=kwargs,
485 )
486 thread.daemon = True
488 # Atomic check-and-start: refuses to spawn a second live thread
489 # for the same research_id. Guards against the double-spawn window
490 # where a post-spawn commit failure in the queue processor could
491 # otherwise cause the retry loop to dispatch the same research twice.
492 try:
493 started = check_and_start_research(
494 research_id,
495 {
496 "thread": thread,
497 "progress": 0,
498 "status": ResearchStatus.IN_PROGRESS,
499 "log": [],
500 "settings": kwargs,
501 },
502 )
503 except Exception:
504 # check_and_start_research raised before the thread ran — the
505 # semaphore won't be released by the worker, so release it here.
506 _global_research_semaphore.release()
507 raise
508 if not started:
509 # No thread will run → no _release_semaphore_on_exit → release here.
510 _global_research_semaphore.release()
511 raise DuplicateResearchError(
512 f"Research {research_id} already has a live thread"
513 )
515 return thread
518def _generate_report_path(query: str) -> Path:
519 """
520 Generates a path for a new report file based on the query.
522 Args:
523 query: The query used for the report.
525 Returns:
526 The path that it generated.
528 """
529 # Generate a unique filename that does not contain
530 # non-alphanumeric characters.
531 query_hash = hashlib.md5( # DevSkim: ignore DS126858
532 query.encode("utf-8"), usedforsecurity=False
533 ).hexdigest()[:10]
534 return OUTPUT_DIR / (
535 f"research_report_{query_hash}_{int(datetime.now(UTC).timestamp())}.md"
536 )
539def _save_chat_message_and_context(
540 chat_session_id,
541 research_id,
542 username,
543 report_content,
544 streaming_enabled,
545 streaming_state,
546 socket_service,
547 settings_snapshot=None,
548):
549 """Save assistant message to chat and update accumulated context."""
550 from ...chat.service import ChatService
551 from ...chat.context import ChatContextManager
553 chat_service = ChatService(username)
554 # chat_messages.content is NOT NULL. Write report_content
555 # inline (snapshot pattern). Falls back to a placeholder marker only
556 # if report_content is itself empty — the same pattern used by the
557 # terminate path's _STOPPED_BEFORE_OUTPUT_MARKER.
558 snapshot_content = report_content or _NO_OUTPUT_MARKER
559 # allow_archived=True: a multi-tab race can flip the session to
560 # archived between research.status=COMPLETED commit and this write.
561 # Losing the final assistant answer is worse than violating the
562 # "no writes to archived sessions" rule for a system-generated row;
563 # user-message writes from chat/routes.py still keep the default.
564 chat_service.add_message(
565 session_id=chat_session_id,
566 role="assistant",
567 content=snapshot_content,
568 message_type="response",
569 research_id=research_id,
570 allow_archived=True,
571 )
572 # Mark the response row as persisted so the trailing
573 # progress_callback("Research completed successfully", 100) — which
574 # runs the termination check and could fire if the user clicks Stop
575 # in the small window between this write and the final emit — does
576 # NOT write a duplicate row via _save_partial_chat_message_on_terminate.
577 if streaming_state is not None: 577 ↛ 579line 577 didn't jump to line 579 because the condition on line 577 was always true
578 streaming_state["_persisted"] = True
579 logger.info(f"Added research result to chat {chat_session_id[:8]}...")
581 try:
582 # report_content is the answer-only string; no extraction needed.
583 chat_content = report_content or ""
584 ctx_manager = ChatContextManager(
585 session_id=chat_session_id,
586 messages=[],
587 settings_snapshot=settings_snapshot,
588 )
589 updates = ctx_manager.extract_context_updates(new_content=chat_content)
590 chat_service.update_accumulated_context(
591 session_id=chat_session_id, **updates
592 )
593 logger.info(
594 f"Updated accumulated context for chat {chat_session_id[:8]}..."
595 )
596 except Exception:
597 # Bumped from debug to warning: a failed accumulated-context
598 # update silently degrades multi-turn context for the next
599 # follow-up in this chat (entities/topics/source counts are
600 # missing). Ops need visibility to catch widespread breakage
601 # before users notice "the AI forgot what we were talking
602 # about" — debug-level was invisible in production.
603 logger.opt(exception=True).warning(
604 "Could not update accumulated context"
605 )
607 if streaming_enabled and streaming_state.get("chunks_sent", 0) > 0:
608 # Flush any partial-bracket fragment held in the citation carry
609 # buffer before sending the final empty sentinel. Without this,
610 # a stream that ends mid-token (LLM emits "[12" as its last
611 # bytes before closing) silently drops the leading "[" from
612 # what the client renders — the carry would be discarded when
613 # the callback's closure goes out of scope.
614 flush = streaming_state.get("_flush_carry")
615 if flush:
616 leftover = flush()
617 if leftover: 617 ↛ 618line 617 didn't jump to line 618 because the condition on line 617 was never true
618 try:
619 socket_service.emit_to_subscribers(
620 "response_chunk",
621 research_id,
622 {
623 "chunk": leftover,
624 "is_streaming": True,
625 "is_final": False,
626 },
627 owner=username,
628 )
629 except Exception:
630 logger.debug(
631 "Carry-buffer flush emit failed (non-critical)"
632 )
633 socket_service.emit_to_subscribers(
634 "response_chunk",
635 research_id,
636 {"chunk": "", "is_streaming": True, "is_final": True},
637 owner=username,
638 )
641_STOPPED_BEFORE_OUTPUT_MARKER = "[Stopped before any output was generated.]"
642_STOPPED_FOOTER = "\n\n_— Stopped by user._"
643_NO_OUTPUT_MARKER = "_(Research completed without producing output.)_"
646# Match a trailing incomplete citation opener at the chunk boundary.
647# Covers both ASCII "[" and the lenticular "【" (U+3010) — some LLMs
648# emit Chinese-style brackets and the citation formatter accepts them,
649# so we have to hold those back the same way to avoid breaking a token
650# across the next chunk.
651_PARTIAL_BRACKET_RE = re.compile(r"[\[【]\d*$")
653# Upper bound on the inline-citation carry buffer. A real citation token
654# is a handful of bytes ("[123"); anything longer means the "[" wasn't a
655# citation opener or the stream is pathological. Flush raw past this so a
656# never-closing "[" + endless digits can't grow the buffer without limit.
657_MAX_CARRY_BYTES = 64
660def _make_chat_stream_callback(
661 research_id,
662 streaming_state,
663 socket_service,
664 username,
665 source_resolver=None,
666 formatter=None,
667):
668 """Build the chat-mode streaming callback.
670 The callback:
671 * Counts chunks (``chunks_sent``).
672 * Buffers RAW chunks in ``streaming_state['chunks']`` so partial
673 content survives termination — the citation handler's local list
674 is discarded on raise. Capped at ``_MAX_PARTIAL_BUFFER_BYTES``;
675 once capped, ``streaming_state['_truncated']`` flips to True and
676 further chunks aren't accumulated server-side.
677 * Raises ``ResearchTerminatedException`` if the user clicked Stop
678 mid-stream — fails fast instead of letting the LLM finish.
679 ``ResearchTerminatedException`` inherits from ``BaseException``,
680 so the citation handler's ``except Exception`` blocks naturally
681 propagate it.
682 * Emits ``response_chunk`` over Socket.IO for live display, with
683 inline citation hyperlinks applied per-chunk when both
684 ``source_resolver`` and ``formatter`` are provided — so the
685 client sees ``[[arxiv.org-1]](url)`` appearing live rather than
686 plain ``[1]`` brackets that only get hyperlinked after the full
687 response saves. Bracket tokens split across chunk boundaries are
688 held in a small carry buffer until the closing ``]`` arrives.
690 ``source_resolver`` — optional ``() -> list[dict]`` returning the
691 current ``all_links_of_system`` (so we read it lazily; the list
692 grows as the agent collects sources).
693 ``formatter`` — optional :class:`CitationFormatter` instance. Its
694 ``mode`` controls the inline format (``[[arxiv.org-1]](url)``
695 etc.), matching what the final-save formatter produces so the
696 live display doesn't mode-shift when ``handleResearchComplete``
697 swaps in the DB-saved version.
699 Extracted to module level so it can be unit-tested without spinning
700 up the full ``run_research_process``.
701 """
703 # Per-call closure state for the streaming-substitution carry
704 # buffer (holds a trailing incomplete bracket like "[12" so the
705 # closing "]3]" on the next chunk completes the citation token).
706 carry = [""]
708 def _flush_carry() -> str:
709 """Release and clear any held partial-bracket fragment.
711 The completion finalizer (``_save_chat_message_and_context``)
712 lives in a different scope and can't reach ``carry`` directly,
713 so it calls this through ``streaming_state['_flush_carry']`` to
714 avoid silently dropping the tail of a stream that ends mid-token
715 like ``"[12"``.
716 """
717 released, carry[0] = carry[0], ""
718 return released
720 # Expose to the completion path via the shared state dict.
721 streaming_state["_flush_carry"] = _flush_carry
723 def _hyperlink_chunk(chunk: str) -> str:
724 """Apply inline citation hyperlinks to a single chunk.
726 Maintains the closure-level ``carry`` buffer for incomplete
727 ``[N`` tokens straddling chunk boundaries. The carry contains
728 the trailing partial-bracket fragment from the previous chunk
729 that we haven't been able to substitute yet — it's prepended
730 to the next chunk so the regex can see the full token. The
731 DELTA returned is what the client should append next: it
732 includes the just-completed carry (now hyperlinked) plus the
733 new chunk's safe portion, MINUS the new chunk's own trailing
734 partial bracket (which becomes the new carry).
735 """
736 if source_resolver is None or formatter is None:
737 return chunk
738 try:
739 sources = source_resolver() or []
740 if not sources: 740 ↛ 743line 740 didn't jump to line 743 because the condition on line 740 was never true
741 # Reset carry so the leading "[" we held onto doesn't
742 # disappear from the client's accumulated text.
743 released, carry[0] = carry[0], ""
744 return released + chunk
745 text = carry[0] + chunk
746 pending = _PARTIAL_BRACKET_RE.search(text)
747 if pending:
748 safe = text[: pending.start()]
749 new_carry = text[pending.start() :]
750 # Bound the carry. A well-formed citation token is a few
751 # bytes (`[12`); if the held fragment grows past this, the
752 # "[" was not actually opening a citation (or a hostile /
753 # misbehaving LLM is streaming `[` + endless digits with no
754 # closing `]`). Flush it raw rather than buffering without
755 # limit — preserves the text, just doesn't hyperlink it.
756 if len(new_carry) > _MAX_CARRY_BYTES:
757 safe = text
758 carry[0] = ""
759 else:
760 carry[0] = new_carry
761 else:
762 safe = text
763 carry[0] = ""
764 if not safe: 764 ↛ 765line 764 didn't jump to line 765 because the condition on line 764 was never true
765 return ""
766 return formatter.apply_inline_hyperlinks(safe, sources)
767 except Exception:
768 # Hyperlinking is quality-of-life. On any failure fall back
769 # to emitting the raw chunk so the user still sees the text.
770 logger.debug(
771 "Inline citation hyperlinking failed; emitting raw chunk",
772 exc_info=True,
773 )
774 released, carry[0] = carry[0], ""
775 return released + chunk
777 def stream_callback(chunk: str):
778 # Resolve through the module namespace each call so tests can
779 # ``patch("local_deep_research.web.routes.globals.is_termination_requested")``.
780 # Cached in sys.modules — negligible cost.
781 from ..routes.globals import is_termination_requested
783 if not chunk:
784 return
785 streaming_state["chunks_sent"] += 1
786 if not streaming_state["_truncated"]: 786 ↛ 807line 786 didn't jump to line 807 because the condition on line 786 was always true
787 chunk_bytes = len(chunk.encode("utf-8"))
788 if (
789 streaming_state["_bytes"] + chunk_bytes
790 <= _MAX_PARTIAL_BUFFER_BYTES
791 ):
792 # IMPORTANT: store the RAW chunk in the partial-content
793 # buffer (terminate handler joins these and saves them as
794 # the partial assistant message). If we stored the
795 # hyperlinked version, the saved-on-terminate text would
796 # be double-formatted when the user resumes.
797 streaming_state["chunks"].append(chunk)
798 streaming_state["_bytes"] += chunk_bytes
799 else:
800 streaming_state["_truncated"] = True
801 logger.warning(
802 f"Partial-content buffer hit {_MAX_PARTIAL_BUFFER_BYTES} bytes "
803 f"for research {research_id}; further chunks won't be persisted on terminate"
804 )
805 # Mid-stream interrupt — fail fast if the user clicked Stop while
806 # the LLM is still streaming.
807 if is_termination_requested(research_id):
808 raise ResearchTerminatedException( # noqa: TRY301 — propagated through citation handler
809 "Research was terminated by user during streaming"
810 )
811 # Apply citation hyperlinks for the client emit only — the
812 # client accumulates substituted text into the streaming bubble
813 # so the user sees [[arxiv.org-1]](url) appearing live as the
814 # model writes "According to [1]…".
815 display_chunk = _hyperlink_chunk(chunk)
816 if not display_chunk: 816 ↛ 817line 816 didn't jump to line 817 because the condition on line 816 was never true
817 return # nothing safe to emit yet (all held in carry buffer)
818 try:
819 socket_service.emit_to_subscribers(
820 "response_chunk",
821 research_id,
822 {
823 "chunk": display_chunk,
824 "is_streaming": True,
825 "is_final": False,
826 },
827 owner=username,
828 )
829 except Exception:
830 logger.debug("Stream chunk emit failed (non-critical)")
832 return stream_callback
835def _save_partial_chat_message_on_terminate(
836 chat_session_id,
837 research_id,
838 username,
839 partial_content,
840 truncated=False,
841 streaming_state=None,
842):
843 """Persist a chat 'response' row capturing whatever was streamed before
844 termination, and emit a final ``response_chunk`` so the client strips
845 the streaming class from the bubble.
847 Must run BEFORE ``handle_termination()`` because that path runs
848 ``cleanup_research_resources()`` which removes the Socket.IO room
849 subscriptions — anything emitted afterwards goes nowhere.
851 Idempotent: when ``streaming_state`` is supplied, sets a ``_persisted``
852 flag so duplicate calls (one from the progress callback, one from the
853 outer except handler when a stream_callback raises mid-stream) only
854 write a single row.
856 Skips silently when ``chat_session_id`` is falsy (single-turn case).
857 All failures are swallowed — termination cleanup must never crash the
858 worker.
859 """
860 if not chat_session_id:
861 return
862 if streaming_state is not None and streaming_state.get("_persisted"):
863 return
864 try:
865 from ...chat.service import ChatService
867 if partial_content:
868 content = partial_content + _STOPPED_FOOTER
869 if truncated:
870 content += " _(output was very long; truncated.)_"
871 else:
872 content = _STOPPED_BEFORE_OUTPUT_MARKER
874 # allow_archived=True: same rationale as the completion path —
875 # the partial response on Stop is system-generated and must
876 # survive a concurrent archive (see _save_chat_message_and_context).
877 ChatService(username).add_message(
878 session_id=chat_session_id,
879 role="assistant",
880 content=content,
881 message_type="response",
882 research_id=research_id,
883 allow_archived=True,
884 )
885 # Set the idempotency flag ONLY after the write succeeds. If
886 # `add_message` raises (DB lock, encryption error, archived
887 # session), the outer ResearchTerminatedException handler retries
888 # this helper — flipping the flag pre-write would short-circuit
889 # the retry and silently lose the partial response.
890 if streaming_state is not None:
891 streaming_state["_persisted"] = True
892 logger.info(
893 f"Persisted partial chat response for terminated research "
894 f"{research_id} ({len(content)} chars)"
895 )
896 except Exception:
897 logger.opt(exception=True).warning(
898 "Failed to persist partial chat message on terminate"
899 )
901 try:
902 _socket_emitter.emit_to_subscribers(
903 "response_chunk",
904 research_id,
905 {"chunk": "", "is_streaming": True, "is_final": True},
906 owner=username,
907 )
908 except Exception:
909 logger.debug("Final-chunk emit on terminate failed (non-critical)")
912@log_for_research
913@thread_cleanup
914def run_research_process(research_id, query, mode, **kwargs):
915 """
916 Run the research process in the background for a given research ID.
918 Args:
919 research_id: The ID of the research
920 query: The research query
921 mode: The research mode (quick/detailed)
922 **kwargs: Additional parameters for the research (model_provider, model, search_engine, etc.)
923 MUST include 'username' for database access
924 """
925 from ..research_state import (
926 is_research_active,
927 is_termination_requested,
928 update_progress_and_check_active,
929 )
931 # Extract username - required for database access
932 username = kwargs.get("username")
933 if not username:
934 logger.error("No username provided to research thread")
935 raise ValueError("Username is required for research process")
936 # Extract user_password early so it's available for all cleanup paths
937 user_password = kwargs.get("user_password")
939 # Establish thread context FIRST so every subsequent log line in this
940 # thread can be attributed to the correct user/research and persisted
941 # to the user's encrypted ResearchLog. Otherwise the early INFO logs
942 # below ("Research thread started", "Research strategy", "Research
943 # parameters") fire before start_research_process gets to its own
944 # set_search_context call (~line 417) and the daemon can't open the
945 # encrypted DB to write them — silently dropped via the bare-except.
946 set_search_context(
947 {
948 "research_id": research_id,
949 "username": username,
950 "user_password": user_password,
951 # Settings snapshot may be absent here (early init); the
952 # shared context built later overwrites this with the
953 # populated snapshot. Defaults to {} so cache reads in any
954 # intermediate code path see a non-None dict.
955 "settings_snapshot": kwargs.get("settings_snapshot") or {},
956 }
957 )
959 logger.info(f"Research thread started with username: {username}")
961 try:
962 # Check if this research has been terminated before we even start
963 if is_termination_requested(research_id):
964 logger.info(
965 f"Research {research_id} was terminated before starting"
966 )
967 cleanup_research_resources(
968 research_id,
969 username,
970 user_password=user_password,
971 final_status=ResearchStatus.SUSPENDED,
972 )
973 return
975 logger.info(
976 f"Starting research process for ID {research_id}, query: {query}"
977 )
979 # Extract key parameters
980 model_provider = kwargs.get("model_provider")
981 model = kwargs.get("model")
982 custom_endpoint = kwargs.get("custom_endpoint")
983 search_engine = kwargs.get("search_engine")
984 max_results = kwargs.get("max_results")
985 time_period = kwargs.get("time_period")
986 iterations = kwargs.get("iterations")
987 questions_per_iteration = kwargs.get("questions_per_iteration")
988 strategy = kwargs.get("strategy")
989 settings_snapshot = kwargs.get(
990 "settings_snapshot", {}
991 ) # Complete settings snapshot
992 # NOTE: the "_username" injection that lets the agent register the
993 # user's document collections is done downstream in
994 # AdvancedSearchSystem.__init__ (the narrowest common consumer of the
995 # strategy-running paths — web run and the programmatic API), not here.
996 # Run-start egress (below) doesn't need it: it passes username
997 # explicitly. See ensure_snapshot_username in search_system.py.
999 # Log settings snapshot to debug
1000 from ...settings.logger import log_settings
1002 log_settings(settings_snapshot, "Settings snapshot received in thread")
1004 # Strategy should already be saved in the database before thread starts
1005 logger.info(f"Research strategy: {strategy}")
1007 # Log all parameters for debugging
1008 logger.info(
1009 f"Research parameters: provider={model_provider}, model={model}, "
1010 f"search_engine={search_engine}, max_results={max_results}, "
1011 f"time_period={time_period}, iterations={iterations}, "
1012 f"questions_per_iteration={questions_per_iteration}, "
1013 f"custom_endpoint={custom_endpoint}, strategy={strategy}"
1014 )
1016 # Reapply current operator environment policy at actual dispatch time.
1017 # Queued snapshots may predate a restart or configuration change; every
1018 # downstream consumer must see this same effective copy.
1019 settings_snapshot = apply_environment_overrides_to_snapshot(
1020 settings_snapshot
1021 )
1022 if max_results is not None:
1023 settings_snapshot["search.max_results"] = max_results
1024 if time_period is not None:
1025 settings_snapshot["search.time_period"] = time_period
1027 # Create a settings context that uses snapshot - no database access in threads
1028 settings_context = SnapshotSettingsContext(
1029 settings_snapshot, username=username
1030 )
1032 if "model_provider" not in kwargs:
1033 model_provider = settings_context.get_setting("llm.provider")
1034 if "model" not in kwargs:
1035 model = settings_context.get_setting("llm.model")
1036 if (
1037 "custom_endpoint" not in kwargs
1038 and model_provider == "openai_endpoint"
1039 ):
1040 custom_endpoint = settings_context.get_setting(
1041 "llm.openai_endpoint.url"
1042 )
1043 if "search_engine" not in kwargs:
1044 search_engine = settings_context.get_setting("search.tool")
1045 if "iterations" not in kwargs:
1046 iterations = settings_context.get_setting("search.iterations")
1047 if "questions_per_iteration" not in kwargs:
1048 questions_per_iteration = settings_context.get_setting(
1049 "search.questions_per_iteration"
1050 )
1051 if "strategy" not in kwargs:
1052 strategy = settings_context.get_setting(
1053 "search.search_strategy", "source-based"
1054 )
1056 # Only log settings if explicitly enabled via LDR_LOG_SETTINGS env var
1057 from ...settings.logger import log_settings
1059 log_settings(
1060 settings_context.values, "SettingsContext values extracted"
1061 )
1063 # Set the settings context for this thread
1064 from ...config.thread_settings import (
1065 set_settings_context,
1066 )
1068 set_settings_context(settings_context)
1070 # Defense-in-depth: register an EgressContext on the audit-hook
1071 # thread-local so socket.connect calls that bypass our explicit
1072 # PEPs (third-party libraries, future contributors reaching for
1073 # raw requests.get, prompt-injection-steered tools) are gated
1074 # against the same scope. Cleared by the @thread_cleanup exit
1075 # handler in database/thread_local_session.py when the worker
1076 # winds down. If snapshot construction errors here, we log and
1077 # continue — the explicit PEPs still gate the known callers, so
1078 # the run is not less secure than before this hook existed.
1079 try:
1080 from ...security import set_active_context as _set_egress_ctx
1081 from ...security.egress.policy import (
1082 PolicyDeniedError,
1083 build_run_egress_context,
1084 )
1086 try:
1087 _egress_ctx = build_run_egress_context(
1088 settings_snapshot, search_engine, username=username
1089 )
1090 _egress_primary = _egress_ctx.primary_engine
1091 _set_egress_ctx(_egress_ctx)
1092 except (PolicyDeniedError, ValueError) as _ctx_err:
1093 # Corrupted scope or invalid policy config. The run-start
1094 # precheck already rejects this at the API boundary;
1095 # if we get here it is because the run was launched
1096 # without going through the precheck (CLI, scheduler).
1097 # Re-raise so the worker fails fast with a clear reason
1098 # rather than running unprotected.
1099 logger.bind(policy_audit=True).warning(
1100 "research worker refused: egress policy unevaluable",
1101 research_id=research_id,
1102 reason=str(_ctx_err),
1103 )
1104 raise
1106 # Stage C (ADR-0007): enforce the two-axis (sensitivity x exposure)
1107 # rule at this shared chokepoint too. The /api/start_research route
1108 # applies it before the run starts, but follow-up / chat / queue
1109 # runs reach the worker without that precheck — so evaluate it here,
1110 # where every entry point flows through, reusing the context
1111 # resolved just above. UNPROTECTED evaluates permissively (never
1112 # blocks); on an internal error audit_run fails CLOSED (a denying
1113 # decision) so the run is refused, not silently degraded. A refusal
1114 # raises PolicyDeniedError, failing the run like the
1115 # unevaluable-config path.
1116 from ...security.egress.run_classification import (
1117 audit_run_from_snapshot,
1118 )
1120 _two_axis = audit_run_from_snapshot(
1121 settings_snapshot,
1122 _egress_ctx,
1123 _egress_primary,
1124 llm_provider=model_provider,
1125 )
1126 if not _two_axis.allowed: 1126 ↛ 1127line 1126 didn't jump to line 1127 because the condition on line 1126 was never true
1127 logger.bind(policy_audit=True).warning(
1128 "research worker refused: two-axis egress",
1129 research_id=research_id,
1130 reason=_two_axis.reason,
1131 offending=_two_axis.offending,
1132 )
1133 raise PolicyDeniedError(_two_axis, target=_egress_primary)
1134 except ImportError:
1135 # security module unavailable — preserve legacy behaviour.
1136 logger.debug("egress audit hook unavailable for research worker")
1138 # user_password already extracted above (before termination check)
1140 # Create shared research context that can be updated during research
1141 shared_research_context = {
1142 "research_id": research_id,
1143 "research_query": query,
1144 "research_mode": mode,
1145 "research_phase": "init",
1146 "search_iteration": 0,
1147 "search_engines_planned": None,
1148 "search_engine_selected": search_engine,
1149 "username": username, # Add username for queue operations
1150 "user_password": user_password, # Add password for metrics access
1151 # Thread-safe settings snapshot propagated to background search
1152 # threads (engine config, per-user resolution, egress scope).
1153 "settings_snapshot": settings_snapshot,
1154 "chat_session_id": kwargs.get("chat_session_id"),
1155 }
1157 # If this is a follow-up research, include the parent context
1158 if "research_context" in kwargs and kwargs["research_context"]:
1159 logger.info(
1160 f"Adding parent research context with {len(kwargs['research_context'].get('past_findings', ''))} chars of findings"
1161 )
1162 shared_research_context.update(kwargs["research_context"])
1164 # Do not log context keys as they may contain sensitive information
1165 logger.info(f"Created shared_research_context for user: {username}")
1167 # Set search context for search tracking
1168 set_search_context(shared_research_context)
1170 # Per-research dedup state for step message persistence
1171 last_step_phase = None
1173 # Pre-bind streaming_state so progress_callback's closure cell has a
1174 # value even if an exception fires between this point and the full
1175 # initialization later in this function. Without this, the except handler's
1176 # call to progress_callback during a concurrent termination raises
1177 # UnboundLocalError, silently skipping the DB FAILED update and the
1178 # error socket emit. The full streaming_state is reassigned below
1179 # before any real streaming starts; keys must match the canonical shape.
1180 streaming_state: dict = {
1181 "chunks_sent": 0,
1182 "chunks": [],
1183 "_bytes": 0,
1184 "_truncated": False,
1185 }
1186 streaming_enabled = False
1188 # Set up progress callback
1189 def progress_callback(message, progress_percent, metadata):
1190 nonlocal last_step_phase
1191 # Frequent termination check
1192 if is_termination_requested(research_id):
1193 # Persist the partial chat row + emit final chunk BEFORE
1194 # handle_termination — afterwards the Socket.IO room is gone.
1195 _save_partial_chat_message_on_terminate(
1196 shared_research_context.get("chat_session_id"),
1197 research_id,
1198 username,
1199 "".join(streaming_state.get("chunks", [])),
1200 truncated=streaming_state.get("_truncated", False),
1201 streaming_state=streaming_state,
1202 )
1203 handle_termination(research_id, username)
1204 streaming_state["_termination_handled"] = True
1205 raise ResearchTerminatedException( # noqa: TRY301 — inside nested callback, not caught by enclosing try
1206 "Research was terminated by user"
1207 )
1209 # Silent phase — no UI logging or socket emission needed
1210 if metadata.get("phase") == "termination_check":
1211 return
1213 # Bind research_id AND username so the database_sink + queue
1214 # daemon can resolve the per-user encrypted DB. Without username
1215 # the daemon's _write_log_to_database hits "No authenticated
1216 # user", silently swallows the error, and ResearchLog ends up
1217 # with zero milestone rows — leaving /api/research/<id>/status
1218 # without a log_entry to render and the frontend stuck on the
1219 # "Performing research..." fallback.
1220 bound_logger = logger.bind(
1221 research_id=research_id, username=username
1222 )
1223 bound_logger.log("MILESTONE", message)
1225 if "SEARCH_PLAN:" in message:
1226 engines = message.split("SEARCH_PLAN:")[1].strip()
1227 metadata["planned_engines"] = engines
1228 metadata["phase"] = "search_planning" # Use existing phase
1229 # Update shared context for token tracking
1230 shared_research_context["search_engines_planned"] = engines
1231 shared_research_context["research_phase"] = "search_planning"
1233 if "ENGINE_SELECTED:" in message:
1234 engine = message.split("ENGINE_SELECTED:")[1].strip()
1235 metadata["selected_engine"] = engine
1236 metadata["phase"] = "search" # Use existing 'search' phase
1237 # Update shared context for token tracking
1238 shared_research_context["search_engine_selected"] = engine
1239 shared_research_context["research_phase"] = "search"
1241 # Capture other research phases for better context tracking
1242 if metadata.get("phase"): 1242 ↛ 1246line 1242 didn't jump to line 1246 because the condition on line 1242 was always true
1243 shared_research_context["research_phase"] = metadata["phase"]
1245 # Update search iteration if available
1246 if "iteration" in metadata:
1247 shared_research_context["search_iteration"] = metadata[
1248 "iteration"
1249 ]
1251 # Adjust progress based on research mode
1252 adjusted_progress = progress_percent
1253 phase = metadata.get("phase", "")
1255 if mode == "detailed":
1256 # Report phases pass through (already mapped by wrapper).
1257 # All other phases — including "complete" emitted by each
1258 # strategy when its analyze_topic finishes (report
1259 # generation runs analyze_topic per subsection, so a
1260 # strategy "complete" fires mid-report and must NOT be
1261 # treated as the end of the whole run) — are capped.
1262 # update_progress_and_check_active enforces global
1263 # monotonicity, so backwards jumps from non-monotonic
1264 # strategy updates are absorbed there.
1265 # None values (error path, sub-search relays) pass through
1266 # unchanged.
1267 if phase not in _REPORT_PHASES and progress_percent is not None:
1268 adjusted_progress = min(
1269 _DETAILED_SEARCH_PROGRESS_CAP, progress_percent
1270 )
1271 elif (
1272 mode == "quick"
1273 and phase == "output_generation"
1274 and progress_percent is not None
1275 ):
1276 # For quick mode, scale output_generation to 85-95% range
1277 if progress_percent > 0: 1277 ↛ 1280line 1277 didn't jump to line 1280 because the condition on line 1277 was always true
1278 adjusted_progress = 85 + (progress_percent / 100) * 10
1279 else:
1280 adjusted_progress = 85
1282 # Atomically update progress and check if research is still active
1283 if adjusted_progress is not None:
1284 adjusted_progress, still_active = (
1285 update_progress_and_check_active(
1286 research_id, adjusted_progress
1287 )
1288 )
1289 else:
1290 still_active = is_research_active(research_id)
1292 if still_active: 1292 ↛ exitline 1292 didn't return from function 'progress_callback' because the condition on line 1292 was always true
1293 # Queue the progress update to be processed in main thread
1294 if adjusted_progress is not None:
1295 from ..queue.processor_v2 import queue_processor
1297 if username: 1297 ↛ 1302line 1297 didn't jump to line 1302 because the condition on line 1297 was always true
1298 queue_processor.queue_progress_update(
1299 username, research_id, adjusted_progress
1300 )
1301 else:
1302 logger.warning(
1303 f"Cannot queue progress update for research {research_id} - no username available"
1304 )
1306 # Determine socket emit throttling
1307 phase = metadata.get("phase", "")
1308 is_final = (
1309 phase
1310 in (
1311 "complete",
1312 "error",
1313 "report_complete",
1314 )
1315 or adjusted_progress == 100
1316 )
1318 should_emit = is_final
1319 if not is_final:
1320 now = time.monotonic()
1321 with _last_emit_lock:
1322 last = _last_emit_times.get(research_id, 0)
1323 if now - last >= _EMIT_THROTTLE_SECONDS:
1324 _last_emit_times[research_id] = now
1325 should_emit = True
1326 # Periodic TTL cleanup for orphaned entries
1327 global _emit_cleanup_counter # noqa: PLW0603
1328 _emit_cleanup_counter += 1
1329 if _emit_cleanup_counter % 100 == 0:
1330 stale = [
1331 rid
1332 for rid, t in _last_emit_times.items()
1333 if now - t > _EMIT_TTL_SECONDS
1334 ]
1335 for rid in stale: 1335 ↛ 1336line 1335 didn't jump to line 1336 because the loop on line 1335 never started
1336 del _last_emit_times[rid]
1338 # Decide chat-step persistence BEFORE building the event:
1339 # a step that will be persisted must always emit, even
1340 # inside the throttle window above — otherwise a burst of
1341 # back-to-back observations (the agent issuing several tool
1342 # calls in one turn) writes rows the live UI never showed,
1343 # breaking the symmetry invariant in the live direction.
1344 #
1345 # Symmetry invariant: for chat sessions, what the user sees
1346 # live must equal what `loadSession` reconstructs on reload.
1347 # If dedup blocks persistence of a repeat step phase, drop
1348 # the socket emit too (unless this is a final-phase event
1349 # the completion handler depends on) so live UI doesn't
1350 # surface events that vanish on reload.
1351 _chat_session_id = shared_research_context.get(
1352 "chat_session_id"
1353 )
1354 _persist = False
1355 _suppress_emit = False
1356 if _chat_session_id:
1357 _persist, _suppress_emit = _chat_step_decision(
1358 phase, last_step_phase, is_final
1359 )
1360 if _persist:
1361 should_emit = True
1363 # Build event data (before emit, shared scope)
1364 event_data = None
1365 if should_emit:
1366 event_data = {
1367 "progress": adjusted_progress,
1368 "message": message,
1369 "phase": phase,
1370 }
1371 # Include additional metadata for MCP/ReAct strategy display
1372 if metadata.get("thought"): 1372 ↛ 1373line 1372 didn't jump to line 1373 because the condition on line 1372 was never true
1373 event_data["thought"] = metadata["thought"]
1374 if metadata.get("tool"):
1375 event_data["tool"] = metadata["tool"]
1376 if metadata.get("arguments"): 1376 ↛ 1377line 1376 didn't jump to line 1377 because the condition on line 1376 was never true
1377 event_data["arguments"] = metadata["arguments"]
1378 if metadata.get("iteration"): 1378 ↛ 1379line 1378 didn't jump to line 1379 because the condition on line 1378 was never true
1379 event_data["iteration"] = metadata["iteration"]
1380 if metadata.get("error"):
1381 event_data["error"] = metadata["error"]
1382 if metadata.get("content"):
1383 event_data["content"] = metadata["content"]
1385 # Persist step message to chat session BEFORE emitting socket
1386 # (ensures DB has the step when clients receive the event).
1387 if _persist:
1388 try:
1389 from ...chat.service import ChatService
1391 ChatService(username).add_progress_step(
1392 session_id=_chat_session_id,
1393 research_id=research_id,
1394 content=_compose_chat_step_content(
1395 message, phase, metadata
1396 ),
1397 phase=phase,
1398 )
1399 last_step_phase = phase
1400 except Exception:
1401 logger.opt(exception=True).warning(
1402 "Failed to persist progress step"
1403 )
1404 # Symmetry invariant: if persistence failed
1405 # (e.g. OperationalError under DB contention),
1406 # suppress the live emit too — otherwise the
1407 # client sees a step that vanishes on reload.
1408 event_data = None
1409 if _suppress_emit:
1410 event_data = None
1412 # Emit socket event AFTER DB persistence
1413 if event_data is not None:
1414 try:
1415 _sio_emit(
1416 "research_progress",
1417 research_id,
1418 event_data,
1419 owner=username,
1420 )
1421 except Exception:
1422 logger.exception("Socket emit error (non-critical)")
1424 # Function to check termination during long-running operations
1425 def check_termination():
1426 if is_termination_requested(research_id):
1427 _save_partial_chat_message_on_terminate(
1428 shared_research_context.get("chat_session_id"),
1429 research_id,
1430 username,
1431 "".join(streaming_state.get("chunks", [])),
1432 truncated=streaming_state.get("_truncated", False),
1433 streaming_state=streaming_state,
1434 )
1435 handle_termination(research_id, username)
1436 streaming_state["_termination_handled"] = True
1437 raise ResearchTerminatedException( # noqa: TRY301 — inside nested callback, not caught by enclosing try
1438 "Research was terminated by user during long-running operation"
1439 )
1440 return False # Not terminated
1442 # Configure the system with the specified parameters
1443 use_llm = None
1444 if model or search_engine or model_provider: 1444 ↛ 1451line 1444 didn't jump to line 1451 because the condition on line 1444 was always true
1445 # Log that we're overriding system settings
1446 logger.info(
1447 f"Overriding system settings with: provider={model_provider}, model={model}, search_engine={search_engine}"
1448 )
1450 # Override LLM if model or model_provider specified
1451 if model or model_provider:
1452 try:
1453 # Get LLM with the overridden settings.
1454 # Pass settings_snapshot explicitly — without it the LLM
1455 # PEP (llm_config.py:295) skips evaluate_llm_endpoint
1456 # because the gate is "if settings_snapshot is not None
1457 # and provider:". The snapshot is already stored on
1458 # shared_research_context as "settings_snapshot".
1459 use_llm = get_llm(
1460 model_name=model,
1461 provider=model_provider,
1462 openai_endpoint_url=custom_endpoint,
1463 research_id=research_id,
1464 research_context=shared_research_context,
1465 settings_snapshot=shared_research_context.get(
1466 "settings_snapshot"
1467 ),
1468 # Thread the run's user so the LLM PEP can resolve a
1469 # per-user registered provider and enforce local-only
1470 # policy. The shared_research_context snapshot is captured
1471 # before AdvancedSearchSystem injects `_username`, so
1472 # without this the whole overridden run (question gen,
1473 # summaries, final report) could fail open to a cloud LLM
1474 # for a user whose primary is a private retriever.
1475 username=username,
1476 )
1478 logger.info(
1479 f"Successfully set LLM to: provider={model_provider}, model={model}"
1480 )
1481 except Exception as e:
1482 logger.exception(
1483 f"Error setting LLM provider={model_provider}, model={model}"
1484 )
1485 error_msg = str(e)
1486 # Surface configuration errors to user instead of silently continuing
1487 config_error_keywords = [
1488 "model path",
1489 "llamacpp",
1490 "cannot connect",
1491 "server",
1492 "not configured",
1493 "not responding",
1494 "directory",
1495 ".gguf",
1496 ]
1497 if any(
1498 keyword in error_msg.lower()
1499 for keyword in config_error_keywords
1500 ):
1501 # This is a configuration error the user can fix
1502 raise ValueError(
1503 f"LLM Configuration Error: {error_msg}"
1504 ) from e
1505 # For other errors, re-raise to avoid silent failures
1506 raise
1508 # Create search engine first if specified, to avoid default creation without username
1509 use_search = None
1510 if search_engine: 1510 ↛ 1549line 1510 didn't jump to line 1549 because the condition on line 1510 was always true
1511 try:
1512 # Create a new search object with these settings
1513 use_search = get_search(
1514 search_tool=search_engine,
1515 llm_instance=use_llm,
1516 username=username,
1517 settings_snapshot=settings_snapshot,
1518 )
1519 logger.info(
1520 f"Successfully created search engine: {search_engine}"
1521 )
1522 except Exception as e:
1523 logger.exception(
1524 f"Error creating search engine {search_engine}"
1525 )
1526 error_msg = str(e)
1527 # Surface configuration errors to user instead of silently continuing
1528 config_error_keywords = [
1529 "searxng",
1530 "instance_url",
1531 "api_key",
1532 "cannot connect",
1533 "connection",
1534 "timeout",
1535 "not configured",
1536 ]
1537 if any(
1538 keyword in error_msg.lower()
1539 for keyword in config_error_keywords
1540 ):
1541 # This is a configuration error the user can fix
1542 raise ValueError(
1543 f"Search Engine Configuration Error ({search_engine}): {error_msg}"
1544 ) from e
1545 # For other errors, re-raise to avoid silent failures
1546 raise
1548 # Set the progress callback in the system
1549 system = AdvancedSearchSystem(
1550 llm=use_llm, # type: ignore[arg-type]
1551 search=use_search, # type: ignore[arg-type]
1552 strategy_name=strategy,
1553 max_iterations=iterations,
1554 questions_per_iteration=questions_per_iteration,
1555 username=username,
1556 settings_snapshot=settings_snapshot,
1557 research_id=research_id,
1558 research_context=shared_research_context,
1559 )
1560 system.set_progress_callback(progress_callback)
1562 # Chat mode: set up LLM streaming callback for real-time response chunks
1563 streaming_enabled = False
1564 # chunks: server-side buffer so partial content survives termination
1565 # (the citation handler's local list is discarded on raise).
1566 # _bytes / _truncated cap the buffer to bound memory on pathologically
1567 # long answers; once capped we still forward to the frontend but stop
1568 # accumulating server-side.
1569 streaming_state = {
1570 "chunks_sent": 0,
1571 "chunks": [],
1572 "_bytes": 0,
1573 "_truncated": False,
1574 }
1575 chat_session_id = shared_research_context.get("chat_session_id")
1577 if chat_session_id:
1578 try:
1579 socket_service = _socket_emitter
1581 # Source resolver returns the strategy's currently-collected
1582 # source list. Late-bound so the streaming callback can
1583 # apply inline hyperlinks as the agent finishes adding to
1584 # all_links_of_system (sources may still be growing when
1585 # synthesis starts; reading via the closure picks up the
1586 # final list at chunk-emit time, not callback-build time).
1587 def _resolve_sources():
1588 if not hasattr(system, "all_links_of_system"): 1588 ↛ 1589line 1588 didn't jump to line 1589 because the condition on line 1588 was never true
1589 return []
1590 return list(system.all_links_of_system or [])
1592 # Build a formatter matching the user's report.citation_format
1593 # so live-display brackets ([[arxiv.org-1]] / [[arxiv-1]] /
1594 # [[1]] etc.) match what the final-save formatter will emit
1595 # — avoids a visible format-flip when handleResearchComplete
1596 # swaps in the DB-saved version.
1597 live_formatter = get_citation_formatter()
1599 stream_callback = _make_chat_stream_callback(
1600 research_id,
1601 streaming_state,
1602 socket_service,
1603 username,
1604 source_resolver=_resolve_sources,
1605 formatter=live_formatter,
1606 )
1608 # Hook into the citation handler's streaming
1609 if hasattr(system, "strategy") and hasattr(
1610 system.strategy, "citation_handler"
1611 ):
1612 handler = system.strategy.citation_handler
1613 if hasattr(handler, "set_stream_callback"): 1613 ↛ 1628line 1613 didn't jump to line 1628 because the condition on line 1613 was always true
1614 handler.set_stream_callback(stream_callback)
1615 streaming_enabled = True
1616 logger.info(
1617 f"Streaming enabled for chat {chat_session_id[:8]}..."
1618 )
1619 except Exception:
1620 # exception=True so the traceback is visible: streaming is
1621 # non-critical (research still completes without it), but a
1622 # silent warning would hide real bugs in the setup above.
1623 logger.opt(exception=True).warning(
1624 "Could not set up streaming (non-critical)"
1625 )
1627 # Helper to save chat message (closes over outer scope vars)
1628 def _maybe_save_chat_message(content):
1629 _chat_sid = shared_research_context.get("chat_session_id")
1630 if not _chat_sid:
1631 return
1632 try:
1633 _save_chat_message_and_context(
1634 _chat_sid,
1635 research_id,
1636 username,
1637 content,
1638 streaming_enabled,
1639 streaming_state,
1640 _socket_emitter,
1641 settings_snapshot=settings_snapshot,
1642 )
1643 except Exception:
1644 # Promoted from debug→warning: at debug level this is invisible
1645 # in production and the user sees "Research completed but no
1646 # report available" with no operator signal. Same rationale as
1647 # the accumulated_context handler in _save_chat_message_and_context.
1648 logger.opt(exception=True).warning(
1649 "Could not add message to chat session — assistant "
1650 "response NOT persisted; user will see 'no report available'"
1651 )
1652 # The DB write failed, so _save_chat_message_and_context never
1653 # emitted its is_final response_chunk. Emit one here so the
1654 # streaming UI clears its 'thinking' state instead of stalling.
1655 try:
1656 _socket_emitter.emit_to_subscribers(
1657 "response_chunk",
1658 research_id,
1659 {"chunk": "", "is_streaming": True, "is_final": True},
1660 owner=username,
1661 )
1662 except Exception:
1663 logger.opt(exception=True).debug(
1664 "Failed to emit final chunk after chat-persist error"
1665 )
1667 # Run the search
1668 progress_callback("Starting research process", 5, {"phase": "init"})
1670 try:
1671 results = system.analyze_topic(query)
1672 if mode == "quick":
1673 progress_callback(
1674 "Search complete, preparing to generate summary...",
1675 85,
1676 {"phase": "output_generation"},
1677 )
1678 else:
1679 progress_callback(
1680 "Search complete, generating output",
1681 80,
1682 {"phase": "output_generation"},
1683 )
1684 except Exception as search_error:
1685 # Better handling of specific search errors
1686 error_message = str(search_error)
1687 error_type = "unknown"
1689 # OpenAI-compatible runtime failures (LM Studio / vLLM / llama.cpp
1690 # server / OpenRouter / custom endpoint) -- rewrite to a message
1691 # that names the provider, base URL, and model (#3878).
1692 if model_provider in { 1692 ↛ 1704line 1692 didn't jump to line 1704 because the condition on line 1692 was never true
1693 "openai_endpoint",
1694 "lmstudio",
1695 "llamacpp",
1696 "openai",
1697 "openrouter",
1698 "orcarouter",
1699 "atlascloud",
1700 "google",
1701 "ionos",
1702 "xai",
1703 } and is_openai_compat_runtime_error(search_error):
1704 rewritten = friendly_openai_compatible_error(
1705 search_error,
1706 provider=model_provider,
1707 base_url=custom_endpoint,
1708 model=model,
1709 )
1710 raise RuntimeError(rewritten) from search_error
1712 # Extract error details for common issues
1713 if "status code: 503" in error_message:
1714 error_message = "Ollama AI service is unavailable (HTTP 503). Please check that Ollama is running properly on your system."
1715 error_type = "ollama_unavailable"
1716 elif "status code: 404" in error_message:
1717 error_message = "Ollama model not found (HTTP 404). Please check that you have pulled the required model."
1718 error_type = "model_not_found"
1719 elif "status code:" in error_message:
1720 # Extract the status code for other HTTP errors
1721 status_code = error_message.split("status code:")[1].strip()
1722 error_message = f"API request failed with status code {status_code}. Please check your configuration."
1723 error_type = "api_error"
1724 elif "connection" in error_message.lower():
1725 error_message = "Connection error. Please check that your LLM service (Ollama/API) is running and accessible."
1726 error_type = "connection_error"
1728 # Raise with improved error message
1729 raise RuntimeError(
1730 f"{error_message} (Error type: {error_type})"
1731 ) from search_error
1733 # Generate output based on mode
1734 if mode == "quick":
1735 # Quick Summary
1736 if results.get("findings") or results.get("formatted_findings"):
1737 raw_formatted_findings = results["formatted_findings"]
1739 # Check if formatted_findings contains an error message
1740 if isinstance(
1741 raw_formatted_findings, str
1742 ) and raw_formatted_findings.startswith("Error:"):
1743 logger.error(
1744 f"Detected error in formatted findings: {raw_formatted_findings[:100]}..."
1745 )
1747 # Determine error type for better user feedback
1748 error_type = "unknown"
1749 error_message = raw_formatted_findings.lower()
1751 if (
1752 "token limit" in error_message
1753 or "context length" in error_message
1754 ):
1755 error_type = "token_limit"
1756 # Log specific error type
1757 logger.warning(
1758 "Detected token limit error in synthesis"
1759 )
1761 # Update progress with specific error type
1762 progress_callback(
1763 "Synthesis hit token limits. Attempting fallback...",
1764 87,
1765 {
1766 "phase": "synthesis_error",
1767 "error_type": error_type,
1768 },
1769 )
1770 elif (
1771 "timeout" in error_message
1772 or "timed out" in error_message
1773 ):
1774 error_type = "timeout"
1775 logger.warning("Detected timeout error in synthesis")
1776 progress_callback(
1777 "Synthesis timed out. Attempting fallback...",
1778 87,
1779 {
1780 "phase": "synthesis_error",
1781 "error_type": error_type,
1782 },
1783 )
1784 elif "rate limit" in error_message:
1785 error_type = "rate_limit"
1786 logger.warning("Detected rate limit error in synthesis")
1787 progress_callback(
1788 "LLM rate limit reached. Attempting fallback...",
1789 87,
1790 {
1791 "phase": "synthesis_error",
1792 "error_type": error_type,
1793 },
1794 )
1795 elif (
1796 "connection" in error_message
1797 or "network" in error_message
1798 ):
1799 error_type = "connection"
1800 logger.warning("Detected connection error in synthesis")
1801 progress_callback(
1802 "Connection issue with LLM. Attempting fallback...",
1803 87,
1804 {
1805 "phase": "synthesis_error",
1806 "error_type": error_type,
1807 },
1808 )
1809 elif (
1810 "llm error" in error_message
1811 or "final answer synthesis fail" in error_message
1812 ):
1813 error_type = "llm_error"
1814 logger.warning(
1815 "Detected general LLM error in synthesis"
1816 )
1817 progress_callback(
1818 "LLM error during synthesis. Attempting fallback...",
1819 87,
1820 {
1821 "phase": "synthesis_error",
1822 "error_type": error_type,
1823 },
1824 )
1825 else:
1826 # Generic error
1827 logger.warning("Detected unknown error in synthesis")
1828 progress_callback(
1829 "Error during synthesis. Attempting fallback...",
1830 87,
1831 {
1832 "phase": "synthesis_error",
1833 "error_type": "unknown",
1834 },
1835 )
1837 # Extract synthesized content from findings if available
1838 synthesized_content = ""
1839 for finding in results.get("findings", []):
1840 if finding.get("phase") == "Final synthesis":
1841 synthesized_content = finding.get("content", "")
1842 break
1844 # Use synthesized content as fallback
1845 if (
1846 synthesized_content
1847 and not synthesized_content.startswith("Error:")
1848 ):
1849 logger.info(
1850 "Using existing synthesized content as fallback"
1851 )
1852 raw_formatted_findings = synthesized_content
1854 # Or use current_knowledge as another fallback
1855 elif results.get("current_knowledge"):
1856 logger.info("Using current_knowledge as fallback")
1857 raw_formatted_findings = results["current_knowledge"]
1859 # Or combine all finding contents as last resort
1860 elif results.get("findings"):
1861 logger.info("Combining all findings as fallback")
1862 # First try to use any findings that are not errors
1863 valid_findings = [
1864 f"## {finding.get('phase', 'Finding')}\n\n{finding.get('content', '')}"
1865 for finding in results.get("findings", [])
1866 if finding.get("content")
1867 and not finding.get("content", "").startswith(
1868 "Error:"
1869 )
1870 ]
1872 synthesis_error = raw_formatted_findings
1873 if valid_findings:
1874 raw_formatted_findings = (
1875 "# Research Results (Fallback Mode)\n\n"
1876 )
1877 raw_formatted_findings += "\n\n".join(
1878 valid_findings
1879 )
1880 raw_formatted_findings += (
1881 f"\n\n## Error Information\n{synthesis_error}"
1882 )
1883 else:
1884 # Last resort: use everything including errors
1885 raw_formatted_findings = (
1886 "# Research Results (Emergency Fallback)\n\n"
1887 )
1888 raw_formatted_findings += "The system encountered errors during final synthesis.\n\n"
1889 raw_formatted_findings += "\n\n".join(
1890 f"## {finding.get('phase', 'Finding')}\n\n{finding.get('content', '')}"
1891 for finding in results.get("findings", [])
1892 if finding.get("content")
1893 )
1895 progress_callback(
1896 f"Using fallback synthesis due to {error_type} error",
1897 88,
1898 {
1899 "phase": "synthesis_fallback",
1900 "error_type": error_type,
1901 },
1902 )
1904 logger.info(
1905 "Found formatted_findings of length: {}",
1906 len(str(raw_formatted_findings)),
1907 )
1909 try:
1910 # Check if we have an error in the findings and use enhanced error handling
1911 if isinstance(
1912 raw_formatted_findings, str
1913 ) and raw_formatted_findings.startswith("Error:"):
1914 logger.info(
1915 "Generating enhanced error report using ErrorReportGenerator"
1916 )
1918 # Generate comprehensive error report
1919 # ErrorReportGenerator does not use LLM (kept for compat)
1920 error_generator = ErrorReportGenerator()
1921 clean_markdown = error_generator.generate_error_report(
1922 error_message=raw_formatted_findings,
1923 query=query,
1924 partial_results=results,
1925 search_iterations=results.get("iterations", 0),
1926 research_id=research_id,
1927 )
1929 logger.info(
1930 "Generated enhanced error report with {} characters",
1931 len(clean_markdown),
1932 )
1933 else:
1934 # report_content stores the synthesized answer
1935 # only (see _extract_synthesized_answer for the
1936 # full rationale). Fall back to the formatted
1937 # blob only when neither Final synthesis nor
1938 # current_knowledge is populated — leaks
1939 # sources, but at least we save *something*.
1940 clean_markdown = (
1941 _extract_synthesized_answer(results)
1942 or raw_formatted_findings
1943 )
1945 # Pull sources from the search-system's accumulated link
1946 # buffer first — same source the detailed-report path
1947 # uses at the equivalent point below. Wrapper
1948 # strategies (e.g. EnhancedContextualFollowUpStrategy,
1949 # IterativeRefinementStrategy) delegate the actual
1950 # search to an inner strategy that populates
1951 # `self.all_links_of_system`, but they don't bubble
1952 # that list back into the result dict's `findings`. So
1953 # the legacy `findings[*].search_results` extraction
1954 # below stays empty for chat follow-ups, leaving the
1955 # citation formatter with no urls to hyperlink. Prefer
1956 # the system-level accumulator; fall back to the
1957 # legacy extraction so direct strategies that bypass
1958 # the system buffer still work.
1959 all_links = list(
1960 getattr(system, "all_links_of_system", None) or []
1961 )
1962 if not all_links:
1963 for finding in results.get("findings", []):
1964 search_results = finding.get("search_results", [])
1965 if search_results:
1966 try:
1967 links = extract_links_from_search_results(
1968 search_results
1969 )
1970 all_links.extend(links)
1971 except Exception:
1972 logger.exception(
1973 "Error processing search results/links"
1974 )
1976 logger.info(
1977 "Successfully converted to clean markdown of length: {}",
1978 len(clean_markdown),
1979 )
1981 # First send a progress update for generating the summary
1982 progress_callback(
1983 "Generating clean summary from research data...",
1984 90,
1985 {"phase": "output_generation"},
1986 )
1988 # Send progress update for saving report
1989 progress_callback(
1990 "Saving research report to database...",
1991 95,
1992 {"phase": "report_complete"},
1993 )
1995 # Format citations in the markdown content. The
1996 # split returns the answer-with-hyperlinks half
1997 # separately from the trailing sources section the
1998 # LLM may have emitted (which is discarded — sources
1999 # live in research_resources, the canonical store).
2000 # When no Sources section is found, fall back to
2001 # structured-source hyperlinking — never re-parse
2002 # concatenated formatter output downstream.
2003 formatter = get_citation_formatter()
2004 try:
2005 # Quick-mode (chat summary) input is raw LLM
2006 # output; the report generator never wraps it
2007 # in the appended-sources sentinel, so disable
2008 # sentinel trust. Otherwise an LLM that quotes
2009 # the marker verbatim would silently trip the
2010 # over-strip safety bypass on a path where the
2011 # bypass can only ever fire as a false-positive.
2012 answer_with_links, llm_sources, on_sentinel = (
2013 formatter.format_document_split(
2014 clean_markdown, trust_sentinel=False
2015 )
2016 )
2017 if not llm_sources:
2018 answer_with_links = (
2019 formatter.apply_inline_hyperlinks(
2020 clean_markdown, all_links
2021 )
2022 )
2023 # Safety check: a >50% strip on a long input
2024 # likely means the regex over-stripped on a
2025 # "Sources:" header inside the answer body.
2026 # Fall back to structured-source hyperlinking
2027 # on the full text. Min-length floor prevents
2028 # false-fires on legitimately short answers.
2029 # Skip the check when the boundary came from
2030 # the unique appended-sources sentinel emitted
2031 # by IntegratedReportGenerator — that boundary
2032 # is correct by construction (e.g. a 1380-source
2033 # detailed-mode run legitimately pushes the
2034 # answer/sources ratio below 50% without any
2035 # over-strip having occurred). Quick-mode
2036 # always passes trust_sentinel=False above, so
2037 # ``on_sentinel`` can never be True here.
2038 SAFETY_MIN_LEN = 800
2039 if ( 2039 ↛ 2046line 2039 didn't jump to line 2046 because the condition on line 2039 was never true
2040 not on_sentinel
2041 and llm_sources
2042 and len(clean_markdown) > SAFETY_MIN_LEN
2043 and len(answer_with_links)
2044 < len(clean_markdown) * 0.5
2045 ):
2046 logger.warning(
2047 "format_document_split appears to have "
2048 "over-stripped (answer={} chars, "
2049 "original={} chars) for research {}. "
2050 "Falling back to structured-source "
2051 "hyperlinking on full input.",
2052 len(answer_with_links),
2053 len(clean_markdown),
2054 research_id,
2055 )
2056 answer_with_links = (
2057 formatter.apply_inline_hyperlinks(
2058 clean_markdown, all_links
2059 )
2060 )
2061 except Exception:
2062 # Hyperlinking is quality-of-life, not a hard
2063 # requirement. If anything blows up, save the
2064 # raw LLM text rather than fail the research.
2065 logger.exception(
2066 "Citation formatter failed for research {}; saving raw answer",
2067 research_id,
2068 )
2069 answer_with_links = clean_markdown
2071 # report_content stores ONLY the synthesized answer.
2072 # The legacy "answer + ## Sources + ## Research
2073 # Metrics" view is reconstructed at render time by
2074 # report_assembly_service.assemble_full_report.
2075 full_report_content = answer_with_links
2077 # Save report FIRST, then sources:
2078 # a chat read between commits sees a report with no
2079 # sources (assembler renders just the answer) — better
2080 # failure mode than partial assembly with sources but
2081 # no answer body.
2082 from ...storage import get_report_storage
2084 with get_user_db_session(username) as db_session:
2085 storage = get_report_storage(session=db_session)
2087 # Prepare metadata
2088 metadata = {
2089 "iterations": results["iterations"],
2090 "generated_at": datetime.now(UTC).isoformat(),
2091 }
2093 # Save report using storage abstraction
2094 success = storage.save_report(
2095 research_id=research_id,
2096 content=full_report_content,
2097 metadata=metadata,
2098 username=username,
2099 )
2101 if not success:
2102 raise RuntimeError("Failed to save research report") # noqa: TRY301 — triggers research failure handling in outer except
2104 # Save sources to database (non-fatal - report
2105 # already saved; sources missing is recoverable
2106 # because the assembler omits empty Sources blocks)
2107 try:
2108 from .research_sources_service import (
2109 ResearchSourcesService,
2110 )
2112 sources_service = ResearchSourcesService()
2113 if all_links:
2114 logger.info(
2115 f"Quick summary: Saving {len(all_links)} sources to database"
2116 )
2117 sources_saved = (
2118 sources_service.save_research_sources(
2119 research_id=research_id,
2120 sources=all_links,
2121 username=username,
2122 )
2123 )
2124 logger.info(
2125 f"Quick summary: Saved {sources_saved} sources for research {research_id}"
2126 )
2127 except Exception:
2128 logger.exception(
2129 f"Failed to save sources for research {research_id} (continuing with report save)"
2130 )
2132 logger.info(f"Report saved for research_id: {research_id}")
2134 # Skip export to additional formats - we're storing in database only
2136 # Update research status in database
2137 completed_at = datetime.now(UTC).isoformat()
2139 with get_user_db_session(username) as db_session:
2140 research = (
2141 db_session.query(ResearchHistory)
2142 .filter_by(id=research_id)
2143 .first()
2144 )
2146 # Preserve existing metadata and update with new values
2147 metadata = _parse_research_metadata(
2148 research.research_meta
2149 )
2151 metadata.update(
2152 {
2153 "iterations": results["iterations"],
2154 "generated_at": datetime.now(UTC).isoformat(),
2155 }
2156 )
2158 # Use the helper function for consistent duration calculation
2159 duration_seconds = calculate_duration(
2160 research.created_at, completed_at
2161 )
2163 research.status = ResearchStatus.COMPLETED
2164 research.completed_at = completed_at
2165 research.duration_seconds = duration_seconds
2166 # report_content was already saved above via the report
2167 # storage abstraction; this block only updates
2168 # status/metadata. report_path is not used in the
2169 # encrypted-database version.
2171 # Generate headline and topics only for news searches
2172 if (
2173 metadata.get("is_news_search")
2174 or metadata.get("search_type") == "news_analysis"
2175 ):
2176 try:
2177 from ...news.utils.headline_generator import (
2178 generate_headline,
2179 )
2180 from ...news.utils.topic_generator import (
2181 generate_topics,
2182 )
2183 from ...search_system import (
2184 ensure_snapshot_username,
2185 )
2187 # Scope the snapshot to this run's owner so
2188 # the LLM PEP resolves a per-user provider
2189 # and honors local-only policy, mirroring
2190 # the scheduler's _store_research_result fix
2191 # (background.py). Without it, headline/topic
2192 # generation silently no-ops for cloud-
2193 # provider users here (fails closed, not a
2194 # security hole, but inconsistent).
2195 headline_topic_snapshot = (
2196 ensure_snapshot_username(
2197 settings_snapshot, username
2198 )
2199 )
2201 # Get the report content from database for better headline/topic generation
2202 report_content = ""
2203 try:
2204 research = (
2205 db_session.query(ResearchHistory)
2206 .filter_by(id=research_id)
2207 .first()
2208 )
2209 if research and research.report_content: 2209 ↛ 2215line 2209 didn't jump to line 2215 because the condition on line 2209 was always true
2210 report_content = research.report_content
2211 logger.info(
2212 f"Retrieved {len(report_content)} chars from database for headline generation"
2213 )
2214 else:
2215 logger.warning(
2216 f"No report content found in database for research_id: {research_id}"
2217 )
2218 except Exception:
2219 logger.warning(
2220 "Could not retrieve report content from database"
2221 )
2223 # Generate headline
2224 logger.info(
2225 f"Generating headline for query: {query[:100]}"
2226 )
2227 headline = generate_headline(
2228 query,
2229 report_content,
2230 settings_snapshot=headline_topic_snapshot,
2231 )
2232 metadata["generated_headline"] = headline
2234 # Generate topics
2235 logger.info(
2236 f"Generating topics with category: {metadata.get('category', 'News')}"
2237 )
2238 topics = generate_topics(
2239 query=query,
2240 findings=report_content,
2241 category=metadata.get("category", "News"),
2242 max_topics=6,
2243 settings_snapshot=headline_topic_snapshot,
2244 )
2245 metadata["generated_topics"] = topics
2247 logger.info(f"Generated headline: {headline}")
2248 logger.info(f"Generated topics: {topics}")
2250 except Exception:
2251 logger.warning(
2252 "Could not generate headline/topics"
2253 )
2255 research.research_meta = metadata
2257 db_session.commit()
2258 logger.info(
2259 f"Database commit completed for research_id: {research_id}"
2260 )
2262 # Update subscription if this was triggered by a subscription
2263 if metadata.get("subscription_id"):
2264 try:
2265 from ...news.subscription_runner import (
2266 advance_refresh_schedule_by_id,
2267 )
2269 subscription_id = metadata["subscription_id"]
2270 if advance_refresh_schedule_by_id( 2270 ↛ 2282line 2270 didn't jump to line 2282
2271 db_session, subscription_id
2272 ):
2273 db_session.commit()
2274 logger.info(
2275 f"Updated subscription {subscription_id} refresh times"
2276 )
2277 except Exception:
2278 logger.warning(
2279 "Could not update subscription refresh time"
2280 )
2282 logger.info(
2283 f"Database updated successfully for research_id: {research_id}"
2284 )
2286 _maybe_save_chat_message(full_report_content)
2288 # Send the final completion message
2289 progress_callback(
2290 "Research completed successfully",
2291 100,
2292 {"phase": "complete"},
2293 )
2295 # Clean up resources
2296 logger.info(
2297 "Cleaning up resources for research_id: {}", research_id
2298 )
2299 cleanup_research_resources(
2300 research_id, username, user_password=user_password
2301 )
2302 logger.info(
2303 "Resources cleaned up for research_id: {}", research_id
2304 )
2306 except Exception as inner_e:
2307 logger.exception("Error during quick summary generation")
2308 raise RuntimeError(
2309 f"Error generating quick summary: {inner_e!s}"
2310 )
2311 else:
2312 raise RuntimeError( # noqa: TRY301 — triggers research failure handling in outer except
2313 "No research findings were generated. Please try again."
2314 )
2315 else:
2316 # Full Report
2317 progress_callback(
2318 "Generating detailed report...",
2319 _DETAILED_REPORT_PROGRESS_START,
2320 {"phase": "report_generation"},
2321 )
2323 # Extract the search system from the results if available
2324 search_system = results.get("search_system", None)
2326 # Wrapper that maps report generator's 0-100% to the configured
2327 # detailed-mode range and relays cancellation checks through the
2328 # outer progress_callback
2329 _report_range = (
2330 _DETAILED_REPORT_PROGRESS_END - _DETAILED_REPORT_PROGRESS_START
2331 )
2333 def report_progress_callback(message, progress_percent, metadata):
2334 if progress_percent is not None:
2335 adjusted = (
2336 _DETAILED_REPORT_PROGRESS_START
2337 + (progress_percent / 100) * _report_range
2338 )
2339 else:
2340 adjusted = progress_percent
2341 progress_callback(message, adjusted, metadata)
2343 # Pass the existing search system to maintain citation indices
2344 report_generator = IntegratedReportGenerator(
2345 search_system=search_system,
2346 settings_snapshot=settings_snapshot,
2347 )
2348 final_report = report_generator.generate_report(
2349 results, query, progress_callback=report_progress_callback
2350 )
2352 progress_callback(
2353 "Report generation complete",
2354 _DETAILED_REPORT_PROGRESS_END,
2355 {"phase": "report_complete"},
2356 )
2358 # Format citations and split off the trailing Sources
2359 # section. Save only the answer half — sources are
2360 # persisted structurally to research_resources.
2361 all_links = (
2362 getattr(search_system, "all_links_of_system", None) or []
2363 )
2364 formatter = get_citation_formatter()
2365 try:
2366 # Detailed-mode input comes from
2367 # ``IntegratedReportGenerator._format_final_report``,
2368 # which wraps the appended ``## Sources`` block in the
2369 # unique sentinel. Trust the sentinel so a legitimate
2370 # large-tail run (e.g. 1380 sources) doesn't trip the
2371 # over-strip safety check on the right answer half.
2372 answer_with_links, llm_sources, on_sentinel = (
2373 formatter.format_document_split(final_report["content"])
2374 )
2375 if not llm_sources: 2375 ↛ 2376line 2375 didn't jump to line 2376 because the condition on line 2375 was never true
2376 answer_with_links = formatter.apply_inline_hyperlinks(
2377 final_report["content"], all_links
2378 )
2379 # Skip the over-strip safety check when the boundary
2380 # came from the unique appended-sources sentinel emitted
2381 # by IntegratedReportGenerator — that boundary is
2382 # correct by construction (e.g. a 1380-source detailed
2383 # run legitimately pushes the answer/sources ratio
2384 # below 50% without any over-strip having occurred).
2385 SAFETY_MIN_LEN = 800
2386 if (
2387 not on_sentinel
2388 and llm_sources
2389 and len(final_report["content"]) > SAFETY_MIN_LEN
2390 and len(answer_with_links)
2391 < len(final_report["content"]) * 0.5
2392 ):
2393 logger.warning(
2394 "format_document_split appears to have over-stripped "
2395 "(answer={} chars, original={} chars) for research {}.",
2396 len(answer_with_links),
2397 len(final_report["content"]),
2398 research_id,
2399 )
2400 answer_with_links = formatter.apply_inline_hyperlinks(
2401 final_report["content"], all_links
2402 )
2403 except Exception:
2404 logger.exception(
2405 "Citation formatter failed for research {}; saving raw answer",
2406 research_id,
2407 )
2408 answer_with_links = final_report["content"]
2409 formatted_content = answer_with_links
2411 # Save report FIRST, sources after.
2412 # See quick-summary path for rationale.
2413 from ...storage import get_report_storage
2415 with get_user_db_session(username) as db_session:
2416 storage = get_report_storage(session=db_session)
2418 # Update metadata. Include generated_at like the quick path so
2419 # the detailed file backup's _metadata.json has parity with
2420 # quick-mode (the detailed path previously omitted it).
2421 metadata = final_report["metadata"]
2422 metadata["iterations"] = results["iterations"]
2423 metadata["generated_at"] = datetime.now(UTC).isoformat()
2425 # Save the report through the storage abstraction, exactly as
2426 # the quick-summary branch does, so detailed reports also honor
2427 # the report.enable_file_backup setting. This previously did a
2428 # raw ORM write of report_content that silently skipped the
2429 # file backup, so a user who enabled file backup got it for
2430 # quick research but never for detailed research.
2431 success = storage.save_report(
2432 research_id=research_id,
2433 content=formatted_content,
2434 metadata=metadata,
2435 username=username,
2436 )
2438 if not success: 2438 ↛ 2439line 2438 didn't jump to line 2439 because the condition on line 2438 was never true
2439 raise RuntimeError("Failed to save research report") # noqa: TRY301 — triggers research failure handling in outer except
2441 logger.info(
2442 f"Report saved to database for research_id: {research_id}"
2443 )
2445 # Save sources AFTER report (non-fatal; assembler omits
2446 # empty Sources blocks if this fails).
2447 try:
2448 from .research_sources_service import ResearchSourcesService
2450 sources_service = ResearchSourcesService()
2451 if all_links:
2452 logger.info(f"Saving {len(all_links)} sources to database")
2453 sources_saved = sources_service.save_research_sources(
2454 research_id=research_id,
2455 sources=all_links,
2456 username=username,
2457 )
2458 logger.info(
2459 f"Saved {sources_saved} sources for research {research_id}"
2460 )
2461 except Exception:
2462 logger.exception(
2463 f"Failed to save sources for research {research_id} (continuing)"
2464 )
2466 # Update research status in database
2467 completed_at = datetime.now(UTC).isoformat()
2469 with get_user_db_session(username) as db_session:
2470 research = (
2471 db_session.query(ResearchHistory)
2472 .filter_by(id=research_id)
2473 .first()
2474 )
2476 # Preserve existing metadata and merge with report metadata
2477 metadata = _parse_research_metadata(research.research_meta)
2479 metadata.update(final_report["metadata"])
2480 metadata["iterations"] = results["iterations"]
2482 # Use the helper function for consistent duration calculation
2483 duration_seconds = calculate_duration(
2484 research.created_at, completed_at
2485 )
2487 research.status = ResearchStatus.COMPLETED
2488 research.completed_at = completed_at
2489 research.duration_seconds = duration_seconds
2490 # report_content was already saved above via the report storage
2491 # abstraction; this block only updates status/metadata.
2492 # report_path is not used in the encrypted-database version.
2494 # Generate headline and topics only for news searches
2495 if (
2496 metadata.get("is_news_search")
2497 or metadata.get("search_type") == "news_analysis"
2498 ):
2499 try:
2500 from ...news.utils.headline_generator import (
2501 generate_headline, # type: ignore[no-redef]
2502 )
2503 from ...news.utils.topic_generator import (
2504 generate_topics, # type: ignore[no-redef]
2505 )
2506 from ...search_system import (
2507 ensure_snapshot_username,
2508 )
2510 # Scope the snapshot to this run's owner so the LLM
2511 # PEP resolves a per-user provider and honors
2512 # local-only policy. Mirrors the quick-summary
2513 # branch above and the scheduler's
2514 # _store_research_result fix (background.py).
2515 headline_topic_snapshot = ensure_snapshot_username(
2516 settings_snapshot, username
2517 )
2519 # Get the report content from database for better headline/topic generation
2520 report_content = ""
2521 try:
2522 research = (
2523 db_session.query(ResearchHistory)
2524 .filter_by(id=research_id)
2525 .first()
2526 )
2527 if research and research.report_content: 2527 ↛ 2530line 2527 didn't jump to line 2530 because the condition on line 2527 was always true
2528 report_content = research.report_content
2529 else:
2530 logger.warning(
2531 f"No report content found in database for research_id: {research_id}"
2532 )
2533 except Exception:
2534 logger.warning(
2535 "Could not retrieve report content from database"
2536 )
2538 # Generate headline
2539 headline = generate_headline(
2540 query,
2541 report_content,
2542 settings_snapshot=headline_topic_snapshot,
2543 )
2544 metadata["generated_headline"] = headline
2546 # Generate topics
2547 topics = generate_topics(
2548 query=query,
2549 findings=report_content,
2550 category=metadata.get("category", "News"),
2551 max_topics=6,
2552 settings_snapshot=headline_topic_snapshot,
2553 )
2554 metadata["generated_topics"] = topics
2556 logger.info(f"Generated headline: {headline}")
2557 logger.info(f"Generated topics: {topics}")
2559 except Exception:
2560 logger.warning("Could not generate headline/topics")
2562 research.research_meta = metadata
2564 db_session.commit()
2566 # Update subscription if this was triggered by a subscription
2567 if metadata.get("subscription_id"): 2567 ↛ 2568line 2567 didn't jump to line 2568 because the condition on line 2567 was never true
2568 try:
2569 from ...news.subscription_runner import (
2570 advance_refresh_schedule_by_id,
2571 )
2573 subscription_id = metadata["subscription_id"]
2574 if advance_refresh_schedule_by_id(
2575 db_session, subscription_id
2576 ):
2577 db_session.commit()
2578 logger.info(
2579 f"Updated subscription {subscription_id} refresh times"
2580 )
2581 except Exception:
2582 logger.warning(
2583 "Could not update subscription refresh time"
2584 )
2586 _maybe_save_chat_message(formatted_content)
2588 progress_callback(
2589 "Research completed successfully",
2590 100,
2591 {"phase": "complete"},
2592 )
2594 # Clean up resources
2595 cleanup_research_resources(
2596 research_id, username, user_password=user_password
2597 )
2599 except ResearchTerminatedException:
2600 logger.info(f"Research {research_id} terminated by user")
2601 # Fallback path: when termination was raised from the streaming
2602 # callback (mid-stream interrupt), progress_callback hasn't run
2603 # since the flag was set, so handle_termination() was NOT called
2604 # and the partial row hasn't been persisted yet. The helper is
2605 # idempotent via streaming_state["_persisted"]; if the in-callback
2606 # path already ran, both calls are no-ops.
2607 _save_partial_chat_message_on_terminate(
2608 shared_research_context.get("chat_session_id"),
2609 research_id,
2610 username,
2611 "".join(streaming_state.get("chunks", [])),
2612 truncated=streaming_state.get("_truncated", False),
2613 streaming_state=streaming_state,
2614 )
2615 # Ensure the SUSPENDED status update + cleanup runs even when the
2616 # exception was raised mid-stream. The in-callback termination paths
2617 # set "_termination_handled"; only run here when they did NOT, so a
2618 # single termination doesn't queue two SUSPENDED updates, emit two
2619 # final socket messages, and (in test mode) sleep twice.
2620 if not streaming_state.get("_termination_handled"):
2621 try:
2622 handle_termination(research_id, username)
2623 except Exception:
2624 logger.opt(exception=True).debug(
2625 "handle_termination in except block failed"
2626 )
2628 except Exception as e:
2629 # Handle error
2630 error_message = f"Research failed: {e!s}"
2631 logger.exception(error_message)
2633 try:
2634 # Check for common Ollama error patterns in the exception and provide more user-friendly errors
2635 user_friendly_error = str(e)
2636 error_context = {}
2638 # Egress policy refusal (two-axis rule, or a fail-closed audit_error
2639 # from the worker chokepoint) — surface the reason and the override
2640 # instead of a generic "unexpected error".
2641 from ...security.egress.policy import PolicyDeniedError
2643 if isinstance(e, PolicyDeniedError): 2643 ↛ 2644line 2643 didn't jump to line 2644 because the condition on line 2643 was never true
2644 _reason = getattr(e.decision, "reason", "") or "egress_denied"
2645 if _reason == "audit_error":
2646 user_friendly_error = (
2647 "Egress policy could not verify this run, so it was "
2648 "refused (fail-closed)."
2649 )
2650 elif _reason.startswith("sensitive_to_exposing"):
2651 # The two-axis rule specifically: a sensitive source would
2652 # reach an exposing sink.
2653 user_friendly_error = (
2654 "Egress policy refused this run: a sensitive source "
2655 f"would reach an exposing destination ({_reason})."
2656 )
2657 else:
2658 # Other egress PEPs (scope mismatch, require-local
2659 # inference, etc.) — no "sensitive source" is necessarily
2660 # involved, so don't claim one.
2661 user_friendly_error = (
2662 f"Egress policy refused this run ({_reason})."
2663 )
2664 error_context = {
2665 "solution": "Set Egress Scope to 'Unprotected' to override, or use local inference / non-sensitive sources."
2666 }
2667 elif "Error type: ollama_unavailable" in user_friendly_error:
2668 user_friendly_error = "Ollama AI service is unavailable. Please check that Ollama is running properly on your system."
2669 error_context = {
2670 "solution": "Start Ollama with 'ollama serve' or check if it's installed correctly."
2671 }
2672 elif "Error type: model_not_found" in user_friendly_error:
2673 user_friendly_error = "Required Ollama model not found. Please pull the model first."
2674 error_context = {
2675 "solution": "Run 'ollama pull mistral' to download the required model."
2676 }
2677 elif "Error type: connection_error" in user_friendly_error:
2678 user_friendly_error = "Connection error with LLM service. Please check that your AI service is running."
2679 error_context = {
2680 "solution": "Ensure Ollama or your API service is running and accessible."
2681 }
2682 elif "Error type: api_error" in user_friendly_error:
2683 user_friendly_error = (
2684 "The language model API rejected the request."
2685 )
2686 error_context = {
2687 "solution": "Check API configuration and credentials."
2688 }
2689 # OpenAI-compatible runtime tokens (#3878). The friendly message
2690 # built by friendly_openai_compatible_error() names the provider,
2691 # base URL and model and appends the raw provider error (only
2692 # credential-scrubbed). The base URL can be a server-level endpoint
2693 # (settings_snapshot bakes in LDR_* env overrides) and the appended
2694 # detail can carry internal hosts/paths, so it must not reach the
2695 # client (CWE-209). Replace it with a safe category message; full
2696 # detail stays in the logs above and the actionable hint is in
2697 # ``solution``.
2698 elif "Error type: openai_connection_refused" in user_friendly_error: 2698 ↛ 2699line 2698 didn't jump to line 2699 because the condition on line 2698 was never true
2699 user_friendly_error = (
2700 "Could not connect to the configured LLM server."
2701 )
2702 error_context = {
2703 "solution": "Start your LLM server (LM Studio / vLLM / llama.cpp server) and verify the base URL in Settings -> LLM Providers."
2704 }
2705 elif "Error type: openai_timeout" in user_friendly_error:
2706 user_friendly_error = "The configured LLM server timed out."
2707 error_context = {
2708 "solution": "The server is reachable but slow -- it may be loading a model. Retry, or increase the request timeout."
2709 }
2710 elif "Error type: openai_auth" in user_friendly_error: 2710 ↛ 2711line 2710 didn't jump to line 2711 because the condition on line 2710 was never true
2711 user_friendly_error = (
2712 "Authentication with the configured LLM provider failed."
2713 )
2714 error_context = {
2715 "solution": "Set or correct the API key for this provider in Settings -> LLM Providers. Local servers usually accept any non-empty key."
2716 }
2717 elif "Error type: openai_permission_denied" in user_friendly_error: 2717 ↛ 2718line 2717 didn't jump to line 2718 because the condition on line 2717 was never true
2718 user_friendly_error = (
2719 "The configured LLM provider denied access to the model."
2720 )
2721 error_context = {
2722 "solution": "Your API key is valid but lacks access to this model. Pick a model your account/server is permitted to use."
2723 }
2724 elif "Error type: openai_model_not_found" in user_friendly_error: 2724 ↛ 2725line 2724 didn't jump to line 2725 because the condition on line 2724 was never true
2725 user_friendly_error = (
2726 "The configured model was not found on the LLM server."
2727 )
2728 error_context = {
2729 "solution": "The model id is not loaded on this server. Pick a currently-loaded model in the provider's UI/config."
2730 }
2731 elif "Error type: openai_bad_request" in user_friendly_error: 2731 ↛ 2732line 2731 didn't jump to line 2732 because the condition on line 2731 was never true
2732 user_friendly_error = "The LLM server rejected the request."
2733 error_context = {
2734 "solution": "The server rejected the request. Check the model id and any provider-specific parameters."
2735 }
2736 elif "Error type: openai_unknown" in user_friendly_error: 2736 ↛ 2737line 2736 didn't jump to line 2737 because the condition on line 2736 was never true
2737 user_friendly_error = (
2738 "The configured LLM provider returned an error."
2739 )
2740 error_context = {
2741 "solution": "Check the provider's logs for the full error and verify the base URL / model id."
2742 }
2743 elif "Error type: openai_rate_limit" in user_friendly_error: 2743 ↛ 2744line 2743 didn't jump to line 2744 because the condition on line 2743 was never true
2744 user_friendly_error = (
2745 "The LLM provider rate-limited the request."
2746 )
2747 error_context = {
2748 "solution": "The provider rate-limited the request. Wait a moment and retry, or enable LLM Rate Limiting in Settings."
2749 }
2750 elif "LLM Configuration Error:" in user_friendly_error:
2751 # The raw text here is str(e) from LLM setup and can carry
2752 # server-level endpoints/paths (settings_snapshot includes
2753 # LDR_* env overrides), so it must not be surfaced to the client
2754 # (CWE-209) -- it is captured in the logs above. Keep only the
2755 # safe category + actionable hint.
2756 user_friendly_error = (
2757 "There was a problem with the LLM configuration."
2758 )
2759 error_context = {
2760 "solution": "Review your LLM model settings (or, on a shared server, contact your administrator) and ensure they are correct."
2761 }
2762 elif "Search Engine Configuration Error" in user_friendly_error:
2763 # Same rationale as the LLM config branch above.
2764 user_friendly_error = (
2765 "There was a problem with the search engine configuration."
2766 )
2767 error_context = {
2768 "solution": "Review your search engine settings (or, on a shared server, contact your administrator) and ensure they are correct."
2769 }
2770 else:
2771 # Unrecognized exception. The raw str(e) here is server-side
2772 # internal detail (file paths, DB/driver text, Python tracebacks)
2773 # with no curated, user-actionable form, so it must not be
2774 # surfaced to the client (CWE-209) — it is already captured by
2775 # the logger.exception above. The branches above classify known
2776 # errors and replace the message with a safe category string +
2777 # hint; this branch handles everything that wasn't classified.
2778 user_friendly_error = (
2779 "Research failed due to an unexpected error. Contact your "
2780 "administrator or check the server logs for details."
2781 )
2783 # Generate enhanced error report for failed research
2784 enhanced_report_content = None
2785 try:
2786 # Get partial results if they exist
2787 partial_results = results if "results" in locals() else None
2788 search_iterations = (
2789 results.get("iterations", 0) if partial_results else 0
2790 )
2792 # Generate comprehensive error report
2793 # ErrorReportGenerator does not use LLM (kept for compat)
2794 error_generator = ErrorReportGenerator()
2795 enhanced_report_content = error_generator.generate_error_report(
2796 # Use the sanitized user_friendly_error (curated for known
2797 # errors, generic for unexpected ones) instead of raw {e!s}:
2798 # this report is persisted and retrievable via the report
2799 # routes, so embedding raw exception text would leak server
2800 # internals (CWE-209). Full detail stays in the logs.
2801 error_message=f"Research failed: {user_friendly_error}",
2802 query=query,
2803 partial_results=partial_results,
2804 search_iterations=search_iterations,
2805 research_id=research_id,
2806 )
2808 logger.info(
2809 "Generated enhanced error report for failed research (length: {})",
2810 len(enhanced_report_content),
2811 )
2813 # Save enhanced error report to encrypted database
2814 try:
2815 # username already available from function scope (line 281)
2816 if username: 2816 ↛ 2838line 2816 didn't jump to line 2838 because the condition on line 2816 was always true
2817 from ...storage import get_report_storage
2819 with get_user_db_session(username) as db_session:
2820 storage = get_report_storage(session=db_session)
2821 success = storage.save_report(
2822 research_id=research_id,
2823 content=enhanced_report_content,
2824 metadata={"error_report": True},
2825 username=username,
2826 )
2827 if success:
2828 logger.info(
2829 "Saved enhanced error report to encrypted database for research {}",
2830 research_id,
2831 )
2832 else:
2833 logger.warning(
2834 "Failed to save enhanced error report to database for research {}",
2835 research_id,
2836 )
2837 else:
2838 logger.warning(
2839 "Cannot save error report: username not available"
2840 )
2842 except Exception as report_error:
2843 logger.exception(
2844 "Failed to save enhanced error report: {}", report_error
2845 )
2847 except Exception as error_gen_error:
2848 logger.exception(
2849 "Failed to generate enhanced error report: {}",
2850 error_gen_error,
2851 )
2852 enhanced_report_content = None
2854 # Get existing metadata from database first
2855 existing_metadata = {}
2856 try:
2857 # username already available from function scope (line 281)
2858 if username: 2858 ↛ 2871line 2858 didn't jump to line 2871 because the condition on line 2858 was always true
2859 with get_user_db_session(username) as db_session:
2860 research = (
2861 db_session.query(ResearchHistory)
2862 .filter_by(id=research_id)
2863 .first()
2864 )
2865 if research and research.research_meta:
2866 existing_metadata = dict(research.research_meta)
2867 except Exception:
2868 logger.exception("Failed to get existing metadata")
2870 # Update metadata with more context about the error while preserving existing values
2871 metadata = existing_metadata
2872 metadata.update({"phase": "error", "error": user_friendly_error})
2873 if error_context:
2874 metadata.update(error_context)
2875 if enhanced_report_content: 2875 ↛ 2879line 2875 didn't jump to line 2879 because the condition on line 2875 was always true
2876 metadata["has_enhanced_report"] = True
2878 # If we still have an active research record, update its log
2879 if is_research_active(research_id):
2880 progress_callback(user_friendly_error, None, metadata)
2882 # If termination was requested, mark as suspended instead of failed
2883 status = (
2884 ResearchStatus.SUSPENDED
2885 if is_termination_requested(research_id)
2886 else ResearchStatus.FAILED
2887 )
2888 message = (
2889 "Research was terminated by user"
2890 if status == ResearchStatus.SUSPENDED
2891 else user_friendly_error
2892 )
2894 # A subscription-triggered run that FAILED must be made due again.
2895 # run_subscription_now / the overdue sweep advance next_refresh at
2896 # spawn time (to avoid the scheduler double-running an in-flight
2897 # subscription), but the completion-time advance only runs on
2898 # success. Without this reset a failed run would leave next_refresh
2899 # pushed a full interval out, silently hiding the subscription from
2900 # the scheduler. Skipped for SUSPENDED (user-terminated) runs.
2901 if (
2902 status == ResearchStatus.FAILED
2903 and username
2904 and metadata.get("subscription_id")
2905 ):
2906 try:
2907 from ...news.subscription_runner import (
2908 mark_subscription_due_by_id,
2909 )
2911 with get_user_db_session(username) as sub_db:
2912 if mark_subscription_due_by_id( 2912 ↛ 2927line 2912 didn't jump to line 2927
2913 sub_db, metadata["subscription_id"]
2914 ):
2915 sub_db.commit()
2916 logger.info(
2917 f"Reset subscription {metadata['subscription_id']} "
2918 "to due after failed run"
2919 )
2920 except Exception:
2921 logger.warning(
2922 "Could not reset subscription refresh time after "
2923 "failed run"
2924 )
2926 # Calculate duration up to termination point - using UTC consistently
2927 now = datetime.now(UTC)
2928 completed_at = now.isoformat()
2930 # NOTE: Database updates from threads are handled by queue processor
2931 # The queue_processor.queue_error_update() method is already being used below
2932 # to safely update the database from the main thread
2934 # Queue the error update to be processed in main thread
2935 # Using the queue processor v2 system
2936 from ..queue.processor_v2 import queue_processor
2938 if username: 2938 ↛ 2952line 2938 didn't jump to line 2952 because the condition on line 2938 was always true
2939 queue_processor.queue_error_update(
2940 username=username,
2941 research_id=research_id,
2942 status=status,
2943 error_message=message,
2944 metadata=metadata,
2945 completed_at=completed_at,
2946 report_path=None,
2947 )
2948 logger.info(
2949 f"Queued error update for research {research_id} with status '{status}'"
2950 )
2951 else:
2952 logger.error(
2953 f"Cannot queue error update for research {research_id} - no username provided. "
2954 f"Status: '{status}', Message: {message}"
2955 )
2957 try:
2958 _sio_emit(
2959 "research_progress",
2960 research_id,
2961 {"status": status, "error": message},
2962 owner=username,
2963 )
2964 except Exception:
2965 logger.exception("Failed to emit error via socket")
2967 # Add error message to chat session if applicable.
2968 # Read chat_session_id from shared_research_context (the canonical
2969 # source — kept consistent with the six other reads in this file).
2970 chat_session_id = shared_research_context.get("chat_session_id")
2971 if chat_session_id and username:
2972 try:
2973 from ...chat.service import ChatService
2975 chat_service = ChatService(username)
2976 # allow_archived=True: same multi-tab race rationale as
2977 # the completion / stop-and-partial paths — if the user
2978 # archived (or deleted) the session between research
2979 # start and failure, the error message would otherwise
2980 # be silently dropped by the active-only insert guard
2981 # and the user sees nothing in the chat.
2982 chat_service.add_message(
2983 session_id=chat_session_id,
2984 role="assistant",
2985 content=f"Sorry, the research failed: {message}",
2986 message_type="response",
2987 allow_archived=True,
2988 )
2989 except Exception:
2990 # Promoted from debug → warning to match the success-path
2991 # rationale: if this write fails the user never sees the
2992 # error and operators get no signal at debug-off level.
2993 logger.opt(exception=True).warning(
2994 "Could not add error message to chat session"
2995 )
2997 except Exception:
2998 logger.exception("Error in error handler")
3000 # Clean up resources. This is the error path, so report FAILED on
3001 # the final socket message rather than a spurious "completed".
3002 cleanup_research_resources(
3003 research_id,
3004 username,
3005 user_password=user_password,
3006 final_status=ResearchStatus.FAILED,
3007 )
3009 finally:
3010 # RESOURCE CLEANUP: Close search engine HTTP sessions.
3011 #
3012 # Search engines (created via get_search()) may hold HTTP connection
3013 # pools. Currently only SemanticScholarSearchEngine creates a
3014 # persistent SafeSession; other engines use stateless safe_get()/
3015 # safe_post() utility functions. However, BaseSearchEngine.close()
3016 # is safe to call on any engine — it checks for a 'session'
3017 # attribute and is fully idempotent (SemanticScholar sets
3018 # self.session = None after close).
3019 #
3020 # Neither @thread_cleanup nor cleanup_research_resources() close
3021 # the search engine — @thread_cleanup only handles database sessions
3022 # and context cleanup, and cleanup_research_resources() only handles
3023 # status updates, notifications, and tracking dict removal.
3024 #
3025 # Without this explicit close, search engine sessions rely on
3026 # Python's non-deterministic garbage collection (__del__) for
3027 # cleanup, which can cause file descriptor exhaustion under
3028 # sustained load.
3029 from ...utilities.resource_utils import safe_close
3031 if "use_search" in locals():
3032 safe_close(use_search, "research search engine")
3033 # Close search system (cascades to strategy thread pools).
3034 # See AdvancedSearchSystem.close() for details.
3035 if "system" in locals():
3036 safe_close(system, "research system")
3037 # Close the LLM instance created for model/provider overrides.
3038 # system.close() does NOT close the LLM passed to it via system.model,
3039 # so we must close it explicitly here.
3040 if "use_llm" in locals():
3041 safe_close(use_llm, "research LLM")
3044def cleanup_research_resources(
3045 research_id,
3046 username=None,
3047 user_password=None,
3048 final_status=ResearchStatus.COMPLETED,
3049 preserve_termination_flag=False,
3050):
3051 """
3052 Clean up resources for a completed research.
3054 Args:
3055 research_id: The ID of the research
3056 username: The username for database access (required for thread context)
3057 final_status: The terminal status to report on the final socket
3058 message. Callers that end a research for a reason other than
3059 normal completion MUST pass the real status (e.g. SUSPENDED on
3060 user termination, FAILED on error) so the final ``progress``
3061 event matches reality. Defaulting this to COMPLETED — and
3062 previously hard-coding it — caused the stop/error paths to emit
3063 a spurious "completed" signal to subscribers.
3064 preserve_termination_flag: Keep a cancellation signal visible after
3065 removing the active registry entry. Used only by the HTTP cancel
3066 handoff; the worker's own termination cleanup uses the default and
3067 removes the flag.
3068 """
3069 from ..research_state import cleanup_research
3071 logger.info("Cleaning up resources for research {}", research_id)
3073 # For testing: Add a small delay to simulate research taking time
3074 # This helps test concurrent research limits
3075 from ...settings.env_registry import is_test_mode
3077 if is_test_mode():
3078 import time
3080 logger.info(
3081 f"Test mode: Adding 5 second delay before cleanup for {research_id}"
3082 )
3083 time.sleep(5)
3085 # The terminal status to report on the final socket message. This comes
3086 # from the caller (which knows why the research ended) rather than a
3087 # hard-coded COMPLETED, so termination (SUSPENDED) and error (FAILED)
3088 # paths no longer emit a false "completed" signal to subscribers.
3089 current_status = final_status
3091 # NOTE: Queue processor already handles database updates from the main thread
3092 # The notify_research_completed() method is called at the end of this function
3093 # which safely updates the database status
3095 # Notify queue processor that research completed
3096 # This uses processor_v2 which handles database updates in the main thread
3097 # avoiding the Flask request context issues that occur in background threads
3098 from ..queue.processor_v2 import queue_processor
3100 if username:
3101 queue_processor.notify_research_completed(
3102 username, research_id, user_password=user_password
3103 )
3104 logger.info(
3105 f"Notified queue processor of completion for research {research_id} (user: {username})"
3106 )
3107 else:
3108 logger.warning(
3109 f"Cannot notify completion for research {research_id} - no username provided"
3110 )
3112 # Remove from active research and termination flags atomically
3113 if preserve_termination_flag:
3114 cleanup_research(research_id, preserve_termination_flag=True)
3115 else:
3116 cleanup_research(research_id)
3118 # Clean up throttle state for this research
3119 with _last_emit_lock:
3120 _last_emit_times.pop(research_id, None)
3122 # Send a final message to subscribers
3123 try:
3124 # Send a final message to any remaining subscribers with explicit status
3125 # Use the proper status message based on database status
3126 if current_status in (
3127 ResearchStatus.SUSPENDED,
3128 ResearchStatus.FAILED,
3129 ):
3130 final_message = {
3131 "status": current_status,
3132 "message": f"Research was {current_status}",
3133 "progress": 0, # For suspended research, show 0% not 100%
3134 }
3135 else:
3136 final_message = {
3137 "status": ResearchStatus.COMPLETED,
3138 "message": "Research process has ended and resources have been cleaned up",
3139 "progress": 100,
3140 }
3142 logger.info(
3143 "Sending final {} socket message for research {}",
3144 current_status,
3145 research_id,
3146 )
3148 _sio_emit(
3149 "research_progress", research_id, final_message, owner=username
3150 )
3152 # Clean up socket subscriptions for this research
3153 _sio_remove(research_id, username)
3155 except Exception:
3156 logger.exception("Error sending final cleanup message")
3159def handle_termination(
3160 research_id, username=None, *, preserve_termination_flag=False
3161):
3162 """
3163 Handle the termination of a research process.
3165 Args:
3166 research_id: The ID of the research
3167 username: The username for database access (required for thread context)
3168 """
3169 logger.info(f"Handling termination for research {research_id}")
3171 # Queue the status update to be processed in the main thread
3172 # This avoids Flask request context errors in background threads
3173 try:
3174 from ..queue.processor_v2 import queue_processor
3176 now = datetime.now(UTC)
3177 completed_at = now.isoformat()
3179 # Queue the suspension update
3180 queue_processor.queue_error_update(
3181 username=username,
3182 research_id=research_id,
3183 status=ResearchStatus.SUSPENDED,
3184 error_message="Research was terminated by user",
3185 metadata={"terminated_at": completed_at},
3186 completed_at=completed_at,
3187 report_path=None,
3188 )
3190 logger.info(f"Queued suspension update for research {research_id}")
3191 except Exception:
3192 logger.exception(
3193 f"Error queueing termination update for research {research_id}"
3194 )
3196 # Clean up resources (this already handles things properly).
3197 # Pass SUSPENDED so the final socket message reports the real terminal
3198 # status — not a spurious "completed" — to chat/progress subscribers.
3199 cleanup_kwargs = {"final_status": ResearchStatus.SUSPENDED}
3200 if preserve_termination_flag:
3201 cleanup_kwargs["preserve_termination_flag"] = True
3202 cleanup_research_resources(research_id, username, **cleanup_kwargs)
3205def cancel_research(research_id, username):
3206 """
3207 Cancel/terminate a research process using ORM.
3209 Args:
3210 research_id: The ID of the research to cancel
3211 username: The username of the user cancelling the research
3213 Returns:
3214 bool: True if the research was found and cancelled, False otherwise
3215 """
3216 try:
3217 from ..research_state import is_research_active, set_termination_flag
3219 # Ownership gate. research ids are per-user, but the global termination
3220 # registry (_termination_flags / _active_research, keyed by id alone) is
3221 # shared across all users. Without this check any signed-in user could
3222 # terminate ANOTHER user's active research by id: the active branch
3223 # below sets the flag and calls handle_termination before the only
3224 # pre-existing ownership query (which lives in the not-active branch).
3225 # Confirm the caller owns this research in their OWN database first.
3226 with get_user_db_session(username) as _owner_check_session:
3227 owns_research = (
3228 _owner_check_session.query(ResearchHistory.id)
3229 .filter_by(id=research_id)
3230 .first()
3231 is not None
3232 )
3233 if not owns_research:
3234 logger.info(
3235 f"Research {research_id} not found for this user; "
3236 "refusing to cancel."
3237 )
3238 return False
3240 # Set termination flag
3241 set_termination_flag(research_id)
3243 # Check if the research is active
3244 if is_research_active(research_id):
3245 # Update the database and release the active slot immediately, but
3246 # keep the signal set until the still-running worker sees it. The
3247 # worker's own handle_termination call performs final flag cleanup.
3248 handle_termination(
3249 research_id,
3250 username,
3251 preserve_termination_flag=True,
3252 )
3253 return True
3254 try:
3255 with get_user_db_session(username) as db_session:
3256 research = (
3257 db_session.query(ResearchHistory)
3258 .filter_by(id=research_id)
3259 .first()
3260 )
3261 if not research: 3261 ↛ 3262line 3261 didn't jump to line 3262 because the condition on line 3261 was never true
3262 logger.info(f"Research {research_id} not found in database")
3263 return False
3265 # Check if already in a terminal state
3266 if research.status in (
3267 ResearchStatus.COMPLETED,
3268 ResearchStatus.SUSPENDED,
3269 ResearchStatus.FAILED,
3270 ResearchStatus.ERROR,
3271 ):
3272 logger.info(
3273 f"Research {research_id} already in terminal state: {research.status}"
3274 )
3275 return True # Consider this a success since it's already stopped
3277 # If it exists but isn't in active_research, still update status
3278 if research.status == ResearchStatus.QUEUED:
3279 from ...database.models import QueuedResearch
3281 claimed_queue_row = (
3282 db_session.query(QueuedResearch.id)
3283 .filter(
3284 QueuedResearch.research_id == research_id,
3285 QueuedResearch.is_processing.is_(True),
3286 )
3287 .exists()
3288 )
3289 suspended = (
3290 db_session.query(ResearchHistory)
3291 .filter(
3292 ResearchHistory.id == research_id,
3293 ResearchHistory.status == ResearchStatus.QUEUED,
3294 ~claimed_queue_row,
3295 )
3296 .update(
3297 {ResearchHistory.status: ResearchStatus.SUSPENDED},
3298 synchronize_session=False,
3299 )
3300 )
3301 if suspended:
3302 from ..queue.lifecycle_cleanup import (
3303 cleanup_queued_research_state,
3304 )
3306 cleanup_queued_research_state(db_session, [research_id])
3307 else:
3308 research.status = ResearchStatus.SUSPENDED
3309 db_session.commit()
3310 logger.info(f"Successfully suspended research {research_id}")
3311 except Exception:
3312 logger.exception(
3313 f"Error accessing database for research {research_id}"
3314 )
3315 return False
3317 return True
3318 except Exception:
3319 logger.exception(
3320 f"Unexpected error in cancel_research for {research_id}"
3321 )
3322 return False