Coverage for src/local_deep_research/web/routers/chat.py: 99%
588 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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
23from urllib.parse import quote
25from fastapi import APIRouter, Depends, HTTPException, Request
26from fastapi.responses import JSONResponse
27from loguru import logger
28from sqlalchemy import update as sa_update
29from sqlalchemy.exc import IntegrityError, SQLAlchemyError
31from ...chat.service import (
32 ArchiveBlockedError,
33 AttemptInProgress,
34 AttemptNotFound,
35 ChatCursorError,
36 ChatService,
37 ChatSessionNotFound,
38 DB_EXCEPTIONS,
39)
40from ...chat.context import ChatContextManager
41from ...constants import ResearchStatus
42from ...database.models import (
43 ChatMessage,
44 ChatSession,
45 ChatSessionStatus,
46 ResearchHistory,
47 UserActiveResearch,
48)
49from ...database.session_context import get_user_db_session
50from ...exceptions import DuplicateResearchError, SystemAtCapacityError
51from ...llm.providers.base import normalize_provider
52from ...settings.manager import SettingsManager
53from ...utilities.formatting import short_id
54from ..auth.password_utils import get_user_password, resolve_user_password
55from ..routes.globals import (
56 cleanup_research,
57 is_research_thread_alive,
58 reclaim_stale_user_active_research,
59)
60from ..research_state import (
61 get_user_research_start_lock,
62 user_research_start_gate,
63)
64from ..services.research_service import (
65 clamp_user_max_concurrent,
66 run_research_process,
67 start_research_process,
68)
69from ..dependencies.auth import require_auth
70from ..dependencies.rate_limit import limiter, _get_client_ip
71from ..dependencies.template_helpers import render_template
72from ..dependencies.threadpool import run_db_sync
73from typing import Annotated
75router = APIRouter(tags=["chat"])
77# Valid status values for sessions (built from the enum so a typo never
78# silently passes validation; the literal "all" sentinel widens the list
79# filter to every status without bypassing the whitelist).
80VALID_UPDATE_STATUSES = {
81 ChatSessionStatus.ACTIVE.value,
82 ChatSessionStatus.ARCHIVED.value,
83}
84VALID_LIST_STATUSES = {*(s.value for s in ChatSessionStatus), "all"}
86# Input length limits
87MAX_QUERY_LENGTH = 10_000
88MAX_TITLE_LENGTH = 500
89MAX_MESSAGE_LENGTH = 10_000
90# Hard cap on `offset` to prevent server-side DoS: get_session_messages
91# fetches `limit + offset` rows from BOTH chat_messages and chat_progress_steps
92# tables, so unbounded offset means unbounded SQL LIMIT. With cursor-based
93# pagination (`before_created_at`) as the recommended path, offset above a few
94# pages is not a normal access pattern.
95MAX_OFFSET = 1_000
97# Wider exception tuple used by HTTP route handlers (subsumes
98# service.DB_EXCEPTIONS plus the attribute/type errors that can escape
99# request-shape coercion code). DB_EXCEPTIONS itself is single-sourced
100# from chat.service so the two never drift.
101ROUTE_EXCEPTIONS = (
102 ValueError,
103 RuntimeError,
104 SQLAlchemyError,
105 AttributeError,
106 TypeError,
107)
110def _bad_body() -> JSONResponse:
111 """Build the 400 response for a non-JSON / non-object request body.
113 Returns a FRESH ``JSONResponse`` each call — a shared module-level
114 instance can't be reused safely across concurrent requests. This is the
115 FastAPI equivalent of the old ``@require_json_body(error_format="success",
116 ...)`` decorator.
118 It does NOT reject form-encoded posts, and an earlier version of this
119 docstring wrongly claimed it "hardens CSRF" by doing so. Flask's
120 ``get_json(silent=True)`` returned ``None`` for a non-JSON Content-Type,
121 which had that side effect; ``request.json()`` parses the body regardless
122 of Content-Type, so a form-encoded POST carrying a JSON object body is
123 accepted here. Verified: ``Content-Type: text/plain`` with a JSON object
124 creates a session.
126 That is safe, but for a different reason: ``CSRFMiddleware``
127 (web/dependencies/csrf.py) is the fail-closed choke point and rejects
128 these requests with 403 before they reach any handler. Recording the
129 correction because a reviewer could otherwise rely on a guarantee this
130 function does not provide.
131 """
132 return JSONResponse(
133 {"success": False, "error": "Request body must be a JSON object"},
134 status_code=400,
135 )
138def _chat_user_key(request: Request) -> str:
139 """Per-user rate-limit key (default slowapi key is per-IP).
141 Without this, users behind a shared NAT/proxy share one bucket and can
142 DoS each other for legitimate chat use. Falls back to the client IP for
143 any unauthenticated request that somehow reaches a limited route.
145 The username branch is prefixed (``user:<name>``, the shape
146 ``_user_key`` uses) so a username shaped like an address is not the
147 same bucket as an anonymous caller from that address; the IP branch is
148 always a validated address or a fixed token, never ``user:...``.
149 """
150 username = (
151 request.session.get("username") if "session" in request.scope else None
152 )
153 return f"user:{username}" if username else _get_client_ip(request)
156async def _json_object_body(request: Request):
157 """Parse the request body as a JSON object, or return a 400 response.
159 Returns the parsed dict on success, or a ``JSONResponse`` (400) the
160 caller must return as-is. Mirrors the old ``@require_json_body`` gate.
161 """
162 try:
163 data = await request.json()
164 except Exception:
165 return _bad_body()
166 if not isinstance(data, dict):
167 return _bad_body()
168 return data
171def _load_settings(username):
172 """Load all settings for a user.
174 ``bypass_cache=True`` matches the call pattern in
175 ``research_routes.start_research``: a setting changed via the UI
176 moments before the user sends a chat message must take effect on the
177 next research, not be served from a stale cache.
178 """
179 with get_user_db_session(username) as db:
180 snapshot = SettingsManager(db_session=db).get_all_settings(
181 bypass_cache=True
182 )
183 # Inject the username so snapshot-driven consumers (get_llm, and ADAPTIVE
184 # egress-scope resolution for a per-user retriever primary) can resolve it.
185 # Without it, chat follow-up summarization builds get_llm() with
186 # username=None -> a private-retriever primary is invisible -> scope falls
187 # open to BOTH and a cloud LLM summarizes private-corpus turns. Mirrors
188 # AdvancedSearchSystem.ensure_snapshot_username. Ported from #5539, which
189 # landed on the Flask chat/routes.py this router replaced.
190 from ...search_system import ensure_snapshot_username
192 return ensure_snapshot_username(snapshot, username)
195def _parse_int_param(
196 value: str | None,
197 default: int,
198 min_val: int = 0,
199 max_val: int | None = None,
200) -> int:
201 """Safely parse an integer parameter with bounds checking."""
202 try:
203 result = int(value) if value is not None else default
204 if result < min_val:
205 return min_val
206 if max_val is not None and result > max_val:
207 return max_val
208 return result
209 except (ValueError, TypeError):
210 return default
213_INVISIBLE_UNICODE_CATEGORIES = {"Cf", "Zl", "Zp"}
216def _validate_title(title) -> tuple[str, int] | None:
217 """Return (error_message, http_status) when *title* is invalid, else None.
219 A title is invalid when it is not a non-empty string or exceeds
220 ``MAX_TITLE_LENGTH``. Callers that allow ``None`` (e.g. create_session
221 where omitting the title is fine) should short-circuit on ``None``
222 before calling this helper.
224 Strips Unicode format / line-separator characters (``Cf``/``Zl``/``Zp``,
225 including zero-width spaces U+200B-U+200D and BOM U+FEFF) before the
226 emptiness check so an "invisible" title like 500 zero-width chars is
227 rejected instead of saving a session that looks blank in the UI.
228 """
229 if not isinstance(title, str):
230 return ("Title cannot be empty", 400)
231 visible = "".join(
232 c
233 for c in title
234 if unicodedata.category(c) not in _INVISIBLE_UNICODE_CATEGORIES
235 )
236 if not visible.strip():
237 return ("Title cannot be empty", 400)
238 if len(title) > MAX_TITLE_LENGTH:
239 return (
240 f"Title too long (max {MAX_TITLE_LENGTH} characters)",
241 400,
242 )
243 return None
246def _cleanup_chat_send_rows(
247 username,
248 research_id,
249 message_id,
250 session_id,
251 reason: str,
252 *,
253 preserve_attempt=False,
254) -> None:
255 """Undo the user-message + research_history rows committed by send_message
256 when ``start_research_process`` rejects the spawn.
258 Used by the ``DuplicateResearchError`` (409), ``SystemAtCapacityError``
259 (429), and generic spawn-failure (500) paths. Failure to clean up is logged
260 at ERROR level so orphan rows + inflated message_count are visible to ops.
261 Retry preserves the replacement query as a failed attempt because the old
262 attempt has already been removed. Its active claim is still released.
263 """
264 try:
265 with get_user_db_session(username) as cleanup_db:
266 if preserve_attempt:
267 research = (
268 cleanup_db.query(ResearchHistory)
269 .filter_by(id=research_id, chat_session_id=session_id)
270 .first()
271 )
272 if research is not None: 272 ↛ 293line 272 didn't jump to line 293 because the condition on line 272 was always true
273 research.status = ResearchStatus.FAILED.value
274 research.completed_at = datetime.now(UTC).isoformat()
275 # The failed replacement has no assistant response to
276 # expose its ID. Link its retained query directly so the
277 # reloaded timeline can offer Retry and Delete actions.
278 message = (
279 cleanup_db.query(ChatMessage)
280 .filter_by(id=message_id, session_id=session_id)
281 .first()
282 )
283 if message is not None: 283 ↛ 293line 283 didn't jump to line 293 because the condition on line 283 was always true
284 message.research_id = research_id
285 else:
286 cleanup_db.query(ResearchHistory).filter_by(
287 id=research_id
288 ).delete()
289 cleanup_db.query(ChatMessage).filter_by(id=message_id).delete()
290 # Drop the per-user-cap tracking row too (the spawn never
291 # started, so no live thread owns it). research_id is a fresh
292 # UUID, so this only ever matches our own just-inserted row.
293 cleanup_db.query(UserActiveResearch).filter_by(
294 username=username, research_id=research_id
295 ).delete()
296 if not preserve_attempt:
297 cleanup_db.execute(
298 sa_update(ChatSession)
299 .where(ChatSession.id == session_id)
300 .values(message_count=ChatSession.message_count - 1)
301 )
302 cleanup_db.commit()
303 except DB_EXCEPTIONS:
304 logger.exception(
305 f"Cleanup after {reason} chat-send rejection FAILED "
306 f"for research {short_id(research_id)} in chat "
307 f"{short_id(session_id)}; orphan rows + inflated "
308 f"message_count may persist until next sweep."
309 )
312# ============================================================================
313# Page Routes
314# ============================================================================
316# Sec-Fetch-Site values that mean another origin started the navigation.
317# "same-site" is still another origin, such as a sibling subdomain or
318# another port on the same host.
319_FOREIGN_FETCH_SITES = frozenset({"cross-site", "same-site"})
322def _drop_foreign_chat_query(request: Request) -> None:
323 """Redirect a chat link from another origin to the same page without ``q``.
325 chat.js pre-fills the chat box from ``?q=`` and never sends that text
326 by itself. The SameSite=Lax session cookie lets a link from another site
327 open the page signed in, so when the browser says that another origin
328 started the navigation, the text is not even pre-filled. Browsers send
329 ``Sec-Fetch-Site`` to HTTPS and localhost origins. ``same-origin`` and
330 ``none`` (a typed or bookmarked URL, or a link opened from another app
331 such as a mail or chat client) keep ``q``, and so does a request without
332 the header. After a login the browser arrives here through this app's
333 own redirect, so the login page drops a chat ``q`` from its ``next``
334 itself (``routers/auth.py::_drop_chat_query``).
336 Declared as ``chat_page``'s first dependency parameter, ahead of
337 ``username``. FastAPI solves dependencies in parameter order, so this
338 runs before ``require_auth``: a signed-out user is sent on to the login
339 page without ``q`` as well.
340 """
341 if "q" not in request.query_params:
342 return
343 site = request.headers.get("sec-fetch-site", "").strip().lower()
344 if site not in _FOREIGN_FETCH_SITES:
345 return
346 session_id = request.path_params.get("session_id")
347 location = "/chat/" + (quote(session_id, safe="") if session_id else "")
348 remaining = request.url.remove_query_params("q").query
349 if remaining:
350 location += "?" + remaining
351 raise HTTPException(status_code=303, headers={"Location": location})
354@router.get("/chat/", include_in_schema=False)
355@router.get("/chat/{session_id}", include_in_schema=False)
356def chat_page(
357 request: Request,
358 # A parameter, not a decorator ``dependencies=[...]`` entry, so every
359 # dependency of this route stays in its signature (where the
360 # cross-user isolation census reads auth). It must stay ahead of
361 # ``username``: dependencies run in parameter order.
362 _foreign_query_dropped: Annotated[None, Depends(_drop_foreign_chat_query)],
363 username: Annotated[str, Depends(require_auth)],
364 session_id: str | None = None,
365):
366 """Render the chat page.
368 Args:
369 session_id: Optional session ID to load existing session
370 """
371 return render_template(
372 request, "pages/chat.html", {"session_id": session_id}
373 )
376# ============================================================================
377# Session API Routes
378# ============================================================================
381@router.post("/api/chat/sessions")
382# Per-user keying (default is per-IP). Without this, users behind a shared
383# NAT/proxy share one bucket and can DoS each other for legitimate chat use.
384@limiter.limit("20 per minute", key_func=_chat_user_key)
385async def create_session(
386 request: Request, username: Annotated[str, Depends(require_auth)]
387):
388 """Create a new chat session."""
389 data = await _json_object_body(request)
390 if isinstance(data, JSONResponse):
391 return data
393 def _impl():
394 try:
395 # Validate input lengths
396 initial_query = data.get("initial_query")
397 title = data.get("title")
399 # Reject non-string initial_query early so len() / downstream
400 # string ops don't raise TypeError → 500.
401 if initial_query is not None and not isinstance(initial_query, str):
402 return JSONResponse(
403 {
404 "success": False,
405 "error": "initial_query must be a string",
406 },
407 status_code=400,
408 )
410 if initial_query and len(initial_query) > MAX_QUERY_LENGTH:
411 return JSONResponse(
412 {
413 "success": False,
414 "error": f"Initial query too long (max {MAX_QUERY_LENGTH} characters)",
415 },
416 status_code=400,
417 )
419 if title is not None:
420 err = _validate_title(title)
421 if err is not None:
422 msg, status = err
423 return JSONResponse(
424 {"success": False, "error": msg}, status_code=status
425 )
427 settings_snapshot = _load_settings(username)
429 service = ChatService(username)
430 session_id = service.create_session(
431 initial_query=initial_query,
432 title=title,
433 settings_snapshot=settings_snapshot,
434 )
436 # Get the created session
437 try:
438 session_data = service.get_session(session_id)
439 except ChatSessionNotFound:
440 # Session was just created in this request — getting "not
441 # found" here means a delete-race or storage failure.
442 logger.exception(
443 "Just-created chat session missing on read-back"
444 )
445 return JSONResponse(
446 {
447 "success": False,
448 "error": "Failed to load created session",
449 },
450 status_code=500,
451 )
453 return JSONResponse(
454 {
455 "success": True,
456 "session_id": session_id,
457 "session": session_data,
458 }
459 )
461 except ROUTE_EXCEPTIONS:
462 logger.exception("Error creating chat session")
463 return JSONResponse(
464 {
465 "success": False,
466 "error": "Failed to create chat session",
467 },
468 status_code=500,
469 )
471 return await run_db_sync(_impl)
474@router.post("/api/chat/sessions/{session_id}/generate-title")
475# Per-user keying + lower limit than create_session because each call is a
476# real LLM round-trip on a server-paid endpoint (vs create_session which is
477# zero-LLM DB work). Without per-user keying, shared-IP users share the bucket.
478@limiter.limit("10 per minute", key_func=_chat_user_key)
479async def generate_session_title(
480 request: Request,
481 session_id: str,
482 username: Annotated[str, Depends(require_auth)],
483):
484 """Regenerate the session title using the configured LLM.
486 Fire-and-forget endpoint the frontend calls asynchronously right after
487 creating a session, so the synchronous POST /api/chat/sessions response
488 isn't blocked on an LLM round-trip.
489 """
490 data = await _json_object_body(request)
491 if isinstance(data, JSONResponse):
492 return data
494 def _impl():
495 try:
496 query = data.get("query")
498 if not query:
499 return JSONResponse(
500 {"success": False, "error": "query is required"},
501 status_code=400,
502 )
503 if not isinstance(query, str) or len(query) > MAX_QUERY_LENGTH:
504 return JSONResponse(
505 {
506 "success": False,
507 "error": f"query must be a string up to {MAX_QUERY_LENGTH} chars",
508 },
509 status_code=400,
510 )
512 service = ChatService(username)
513 try:
514 service.get_session(session_id)
515 except ChatSessionNotFound:
516 return JSONResponse(
517 {"success": False, "error": "Session not found"},
518 status_code=404,
519 )
521 settings_snapshot = _load_settings(username)
522 new_title = service.regenerate_title_with_llm(
523 session_id, query, settings_snapshot
524 )
525 if not new_title:
526 # LLM disabled, or LLM call failed — keep existing fallback
527 return JSONResponse({"success": False, "title": None})
529 return JSONResponse({"success": True, "title": new_title})
531 except ROUTE_EXCEPTIONS:
532 logger.exception("Error regenerating chat title")
533 return JSONResponse(
534 {"success": False, "error": "Failed to regenerate title"},
535 status_code=500,
536 )
538 return await run_db_sync(_impl)
541@router.get("/api/chat/sessions")
542async def list_sessions(
543 request: Request, username: Annotated[str, Depends(require_auth)]
544):
545 """List chat sessions for the current user.
547 Query params: status (active/archived/deleted/all, default active),
548 limit (default 20), offset (default 0).
549 """
551 def _impl():
552 try:
553 status = request.query_params.get(
554 "status", ChatSessionStatus.ACTIVE.value
555 )
556 # Validate status parameter
557 if status not in VALID_LIST_STATUSES:
558 status = ChatSessionStatus.ACTIVE.value
559 limit = _parse_int_param(
560 request.query_params.get("limit"), 20, min_val=1, max_val=100
561 )
562 offset = _parse_int_param(
563 request.query_params.get("offset"),
564 0,
565 min_val=0,
566 max_val=MAX_OFFSET,
567 )
569 service = ChatService(username)
570 sessions = service.list_sessions(
571 status=status, limit=limit, offset=offset
572 )
574 return JSONResponse({"success": True, "sessions": sessions})
576 except ROUTE_EXCEPTIONS:
577 logger.exception("Error listing chat sessions")
578 return JSONResponse(
579 {"success": False, "error": "Failed to list chat sessions"},
580 status_code=500,
581 )
583 return await run_db_sync(_impl)
586@router.get("/api/chat/sessions/{session_id}")
587async def get_session(
588 request: Request,
589 session_id: str,
590 username: Annotated[str, Depends(require_auth)],
591):
592 """Get a specific chat session."""
594 def _impl():
595 try:
596 service = ChatService(username)
597 try:
598 session_data = service.get_session(session_id)
599 except ChatSessionNotFound:
600 return JSONResponse(
601 {"success": False, "error": "Session not found"},
602 status_code=404,
603 )
605 return JSONResponse({"success": True, "session": session_data})
607 except ROUTE_EXCEPTIONS:
608 logger.exception("Error getting chat session")
609 return JSONResponse(
610 {"success": False, "error": "Failed to get chat session"},
611 status_code=500,
612 )
614 return await run_db_sync(_impl)
617@router.patch("/api/chat/sessions/{session_id}")
618# Per-user keying, like the other state-changing chat routes.
619@limiter.limit("30 per minute", key_func=_chat_user_key)
620async def update_session(
621 request: Request,
622 session_id: str,
623 username: Annotated[str, Depends(require_auth)],
624):
625 """Update a chat session (title, archive, delete)."""
626 data = await _json_object_body(request)
627 if isinstance(data, JSONResponse):
628 return data
630 def _impl():
631 try:
632 # Require at least one valid field
633 valid_fields = {"title", "status"}
634 if not any(field in data for field in valid_fields):
635 return JSONResponse(
636 {
637 "success": False,
638 "error": "Request must include at least one of: title, status",
639 },
640 status_code=400,
641 )
643 service = ChatService(username)
645 try:
646 service.get_session(session_id)
647 except ChatSessionNotFound:
648 return JSONResponse(
649 {"success": False, "error": "Session not found"},
650 status_code=404,
651 )
653 ops_ok = True
655 if "title" in data:
656 title = data["title"]
657 err = _validate_title(title)
658 if err is not None:
659 msg, status = err
660 return JSONResponse(
661 {"success": False, "error": msg}, status_code=status
662 )
663 ops_ok = (
664 service.update_session_title(session_id, title) and ops_ok
665 )
667 if "status" in data:
668 new_status = data["status"]
669 if new_status not in VALID_UPDATE_STATUSES:
670 return JSONResponse(
671 {"success": False, "error": "Invalid status value"},
672 status_code=400,
673 )
674 if new_status == ChatSessionStatus.ACTIVE.value:
675 ops_ok = service.reactivate_session(session_id) and ops_ok
676 elif new_status == ChatSessionStatus.ARCHIVED.value: 676 ↛ 690line 676 didn't jump to line 690 because the condition on line 676 was always true
677 try:
678 ops_ok = service.archive_session(session_id) and ops_ok
679 except ArchiveBlockedError:
680 # Symmetric with send-to-archived (also 409). Hard-coded
681 # message — never echo str(exc) here (CWE-209).
682 return JSONResponse(
683 {
684 "success": False,
685 "error": "Cannot archive: research in_progress. Stop it first.",
686 },
687 status_code=409,
688 )
690 try:
691 session_data = service.get_session(session_id)
692 except ChatSessionNotFound:
693 # Session was deleted by a concurrent request between the
694 # update above and this read-back. Treat as 404.
695 return JSONResponse(
696 {"success": False, "error": "Session not found"},
697 status_code=404,
698 )
700 if not ops_ok:
701 # The read-back above succeeded, so the session still exists,
702 # yet an update reported failure — a DB write error swallowed
703 # into a False return. Surface it instead of reporting success.
704 logger.error(
705 f"Chat session update failed at DB layer for "
706 f"{session_id[:8]}..."
707 )
708 return JSONResponse(
709 {"success": False, "error": "Failed to update session"},
710 status_code=500,
711 )
713 return JSONResponse({"success": True, "session": session_data})
715 except ROUTE_EXCEPTIONS:
716 logger.exception("Error updating chat session")
717 return JSONResponse(
718 {"success": False, "error": "Failed to update chat session"},
719 status_code=500,
720 )
722 return await run_db_sync(_impl)
725@router.delete("/api/chat/sessions/{session_id}")
726# Per-user keying, like the other state-changing chat routes.
727@limiter.limit("30 per minute", key_func=_chat_user_key)
728async def delete_session(
729 request: Request,
730 session_id: str,
731 username: Annotated[str, Depends(require_auth)],
732):
733 """Delete a chat session permanently."""
735 def _impl():
736 try:
737 service = ChatService(username)
738 success = service.delete_session(session_id)
740 if not success:
741 return JSONResponse(
742 {"success": False, "error": "Session not found"},
743 status_code=404,
744 )
746 return JSONResponse({"success": True})
748 except ROUTE_EXCEPTIONS:
749 logger.exception("Error deleting chat session")
750 return JSONResponse(
751 {"success": False, "error": "Failed to delete chat session"},
752 status_code=500,
753 )
755 return await run_db_sync(_impl)
758# ============================================================================
759# Message API Routes
760# ============================================================================
763@router.get("/api/chat/sessions/{session_id}/messages")
764async def get_messages(
765 request: Request,
766 session_id: str,
767 username: Annotated[str, Depends(require_auth)],
768):
769 """Get messages for a chat session (cursor-paginated, ASC)."""
771 def _impl():
772 try:
773 limit = _parse_int_param(
774 request.query_params.get("limit"), 50, min_val=1, max_val=100
775 )
776 offset = _parse_int_param(
777 request.query_params.get("offset"),
778 0,
779 min_val=0,
780 max_val=MAX_OFFSET,
781 )
782 before_created_at = (
783 request.query_params.get("before_created_at") or None
784 )
785 before_id = request.query_params.get("before_id") or None
787 service = ChatService(username)
789 try:
790 service.get_session(session_id)
791 except ChatSessionNotFound:
792 return JSONResponse(
793 {"success": False, "error": "Session not found"},
794 status_code=404,
795 )
797 # Fetch one extra row so we can tell the client whether more
798 # older entries exist without a second round-trip.
799 peek_limit = limit + 1
800 page = service.get_session_messages(
801 session_id,
802 limit=peek_limit,
803 offset=offset,
804 before_created_at=before_created_at,
805 before_id=before_id,
806 )
807 has_more = len(page) > limit
808 messages = page[-limit:] if has_more else page
810 # The client (chat.js loadSession) restores the live "thinking"
811 # indicator from this field. O(1) via the partial-unique index
812 # ux_research_history_chat_session_in_progress.
813 in_progress_research_id = service.get_in_progress_research_id(
814 session_id
815 )
817 return JSONResponse(
818 {
819 "success": True,
820 "messages": messages,
821 "has_more": has_more,
822 "in_progress_research_id": in_progress_research_id,
823 }
824 )
826 except ChatCursorError:
827 return JSONResponse(
828 {
829 "success": False,
830 "error": "Invalid pagination cursor. before_created_at must "
831 "include a timezone; before_id must identify an entry in "
832 "this session at before_created_at.",
833 },
834 status_code=400,
835 )
836 except ROUTE_EXCEPTIONS:
837 logger.exception("Error getting chat messages")
838 return JSONResponse(
839 {"success": False, "error": "Failed to get chat messages"},
840 status_code=500,
841 )
843 return await run_db_sync(_impl)
846@router.post("/api/chat/sessions/{session_id}/messages")
847# Per-user keying. send_message launches a full research run, so this is the
848# heaviest chat endpoint; shared-IP users sharing the bucket would lock each
849# other out.
850@limiter.limit("10 per minute", key_func=_chat_user_key)
851async def send_message(
852 request: Request,
853 session_id: str,
854 username: Annotated[str, Depends(require_auth)],
855):
856 """Send a message in a chat session, optionally triggering research.
858 Research mode is always "quick" in chat (intentional v1 scope).
859 """
860 data = await _json_object_body(request)
861 if isinstance(data, JSONResponse):
862 return data
864 def _impl():
865 admission_lock = None
866 try:
867 if not data or not data.get("content"):
868 return JSONResponse(
869 {"success": False, "error": "Message content is required"},
870 status_code=400,
871 )
873 # Reject non-string content before .strip() raises AttributeError
874 # → 500. Mirrors the isinstance guard in _validate_title.
875 if not isinstance(data["content"], str):
876 return JSONResponse(
877 {"success": False, "error": "content must be a string"},
878 status_code=400,
879 )
881 content = data["content"].strip()
883 # Reject whitespace-only content
884 if not content:
885 return JSONResponse(
886 {"success": False, "error": "Message content is required"},
887 status_code=400,
888 )
890 if len(content) > MAX_MESSAGE_LENGTH:
891 return JSONResponse(
892 {
893 "success": False,
894 "error": f"Message too long (max {MAX_MESSAGE_LENGTH} characters)",
895 },
896 status_code=400,
897 )
899 # trigger_research is a strict boolean. Reject non-bool values
900 # with a 400 instead of silently coercing them to True: a client
901 # sending {"trigger_research": "no"} or {"trigger_research": 0}
902 # intends to SUPPRESS research, and coercing the truthy-string to
903 # True would launch an unwanted (paid) research run against their
904 # intent (#4427).
905 raw = data.get("trigger_research", True)
906 if not isinstance(raw, bool):
907 return JSONResponse(
908 {
909 "success": False,
910 "error": "trigger_research must be a boolean",
911 },
912 status_code=400,
913 )
914 trigger_research = raw
916 service = ChatService(username)
918 # Verify session exists (informational fast-fail; the
919 # UPDATE...RETURNING inside insert_message_in_db is the
920 # authoritative check that survives a delete-race).
921 try:
922 session_data = service.get_session(session_id)
923 except ChatSessionNotFound:
924 return JSONResponse(
925 {"success": False, "error": "Session not found"},
926 status_code=404,
927 )
929 # Reject sends to non-active sessions. Archived/deleted sessions
930 # are intentionally read-only — users must reactivate first.
931 if session_data.get("status") != ChatSessionStatus.ACTIVE.value:
932 return JSONResponse(
933 {
934 "success": False,
935 "error": "This chat is archived. Reactivate it to continue.",
936 },
937 status_code=409,
938 )
940 # Pre-fetch existing messages for context decisions.
941 # include_steps=False keeps `limit` spent entirely on genuine
942 # ChatMessage conversation turns. Progress-step rows would
943 # otherwise dominate this 20-entry window on a research-heavy
944 # session and push real conversation history out of it before
945 # ChatContextManager ever sees it — and ChatContextManager
946 # discards step rows anyway, so there is no reason to spend the
947 # window budget on them here.
948 messages = service.get_session_messages(
949 session_id, limit=20, include_steps=False
950 )
952 research_id = None
953 research_mode = "none"
954 message_id = None
955 settings_snapshot = None
956 research_context = None
958 if trigger_research:
959 # Always quick mode in chat (intentional v1 scope).
960 research_mode = "quick"
962 # Verify the DB password is available BEFORE creating any rows
963 # or spawning a worker. Chat-triggered research runs on a
964 # background thread that writes token/search metrics to the
965 # user's encrypted database; without the password every metric
966 # write is silently dropped, leaving the metrics dashboard empty
967 # while the research still completes (issue #4457). The password
968 # is re-fetched at the spawn site below, so discard it here.
969 _pw, session_expired = resolve_user_password(username)
970 if session_expired:
971 # Use this router's {success, error} shape (the chat
972 # frontend reads `err.error`); a {status, message} body
973 # would surface only a generic "Request failed (401)".
974 return JSONResponse(
975 {
976 "success": False,
977 "error": "Your session has expired. Please log out and log back in to continue.",
978 },
979 status_code=401,
980 )
982 # ChatService's reads above reuse a thread-local Session and
983 # leave its read transaction open. Release that connection
984 # before taking the admission gate: every producer follows
985 # gate -> DB, and reversing the order here can deadlock under
986 # pool exhaustion with a gate holder waiting for a connection.
987 with get_user_db_session(username) as ordering_db:
988 ordering_db.rollback()
990 # Short gated fast-fail before the potentially slow context
991 # build below. The authoritative check is repeated after that
992 # work and retained through claim + worker registration.
993 with user_research_start_gate(username):
994 cap_error = _check_chat_research_slot_locked(
995 username, session_id
996 )
997 if cap_error is not None:
998 message, status, active_research_id = cap_error
999 payload = {"success": False, "error": message}
1000 if active_research_id is not None:
1001 payload["active_research_id"] = active_research_id
1002 return JSONResponse(payload, status_code=status)
1004 # Settings + context (read-only — fine to do after the cap
1005 # check, before the atomic write).
1006 if trigger_research:
1007 settings_snapshot = _load_settings(username)
1008 context_manager = ChatContextManager(
1009 session_id,
1010 messages,
1011 session_data.get("accumulated_context"),
1012 settings_snapshot=settings_snapshot,
1013 )
1014 # Pass the new user message so prior conversation is condensed
1015 # into a summary focused on this question.
1016 research_context = context_manager.build_research_context(
1017 current_query=content
1018 )
1019 research_id = str(uuid.uuid4())
1021 # Parse numeric search settings up-front. A malformed value
1022 # must return a clean 400 HERE — before the atomic write
1023 # below commits the user message + IN_PROGRESS research row.
1024 try:
1025 iterations = int(
1026 settings_snapshot.get("search.iterations", {}).get(
1027 "value", 3
1028 )
1029 )
1030 questions = int(
1031 settings_snapshot.get(
1032 "search.questions_per_iteration", {}
1033 ).get("value", 1)
1034 )
1035 except (ValueError, TypeError):
1036 return JSONResponse(
1037 {
1038 "success": False,
1039 "error": (
1040 "Invalid numeric value in search settings "
1041 "(iterations / questions_per_iteration)."
1042 ),
1043 },
1044 status_code=400,
1045 )
1047 # End the cached read transaction before taking the stable
1048 # per-user gate (gate -> DB is the process-wide lock order).
1049 with get_user_db_session(username) as ordering_db:
1050 ordering_db.rollback()
1052 pending_admission_lock = get_user_research_start_lock(username)
1053 pending_admission_lock.acquire()
1054 admission_lock = pending_admission_lock
1056 # Context preparation can take long enough for another start
1057 # to consume the last slot. Reconcile and recheck under the
1058 # gate that remains held through commit, registration, and
1059 # every spawn-failure cleanup return below.
1060 cap_error = _check_chat_research_slot_locked(
1061 username, session_id
1062 )
1063 if cap_error is not None:
1064 message, status, active_research_id = cap_error
1065 payload = {"success": False, "error": message}
1066 if active_research_id is not None: 1066 ↛ 1068line 1066 didn't jump to line 1068 because the condition on line 1066 was always true
1067 payload["active_research_id"] = active_research_id
1068 return JSONResponse(payload, status_code=status)
1070 # ---- Atomic write: user message + research row in ONE txn ----
1071 # Closes the orphan window: any IntegrityError or
1072 # concurrent-delete on the research insert rolls back the user
1073 # message too. The UPDATE...RETURNING inside insert_message_in_db
1074 # doubles as the authoritative "session-still-exists" check.
1075 try:
1076 with get_user_db_session(username) as db_session:
1077 message_id = service.insert_message_in_db(
1078 db_session,
1079 session_id=session_id,
1080 role="user",
1081 content=content,
1082 message_type="query"
1083 if len(messages) == 0
1084 else "followup",
1085 )
1087 if trigger_research:
1088 created_at = datetime.now(UTC).isoformat()
1089 research_meta = {
1090 "submission": {
1091 "chat_session_id": session_id,
1092 "message_id": message_id,
1093 "research_mode": research_mode,
1094 },
1095 # Persist the run's settings snapshot so
1096 # view/export (report_assembly_service) renders
1097 # the same report.uncited_sources_mode the
1098 # generation-time render used. Mirrors the
1099 # main-form creation path; completion merge
1100 # back-fills when missing.
1101 "settings_snapshot": settings_snapshot,
1102 }
1103 research = ResearchHistory(
1104 id=research_id,
1105 query=content,
1106 mode=research_mode,
1107 status=ResearchStatus.IN_PROGRESS.value,
1108 created_at=created_at,
1109 progress_log=[{"time": created_at, "progress": 0}],
1110 research_meta=research_meta,
1111 chat_session_id=session_id,
1112 )
1113 db_session.add(research)
1115 # Count this research toward the per-user concurrent
1116 # cap. Mirrors research_routes.start_research. Added in
1117 # the SAME transaction as the research row so the
1118 # IntegrityError rollback below undoes both.
1119 db_session.add(
1120 UserActiveResearch(
1121 username=username,
1122 research_id=research_id,
1123 status=ResearchStatus.IN_PROGRESS,
1124 thread_id=str(threading.current_thread().ident),
1125 settings_snapshot=settings_snapshot,
1126 )
1127 )
1129 try:
1130 db_session.commit()
1131 except IntegrityError:
1132 # The admission gate is held from the recheck through
1133 # this commit, so no other chat send or retry in this
1134 # process can pass the per-session guard meanwhile.
1135 # The partial unique index on (chat_session_id) WHERE
1136 # status='in_progress' (migration 0010) is the
1137 # backstop for a writer outside that gate and catches
1138 # the loser here. Rolling back also undoes the
1139 # user-message INSERT and the message_count
1140 # increment — no orphan.
1141 db_session.rollback()
1142 logger.warning(
1143 f"Concurrent in-progress research race for chat {session_id[:8]}..."
1144 )
1145 return JSONResponse(
1146 {
1147 "success": False,
1148 "error": "Research already in progress on this chat session.",
1149 },
1150 status_code=409,
1151 )
1152 except ValueError as exc:
1153 # `insert_message_in_db` raises ValueError("not found") when
1154 # the session row was deleted between the existence check and
1155 # the UPDATE...RETURNING. Map to 404.
1156 if "not found" in str(exc).lower():
1157 return JSONResponse(
1158 {"success": False, "error": "Session not found"},
1159 status_code=404,
1160 )
1161 raise
1162 # ---- end atomic write ----
1164 if trigger_research:
1165 # Type narrowing: variables initialized to None at the top
1166 # and assigned inside the matching `if trigger_research:`
1167 # block above. Real runtime check (not assert) so it survives
1168 # ``python -O`` (bandit S101).
1169 if (
1170 settings_snapshot is None
1171 or research_context is None
1172 or research_id is None
1173 or message_id is None
1174 ): # pragma: no cover — unreachable invariant guard
1175 raise RuntimeError(
1176 "trigger_research path entered with unset state"
1177 )
1178 # Get user password for metrics
1179 pw = get_user_password(username)
1181 # Extract settings values with safe .get() defaults
1182 model_provider = settings_snapshot.get("llm.provider", {}).get(
1183 "value", ""
1184 )
1185 model = settings_snapshot.get("llm.model", {}).get("value", "")
1186 search_engine = settings_snapshot.get("search.tool", {}).get(
1187 "value", ""
1188 )
1189 # llm.openai_endpoint.url is only the client's base URL for
1190 # the openai_endpoint provider; passing it for any other
1191 # provider mislabels that provider's endpoint downstream.
1192 custom_endpoint = (
1193 settings_snapshot.get("llm.openai_endpoint.url", {}).get(
1194 "value"
1195 )
1196 if normalize_provider(model_provider) == "openai_endpoint"
1197 else None
1198 )
1199 # Defensive fallbacks kept in sync with default_settings.json.
1200 user_strategy = settings_snapshot.get(
1201 "search.search_strategy", {}
1202 ).get("value", "langgraph-agent")
1203 # `iterations` and `questions` were parsed + validated up-front
1204 # (before the atomic write) so a malformed setting returns a
1205 # clean 400 instead of orphaning a committed research row.
1207 # For follow-up messages, use the contextual follow-up strategy
1208 # which wraps the user's preferred strategy as a delegate.
1209 if research_context.get("is_multi_turn"):
1210 strategy = "enhanced-contextual-followup"
1211 research_context["delegate_strategy"] = user_strategy
1212 else:
1213 strategy = user_strategy
1215 # Spawn the worker thread. ``DuplicateResearchError`` inherits
1216 # from ``Exception`` (not RuntimeError) so it is NOT in
1217 # ROUTE_EXCEPTIONS and would otherwise escape to a generic 500
1218 # — leaving the rows we just committed orphaned. Catch here,
1219 # undo our side effects, and return 409.
1220 try:
1221 start_research_process(
1222 research_id,
1223 content,
1224 research_mode,
1225 run_research_process,
1226 username=username,
1227 user_password=pw,
1228 model_provider=model_provider,
1229 model=model,
1230 search_engine=search_engine,
1231 custom_endpoint=custom_endpoint,
1232 strategy=strategy,
1233 iterations=iterations,
1234 questions_per_iteration=questions,
1235 research_context=research_context,
1236 chat_session_id=session_id,
1237 chat_message_id=message_id,
1238 settings_snapshot=settings_snapshot,
1239 )
1240 except DuplicateResearchError:
1241 logger.warning(
1242 f"DuplicateResearchError on chat send_message "
1243 f"for {research_id[:8]}... (chat {session_id[:8]}...)"
1244 )
1245 # Per DuplicateResearchError docstring: do NOT mutate
1246 # UserActiveResearch or the existing ResearchHistory row —
1247 # those belong to the live thread. Only undo our own rows.
1248 _cleanup_chat_send_rows(
1249 username,
1250 research_id,
1251 message_id,
1252 session_id,
1253 "duplicate",
1254 )
1255 return JSONResponse(
1256 {
1257 "success": False,
1258 "error": "Research already in progress on this chat session.",
1259 },
1260 status_code=409,
1261 )
1262 except SystemAtCapacityError:
1263 # System at concurrent-research capacity. Undo the rows we
1264 # committed above and return 429 so the client can retry.
1265 logger.warning(
1266 f"SystemAtCapacityError on chat send_message "
1267 f"for {research_id[:8]}... (chat {session_id[:8]}...)"
1268 )
1269 _cleanup_chat_send_rows(
1270 username,
1271 research_id,
1272 message_id,
1273 session_id,
1274 "capacity",
1275 )
1276 return JSONResponse(
1277 {
1278 "success": False,
1279 "error": "Server is at research capacity. Please retry shortly.",
1280 },
1281 status_code=429,
1282 )
1283 except Exception:
1284 # Thread construction/registration can fail for reasons
1285 # other than duplicate ownership or capacity (for example,
1286 # RuntimeError from threading.Thread.start). The durable
1287 # message, research, and per-user claim already exist, so
1288 # undo all of them before releasing the admission gate.
1289 logger.exception(
1290 f"Failed to spawn chat research "
1291 f"{research_id[:8]}... (chat {session_id[:8]}...)"
1292 )
1293 _cleanup_chat_send_rows(
1294 username,
1295 research_id,
1296 message_id,
1297 session_id,
1298 "spawn failure",
1299 )
1300 return JSONResponse(
1301 {
1302 "success": False,
1303 "error": "Failed to start research. Please try again.",
1304 },
1305 status_code=500,
1306 )
1308 logger.info(
1309 f"Started chat research {research_id[:8]}... for chat {session_id[:8]}..."
1310 )
1312 return JSONResponse(
1313 {
1314 "success": True,
1315 "message_id": message_id,
1316 "session_id": session_id,
1317 "research_id": research_id,
1318 "research_mode": research_mode,
1319 }
1320 )
1322 except ROUTE_EXCEPTIONS:
1323 logger.exception("Error sending chat message")
1324 return JSONResponse(
1325 {"success": False, "error": "Failed to send message"},
1326 status_code=500,
1327 )
1328 finally:
1329 if admission_lock is not None:
1330 admission_lock.release()
1332 return await run_db_sync(_impl)
1335def _check_chat_research_slot_locked(
1336 username: str, session_id: str
1337) -> tuple[str, int, str | None] | None:
1338 """Reconcile and check chat admission while the caller holds its gate.
1340 The caller must acquire ``user_research_start_gate(username)`` before
1341 entering this helper. Opening the DB here preserves the process-wide
1342 gate-before-session order. The read transaction is explicitly ended
1343 before returning so a later gate acquisition cannot form a pool cycle.
1345 Rows older than the 30-second spawn grace window whose worker is dead
1346 are reclaimed before the slot checks run: this session's IN_PROGRESS
1347 research rows first, then the user's ``UserActiveResearch`` slots. Once
1348 that window has passed, a crashed worker's rows therefore cannot trip
1349 the per-session 409 or the per-user 429. Returns
1350 ``(error_message, http_status, active_research_id)`` to reject the
1351 request, else ``None``.
1352 """
1353 grace_cutoff_dt = datetime.now(UTC) - timedelta(seconds=30)
1354 grace_cutoff_iso = grace_cutoff_dt.isoformat()
1356 with get_user_db_session(username) as cap_db:
1357 try:
1358 stale_chat = (
1359 cap_db.query(ResearchHistory)
1360 .filter(
1361 ResearchHistory.chat_session_id == session_id,
1362 ResearchHistory.status == ResearchStatus.IN_PROGRESS,
1363 ResearchHistory.created_at < grace_cutoff_iso,
1364 )
1365 .all()
1366 )
1367 reclaimed_chat = False
1368 for row in stale_chat:
1369 if not is_research_thread_alive(row.id):
1370 logger.warning(
1371 f"Reclaiming stale chat research {row.id[:8]}... "
1372 f"(thread dead) on chat {session_id[:8]}..."
1373 )
1374 row.status = ResearchStatus.FAILED
1375 cleanup_research(row.id)
1376 reclaimed_chat = True
1377 if reclaimed_chat:
1378 cap_db.commit()
1380 if reclaim_stale_user_active_research(
1381 cap_db,
1382 username,
1383 grace_cutoff_dt=grace_cutoff_dt,
1384 logger=logger,
1385 ):
1386 cap_db.commit()
1388 cap_error = _enforce_chat_session_research_slot(
1389 cap_db, username, session_id
1390 )
1391 if cap_error is None:
1392 return None
1394 message, status = cap_error
1395 active_research_id = None
1396 if status == 409:
1397 active_row = (
1398 cap_db.query(ResearchHistory.id)
1399 .filter_by(
1400 chat_session_id=session_id,
1401 status=ResearchStatus.IN_PROGRESS,
1402 )
1403 .first()
1404 )
1405 if active_row is not None: 1405 ↛ 1407line 1405 didn't jump to line 1407 because the condition on line 1405 was always true
1406 active_research_id = active_row[0]
1407 return message, status, active_research_id
1408 finally:
1409 cap_db.rollback()
1412def _enforce_chat_session_research_slot(
1413 cap_db, username: str, session_id: str
1414) -> tuple[str, int] | None:
1415 """Reject send/retry when this chat session or user is at capacity.
1417 Runs inside the caller's ``cap_db`` transaction after any stale-row
1418 reclaims, so those writes are visible to the count check below it.
1420 Returns ``(error_message, http_status)`` when the request should be
1421 rejected, else ``None``.
1423 Two checks:
1424 1. Per-session guard: at most one live research per chat session.
1425 2. Per-user global cap: at most ``app.max_concurrent_researches``
1426 researches per user across ALL sessions. Mirrors
1427 research start_research.
1429 DOES NOT include the stale-row reclaim sweep. Its only caller,
1430 ``_check_chat_research_slot_locked``, first reclaims dead rows older
1431 than its grace window so neither check counts them.
1432 """
1433 # Per-session guard. Note: this fires for ANY in-progress research
1434 # on the session, INCLUDING the target research on a retry path.
1435 # That is how retry refuses a target whose worker is still live,
1436 # before anything is deleted. A target whose worker died more than the
1437 # 30-second grace window ago has already been reclaimed by the caller's
1438 # sweep, so it cannot block its own retry.
1439 existing_session_research = (
1440 cap_db.query(ResearchHistory)
1441 .filter_by(
1442 chat_session_id=session_id,
1443 status=ResearchStatus.IN_PROGRESS,
1444 )
1445 .first()
1446 )
1447 if existing_session_research:
1448 return (
1449 "Research already in progress on this chat session. "
1450 "Stop it before sending a new message.",
1451 409,
1452 )
1454 active_count = (
1455 cap_db.query(UserActiveResearch)
1456 .filter_by(
1457 username=username,
1458 status=ResearchStatus.IN_PROGRESS,
1459 )
1460 .count()
1461 )
1462 # Clamp to the server-wide semaphore ceiling (#5549):
1463 # app.max_concurrent_researches is user-editable but the semaphore it
1464 # gates is global, so an unclamped read would let one user raise their
1465 # own limit past the operator's and starve everyone else.
1466 max_concurrent = clamp_user_max_concurrent(
1467 SettingsManager(db_session=cap_db).get_setting(
1468 "app.max_concurrent_researches", 3
1469 )
1470 )
1471 if active_count >= max_concurrent:
1472 return (
1473 f"Concurrent research limit reached ({active_count}/"
1474 f"{max_concurrent}). Wait for an existing research to finish.",
1475 429,
1476 )
1478 return None
1481# ============================================================================
1482# Per-attempt API Routes (delete + retry a single chat turn)
1483# ============================================================================
1486@router.delete("/api/chat/sessions/{session_id}/attempts/{research_id}")
1487# Per-user keying, like the other state-changing chat routes. Caps bulk
1488# delete attempts that the global limiter alone left under-constrained.
1489@limiter.limit("30 per minute", key_func=_chat_user_key)
1490async def delete_attempt(
1491 request: Request,
1492 session_id: str,
1493 research_id: str,
1494 username: Annotated[str, Depends(require_auth)],
1495):
1496 """Delete a single chat attempt (user message + research + response).
1498 Refuses with 409 if the target research is IN_PROGRESS and its worker
1499 thread is alive — the client must Stop it first (or wait for it to
1500 fail naturally). Stale IN_PROGRESS rows whose thread is dead are
1501 reclaimed and deleted.
1502 """
1504 def _impl():
1505 try:
1506 service = ChatService(username)
1507 try:
1508 service.delete_attempt(session_id, research_id)
1509 except AttemptNotFound:
1510 return JSONResponse(
1511 {"success": False, "error": "Attempt not found"},
1512 status_code=404,
1513 )
1514 except AttemptInProgress:
1515 return JSONResponse(
1516 {
1517 "success": False,
1518 "error": (
1519 "Research is in progress. Stop it before deleting "
1520 "the attempt."
1521 ),
1522 "active_research_id": research_id,
1523 },
1524 status_code=409,
1525 )
1527 return JSONResponse({"success": True})
1529 except ROUTE_EXCEPTIONS:
1530 logger.exception("Error deleting chat attempt")
1531 return JSONResponse(
1532 {"success": False, "error": "Failed to delete attempt"},
1533 status_code=500,
1534 )
1536 return await run_db_sync(_impl)
1539@router.post("/api/chat/sessions/{session_id}/attempts/{research_id}/retry")
1540# Same per-minute cap as send_message: retry spawns a full research run,
1541# so it shares the spawn budget rather than getting a separate one (else
1542# alternating send/retry would let a user exceed the intended rate).
1543@limiter.limit("10 per minute", key_func=_chat_user_key)
1544async def retry_attempt(
1545 request: Request,
1546 session_id: str,
1547 research_id: str,
1548 username: Annotated[str, Depends(require_auth)],
1549):
1550 """Retry a chat attempt: delete the old turn, re-submit same content.
1552 Looks up the original user message content via
1553 ``ChatService.get_original_attempt_query`` (uses
1554 ``research_meta.submission.message_id`` first, falls back to a
1555 ChatMessage query for older rows), deletes the old attempt
1556 atomically, then runs the same spawn path as ``send_message``.
1558 Returns the SAME shape as ``send_message`` so the client can
1559 subscribe to the new research_id via its existing post-send flow.
1560 """
1562 def _impl():
1563 admission_lock = None
1564 service = None
1565 rag_targets: list[tuple[str, list[str]]] = []
1566 try:
1567 service = ChatService(username)
1569 # Verify session exists + is active BEFORE doing anything
1570 # destructive. The get_session call also seeds session_data for
1571 # the context-builder below.
1572 try:
1573 session_data = service.get_session(session_id)
1574 except ChatSessionNotFound:
1575 return JSONResponse(
1576 {"success": False, "error": "Session not found"},
1577 status_code=404,
1578 )
1580 if session_data.get("status") != ChatSessionStatus.ACTIVE.value:
1581 return JSONResponse(
1582 {
1583 "success": False,
1584 "error": (
1585 "This chat is archived. Reactivate it to continue."
1586 ),
1587 },
1588 status_code=409,
1589 )
1591 # Resolve DB password BEFORE any destructive op: the spawn path
1592 # needs it for metrics writes, and re-checking AFTER
1593 # delete_attempt would leave the attempt gone on a
1594 # session-expired 401.
1595 _pw, session_expired = resolve_user_password(username)
1596 if session_expired:
1597 return JSONResponse(
1598 {
1599 "success": False,
1600 "error": (
1601 "Your session has expired. Please log out and "
1602 "log back in to continue."
1603 ),
1604 },
1605 status_code=401,
1606 )
1608 # get_session() leaves the reused thread-local Session in a read
1609 # transaction. End it before the early gate acquisition so retry
1610 # follows the same gate -> DB order as every other producer.
1611 with get_user_db_session(username) as ordering_db:
1612 ordering_db.rollback()
1614 # Cheap gated fast-fail before context preparation. If the target
1615 # is still live this returns 409 before any destructive work.
1616 with user_research_start_gate(username):
1617 cap_error = _check_chat_research_slot_locked(
1618 username, session_id
1619 )
1620 if cap_error is not None:
1621 message, status, active_research_id = cap_error
1622 payload = {"success": False, "error": message}
1623 if active_research_id is not None:
1624 payload["active_research_id"] = active_research_id
1625 return JSONResponse(payload, status_code=status)
1627 # Fetch original user content BEFORE deleting the old attempt.
1628 # After delete_attempt the row is gone, so this lookup must
1629 # succeed first. Raises AttemptNotFound → 404.
1630 try:
1631 original_content = service.get_original_attempt_query(
1632 session_id, research_id
1633 )
1634 except AttemptNotFound:
1635 return JSONResponse(
1636 {"success": False, "error": "Attempt not found"},
1637 status_code=404,
1638 )
1640 if not original_content.strip():
1641 # Empty/whitespace-only original — shouldn't happen since
1642 # send_message rejects these, but guard against a corrupt
1643 # row before we destroy the only copy.
1644 return JSONResponse(
1645 {
1646 "success": False,
1647 "error": (
1648 "Original attempt has no query content to retry."
1649 ),
1650 },
1651 status_code=400,
1652 )
1654 # Build the replacement context before deleting anything, but
1655 # exclude the target attempt in memory so it cannot become its
1656 # own prior context. Current rows link the triggering user message
1657 # through research_meta; legacy rows carry research_id directly.
1658 original_message_id = None
1659 with get_user_db_session(username) as attempt_db:
1660 attempt_row = (
1661 attempt_db.query(ResearchHistory)
1662 .filter(
1663 ResearchHistory.id == research_id,
1664 ResearchHistory.chat_session_id == session_id,
1665 )
1666 .first()
1667 )
1668 if attempt_row is None: 1668 ↛ 1669line 1668 didn't jump to line 1669 because the condition on line 1668 was never true
1669 return JSONResponse(
1670 {"success": False, "error": "Attempt not found"},
1671 status_code=404,
1672 )
1673 submission = (attempt_row.research_meta or {}).get(
1674 "submission", {}
1675 )
1676 if isinstance(submission, dict):
1677 candidate = submission.get("message_id")
1678 if isinstance(candidate, str) and candidate: 1678 ↛ 1689line 1678 didn't jump to line 1689
1679 original_message_id = candidate
1681 # include_steps=False keeps `limit` spent entirely on genuine
1682 # ChatMessage conversation turns. Progress-step rows (there can
1683 # be dozens per attempt, most recently from the very attempt
1684 # being retried) would otherwise dominate the merged window used
1685 # by get_session_messages() for client rendering and push real
1686 # conversation history out of it before this filter ever runs —
1687 # ChatContextManager discards step rows anyway, so there is no
1688 # reason to spend the window budget on them here.
1689 messages = service.get_session_messages(
1690 session_id, limit=50, include_steps=False
1691 )
1692 messages = [
1693 message
1694 for message in messages
1695 if message.get("research_id") != research_id
1696 and message.get("id") != original_message_id
1697 ][-20:]
1699 settings_snapshot = _load_settings(username)
1700 context_manager = ChatContextManager(
1701 session_id,
1702 messages,
1703 session_data.get("accumulated_context"),
1704 settings_snapshot=settings_snapshot,
1705 )
1706 research_context = context_manager.build_research_context(
1707 current_query=original_content
1708 )
1710 # Parse numeric search settings up-front. A malformed value
1711 # must return a clean 400 HERE — before the atomic write
1712 # below commits the user message + IN_PROGRESS research row.
1713 try:
1714 iterations = int(
1715 settings_snapshot.get("search.iterations", {}).get(
1716 "value", 3
1717 )
1718 )
1719 questions = int(
1720 settings_snapshot.get(
1721 "search.questions_per_iteration", {}
1722 ).get("value", 1)
1723 )
1724 except (ValueError, TypeError):
1725 return JSONResponse(
1726 {
1727 "success": False,
1728 "error": (
1729 "Invalid numeric value in search settings "
1730 "(iterations / questions_per_iteration)."
1731 ),
1732 },
1733 status_code=400,
1734 )
1736 # End all cached read transactions before taking the stable gate.
1737 with get_user_db_session(username) as ordering_db:
1738 ordering_db.rollback()
1740 pending_admission_lock = get_user_research_start_lock(username)
1741 pending_admission_lock.acquire()
1742 admission_lock = pending_admission_lock
1744 # Nothing destructive has happened yet. Recheck under the gate
1745 # retained through delete, replacement claim, spawn, and cleanup.
1746 cap_error = _check_chat_research_slot_locked(username, session_id)
1747 if cap_error is not None:
1748 message, status, active_research_id = cap_error
1749 payload = {"success": False, "error": message}
1750 if active_research_id is not None: 1750 ↛ 1752line 1750 didn't jump to line 1752 because the condition on line 1750 was always true
1751 payload["active_research_id"] = active_research_id
1752 return JSONResponse(payload, status_code=status)
1754 # Delete only after the authoritative gated recheck, so losing a
1755 # race cannot destroy the user's old attempt without admitting its
1756 # replacement. The RAG purge for documents the old attempt owned
1757 # is deferred until the gate is released (see finally below): it
1758 # does vector-store IO under FAISS locks, and the queue tick
1759 # acquires every user's gate in turn.
1760 try:
1761 rag_targets = service.delete_attempt_rows(
1762 session_id, research_id
1763 )
1764 except AttemptInProgress:
1765 return JSONResponse(
1766 {
1767 "success": False,
1768 "error": (
1769 "Research is in progress. Stop it before retrying."
1770 ),
1771 "active_research_id": research_id,
1772 },
1773 status_code=409,
1774 )
1775 except AttemptNotFound:
1776 return JSONResponse(
1777 {"success": False, "error": "Attempt not found"},
1778 status_code=404,
1779 )
1781 research_mode = "quick"
1782 new_research_id = str(uuid.uuid4())
1784 # ---- Atomic write: user message + research row in ONE txn ----
1785 # Same pattern as send_message: any IntegrityError rolls back
1786 # the user message too, so no orphan rows.
1787 try:
1788 with get_user_db_session(username) as db_session:
1789 new_message_id = service.insert_message_in_db(
1790 db_session,
1791 session_id=session_id,
1792 role="user",
1793 content=original_content,
1794 message_type="query"
1795 if len(messages) == 0
1796 else "followup",
1797 )
1799 created_at = datetime.now(UTC).isoformat()
1800 research_meta = {
1801 "submission": {
1802 "chat_session_id": session_id,
1803 "message_id": new_message_id,
1804 "research_mode": research_mode,
1805 },
1806 # Persist the run's settings snapshot so
1807 # view/export renders the same uncited-sources
1808 # mode as generation. Mirrors the main-form path;
1809 # completion merge back-fills when missing.
1810 "settings_snapshot": settings_snapshot,
1811 }
1812 db_session.add(
1813 ResearchHistory(
1814 id=new_research_id,
1815 query=original_content,
1816 mode=research_mode,
1817 status=ResearchStatus.IN_PROGRESS.value,
1818 created_at=created_at,
1819 progress_log=[{"time": created_at, "progress": 0}],
1820 research_meta=research_meta,
1821 chat_session_id=session_id,
1822 )
1823 )
1824 db_session.add(
1825 UserActiveResearch(
1826 username=username,
1827 research_id=new_research_id,
1828 status=ResearchStatus.IN_PROGRESS,
1829 thread_id=str(threading.current_thread().ident),
1830 settings_snapshot=settings_snapshot,
1831 )
1832 )
1834 try:
1835 db_session.commit()
1836 except IntegrityError:
1837 db_session.rollback()
1838 logger.warning(
1839 f"Concurrent in-progress research race for "
1840 f"chat {session_id[:8]}... (retry)"
1841 )
1842 return JSONResponse(
1843 {
1844 "success": False,
1845 "error": (
1846 "Research already in progress on this "
1847 "chat session."
1848 ),
1849 },
1850 status_code=409,
1851 )
1852 except ValueError as exc:
1853 if "not found" in str(exc).lower():
1854 return JSONResponse(
1855 {"success": False, "error": "Session not found"},
1856 status_code=404,
1857 )
1858 raise
1859 # ---- end atomic write ----
1861 pw = get_user_password(username)
1863 model_provider = settings_snapshot.get("llm.provider", {}).get(
1864 "value", ""
1865 )
1866 model = settings_snapshot.get("llm.model", {}).get("value", "")
1867 search_engine = settings_snapshot.get("search.tool", {}).get(
1868 "value", ""
1869 )
1870 # Only meaningful for the openai_endpoint provider (see above).
1871 custom_endpoint = (
1872 settings_snapshot.get("llm.openai_endpoint.url", {}).get(
1873 "value"
1874 )
1875 if normalize_provider(model_provider) == "openai_endpoint"
1876 else None
1877 )
1878 user_strategy = settings_snapshot.get(
1879 "search.search_strategy", {}
1880 ).get("value", "langgraph-agent")
1882 if research_context.get("is_multi_turn"):
1883 strategy = "enhanced-contextual-followup"
1884 research_context["delegate_strategy"] = user_strategy
1885 else:
1886 strategy = user_strategy
1888 try:
1889 start_research_process(
1890 new_research_id,
1891 original_content,
1892 research_mode,
1893 run_research_process,
1894 username=username,
1895 user_password=pw,
1896 model_provider=model_provider,
1897 model=model,
1898 search_engine=search_engine,
1899 custom_endpoint=custom_endpoint,
1900 strategy=strategy,
1901 iterations=iterations,
1902 questions_per_iteration=questions,
1903 research_context=research_context,
1904 chat_session_id=session_id,
1905 chat_message_id=new_message_id,
1906 settings_snapshot=settings_snapshot,
1907 )
1908 except DuplicateResearchError:
1909 logger.warning(
1910 f"DuplicateResearchError on chat retry_attempt "
1911 f"for {new_research_id[:8]}... (chat {session_id[:8]}...)"
1912 )
1913 _cleanup_chat_send_rows(
1914 username,
1915 new_research_id,
1916 new_message_id,
1917 session_id,
1918 "duplicate",
1919 preserve_attempt=True,
1920 )
1921 return JSONResponse(
1922 {
1923 "success": False,
1924 "error": (
1925 "Research already in progress on this chat session."
1926 ),
1927 },
1928 status_code=409,
1929 )
1930 except SystemAtCapacityError:
1931 logger.warning(
1932 f"SystemAtCapacityError on chat retry_attempt "
1933 f"for {new_research_id[:8]}... (chat {session_id[:8]}...)"
1934 )
1935 _cleanup_chat_send_rows(
1936 username,
1937 new_research_id,
1938 new_message_id,
1939 session_id,
1940 "capacity",
1941 preserve_attempt=True,
1942 )
1943 return JSONResponse(
1944 {
1945 "success": False,
1946 "error": (
1947 "Server is at research capacity. Please retry "
1948 "shortly."
1949 ),
1950 },
1951 status_code=429,
1952 )
1953 except Exception:
1954 logger.exception(
1955 f"Failed to spawn retried chat research "
1956 f"{new_research_id[:8]}... "
1957 f"(chat {session_id[:8]}...)"
1958 )
1959 _cleanup_chat_send_rows(
1960 username,
1961 new_research_id,
1962 new_message_id,
1963 session_id,
1964 "retry spawn failure",
1965 preserve_attempt=True,
1966 )
1967 return JSONResponse(
1968 {
1969 "success": False,
1970 "error": "Failed to retry research. Please try again.",
1971 },
1972 status_code=500,
1973 )
1975 logger.info(
1976 f"Retried chat attempt as research "
1977 f"{new_research_id[:8]}... for chat {session_id[:8]}..."
1978 )
1980 return JSONResponse(
1981 {
1982 "success": True,
1983 "session_id": session_id,
1984 "research_id": new_research_id,
1985 "message_id": new_message_id,
1986 "research_mode": research_mode,
1987 }
1988 )
1990 except ROUTE_EXCEPTIONS:
1991 logger.exception("Error retrying chat attempt")
1992 return JSONResponse(
1993 {"success": False, "error": "Failed to retry attempt"},
1994 status_code=500,
1995 )
1996 finally:
1997 if admission_lock is not None:
1998 admission_lock.release()
1999 # The old attempt's rows are committed as deleted whatever
2000 # happened after, so purge its RAG data now that the gate is
2001 # free. Best-effort: purge_attempt_rag logs its own failures.
2002 if rag_targets and service is not None:
2003 service.purge_attempt_rag(rag_targets)
2005 return await run_db_sync(_impl)