Coverage for src / local_deep_research / followup_research / routes.py: 83%
93 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-25 01:07 +0000
1"""
2Flask routes for follow-up research functionality.
3"""
5from flask import Blueprint, request, jsonify, session, g
6from loguru import logger
8from ..constants import ResearchStatus
9from .service import FollowUpResearchService
10from .models import FollowUpRequest
11from ..web.auth.decorators import login_required
13# Create blueprint
14followup_bp = Blueprint("followup", __name__, url_prefix="/api/followup")
17@followup_bp.route("/prepare", methods=["POST"])
18@login_required
19def prepare_followup():
20 """
21 Prepare a follow-up research by loading parent context.
23 Request body:
24 {
25 "parent_research_id": "uuid",
26 "question": "follow-up question"
27 }
29 Returns:
30 {
31 "success": true,
32 "parent_summary": "...",
33 "available_sources": 10,
34 "suggested_strategy": "source-based"
35 }
36 """
37 try:
38 data = request.get_json()
39 parent_id = data.get("parent_research_id")
40 question = data.get("question")
42 if not parent_id or not question:
43 return jsonify(
44 {
45 "success": False,
46 "error": "Missing parent_research_id or question",
47 }
48 ), 400
50 # Get username from session
51 username = session.get("username")
53 # Get settings snapshot to use for suggested strategy
54 from ..settings.manager import SettingsManager
55 from ..database.session_context import get_user_db_session
57 with get_user_db_session(username) as db_session:
58 settings_manager = SettingsManager(db_session=db_session)
59 settings_snapshot = settings_manager.get_all_settings()
61 # Get strategy from settings
62 strategy_from_settings = settings_snapshot.get(
63 "search.search_strategy", {}
64 ).get("value", "source-based")
66 # Initialize service
67 service = FollowUpResearchService(username=username)
69 # Load parent context
70 parent_data = service.load_parent_research(parent_id)
72 if not parent_data:
73 # For now, return success with empty data to allow testing
74 logger.warning(
75 f"Parent research {parent_id} not found, returning empty context"
76 )
77 return jsonify(
78 {
79 "success": True,
80 "parent_summary": "Previous research context",
81 "available_sources": 0,
82 "suggested_strategy": strategy_from_settings, # Use strategy from settings
83 "parent_research": {
84 "id": parent_id,
85 "query": "Previous query",
86 "sources_count": 0,
87 },
88 }
89 )
91 # Prepare response with parent context summary
92 response = {
93 "success": True,
94 "parent_summary": parent_data.get("query", ""),
95 "available_sources": len(parent_data.get("resources", [])),
96 "suggested_strategy": strategy_from_settings, # Use strategy from settings
97 "parent_research": {
98 "id": parent_id,
99 "query": parent_data.get("query", ""),
100 "sources_count": len(parent_data.get("resources", [])),
101 },
102 }
104 return jsonify(response)
106 except Exception:
107 logger.exception("Error preparing follow-up")
108 return jsonify(
109 {"success": False, "error": "An internal error has occurred."}
110 ), 500
113@followup_bp.route("/start", methods=["POST"])
114@login_required
115def start_followup():
116 """
117 Start a follow-up research.
119 Request body:
120 {
121 "parent_research_id": "uuid",
122 "question": "follow-up question",
123 "strategy": "source-based", # optional
124 "max_iterations": 1, # optional
125 "questions_per_iteration": 3 # optional
126 }
128 Returns:
129 {
130 "success": true,
131 "research_id": "new-uuid",
132 "message": "Follow-up research started"
133 }
134 """
135 try:
136 from ..web.services.research_service import (
137 start_research_process,
138 run_research_process,
139 )
140 from ..web.routes.globals import active_research, termination_flags
141 import uuid
143 data = request.get_json()
145 # Get username from session
146 username = session.get("username")
148 # Get settings snapshot first to use database values
149 from ..settings.manager import SettingsManager
150 from ..database.session_context import get_user_db_session
152 with get_user_db_session(username) as db_session:
153 settings_manager = SettingsManager(db_session=db_session)
154 settings_snapshot = settings_manager.get_all_settings()
156 # Get strategy from settings snapshot, fallback to source-based if not set
157 strategy_from_settings = settings_snapshot.get(
158 "search.search_strategy", {}
159 ).get("value", "source-based")
161 # Get iterations and questions from settings snapshot
162 iterations_from_settings = settings_snapshot.get(
163 "search.iterations", {}
164 ).get("value", 1)
165 questions_from_settings = settings_snapshot.get(
166 "search.questions_per_iteration", {}
167 ).get("value", 3)
169 # Create follow-up request using settings values
170 followup_request = FollowUpRequest(
171 parent_research_id=data.get("parent_research_id"),
172 question=data.get("question"),
173 strategy=strategy_from_settings, # Use strategy from settings
174 max_iterations=iterations_from_settings, # Use iterations from settings
175 questions_per_iteration=questions_from_settings, # Use questions from settings
176 )
178 # Initialize service
179 service = FollowUpResearchService(username=username)
181 # Prepare research parameters
182 research_params = service.perform_followup(followup_request)
184 logger.info(f"Research params type: {type(research_params)}")
185 logger.info(
186 f"Research params keys: {research_params.keys() if isinstance(research_params, dict) else 'Not a dict'}"
187 )
188 logger.info(
189 f"Query value: {research_params.get('query') if isinstance(research_params, dict) else 'N/A'}"
190 )
191 logger.info(
192 f"Query type: {type(research_params.get('query')) if isinstance(research_params, dict) else 'N/A'}"
193 )
195 # Generate new research ID
196 research_id = str(uuid.uuid4())
198 # Create database entry (settings_snapshot already captured above)
199 from ..database.models import ResearchHistory
200 from datetime import datetime, UTC
202 created_at = datetime.now(UTC).isoformat()
204 with get_user_db_session(username) as db_session:
205 # Create the database entry (required for tracking)
206 research_meta = {
207 "submission": {
208 "parent_research_id": data.get("parent_research_id"),
209 "question": data.get("question"),
210 "strategy": "contextual-followup",
211 },
212 }
214 research = ResearchHistory(
215 id=research_id,
216 query=research_params["query"],
217 mode="quick", # Use 'quick' not 'quick_summary'
218 status=ResearchStatus.IN_PROGRESS,
219 created_at=created_at,
220 progress_log=[{"time": created_at, "progress": 0}],
221 research_meta=research_meta,
222 )
223 db_session.add(research)
224 db_session.commit()
225 logger.info(
226 f"Created follow-up research entry with ID: {research_id}"
227 )
229 # Start the research process using the existing infrastructure
230 # Use quick_summary mode for follow-ups by default
231 logger.info(
232 f"Starting follow-up research for query of type: {type(research_params.get('query'))}"
233 )
235 # Get user password for metrics database access
236 user_password = None
237 session_id = session.get("session_id")
238 if session_id: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 from ..database.session_passwords import session_password_store
241 user_password = session_password_store.retrieve(
242 username, session_id
243 )
245 # Fallback to g.user_password (set by middleware if temp_auth was used)
246 if not user_password: 246 ↛ 250line 246 didn't jump to line 250 because the condition on line 246 was always true
247 user_password = getattr(g, "user_password", None)
249 # Last resort: try temp_auth_store
250 if not user_password: 250 ↛ 260line 250 didn't jump to line 260 because the condition on line 250 was always true
251 from ..database.temp_auth import temp_auth_store
253 auth_token = session.get("temp_auth_token")
254 if auth_token: 254 ↛ 256line 254 didn't jump to line 256 because the condition on line 254 was never true
255 # Use peek_auth to avoid consuming the token
256 auth_data = temp_auth_store.peek_auth(auth_token)
257 if auth_data and auth_data[0] == username:
258 user_password = auth_data[1]
260 if not user_password: 260 ↛ 266line 260 didn't jump to line 266 because the condition on line 260 was always true
261 logger.warning(
262 f"No password available for metrics access for user {username}"
263 )
265 # Get model and search settings from user's settings
266 model_provider = settings_snapshot.get("llm.provider", {}).get(
267 "value", "OLLAMA"
268 )
269 model = settings_snapshot.get("llm.model", {}).get(
270 "value", "gemma3:12b"
271 )
272 search_engine = settings_snapshot.get("search.tool", {}).get(
273 "value", "searxng"
274 )
275 custom_endpoint = settings_snapshot.get(
276 "llm.openai_endpoint.url", {}
277 ).get("value")
279 start_research_process(
280 research_id,
281 research_params["query"],
282 "quick", # Use 'quick' for quick summary mode
283 active_research,
284 termination_flags,
285 run_research_process,
286 username=username,
287 user_password=user_password, # Pass password for metrics database access
288 model_provider=model_provider, # Pass model provider
289 model=model, # Pass model name
290 search_engine=search_engine, # Pass search engine
291 custom_endpoint=custom_endpoint, # Pass custom endpoint if any
292 strategy="enhanced-contextual-followup", # Use enhanced contextual follow-up strategy
293 iterations=research_params["max_iterations"],
294 questions_per_iteration=research_params["questions_per_iteration"],
295 delegate_strategy=research_params.get(
296 "delegate_strategy", "source-based"
297 ),
298 research_context=research_params["research_context"],
299 parent_research_id=research_params[
300 "parent_research_id"
301 ], # Pass parent research ID
302 settings_snapshot=settings_snapshot,
303 )
305 return jsonify(
306 {
307 "success": True,
308 "research_id": research_id,
309 "message": "Follow-up research started",
310 }
311 )
313 except Exception:
314 logger.exception("Error starting follow-up")
315 return jsonify(
316 {"success": False, "error": "An internal error has occurred."}
317 ), 500