Coverage for src / local_deep_research / followup_research / routes.py: 100%
89 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1"""
2Flask routes for follow-up research functionality.
3"""
5from flask import Blueprint, request, jsonify, session
6from loguru import logger
8from ..constants import ResearchStatus
9from ..llm.providers.base import normalize_provider
10from .service import FollowUpResearchService
11from .models import FollowUpRequest
12from ..security.decorators import require_json_body
13from ..web.auth.decorators import login_required
14from ..web.auth.password_utils import get_user_password
16# Create blueprint
17followup_bp = Blueprint("followup", __name__, url_prefix="/api/followup")
19# NOTE: Routes use session["username"] (not .get()) intentionally.
20# @login_required guarantees the key exists; direct access fails fast
21# if the decorator is ever removed.
24@followup_bp.route("/prepare", methods=["POST"])
25@login_required
26@require_json_body(error_format="success")
27def prepare_followup():
28 """
29 Prepare a follow-up research by loading parent context.
31 Request body:
32 {
33 "parent_research_id": "uuid",
34 "question": "follow-up question"
35 }
37 Returns:
38 {
39 "success": true,
40 "parent_summary": "...",
41 "available_sources": 10,
42 "suggested_strategy": "source-based"
43 }
44 """
45 try:
46 data = request.get_json()
47 parent_id = data.get("parent_research_id")
48 question = data.get("question")
50 if not parent_id or not question:
51 return jsonify(
52 {
53 "success": False,
54 "error": "Missing parent_research_id or question",
55 }
56 ), 400
58 # Get username from session
59 username = session["username"]
61 # Get settings snapshot to use for suggested strategy
62 from ..settings.manager import SettingsManager
63 from ..database.session_context import get_user_db_session
65 with get_user_db_session(username) as db_session:
66 settings_manager = SettingsManager(db_session=db_session)
67 settings_snapshot = settings_manager.get_all_settings()
69 # Get strategy from settings
70 strategy_from_settings = settings_snapshot.get(
71 "search.search_strategy", {}
72 ).get("value", "source-based")
74 # Initialize service
75 service = FollowUpResearchService(username=username)
77 # Load parent context
78 parent_data = service.load_parent_research(parent_id)
80 if not parent_data:
81 # For now, return success with empty data to allow testing
82 logger.warning(
83 f"Parent research {parent_id} not found, returning empty context"
84 )
85 return jsonify(
86 {
87 "success": True,
88 "parent_summary": "Previous research context",
89 "available_sources": 0,
90 "suggested_strategy": strategy_from_settings, # Use strategy from settings
91 "parent_research": {
92 "id": parent_id,
93 "query": "Previous query",
94 "sources_count": 0,
95 },
96 }
97 )
99 # Prepare response with parent context summary
100 response = {
101 "success": True,
102 "parent_summary": parent_data.get("query", ""),
103 "available_sources": len(parent_data.get("resources", [])),
104 "suggested_strategy": strategy_from_settings, # Use strategy from settings
105 "parent_research": {
106 "id": parent_id,
107 "query": parent_data.get("query", ""),
108 "sources_count": len(parent_data.get("resources", [])),
109 },
110 }
112 return jsonify(response)
114 except Exception:
115 logger.exception("Error preparing follow-up")
116 return jsonify(
117 {"success": False, "error": "An internal error has occurred."}
118 ), 500
121@followup_bp.route("/start", methods=["POST"])
122@login_required
123@require_json_body(error_format="success")
124def start_followup():
125 """
126 Start a follow-up research.
128 Request body:
129 {
130 "parent_research_id": "uuid",
131 "question": "follow-up question",
132 "strategy": "source-based", # optional
133 "max_iterations": 1, # optional
134 "questions_per_iteration": 3 # optional
135 }
137 Returns:
138 {
139 "success": true,
140 "research_id": "new-uuid",
141 "message": "Follow-up research started"
142 }
143 """
144 try:
145 from ..web.services.research_service import (
146 start_research_process,
147 run_research_process,
148 )
149 import uuid
151 data = request.get_json()
153 # Get username from session
154 username = session["username"]
156 # Get settings snapshot first to use database values
157 from ..settings.manager import SettingsManager
158 from ..database.session_context import get_user_db_session
160 with get_user_db_session(username) as db_session:
161 settings_manager = SettingsManager(db_session=db_session)
162 settings_snapshot = settings_manager.get_all_settings()
164 # Get strategy from settings snapshot, fallback to source-based if not set
165 strategy_from_settings = settings_snapshot.get(
166 "search.search_strategy", {}
167 ).get("value", "source-based")
169 # Get iterations and questions from settings snapshot
170 iterations_from_settings = settings_snapshot.get(
171 "search.iterations", {}
172 ).get("value", 1)
173 questions_from_settings = settings_snapshot.get(
174 "search.questions_per_iteration", {}
175 ).get("value", 3)
177 # Create follow-up request using settings values
178 followup_request = FollowUpRequest(
179 parent_research_id=data.get("parent_research_id"),
180 question=data.get("question"),
181 strategy=strategy_from_settings, # Use strategy from settings
182 max_iterations=iterations_from_settings, # Use iterations from settings
183 questions_per_iteration=questions_from_settings, # Use questions from settings
184 )
186 # Initialize service
187 service = FollowUpResearchService(username=username)
189 # Prepare research parameters
190 research_params = service.perform_followup(followup_request)
192 logger.info(f"Research params type: {type(research_params)}")
193 logger.info(
194 f"Research params keys: {research_params.keys() if isinstance(research_params, dict) else 'Not a dict'}"
195 )
196 logger.info(
197 f"Query value: {research_params.get('query') if isinstance(research_params, dict) else 'N/A'}"
198 )
199 logger.info(
200 f"Query type: {type(research_params.get('query')) if isinstance(research_params, dict) else 'N/A'}"
201 )
203 # Get user password for metrics database access.
204 # Uses the shared helper (password_utils) so every route checks the
205 # same three sources in the same order — avoids subtle divergence.
206 # Must check BEFORE creating ResearchHistory to avoid orphaned records.
207 user_password = get_user_password(username)
209 if not user_password:
210 from ..database.encrypted_db import db_manager
212 if db_manager.has_encryption:
213 logger.error(
214 f"No password available for user {username} with encrypted database - "
215 "cannot start follow-up research (session password expired or lost after server restart)"
216 )
217 # Use success/error keys to match followup API convention
218 # (the followup frontend checks data.success and data.error)
219 return jsonify(
220 {
221 "success": False,
222 "error": "Your session has expired. Please log out and log back in to start research.",
223 }
224 ), 401
225 logger.warning(
226 f"No password available for metrics access for user {username}"
227 )
229 # Generate new research ID
230 research_id = str(uuid.uuid4())
232 # Create database entry (settings_snapshot already captured above)
233 from ..database.models import ResearchHistory
234 from datetime import datetime, UTC
236 created_at = datetime.now(UTC).isoformat()
238 with get_user_db_session(username) as db_session:
239 # Create the database entry (required for tracking)
240 research_meta = {
241 "submission": {
242 "parent_research_id": data.get("parent_research_id"),
243 "question": data.get("question"),
244 "strategy": "contextual-followup",
245 },
246 }
248 research = ResearchHistory(
249 id=research_id,
250 query=research_params["query"],
251 mode="quick", # Use 'quick' not 'quick_summary'
252 status=ResearchStatus.IN_PROGRESS,
253 created_at=created_at,
254 progress_log=[{"time": created_at, "progress": 0}],
255 research_meta=research_meta,
256 )
257 db_session.add(research)
258 db_session.commit()
259 logger.info(
260 f"Created follow-up research entry with ID: {research_id}"
261 )
263 # Start the research process using the existing infrastructure
264 # Use quick_summary mode for follow-ups by default
265 logger.info(
266 f"Starting follow-up research for query of type: {type(research_params.get('query'))}"
267 )
269 # Get model and search settings from user's settings
270 model_provider = settings_snapshot.get("llm.provider", {}).get(
271 "value", "ollama"
272 )
273 # Normalize provider to lowercase canonical form
274 model_provider = normalize_provider(model_provider)
275 model = settings_snapshot.get("llm.model", {}).get(
276 "value", "gemma3:12b"
277 )
278 search_engine = settings_snapshot.get("search.tool", {}).get(
279 "value", "searxng"
280 )
281 custom_endpoint = settings_snapshot.get(
282 "llm.openai_endpoint.url", {}
283 ).get("value")
285 start_research_process(
286 research_id,
287 research_params["query"],
288 "quick", # Use 'quick' for quick summary mode
289 run_research_process,
290 username=username,
291 user_password=user_password, # Pass password for metrics database access
292 model_provider=model_provider, # Pass model provider
293 model=model, # Pass model name
294 search_engine=search_engine, # Pass search engine
295 custom_endpoint=custom_endpoint, # Pass custom endpoint if any
296 strategy="enhanced-contextual-followup", # Use enhanced contextual follow-up strategy
297 iterations=research_params["max_iterations"],
298 questions_per_iteration=research_params["questions_per_iteration"],
299 delegate_strategy=research_params.get(
300 "delegate_strategy", "source-based"
301 ),
302 research_context=research_params["research_context"],
303 parent_research_id=research_params[
304 "parent_research_id"
305 ], # Pass parent research ID
306 settings_snapshot=settings_snapshot,
307 )
309 return jsonify(
310 {
311 "success": True,
312 "research_id": research_id,
313 "message": "Follow-up research started",
314 }
315 )
317 except Exception:
318 logger.exception("Error starting follow-up")
319 return jsonify(
320 {"success": False, "error": "An internal error has occurred."}
321 ), 500