Coverage for src/local_deep_research/web/routers/chat.py: 83%
503 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2FastAPI routes for the chat API.
4Provides endpoints for:
5- Chat page rendering
6- Session management (create, list, get, archive, delete)
7- Message management (send, list)
8- Research triggering from chat
10Ported from the pre-FastAPI Flask blueprint ``chat/routes.py``: Flask
11``session``/``@login_required`` became ``Depends(require_auth)``,
12``request.get_json`` became ``await request.json()``, ``jsonify(...), N``
13became ``JSONResponse(..., status_code=N)``, and the blocking DB/service
14work runs in the threadpool via ``run_db_sync`` so it never blocks the
15uvicorn event loop. The framework-agnostic helpers and all concurrency /
16cleanup logic are preserved verbatim.
17"""
19import threading
20import unicodedata
21import uuid
22from datetime import datetime, timedelta, UTC
24from fastapi import APIRouter, Depends, Request
25from fastapi.responses import JSONResponse
26from loguru import logger
27from sqlalchemy import update as sa_update
28from sqlalchemy.exc import IntegrityError, SQLAlchemyError
30from ...chat.service import (
31 ArchiveBlockedError,
32 AttemptInProgress,
33 AttemptNotFound,
34 ChatService,
35 ChatSessionNotFound,
36 DB_EXCEPTIONS,
37)
38from ...chat.context import ChatContextManager
39from ...constants import ResearchStatus
40from ...database.models import (
41 ChatMessage,
42 ChatSession,
43 ChatSessionStatus,
44 ResearchHistory,
45 UserActiveResearch,
46)
47from ...database.session_context import get_user_db_session
48from ...exceptions import DuplicateResearchError, SystemAtCapacityError
49from ...settings.manager import SettingsManager
50from ..auth.password_utils import get_user_password, resolve_user_password
51from ..routes.globals import (
52 cleanup_research,
53 is_research_thread_alive,
54 reclaim_stale_user_active_research,
55)
56from ..services.research_service import (
57 clamp_user_max_concurrent,
58 run_research_process,
59 start_research_process,
60)
61from ..dependencies.auth import require_auth
62from ..dependencies.rate_limit import limiter, _get_client_ip
63from ..dependencies.template_helpers import render_template
64from ..dependencies.threadpool import run_db_sync
65from typing import Annotated
67router = APIRouter(tags=["chat"])
69# Valid status values for sessions (built from the enum so a typo never
70# silently passes validation; the literal "all" sentinel widens the list
71# filter to every status without bypassing the whitelist).
72VALID_UPDATE_STATUSES = {
73 ChatSessionStatus.ACTIVE.value,
74 ChatSessionStatus.ARCHIVED.value,
75}
76VALID_LIST_STATUSES = {*(s.value for s in ChatSessionStatus), "all"}
78# Input length limits
79MAX_QUERY_LENGTH = 10_000
80MAX_TITLE_LENGTH = 500
81MAX_MESSAGE_LENGTH = 10_000
82# Hard cap on `offset` to prevent server-side DoS: get_session_messages
83# fetches `limit + offset` rows from BOTH chat_messages and chat_progress_steps
84# tables, so unbounded offset means unbounded SQL LIMIT. With cursor-based
85# pagination (`before_created_at`) as the recommended path, offset above a few
86# pages is not a normal access pattern.
87MAX_OFFSET = 1_000
89# Wider exception tuple used by HTTP route handlers (subsumes
90# service.DB_EXCEPTIONS plus the attribute/type errors that can escape
91# request-shape coercion code). DB_EXCEPTIONS itself is single-sourced
92# from chat.service so the two never drift.
93ROUTE_EXCEPTIONS = (
94 ValueError,
95 RuntimeError,
96 SQLAlchemyError,
97 AttributeError,
98 TypeError,
99)
102def _bad_body() -> JSONResponse:
103 """Build the 400 response for a non-JSON / non-object request body.
105 Returns a FRESH ``JSONResponse`` each call — a shared module-level
106 instance can't be reused safely across concurrent requests. This is the
107 FastAPI equivalent of the old ``@require_json_body(error_format="success",
108 ...)`` decorator.
110 It does NOT reject form-encoded posts, and an earlier version of this
111 docstring wrongly claimed it "hardens CSRF" by doing so. Flask's
112 ``get_json(silent=True)`` returned ``None`` for a non-JSON Content-Type,
113 which had that side effect; ``request.json()`` parses the body regardless
114 of Content-Type, so a form-encoded POST carrying a JSON object body is
115 accepted here. Verified: ``Content-Type: text/plain`` with a JSON object
116 creates a session.
118 That is safe, but for a different reason: ``CSRFMiddleware``
119 (web/dependencies/csrf.py) is the fail-closed choke point and rejects
120 these requests with 403 before they reach any handler. Recording the
121 correction because a reviewer could otherwise rely on a guarantee this
122 function does not provide.
123 """
124 return JSONResponse(
125 {"success": False, "error": "Request body must be a JSON object"},
126 status_code=400,
127 )
130def _chat_user_key(request: Request) -> str:
131 """Per-user rate-limit key (default slowapi key is per-IP).
133 Without this, users behind a shared NAT/proxy share one bucket and can
134 DoS each other for legitimate chat use. Falls back to the client IP for
135 any unauthenticated request that somehow reaches a limited route.
136 """
137 return request.session.get("username") or _get_client_ip(request)
140async def _json_object_body(request: Request):
141 """Parse the request body as a JSON object, or return a 400 response.
143 Returns the parsed dict on success, or a ``JSONResponse`` (400) the
144 caller must return as-is. Mirrors the old ``@require_json_body`` gate.
145 """
146 try:
147 data = await request.json()
148 except Exception:
149 return _bad_body()
150 if not isinstance(data, dict):
151 return _bad_body()
152 return data
155def _load_settings(username):
156 """Load all settings for a user.
158 ``bypass_cache=True`` matches the call pattern in
159 ``research_routes.start_research``: a setting changed via the UI
160 moments before the user sends a chat message must take effect on the
161 next research, not be served from a stale cache.
162 """
163 with get_user_db_session(username) as db:
164 snapshot = SettingsManager(db_session=db).get_all_settings(
165 bypass_cache=True
166 )
167 # Inject the username so snapshot-driven consumers (get_llm, and ADAPTIVE
168 # egress-scope resolution for a per-user retriever primary) can resolve it.
169 # Without it, chat follow-up summarization builds get_llm() with
170 # username=None -> a private-retriever primary is invisible -> scope falls
171 # open to BOTH and a cloud LLM summarizes private-corpus turns. Mirrors
172 # AdvancedSearchSystem.ensure_snapshot_username. Ported from #5539, which
173 # landed on the Flask chat/routes.py this router replaced.
174 from ...search_system import ensure_snapshot_username
176 return ensure_snapshot_username(snapshot, username)
179def _parse_int_param(
180 value: str | None,
181 default: int,
182 min_val: int = 0,
183 max_val: int | None = None,
184) -> int:
185 """Safely parse an integer parameter with bounds checking."""
186 try:
187 result = int(value) if value is not None else default
188 if result < min_val:
189 return min_val
190 if max_val is not None and result > max_val:
191 return max_val
192 return result
193 except (ValueError, TypeError):
194 return default
197_INVISIBLE_UNICODE_CATEGORIES = {"Cf", "Zl", "Zp"}
200def _validate_title(title) -> tuple[str, int] | None:
201 """Return (error_message, http_status) when *title* is invalid, else None.
203 A title is invalid when it is not a non-empty string or exceeds
204 ``MAX_TITLE_LENGTH``. Callers that allow ``None`` (e.g. create_session
205 where omitting the title is fine) should short-circuit on ``None``
206 before calling this helper.
208 Strips Unicode format / line-separator characters (``Cf``/``Zl``/``Zp``,
209 including zero-width spaces U+200B-U+200D and BOM U+FEFF) before the
210 emptiness check so an "invisible" title like 500 zero-width chars is
211 rejected instead of saving a session that looks blank in the UI.
212 """
213 if not isinstance(title, str):
214 return ("Title cannot be empty", 400)
215 visible = "".join(
216 c
217 for c in title
218 if unicodedata.category(c) not in _INVISIBLE_UNICODE_CATEGORIES
219 )
220 if not visible.strip():
221 return ("Title cannot be empty", 400)
222 if len(title) > MAX_TITLE_LENGTH:
223 return (
224 f"Title too long (max {MAX_TITLE_LENGTH} characters)",
225 400,
226 )
227 return None
230def _cleanup_chat_send_rows(
231 username, research_id, message_id, session_id, reason: str
232) -> None:
233 """Undo the user-message + research_history rows committed by send_message
234 when ``start_research_process`` rejects the spawn.
236 Used by both the ``DuplicateResearchError`` (409) and
237 ``SystemAtCapacityError`` (429) paths. Failure to clean up is logged at
238 ERROR level so orphan rows + inflated message_count are visible to ops.
239 """
240 try:
241 with get_user_db_session(username) as cleanup_db:
242 cleanup_db.query(ResearchHistory).filter_by(id=research_id).delete()
243 cleanup_db.query(ChatMessage).filter_by(id=message_id).delete()
244 # Drop the per-user-cap tracking row too (the spawn never
245 # started, so no live thread owns it). research_id is a fresh
246 # UUID, so this only ever matches our own just-inserted row.
247 cleanup_db.query(UserActiveResearch).filter_by(
248 username=username, research_id=research_id
249 ).delete()
250 cleanup_db.execute(
251 sa_update(ChatSession)
252 .where(ChatSession.id == session_id)
253 .values(message_count=ChatSession.message_count - 1)
254 )
255 cleanup_db.commit()
256 except DB_EXCEPTIONS:
257 logger.exception(
258 f"Cleanup after {reason} chat-send rejection FAILED "
259 f"for research {research_id[:8]}... in chat "
260 f"{session_id[:8]}...; orphan rows + inflated "
261 f"message_count may persist until next sweep."
262 )
265# ============================================================================
266# Page Routes
267# ============================================================================
270@router.get("/chat/", include_in_schema=False)
271@router.get("/chat/{session_id}", include_in_schema=False)
272def chat_page(
273 request: Request,
274 username: Annotated[str, Depends(require_auth)],
275 session_id: str | None = None,
276):
277 """Render the chat page.
279 Args:
280 session_id: Optional session ID to load existing session
281 """
282 return render_template(
283 request, "pages/chat.html", {"session_id": session_id}
284 )
287# ============================================================================
288# Session API Routes
289# ============================================================================
292@router.post("/api/chat/sessions")
293# Per-user keying (default is per-IP). Without this, users behind a shared
294# NAT/proxy share one bucket and can DoS each other for legitimate chat use.
295@limiter.limit("20 per minute", key_func=_chat_user_key)
296async def create_session(
297 request: Request, username: Annotated[str, Depends(require_auth)]
298):
299 """Create a new chat session."""
300 data = await _json_object_body(request)
301 if isinstance(data, JSONResponse):
302 return data
304 def _impl():
305 try:
306 # Validate input lengths
307 initial_query = data.get("initial_query")
308 title = data.get("title")
310 # Reject non-string initial_query early so len() / downstream
311 # string ops don't raise TypeError → 500.
312 if initial_query is not None and not isinstance(initial_query, str):
313 return JSONResponse(
314 {
315 "success": False,
316 "error": "initial_query must be a string",
317 },
318 status_code=400,
319 )
321 if initial_query and len(initial_query) > MAX_QUERY_LENGTH: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 return JSONResponse(
323 {
324 "success": False,
325 "error": f"Initial query too long (max {MAX_QUERY_LENGTH} characters)",
326 },
327 status_code=400,
328 )
330 if title is not None:
331 err = _validate_title(title)
332 if err is not None:
333 msg, status = err
334 return JSONResponse(
335 {"success": False, "error": msg}, status_code=status
336 )
338 settings_snapshot = _load_settings(username)
340 service = ChatService(username)
341 session_id = service.create_session(
342 initial_query=initial_query,
343 title=title,
344 settings_snapshot=settings_snapshot,
345 )
347 # Get the created session
348 try:
349 session_data = service.get_session(session_id)
350 except ChatSessionNotFound:
351 # Session was just created in this request — getting "not
352 # found" here means a delete-race or storage failure.
353 logger.exception(
354 "Just-created chat session missing on read-back"
355 )
356 return JSONResponse(
357 {
358 "success": False,
359 "error": "Failed to load created session",
360 },
361 status_code=500,
362 )
364 return JSONResponse(
365 {
366 "success": True,
367 "session_id": session_id,
368 "session": session_data,
369 }
370 )
372 except ROUTE_EXCEPTIONS:
373 logger.exception("Error creating chat session")
374 return JSONResponse(
375 {
376 "success": False,
377 "error": "Failed to create chat session",
378 },
379 status_code=500,
380 )
382 return await run_db_sync(_impl)
385@router.post("/api/chat/sessions/{session_id}/generate-title")
386# Per-user keying + lower limit than create_session because each call is a
387# real LLM round-trip on a server-paid endpoint (vs create_session which is
388# zero-LLM DB work). Without per-user keying, shared-IP users share the bucket.
389@limiter.limit("10 per minute", key_func=_chat_user_key)
390async def generate_session_title(
391 request: Request,
392 session_id: str,
393 username: Annotated[str, Depends(require_auth)],
394):
395 """Regenerate the session title using the configured LLM.
397 Fire-and-forget endpoint the frontend calls asynchronously right after
398 creating a session, so the synchronous POST /api/chat/sessions response
399 isn't blocked on an LLM round-trip.
400 """
401 data = await _json_object_body(request)
402 if isinstance(data, JSONResponse): 402 ↛ 403line 402 didn't jump to line 403 because the condition on line 402 was never true
403 return data
405 def _impl():
406 try:
407 query = data.get("query")
409 if not query:
410 return JSONResponse(
411 {"success": False, "error": "query is required"},
412 status_code=400,
413 )
414 if not isinstance(query, str) or len(query) > MAX_QUERY_LENGTH: 414 ↛ 415line 414 didn't jump to line 415 because the condition on line 414 was never true
415 return JSONResponse(
416 {
417 "success": False,
418 "error": f"query must be a string up to {MAX_QUERY_LENGTH} chars",
419 },
420 status_code=400,
421 )
423 service = ChatService(username)
424 try:
425 service.get_session(session_id)
426 except ChatSessionNotFound:
427 return JSONResponse(
428 {"success": False, "error": "Session not found"},
429 status_code=404,
430 )
432 settings_snapshot = _load_settings(username)
433 new_title = service.regenerate_title_with_llm(
434 session_id, query, settings_snapshot
435 )
436 if not new_title: 436 ↛ 440line 436 didn't jump to line 440 because the condition on line 436 was always true
437 # LLM disabled, or LLM call failed — keep existing fallback
438 return JSONResponse({"success": False, "title": None})
440 return JSONResponse({"success": True, "title": new_title})
442 except ROUTE_EXCEPTIONS:
443 logger.exception("Error regenerating chat title")
444 return JSONResponse(
445 {"success": False, "error": "Failed to regenerate title"},
446 status_code=500,
447 )
449 return await run_db_sync(_impl)
452@router.get("/api/chat/sessions")
453async def list_sessions(
454 request: Request, username: Annotated[str, Depends(require_auth)]
455):
456 """List chat sessions for the current user.
458 Query params: status (active/archived/deleted/all, default active),
459 limit (default 20), offset (default 0).
460 """
462 def _impl():
463 try:
464 status = request.query_params.get(
465 "status", ChatSessionStatus.ACTIVE.value
466 )
467 # Validate status parameter
468 if status not in VALID_LIST_STATUSES:
469 status = ChatSessionStatus.ACTIVE.value
470 limit = _parse_int_param(
471 request.query_params.get("limit"), 20, min_val=1, max_val=100
472 )
473 offset = _parse_int_param(
474 request.query_params.get("offset"),
475 0,
476 min_val=0,
477 max_val=MAX_OFFSET,
478 )
480 service = ChatService(username)
481 sessions = service.list_sessions(
482 status=status, limit=limit, offset=offset
483 )
485 return JSONResponse({"success": True, "sessions": sessions})
487 except ROUTE_EXCEPTIONS:
488 logger.exception("Error listing chat sessions")
489 return JSONResponse(
490 {"success": False, "error": "Failed to list chat sessions"},
491 status_code=500,
492 )
494 return await run_db_sync(_impl)
497@router.get("/api/chat/sessions/{session_id}")
498async def get_session(
499 request: Request,
500 session_id: str,
501 username: Annotated[str, Depends(require_auth)],
502):
503 """Get a specific chat session."""
505 def _impl():
506 try:
507 service = ChatService(username)
508 try:
509 session_data = service.get_session(session_id)
510 except ChatSessionNotFound:
511 return JSONResponse(
512 {"success": False, "error": "Session not found"},
513 status_code=404,
514 )
516 return JSONResponse({"success": True, "session": session_data})
518 except ROUTE_EXCEPTIONS:
519 logger.exception("Error getting chat session")
520 return JSONResponse(
521 {"success": False, "error": "Failed to get chat session"},
522 status_code=500,
523 )
525 return await run_db_sync(_impl)
528@router.patch("/api/chat/sessions/{session_id}")
529# Per-user keying, like the other state-changing chat routes.
530@limiter.limit("30 per minute", key_func=_chat_user_key)
531async def update_session(
532 request: Request,
533 session_id: str,
534 username: Annotated[str, Depends(require_auth)],
535):
536 """Update a chat session (title, archive, delete)."""
537 data = await _json_object_body(request)
538 if isinstance(data, JSONResponse): 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true
539 return data
541 def _impl():
542 try:
543 # Require at least one valid field
544 valid_fields = {"title", "status"}
545 if not any(field in data for field in valid_fields):
546 return JSONResponse(
547 {
548 "success": False,
549 "error": "Request must include at least one of: title, status",
550 },
551 status_code=400,
552 )
554 service = ChatService(username)
556 try:
557 service.get_session(session_id)
558 except ChatSessionNotFound:
559 return JSONResponse(
560 {"success": False, "error": "Session not found"},
561 status_code=404,
562 )
564 ops_ok = True
566 if "title" in data:
567 title = data["title"]
568 err = _validate_title(title)
569 if err is not None:
570 msg, status = err
571 return JSONResponse(
572 {"success": False, "error": msg}, status_code=status
573 )
574 ops_ok = (
575 service.update_session_title(session_id, title) and ops_ok
576 )
578 if "status" in data:
579 new_status = data["status"]
580 if new_status not in VALID_UPDATE_STATUSES:
581 return JSONResponse(
582 {"success": False, "error": "Invalid status value"},
583 status_code=400,
584 )
585 if new_status == ChatSessionStatus.ACTIVE.value:
586 ops_ok = service.reactivate_session(session_id) and ops_ok
587 elif new_status == ChatSessionStatus.ARCHIVED.value: 587 ↛ 601line 587 didn't jump to line 601 because the condition on line 587 was always true
588 try:
589 ops_ok = service.archive_session(session_id) and ops_ok
590 except ArchiveBlockedError:
591 # Symmetric with send-to-archived (also 409). Hard-coded
592 # message — never echo str(exc) here (CWE-209).
593 return JSONResponse(
594 {
595 "success": False,
596 "error": "Cannot archive: research in_progress. Stop it first.",
597 },
598 status_code=409,
599 )
601 try:
602 session_data = service.get_session(session_id)
603 except ChatSessionNotFound:
604 # Session was deleted by a concurrent request between the
605 # update above and this read-back. Treat as 404.
606 return JSONResponse(
607 {"success": False, "error": "Session not found"},
608 status_code=404,
609 )
611 if not ops_ok: 611 ↛ 615line 611 didn't jump to line 615 because the condition on line 611 was never true
612 # The read-back above succeeded, so the session still exists,
613 # yet an update reported failure — a DB write error swallowed
614 # into a False return. Surface it instead of reporting success.
615 logger.error(
616 f"Chat session update failed at DB layer for "
617 f"{session_id[:8]}..."
618 )
619 return JSONResponse(
620 {"success": False, "error": "Failed to update session"},
621 status_code=500,
622 )
624 return JSONResponse({"success": True, "session": session_data})
626 except ROUTE_EXCEPTIONS:
627 logger.exception("Error updating chat session")
628 return JSONResponse(
629 {"success": False, "error": "Failed to update chat session"},
630 status_code=500,
631 )
633 return await run_db_sync(_impl)
636@router.delete("/api/chat/sessions/{session_id}")
637# Per-user keying, like the other state-changing chat routes.
638@limiter.limit("30 per minute", key_func=_chat_user_key)
639async def delete_session(
640 request: Request,
641 session_id: str,
642 username: Annotated[str, Depends(require_auth)],
643):
644 """Delete a chat session permanently."""
646 def _impl():
647 try:
648 service = ChatService(username)
649 success = service.delete_session(session_id)
651 if not success:
652 return JSONResponse(
653 {"success": False, "error": "Session not found"},
654 status_code=404,
655 )
657 return JSONResponse({"success": True})
659 except ROUTE_EXCEPTIONS:
660 logger.exception("Error deleting chat session")
661 return JSONResponse(
662 {"success": False, "error": "Failed to delete chat session"},
663 status_code=500,
664 )
666 return await run_db_sync(_impl)
669# ============================================================================
670# Message API Routes
671# ============================================================================
674@router.get("/api/chat/sessions/{session_id}/messages")
675async def get_messages(
676 request: Request,
677 session_id: str,
678 username: Annotated[str, Depends(require_auth)],
679):
680 """Get messages for a chat session (cursor-paginated, ASC)."""
682 def _impl():
683 try:
684 limit = _parse_int_param(
685 request.query_params.get("limit"), 50, min_val=1, max_val=100
686 )
687 offset = _parse_int_param(
688 request.query_params.get("offset"),
689 0,
690 min_val=0,
691 max_val=MAX_OFFSET,
692 )
693 before_created_at = (
694 request.query_params.get("before_created_at") or None
695 )
696 before_id = request.query_params.get("before_id") or None
698 service = ChatService(username)
700 try:
701 service.get_session(session_id)
702 except ChatSessionNotFound:
703 return JSONResponse(
704 {"success": False, "error": "Session not found"},
705 status_code=404,
706 )
708 # Fetch one extra row so we can tell the client whether more
709 # older entries exist without a second round-trip.
710 peek_limit = limit + 1
711 page = service.get_session_messages(
712 session_id,
713 limit=peek_limit,
714 offset=offset,
715 before_created_at=before_created_at,
716 before_id=before_id,
717 )
718 has_more = len(page) > limit
719 messages = page[-limit:] if has_more else page
721 # The client (chat.js loadSession) restores the live "thinking"
722 # indicator from this field. O(1) via the partial-unique index
723 # ux_research_history_chat_session_in_progress.
724 in_progress_research_id = service.get_in_progress_research_id(
725 session_id
726 )
728 return JSONResponse(
729 {
730 "success": True,
731 "messages": messages,
732 "has_more": has_more,
733 "in_progress_research_id": in_progress_research_id,
734 }
735 )
737 except ROUTE_EXCEPTIONS:
738 logger.exception("Error getting chat messages")
739 return JSONResponse(
740 {"success": False, "error": "Failed to get chat messages"},
741 status_code=500,
742 )
744 return await run_db_sync(_impl)
747@router.post("/api/chat/sessions/{session_id}/messages")
748# Per-user keying. send_message launches a full research run, so this is the
749# heaviest chat endpoint; shared-IP users sharing the bucket would lock each
750# other out.
751@limiter.limit("10 per minute", key_func=_chat_user_key)
752async def send_message(
753 request: Request,
754 session_id: str,
755 username: Annotated[str, Depends(require_auth)],
756):
757 """Send a message in a chat session, optionally triggering research.
759 Research mode is always "quick" in chat (intentional v1 scope).
760 """
761 data = await _json_object_body(request)
762 if isinstance(data, JSONResponse):
763 return data
765 def _impl():
766 try:
767 if not data or not data.get("content"):
768 return JSONResponse(
769 {"success": False, "error": "Message content is required"},
770 status_code=400,
771 )
773 # Reject non-string content before .strip() raises AttributeError
774 # → 500. Mirrors the isinstance guard in _validate_title.
775 if not isinstance(data["content"], str): 775 ↛ 776line 775 didn't jump to line 776 because the condition on line 775 was never true
776 return JSONResponse(
777 {"success": False, "error": "content must be a string"},
778 status_code=400,
779 )
781 content = data["content"].strip()
783 # Reject whitespace-only content
784 if not content:
785 return JSONResponse(
786 {"success": False, "error": "Message content is required"},
787 status_code=400,
788 )
790 if len(content) > MAX_MESSAGE_LENGTH:
791 return JSONResponse(
792 {
793 "success": False,
794 "error": f"Message too long (max {MAX_MESSAGE_LENGTH} characters)",
795 },
796 status_code=400,
797 )
799 # trigger_research is a strict boolean. Reject non-bool values
800 # with a 400 instead of silently coercing them to True: a client
801 # sending {"trigger_research": "no"} or {"trigger_research": 0}
802 # intends to SUPPRESS research, and coercing the truthy-string to
803 # True would launch an unwanted (paid) research run against their
804 # intent (#4427).
805 raw = data.get("trigger_research", True)
806 if not isinstance(raw, bool):
807 return JSONResponse(
808 {
809 "success": False,
810 "error": "trigger_research must be a boolean",
811 },
812 status_code=400,
813 )
814 trigger_research = raw
816 service = ChatService(username)
818 # Verify session exists (informational fast-fail; the
819 # UPDATE...RETURNING inside insert_message_in_db is the
820 # authoritative check that survives a delete-race).
821 try:
822 session_data = service.get_session(session_id)
823 except ChatSessionNotFound:
824 return JSONResponse(
825 {"success": False, "error": "Session not found"},
826 status_code=404,
827 )
829 # Reject sends to non-active sessions. Archived/deleted sessions
830 # are intentionally read-only — users must reactivate first.
831 if session_data.get("status") != ChatSessionStatus.ACTIVE.value:
832 return JSONResponse(
833 {
834 "success": False,
835 "error": "This chat is archived. Reactivate it to continue.",
836 },
837 status_code=409,
838 )
840 # Pre-fetch existing messages for context decisions.
841 messages = service.get_session_messages(session_id, limit=20)
843 research_id = None
844 research_mode = "none"
845 message_id = None
846 settings_snapshot = None
847 research_context = None
849 if trigger_research:
850 # Always quick mode in chat (intentional v1 scope).
851 research_mode = "quick"
853 # Verify the DB password is available BEFORE creating any rows
854 # or spawning a worker. Chat-triggered research runs on a
855 # background thread that writes token/search metrics to the
856 # user's encrypted database; without the password every metric
857 # write is silently dropped, leaving the metrics dashboard empty
858 # while the research still completes (issue #4457). The password
859 # is re-fetched at the spawn site below, so discard it here.
860 _pw, session_expired = resolve_user_password(username)
861 if session_expired:
862 # Use this router's {success, error} shape (the chat
863 # frontend reads `err.error`); a {status, message} body
864 # would surface only a generic "Request failed (401)".
865 return JSONResponse(
866 {
867 "success": False,
868 "error": "Your session has expired. Please log out and log back in to continue.",
869 },
870 status_code=401,
871 )
873 # ---- Concurrency guards (per-session + global per-user) ----
874 # Both guards run in one transaction so a stale-row reclaim
875 # is visible to the count check below it.
876 #
877 # Sweep AGE NOTE: a brand-new IN_PROGRESS row briefly exists
878 # before its worker registers in `_active_research`. Only
879 # reclaim rows older than the spawn grace cutoff so we don't
880 # kill our own freshly-inserted research from a racing send.
881 _STALE_RESEARCH_GRACE_SECONDS = 30
882 grace_cutoff_dt = datetime.now(UTC) - timedelta(
883 seconds=_STALE_RESEARCH_GRACE_SECONDS
884 )
885 # ResearchHistory.created_at is a String column (ISO-8601);
886 # UserActiveResearch.started_at is a UtcDateTime column.
887 grace_cutoff_iso = grace_cutoff_dt.isoformat()
888 with get_user_db_session(username) as cap_db:
889 # 1. Reclaim stale chat-session research rows whose
890 # worker thread is dead AND older than the grace cutoff.
891 stale_chat = (
892 cap_db.query(ResearchHistory)
893 .filter(
894 ResearchHistory.chat_session_id == session_id,
895 ResearchHistory.status
896 == ResearchStatus.IN_PROGRESS,
897 ResearchHistory.created_at < grace_cutoff_iso,
898 )
899 .all()
900 )
901 reclaimed_chat = False
902 for row in stale_chat: 902 ↛ 903line 902 didn't jump to line 903 because the loop on line 902 never started
903 if not is_research_thread_alive(row.id):
904 logger.warning(
905 f"Reclaiming stale chat research {row.id[:8]}... "
906 f"(thread dead) on chat {session_id[:8]}..."
907 )
908 row.status = ResearchStatus.FAILED
909 cleanup_research(row.id)
910 reclaimed_chat = True
911 if reclaimed_chat: 911 ↛ 912line 911 didn't jump to line 912 because the condition on line 911 was never true
912 cap_db.commit()
914 # 2. Per-session guard: at most one live research per chat.
915 existing_session_research = (
916 cap_db.query(ResearchHistory)
917 .filter_by(
918 chat_session_id=session_id,
919 status=ResearchStatus.IN_PROGRESS,
920 )
921 .first()
922 )
923 if existing_session_research:
924 return JSONResponse(
925 {
926 "success": False,
927 "error": "Research already in progress on this chat session. Stop it before sending a new message.",
928 "active_research_id": existing_session_research.id,
929 },
930 status_code=409,
931 )
933 # 3. Reclaim stale UserActiveResearch rows so the count
934 # below isn't inflated by dead threads. Same grace
935 # window applied via started_at.
936 if reclaim_stale_user_active_research( 936 ↛ 942line 936 didn't jump to line 942 because the condition on line 936 was never true
937 cap_db,
938 username,
939 grace_cutoff_dt=grace_cutoff_dt,
940 logger=logger,
941 ):
942 cap_db.commit()
944 # 4. Global per-user cap (mirrors
945 # research_routes.start_research). Without this,
946 # multiple chat tabs let a user bypass the cap.
947 active_count = (
948 cap_db.query(UserActiveResearch)
949 .filter_by(
950 username=username,
951 status=ResearchStatus.IN_PROGRESS,
952 )
953 .count()
954 )
955 # Clamp to the server-wide semaphore ceiling (#5549):
956 # app.max_concurrent_researches is user-editable but the
957 # semaphore it gates is global, so an unclamped read lets
958 # one user raise their own limit past the operator's and
959 # starve everyone else.
960 max_concurrent = clamp_user_max_concurrent(
961 SettingsManager(db_session=cap_db).get_setting(
962 "app.max_concurrent_researches", 3
963 )
964 )
965 if active_count >= max_concurrent:
966 return JSONResponse(
967 {
968 "success": False,
969 "error": (
970 f"Concurrent research limit reached "
971 f"({active_count}/{max_concurrent}). "
972 "Wait for an existing research to finish."
973 ),
974 },
975 status_code=429,
976 )
977 # ---- end concurrency guards ----
979 # Settings + context (read-only — fine to do after the cap
980 # check, before the atomic write).
981 if trigger_research:
982 settings_snapshot = _load_settings(username)
983 context_manager = ChatContextManager(
984 session_id,
985 messages,
986 session_data.get("accumulated_context"),
987 settings_snapshot=settings_snapshot,
988 )
989 # Pass the new user message so prior conversation is condensed
990 # into a summary focused on this question.
991 research_context = context_manager.build_research_context(
992 current_query=content
993 )
994 research_id = str(uuid.uuid4())
996 # Parse numeric search settings up-front. A malformed value
997 # must return a clean 400 HERE — before the atomic write
998 # below commits the user message + IN_PROGRESS research row.
999 try:
1000 iterations = int(
1001 settings_snapshot.get("search.iterations", {}).get(
1002 "value", 3
1003 )
1004 )
1005 questions = int(
1006 settings_snapshot.get(
1007 "search.questions_per_iteration", {}
1008 ).get("value", 1)
1009 )
1010 except (ValueError, TypeError):
1011 return JSONResponse(
1012 {
1013 "success": False,
1014 "error": (
1015 "Invalid numeric value in search settings "
1016 "(iterations / questions_per_iteration)."
1017 ),
1018 },
1019 status_code=400,
1020 )
1022 # ---- Atomic write: user message + research row in ONE txn ----
1023 # Closes the orphan window: any IntegrityError or
1024 # concurrent-delete on the research insert rolls back the user
1025 # message too. The UPDATE...RETURNING inside insert_message_in_db
1026 # doubles as the authoritative "session-still-exists" check.
1027 try:
1028 with get_user_db_session(username) as db_session:
1029 message_id = service.insert_message_in_db(
1030 db_session,
1031 session_id=session_id,
1032 role="user",
1033 content=content,
1034 message_type="query"
1035 if len(messages) == 0
1036 else "followup",
1037 )
1039 if trigger_research:
1040 created_at = datetime.now(UTC).isoformat()
1041 research_meta = {
1042 "submission": {
1043 "chat_session_id": session_id,
1044 "message_id": message_id,
1045 "research_mode": research_mode,
1046 },
1047 }
1048 research = ResearchHistory(
1049 id=research_id,
1050 query=content,
1051 mode=research_mode,
1052 status=ResearchStatus.IN_PROGRESS.value,
1053 created_at=created_at,
1054 progress_log=[{"time": created_at, "progress": 0}],
1055 research_meta=research_meta,
1056 chat_session_id=session_id,
1057 )
1058 db_session.add(research)
1060 # Count this research toward the per-user concurrent
1061 # cap. Mirrors research_routes.start_research. Added in
1062 # the SAME transaction as the research row so the
1063 # IntegrityError rollback below undoes both.
1064 db_session.add(
1065 UserActiveResearch(
1066 username=username,
1067 research_id=research_id,
1068 status=ResearchStatus.IN_PROGRESS,
1069 thread_id=str(threading.current_thread().ident),
1070 settings_snapshot=settings_snapshot,
1071 )
1072 )
1074 try:
1075 db_session.commit()
1076 except IntegrityError:
1077 # Two near-simultaneous POSTs both passed the
1078 # per-session guard; the partial unique index on
1079 # (chat_session_id) WHERE status='in_progress'
1080 # (migration 0010) catches the loser here. Rolling
1081 # back also undoes the user-message INSERT and the
1082 # message_count increment — no orphan.
1083 db_session.rollback()
1084 logger.warning(
1085 f"Concurrent in-progress research race for chat {session_id[:8]}..."
1086 )
1087 return JSONResponse(
1088 {
1089 "success": False,
1090 "error": "Research already in progress on this chat session.",
1091 },
1092 status_code=409,
1093 )
1094 except ValueError as exc:
1095 # `insert_message_in_db` raises ValueError("not found") when
1096 # the session row was deleted between the existence check and
1097 # the UPDATE...RETURNING. Map to 404.
1098 if "not found" in str(exc).lower(): 1098 ↛ 1103line 1098 didn't jump to line 1103 because the condition on line 1098 was always true
1099 return JSONResponse(
1100 {"success": False, "error": "Session not found"},
1101 status_code=404,
1102 )
1103 raise
1104 # ---- end atomic write ----
1106 if trigger_research:
1107 # Type narrowing: variables initialized to None at the top
1108 # and assigned inside the matching `if trigger_research:`
1109 # block above. Real runtime check (not assert) so it survives
1110 # ``python -O`` (bandit S101).
1111 if (
1112 settings_snapshot is None
1113 or research_context is None
1114 or research_id is None
1115 or message_id is None
1116 ): # pragma: no cover — unreachable invariant guard
1117 raise RuntimeError(
1118 "trigger_research path entered with unset state"
1119 )
1120 # Get user password for metrics
1121 pw = get_user_password(username)
1123 # Extract settings values with safe .get() defaults
1124 model_provider = settings_snapshot.get("llm.provider", {}).get(
1125 "value", ""
1126 )
1127 model = settings_snapshot.get("llm.model", {}).get("value", "")
1128 search_engine = settings_snapshot.get("search.tool", {}).get(
1129 "value", ""
1130 )
1131 custom_endpoint = settings_snapshot.get(
1132 "llm.openai_endpoint.url", {}
1133 ).get("value")
1134 # Defensive fallbacks kept in sync with default_settings.json.
1135 user_strategy = settings_snapshot.get(
1136 "search.search_strategy", {}
1137 ).get("value", "langgraph-agent")
1138 # `iterations` and `questions` were parsed + validated up-front
1139 # (before the atomic write) so a malformed setting returns a
1140 # clean 400 instead of orphaning a committed research row.
1142 # For follow-up messages, use the contextual follow-up strategy
1143 # which wraps the user's preferred strategy as a delegate.
1144 if research_context.get("is_multi_turn"):
1145 strategy = "enhanced-contextual-followup"
1146 research_context["delegate_strategy"] = user_strategy
1147 else:
1148 strategy = user_strategy
1150 # Spawn the worker thread. ``DuplicateResearchError`` inherits
1151 # from ``Exception`` (not RuntimeError) so it is NOT in
1152 # ROUTE_EXCEPTIONS and would otherwise escape to a generic 500
1153 # — leaving the rows we just committed orphaned. Catch here,
1154 # undo our side effects, and return 409.
1155 try:
1156 start_research_process(
1157 research_id,
1158 content,
1159 research_mode,
1160 run_research_process,
1161 username=username,
1162 user_password=pw,
1163 model_provider=model_provider,
1164 model=model,
1165 search_engine=search_engine,
1166 custom_endpoint=custom_endpoint,
1167 strategy=strategy,
1168 iterations=iterations,
1169 questions_per_iteration=questions,
1170 research_context=research_context,
1171 chat_session_id=session_id,
1172 chat_message_id=message_id,
1173 settings_snapshot=settings_snapshot,
1174 )
1175 except DuplicateResearchError:
1176 logger.warning(
1177 f"DuplicateResearchError on chat send_message "
1178 f"for {research_id[:8]}... (chat {session_id[:8]}...)"
1179 )
1180 # Per DuplicateResearchError docstring: do NOT mutate
1181 # UserActiveResearch or the existing ResearchHistory row —
1182 # those belong to the live thread. Only undo our own rows.
1183 _cleanup_chat_send_rows(
1184 username,
1185 research_id,
1186 message_id,
1187 session_id,
1188 "duplicate",
1189 )
1190 return JSONResponse(
1191 {
1192 "success": False,
1193 "error": "Research already in progress on this chat session.",
1194 },
1195 status_code=409,
1196 )
1197 except SystemAtCapacityError:
1198 # System at concurrent-research capacity. Undo the rows we
1199 # committed above and return 429 so the client can retry.
1200 logger.warning(
1201 f"SystemAtCapacityError on chat send_message "
1202 f"for {research_id[:8]}... (chat {session_id[:8]}...)"
1203 )
1204 _cleanup_chat_send_rows(
1205 username,
1206 research_id,
1207 message_id,
1208 session_id,
1209 "capacity",
1210 )
1211 return JSONResponse(
1212 {
1213 "success": False,
1214 "error": "Server is at research capacity. Please retry shortly.",
1215 },
1216 status_code=429,
1217 )
1219 logger.info(
1220 f"Started chat research {research_id[:8]}... for chat {session_id[:8]}..."
1221 )
1223 return JSONResponse(
1224 {
1225 "success": True,
1226 "message_id": message_id,
1227 "session_id": session_id,
1228 "research_id": research_id,
1229 "research_mode": research_mode,
1230 }
1231 )
1233 except ROUTE_EXCEPTIONS:
1234 logger.exception("Error sending chat message")
1235 return JSONResponse(
1236 {"success": False, "error": "Failed to send message"},
1237 status_code=500,
1238 )
1240 return await run_db_sync(_impl)
1243def _enforce_chat_session_research_slot(
1244 cap_db, username: str, session_id: str
1245) -> tuple[str, int] | None:
1246 """Reject send/retry when this chat session or user is at capacity.
1248 Runs inside the caller's ``cap_db`` transaction so the stale-row
1249 reclaims it performs are visible to the count check below it.
1251 Returns ``(error_message, http_status)`` when the request should be
1252 rejected, else ``None``.
1254 Two checks:
1255 1. Per-session guard: at most one live research per chat session.
1256 Mirrors the same check in send_message's inline block.
1257 2. Per-user global cap: at most ``app.max_concurrent_researches``
1258 researches per user across ALL sessions. Mirrors
1259 research start_research.
1261 DOES NOT include the stale-row reclaim sweep — the caller does that
1262 first via ``reclaim_stale_user_active_research`` so the count check
1263 sees accurate numbers.
1264 """
1265 # Per-session guard. Note: this fires for ANY in-progress research
1266 # on the session, INCLUDING the target research on a retry path.
1267 # Retry routes must therefore ensure the target is not in-progress
1268 # (typically via delete_attempt's AttemptInProgress semantics)
1269 # BEFORE calling this.
1270 existing_session_research = (
1271 cap_db.query(ResearchHistory)
1272 .filter_by(
1273 chat_session_id=session_id,
1274 status=ResearchStatus.IN_PROGRESS,
1275 )
1276 .first()
1277 )
1278 if existing_session_research: 1278 ↛ 1279line 1278 didn't jump to line 1279 because the condition on line 1278 was never true
1279 return (
1280 "Research already in progress on this chat session. "
1281 "Stop it before sending a new message.",
1282 409,
1283 )
1285 active_count = (
1286 cap_db.query(UserActiveResearch)
1287 .filter_by(
1288 username=username,
1289 status=ResearchStatus.IN_PROGRESS,
1290 )
1291 .count()
1292 )
1293 # Clamp to the server-wide semaphore ceiling (#5549) — same reasoning
1294 # as the send_message cap above.
1295 max_concurrent = clamp_user_max_concurrent(
1296 SettingsManager(db_session=cap_db).get_setting(
1297 "app.max_concurrent_researches", 3
1298 )
1299 )
1300 if active_count >= max_concurrent:
1301 return (
1302 f"Concurrent research limit reached ({active_count}/"
1303 f"{max_concurrent}). Wait for an existing research to finish.",
1304 429,
1305 )
1307 return None
1310# ============================================================================
1311# Per-attempt API Routes (delete + retry a single chat turn)
1312# ============================================================================
1315@router.delete("/api/chat/sessions/{session_id}/attempts/{research_id}")
1316# Per-user keying, like the other state-changing chat routes. Caps bulk
1317# delete attempts that the global limiter alone left under-constrained.
1318@limiter.limit("30 per minute", key_func=_chat_user_key)
1319async def delete_attempt(
1320 request: Request,
1321 session_id: str,
1322 research_id: str,
1323 username: Annotated[str, Depends(require_auth)],
1324):
1325 """Delete a single chat attempt (user message + research + response).
1327 Refuses with 409 if the target research is IN_PROGRESS and its worker
1328 thread is alive — the client must Stop it first (or wait for it to
1329 fail naturally). Stale IN_PROGRESS rows whose thread is dead are
1330 reclaimed and deleted.
1331 """
1333 def _impl():
1334 try:
1335 service = ChatService(username)
1336 try:
1337 service.delete_attempt(session_id, research_id)
1338 except AttemptNotFound:
1339 return JSONResponse(
1340 {"success": False, "error": "Attempt not found"},
1341 status_code=404,
1342 )
1343 except AttemptInProgress:
1344 return JSONResponse(
1345 {
1346 "success": False,
1347 "error": (
1348 "Research is in progress. Stop it before deleting "
1349 "the attempt."
1350 ),
1351 "active_research_id": research_id,
1352 },
1353 status_code=409,
1354 )
1356 return JSONResponse({"success": True})
1358 except ROUTE_EXCEPTIONS:
1359 logger.exception("Error deleting chat attempt")
1360 return JSONResponse(
1361 {"success": False, "error": "Failed to delete attempt"},
1362 status_code=500,
1363 )
1365 return await run_db_sync(_impl)
1368@router.post("/api/chat/sessions/{session_id}/attempts/{research_id}/retry")
1369# Same per-minute cap as send_message: retry spawns a full research run,
1370# so it shares the spawn budget rather than getting a separate one (else
1371# alternating send/retry would let a user exceed the intended rate).
1372@limiter.limit("10 per minute", key_func=_chat_user_key)
1373async def retry_attempt(
1374 request: Request,
1375 session_id: str,
1376 research_id: str,
1377 username: Annotated[str, Depends(require_auth)],
1378):
1379 """Retry a chat attempt: delete the old turn, re-submit same content.
1381 Looks up the original user message content via
1382 ``ChatService.get_original_attempt_query`` (uses
1383 ``research_meta.submission.message_id`` first, falls back to a
1384 ChatMessage query for older rows), deletes the old attempt
1385 atomically, then runs the same spawn path as ``send_message``.
1387 Returns the SAME shape as ``send_message`` so the client can
1388 subscribe to the new research_id via its existing post-send flow.
1389 """
1391 def _impl():
1392 try:
1393 service = ChatService(username)
1395 # Verify session exists + is active BEFORE doing anything
1396 # destructive. The get_session call also seeds session_data for
1397 # the context-builder below.
1398 try:
1399 session_data = service.get_session(session_id)
1400 except ChatSessionNotFound:
1401 return JSONResponse(
1402 {"success": False, "error": "Session not found"},
1403 status_code=404,
1404 )
1406 if session_data.get("status") != ChatSessionStatus.ACTIVE.value:
1407 return JSONResponse(
1408 {
1409 "success": False,
1410 "error": (
1411 "This chat is archived. Reactivate it to continue."
1412 ),
1413 },
1414 status_code=409,
1415 )
1417 # Resolve DB password BEFORE any destructive op: the spawn path
1418 # needs it for metrics writes, and re-checking AFTER
1419 # delete_attempt would leave the attempt gone on a
1420 # session-expired 401.
1421 _pw, session_expired = resolve_user_password(username)
1422 if session_expired: 1422 ↛ 1423line 1422 didn't jump to line 1423 because the condition on line 1422 was never true
1423 return JSONResponse(
1424 {
1425 "success": False,
1426 "error": (
1427 "Your session has expired. Please log out and "
1428 "log back in to continue."
1429 ),
1430 },
1431 status_code=401,
1432 )
1434 # Per-session + per-user concurrency guards. Same sweep pattern
1435 # as send_message: reclaim stale rows older than the grace
1436 # window first so the count check is accurate, then refuse if
1437 # this session OR the user is at capacity.
1438 #
1439 # If the TARGET research is still IN_PROGRESS, the per-session
1440 # guard catches it here and returns 409 — the user must Stop it
1441 # first. (delete_attempt would also catch this via
1442 # AttemptInProgress, but failing earlier — before any rows are
1443 # touched — keeps the failure cheap.)
1444 _STALE_RESEARCH_GRACE_SECONDS = 30
1445 grace_cutoff_dt = datetime.now(UTC) - timedelta(
1446 seconds=_STALE_RESEARCH_GRACE_SECONDS
1447 )
1448 grace_cutoff_iso = grace_cutoff_dt.isoformat()
1449 with get_user_db_session(username) as cap_db:
1450 # Reclaim stale chat-session research rows whose worker is
1451 # dead AND older than the grace cutoff. Mirrors
1452 # send_message's sweep.
1453 stale_chat = (
1454 cap_db.query(ResearchHistory)
1455 .filter(
1456 ResearchHistory.chat_session_id == session_id,
1457 ResearchHistory.status == ResearchStatus.IN_PROGRESS,
1458 ResearchHistory.created_at < grace_cutoff_iso,
1459 )
1460 .all()
1461 )
1462 reclaimed_chat = False
1463 for row in stale_chat: 1463 ↛ 1464line 1463 didn't jump to line 1464 because the loop on line 1463 never started
1464 if not is_research_thread_alive(row.id):
1465 logger.warning(
1466 f"Reclaiming stale chat research {row.id[:8]}... "
1467 f"(thread dead) on chat {session_id[:8]}..."
1468 )
1469 row.status = ResearchStatus.FAILED
1470 cleanup_research(row.id)
1471 reclaimed_chat = True
1472 if reclaimed_chat: 1472 ↛ 1473line 1472 didn't jump to line 1473 because the condition on line 1472 was never true
1473 cap_db.commit()
1475 # Reclaim stale UserActiveResearch rows globally for this
1476 # user, so the per-user cap count below isn't inflated.
1477 if reclaim_stale_user_active_research( 1477 ↛ 1483line 1477 didn't jump to line 1483 because the condition on line 1477 was never true
1478 cap_db,
1479 username,
1480 grace_cutoff_dt=grace_cutoff_dt,
1481 logger=logger,
1482 ):
1483 cap_db.commit()
1485 cap_error = _enforce_chat_session_research_slot(
1486 cap_db, username, session_id
1487 )
1488 if cap_error is not None: 1488 ↛ 1489line 1488 didn't jump to line 1489 because the condition on line 1488 was never true
1489 msg, status = cap_error
1490 if status == 409:
1491 return JSONResponse(
1492 {
1493 "success": False,
1494 "error": msg,
1495 "active_research_id": research_id,
1496 },
1497 status_code=409,
1498 )
1499 return JSONResponse(
1500 {"success": False, "error": msg}, status_code=status
1501 )
1503 # Fetch original user content BEFORE deleting the old attempt.
1504 # After delete_attempt the row is gone, so this lookup must
1505 # succeed first. Raises AttemptNotFound → 404.
1506 try:
1507 original_content = service.get_original_attempt_query(
1508 session_id, research_id
1509 )
1510 except AttemptNotFound:
1511 return JSONResponse(
1512 {"success": False, "error": "Attempt not found"},
1513 status_code=404,
1514 )
1516 if not original_content.strip(): 1516 ↛ 1520line 1516 didn't jump to line 1520 because the condition on line 1516 was never true
1517 # Empty/whitespace-only original — shouldn't happen since
1518 # send_message rejects these, but guard against a corrupt
1519 # row before we destroy the only copy.
1520 return JSONResponse(
1521 {
1522 "success": False,
1523 "error": (
1524 "Original attempt has no query content to retry."
1525 ),
1526 },
1527 status_code=400,
1528 )
1530 # Delete the old attempt. By this point the target is guaranteed
1531 # not IN_PROGRESS (the per-session guard above would have
1532 # caught it), so this should not raise AttemptInProgress — but
1533 # catch it anyway in case of a narrow race.
1534 try:
1535 service.delete_attempt(session_id, research_id)
1536 except AttemptInProgress:
1537 return JSONResponse(
1538 {
1539 "success": False,
1540 "error": (
1541 "Research is in progress. Stop it before retrying."
1542 ),
1543 "active_research_id": research_id,
1544 },
1545 status_code=409,
1546 )
1547 except AttemptNotFound:
1548 # Concurrent retry already deleted the target.
1549 return JSONResponse(
1550 {"success": False, "error": "Attempt not found"},
1551 status_code=404,
1552 )
1554 # Re-fetch messages list AFTER the delete so the context
1555 # builder doesn't see the just-removed attempt (otherwise the
1556 # new research would treat the retried query as a follow-up to
1557 # itself).
1558 messages = service.get_session_messages(session_id, limit=20)
1560 settings_snapshot = _load_settings(username)
1561 context_manager = ChatContextManager(
1562 session_id,
1563 messages,
1564 session_data.get("accumulated_context"),
1565 settings_snapshot=settings_snapshot,
1566 )
1567 research_context = context_manager.build_research_context(
1568 current_query=original_content
1569 )
1571 # Parse numeric search settings up-front. A malformed value
1572 # must return a clean 400 HERE — before the atomic write
1573 # below commits the user message + IN_PROGRESS research row.
1574 try:
1575 iterations = int(
1576 settings_snapshot.get("search.iterations", {}).get(
1577 "value", 3
1578 )
1579 )
1580 questions = int(
1581 settings_snapshot.get(
1582 "search.questions_per_iteration", {}
1583 ).get("value", 1)
1584 )
1585 except (ValueError, TypeError):
1586 return JSONResponse(
1587 {
1588 "success": False,
1589 "error": (
1590 "Invalid numeric value in search settings "
1591 "(iterations / questions_per_iteration)."
1592 ),
1593 },
1594 status_code=400,
1595 )
1597 research_mode = "quick"
1598 new_research_id = str(uuid.uuid4())
1600 # ---- Atomic write: user message + research row in ONE txn ----
1601 # Same pattern as send_message: any IntegrityError rolls back
1602 # the user message too, so no orphan rows.
1603 try:
1604 with get_user_db_session(username) as db_session:
1605 new_message_id = service.insert_message_in_db(
1606 db_session,
1607 session_id=session_id,
1608 role="user",
1609 content=original_content,
1610 message_type="query"
1611 if len(messages) == 0
1612 else "followup",
1613 )
1615 created_at = datetime.now(UTC).isoformat()
1616 research_meta = {
1617 "submission": {
1618 "chat_session_id": session_id,
1619 "message_id": new_message_id,
1620 "research_mode": research_mode,
1621 },
1622 }
1623 db_session.add(
1624 ResearchHistory(
1625 id=new_research_id,
1626 query=original_content,
1627 mode=research_mode,
1628 status=ResearchStatus.IN_PROGRESS.value,
1629 created_at=created_at,
1630 progress_log=[{"time": created_at, "progress": 0}],
1631 research_meta=research_meta,
1632 chat_session_id=session_id,
1633 )
1634 )
1635 db_session.add(
1636 UserActiveResearch(
1637 username=username,
1638 research_id=new_research_id,
1639 status=ResearchStatus.IN_PROGRESS,
1640 thread_id=str(threading.current_thread().ident),
1641 settings_snapshot=settings_snapshot,
1642 )
1643 )
1645 try:
1646 db_session.commit()
1647 except IntegrityError:
1648 db_session.rollback()
1649 logger.warning(
1650 f"Concurrent in-progress research race for "
1651 f"chat {session_id[:8]}... (retry)"
1652 )
1653 return JSONResponse(
1654 {
1655 "success": False,
1656 "error": (
1657 "Research already in progress on this "
1658 "chat session."
1659 ),
1660 },
1661 status_code=409,
1662 )
1663 except ValueError as exc:
1664 if "not found" in str(exc).lower():
1665 return JSONResponse(
1666 {"success": False, "error": "Session not found"},
1667 status_code=404,
1668 )
1669 raise
1670 # ---- end atomic write ----
1672 pw = get_user_password(username)
1674 model_provider = settings_snapshot.get("llm.provider", {}).get(
1675 "value", ""
1676 )
1677 model = settings_snapshot.get("llm.model", {}).get("value", "")
1678 search_engine = settings_snapshot.get("search.tool", {}).get(
1679 "value", ""
1680 )
1681 custom_endpoint = settings_snapshot.get(
1682 "llm.openai_endpoint.url", {}
1683 ).get("value")
1684 user_strategy = settings_snapshot.get(
1685 "search.search_strategy", {}
1686 ).get("value", "langgraph-agent")
1688 if research_context.get("is_multi_turn"): 1688 ↛ 1689line 1688 didn't jump to line 1689 because the condition on line 1688 was never true
1689 strategy = "enhanced-contextual-followup"
1690 research_context["delegate_strategy"] = user_strategy
1691 else:
1692 strategy = user_strategy
1694 try:
1695 start_research_process(
1696 new_research_id,
1697 original_content,
1698 research_mode,
1699 run_research_process,
1700 username=username,
1701 user_password=pw,
1702 model_provider=model_provider,
1703 model=model,
1704 search_engine=search_engine,
1705 custom_endpoint=custom_endpoint,
1706 strategy=strategy,
1707 iterations=iterations,
1708 questions_per_iteration=questions,
1709 research_context=research_context,
1710 chat_session_id=session_id,
1711 chat_message_id=new_message_id,
1712 settings_snapshot=settings_snapshot,
1713 )
1714 except DuplicateResearchError:
1715 logger.warning(
1716 f"DuplicateResearchError on chat retry_attempt "
1717 f"for {new_research_id[:8]}... (chat {session_id[:8]}...)"
1718 )
1719 _cleanup_chat_send_rows(
1720 username,
1721 new_research_id,
1722 new_message_id,
1723 session_id,
1724 "duplicate",
1725 )
1726 return JSONResponse(
1727 {
1728 "success": False,
1729 "error": (
1730 "Research already in progress on this chat session."
1731 ),
1732 },
1733 status_code=409,
1734 )
1735 except SystemAtCapacityError:
1736 logger.warning(
1737 f"SystemAtCapacityError on chat retry_attempt "
1738 f"for {new_research_id[:8]}... (chat {session_id[:8]}...)"
1739 )
1740 _cleanup_chat_send_rows(
1741 username,
1742 new_research_id,
1743 new_message_id,
1744 session_id,
1745 "capacity",
1746 )
1747 return JSONResponse(
1748 {
1749 "success": False,
1750 "error": (
1751 "Server is at research capacity. Please retry "
1752 "shortly."
1753 ),
1754 },
1755 status_code=429,
1756 )
1758 logger.info(
1759 f"Retried chat attempt as research "
1760 f"{new_research_id[:8]}... for chat {session_id[:8]}..."
1761 )
1763 return JSONResponse(
1764 {
1765 "success": True,
1766 "session_id": session_id,
1767 "research_id": new_research_id,
1768 "message_id": new_message_id,
1769 "research_mode": research_mode,
1770 }
1771 )
1773 except ROUTE_EXCEPTIONS:
1774 logger.exception("Error retrying chat attempt")
1775 return JSONResponse(
1776 {"success": False, "error": "Failed to retry attempt"},
1777 status_code=500,
1778 )
1780 return await run_db_sync(_impl)