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