Coverage for src/local_deep_research/benchmarks/web_api/benchmark_service.py: 96%
468 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""Benchmark service for handling web-based benchmark execution."""
3import hashlib
4import json
5import threading
6import time
7from datetime import UTC, datetime
8from enum import Enum
9from typing import Any, Dict, List, Optional
11from loguru import logger
12from sqlalchemy.exc import SQLAlchemyError
14from ...api.research_functions import quick_summary
15from ...settings.manager import SnapshotSettingsContext
16from ...web.services.research_service import _global_research_semaphore
17from ...database.models.benchmark import (
18 BenchmarkResult,
19 BenchmarkRun,
20 BenchmarkStatus,
21 DatasetType,
22)
23from ...web.services.socket_service import SocketIOService
24from ..datasets import load_dataset
25from ..graders import extract_answer_from_response, grade_single_result
26from ..runners import format_query
27from ...database.thread_local_session import thread_cleanup
29# Generic failure message surfaced to API clients in place of the raw benchmark
30# exception. ``benchmark_run.error_message`` is set from ``str(e)`` of the
31# benchmark run (LLM calls, search engines, grading — the same stack as
32# research) and returned by get_benchmark_status() via
33# GET /benchmark/api/status/<id>. The raw text can carry server-level LLM
34# endpoints / hosts / filesystem paths (settings_snapshot includes LDR_* env
35# overrides), so it must not reach the client (CWE-209). Full detail stays in
36# the logger.exception calls at the failure sites.
37_GENERIC_BENCHMARK_ERROR = (
38 "Benchmark run failed due to an internal error. "
39 "Check the server logs for details."
40)
43class BenchmarkTaskStatus(Enum):
44 """Status values for benchmark tasks in the queue tracker."""
46 QUEUED = "queued"
47 PROCESSING = "processing"
48 COMPLETED = "completed"
49 FAILED = "failed"
50 CANCELLED = "cancelled"
53class BenchmarkQueueTracker:
54 """Simple in-memory tracker for benchmark queue status.
56 This replaces the removed memory_queue functionality for benchmarks.
57 Since benchmarks are temporary and don't need persistence,
58 this simple in-memory solution is sufficient.
60 Thread-safe for concurrent access from multiple benchmark threads.
61 """
63 def __init__(self):
64 self.tasks = {}
65 self._lock = threading.Lock()
67 def add_task(
68 self, task_id: str, username: str, task_type: str = "benchmark"
69 ):
70 """Add a new task to tracking.
72 Also performs opportunistic cleanup of old completed tasks.
73 """
74 # Cleanup old tasks before adding new one (outside lock for better performance)
75 self.cleanup_completed_tasks()
77 with self._lock:
78 self.tasks[task_id] = {
79 "username": username,
80 "task_type": task_type,
81 "status": BenchmarkTaskStatus.QUEUED.value,
82 "created_at": datetime.now(UTC),
83 }
85 def update_task_status(self, task_id: str, status: BenchmarkTaskStatus):
86 """Update the status of a task."""
87 with self._lock:
88 if task_id in self.tasks:
89 self.tasks[task_id]["status"] = status.value
90 self.tasks[task_id]["updated_at"] = datetime.now(UTC)
91 else:
92 logger.warning(
93 f"Attempted to update status for non-existent task: {task_id}"
94 )
96 def get_task_status(self, task_id: str) -> Optional[Dict]:
97 """Get the current status of a task."""
98 with self._lock:
99 return self.tasks.get(task_id)
101 def remove_task(self, task_id: str):
102 """Remove a task from tracking."""
103 with self._lock:
104 self.tasks.pop(task_id, None)
106 def cleanup_completed_tasks(self, max_age_seconds: int = 3600):
107 """Remove completed tasks older than max_age_seconds.
109 Args:
110 max_age_seconds: Maximum age in seconds for completed tasks (default 1 hour)
111 """
112 with self._lock:
113 now = datetime.now(UTC)
114 to_remove = []
115 for task_id, task_data in self.tasks.items():
116 # Only cleanup completed, failed, or cancelled tasks
117 if task_data["status"] in [
118 BenchmarkTaskStatus.COMPLETED.value,
119 BenchmarkTaskStatus.FAILED.value,
120 BenchmarkTaskStatus.CANCELLED.value,
121 ]:
122 # Check if task has updated_at timestamp
123 updated_at = task_data.get(
124 "updated_at", task_data.get("created_at")
125 )
126 if updated_at: 126 ↛ 115line 126 didn't jump to line 115 because the condition on line 126 was always true
127 age = (now - updated_at).total_seconds()
128 if age > max_age_seconds:
129 to_remove.append(task_id)
131 for task_id in to_remove:
132 self.tasks.pop(task_id, None)
133 logger.debug(f"Cleaned up old task: {task_id}")
135 if to_remove:
136 logger.info(f"Cleaned up {len(to_remove)} old benchmark tasks")
139class BenchmarkService:
140 """Service for managing benchmark runs through the web interface."""
142 def __init__(self, socket_service=None):
143 self.active_runs: Dict[int, Dict] = {}
144 self.socket_service = socket_service or self._get_socket_service()
145 self.rate_limit_detected: Dict[
146 int, bool
147 ] = {} # Track rate limiting per benchmark run
148 self.queue_tracker = BenchmarkQueueTracker() # Initialize queue tracker
149 # Serializes benchmark-result persistence. _sync_results_to_database
150 # runs on the worker thread while sync_pending_results runs on the
151 # request thread; without this they can both INSERT the same
152 # (benchmark_run_id, query_hash) row and trip the uix_run_query
153 # unique constraint, which rolls back the whole pending batch.
154 self._results_sync_lock = threading.Lock()
156 def _get_socket_service(self):
157 """Get socket service instance, handling cases where Flask app is not available."""
158 try:
159 return SocketIOService()
160 except Exception:
161 # Return a mock socket service for testing/standalone use
162 class MockSocketService:
163 def emit_to_room(self, room, event, data):
164 pass
166 return MockSocketService()
168 def generate_config_hash(self, search_config: Dict[str, Any]) -> str:
169 """Generate a hash for search configuration compatibility checking."""
170 relevant_params = {
171 "iterations": search_config.get("iterations"),
172 "questions_per_iteration": search_config.get(
173 "questions_per_iteration"
174 ),
175 "search_tool": search_config.get("search_tool"),
176 "search_strategy": search_config.get("search_strategy"),
177 "model_name": search_config.get("model_name"),
178 "provider": search_config.get("provider"),
179 }
180 # Remove None values
181 relevant_params = {
182 k: v for k, v in relevant_params.items() if v is not None
183 }
184 config_str = json.dumps(relevant_params, sort_keys=True)
185 return hashlib.md5( # DevSkim: ignore DS126858
186 config_str.encode(), usedforsecurity=False
187 ).hexdigest()[:8]
189 def generate_query_hash(
190 self, question: str, dataset_type: str, example_id: str
191 ) -> str:
192 """Generate a hash for a query to enable deduplication.
194 The example id is part of the key so two distinct examples that share
195 the same question text (and dataset) get distinct hashes and each
196 persist a row, instead of colliding on the ``uix_run_query`` unique
197 constraint and dropping one. Re-hashing the same example is stable, so
198 the sync dedup still collapses a genuine re-sync of that example.
199 """
200 query_content = (
201 f"{example_id}|{question.strip()}|{dataset_type.lower()}"
202 )
203 return hashlib.md5( # DevSkim: ignore DS126858
204 query_content.encode(), usedforsecurity=False
205 ).hexdigest()
207 def create_benchmark_run(
208 self,
209 run_name: Optional[str],
210 search_config: Dict[str, Any],
211 evaluation_config: Dict[str, Any],
212 datasets_config: Dict[str, Dict],
213 username: Optional[str] = None,
214 user_password: Optional[str] = None,
215 ) -> int:
216 """Create a new benchmark run in the database."""
217 from ...database.session_context import get_user_db_session
219 with get_user_db_session(username, user_password) as session:
220 try:
221 config_hash = self.generate_config_hash(search_config)
223 # Calculate total examples
224 total_examples = sum(
225 config.get("count", 0)
226 for config in datasets_config.values()
227 )
229 benchmark_run = BenchmarkRun(
230 run_name=run_name,
231 config_hash=config_hash,
232 query_hash_list=[], # Will be populated as we process
233 search_config=search_config,
234 evaluation_config=evaluation_config,
235 datasets_config=datasets_config,
236 total_examples=total_examples,
237 status=BenchmarkStatus.PENDING,
238 )
240 session.add(benchmark_run)
241 session.commit()
243 logger.info(
244 f"Created benchmark run {benchmark_run.id} with config hash {config_hash}"
245 )
246 return int(benchmark_run.id)
248 except Exception:
249 session.rollback()
250 logger.exception("Error creating benchmark run")
251 raise
253 def start_benchmark(
254 self,
255 benchmark_run_id: int,
256 username: Optional[str] = None,
257 user_password: Optional[str] = None,
258 ) -> bool:
259 """Start a benchmark run in a background thread."""
260 from ...database.session_context import get_user_db_session
262 try:
263 # Get all data from the database in the main thread
264 # This avoids database access from the background thread
265 with get_user_db_session(username, user_password) as session:
266 # Get benchmark run details
267 benchmark_run = (
268 session.query(BenchmarkRun)
269 .filter(BenchmarkRun.id == benchmark_run_id)
270 .first()
271 )
272 if not benchmark_run:
273 raise ValueError( # noqa: TRY301 — caught by except, sets FAILED status in DB
274 f"Benchmark run {benchmark_run_id} not found"
275 )
277 # Create settings snapshot for thread safety + provenance.
278 # The settings snapshot is BEST-EFFORT provenance, not a
279 # critical path: a recoverable read failure must not break
280 # the user's ability to start a benchmark. We catch only the
281 # failures that can legitimately escape get_all_settings:
282 # SQLAlchemyError (transient DB issues) and LookupError /
283 # ValueError (a stale enum-typed settings row, e.g. a legacy
284 # 'CHAT' value that fails enum coercion). Those degrade to an
285 # empty snapshot — ldr_version is still recorded. Anything
286 # else (a genuine programming error) is deliberately NOT
287 # swallowed here: it propagates to the outer handler, which
288 # logs a full traceback and marks the run FAILED — the right
289 # outcome, since a settings subsystem that breaks unexpectedly
290 # would also corrupt the benchmark's own config reads.
291 from local_deep_research.settings import SettingsManager
293 settings_manager = SettingsManager(session)
294 try:
295 settings_snapshot = settings_manager.get_all_settings()
296 except (SQLAlchemyError, LookupError, ValueError):
297 # Use logger.exception so the traceback goes to stderr
298 # without splatting `repr(exc)` into the message body —
299 # exception details may include user data.
300 logger.exception(
301 "Failed to capture settings snapshot for benchmark "
302 f"{benchmark_run_id}; running with empty snapshot."
303 )
304 settings_snapshot = {}
306 # Get user password for metrics tracking in background thread
307 from flask import session as flask_session
308 from ...database.session_passwords import session_password_store
310 _user_password = None
311 session_id = flask_session.get("session_id")
312 if session_id and username: 312 ↛ 324line 312 didn't jump to line 324 because the condition on line 312 was always true
313 _user_password = (
314 session_password_store.get_session_password(
315 username, session_id
316 )
317 )
318 if not _user_password: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 logger.warning(
320 f"No password found for user {username} in current session"
321 )
323 # Extract all data we need
324 benchmark_data = {
325 "benchmark_run_id": benchmark_run_id,
326 "username": username or "benchmark_user",
327 "user_password": _user_password, # Add password for metrics tracking
328 "config_hash": benchmark_run.config_hash,
329 "datasets_config": benchmark_run.datasets_config,
330 "search_config": benchmark_run.search_config,
331 "evaluation_config": benchmark_run.evaluation_config,
332 "settings_snapshot": settings_snapshot, # Add settings snapshot
333 }
335 # Update status in database. Persist provenance (ldr_version
336 # + redacted settings_snapshot) BEFORE the commit so the
337 # YAML download can later reproduce the run; the snapshot
338 # already exists in memory from line 306, we just store the
339 # redacted form so secrets don't sit in the DB even though
340 # SQLCipher protects at rest.
341 from local_deep_research import __version__
342 from local_deep_research.security.data_sanitizer import (
343 DataSanitizer,
344 )
346 benchmark_run.ldr_version = __version__
347 benchmark_run.settings_snapshot = (
348 DataSanitizer.redact_settings_snapshot(settings_snapshot)
349 )
350 benchmark_run.status = BenchmarkStatus.IN_PROGRESS
351 benchmark_run.start_time = datetime.now(UTC)
352 session.commit()
354 # Store data in memory for the thread
355 self.active_runs[benchmark_run_id] = {
356 "data": benchmark_data,
357 "start_time": datetime.now(UTC),
358 "status": "running",
359 "results": [],
360 }
362 # Start background thread
363 thread = threading.Thread(
364 target=self._run_benchmark_thread,
365 args=(benchmark_run_id,),
366 daemon=True,
367 )
368 thread.start()
370 self.active_runs[benchmark_run_id]["thread"] = thread
372 logger.info(f"Started benchmark run {benchmark_run_id}")
373 return True
375 except Exception as e:
376 logger.exception(f"Error starting benchmark {benchmark_run_id}")
377 # If we populated active_runs before the spawn failed, drop the
378 # stale entry — it has no thread and would mislead subsequent
379 # cancel_benchmark / get_run_status calls.
380 self.active_runs.pop(benchmark_run_id, None)
381 # Update status using user database
382 with get_user_db_session(username, user_password) as session:
383 benchmark_run = (
384 session.query(BenchmarkRun)
385 .filter(BenchmarkRun.id == benchmark_run_id)
386 .first()
387 )
388 if benchmark_run:
389 benchmark_run.status = BenchmarkStatus.FAILED
390 benchmark_run.error_message = str(e)
391 session.commit()
392 return False
394 @thread_cleanup
395 def _run_benchmark_thread(self, benchmark_run_id: int):
396 """Main benchmark execution thread."""
397 # IMPORTANT: This runs in a background thread, so we cannot access the user database
398 # Using in-memory queue tracker for benchmark status tracking
400 task_id = None
402 # Get the benchmark data that was passed to us
403 # We need to retrieve this from the service database or from memory
404 benchmark_data = self.active_runs.get(benchmark_run_id, {}).get("data")
406 try:
407 if not benchmark_data:
408 raise ValueError( # noqa: TRY301
409 f"Benchmark data for run {benchmark_run_id} not found"
410 )
411 # Set up settings context for thread-local access
412 settings_snapshot = benchmark_data.get("settings_snapshot", {})
413 username = benchmark_data.get("username", "benchmark_user")
415 # Create a settings context that threads can use
416 settings_context = SnapshotSettingsContext(
417 settings_snapshot,
418 username=username,
419 missing_key_log_level="WARNING",
420 )
422 # Set the context in thread-local storage
423 from ...config.thread_settings import set_settings_context
425 set_settings_context(settings_context)
427 # Extract all the data we need
428 datasets_config = benchmark_data["datasets_config"]
429 search_config = benchmark_data["search_config"]
430 evaluation_config = benchmark_data["evaluation_config"]
432 # Create task queue
433 task_queue = self._create_task_queue(
434 datasets_config,
435 benchmark_run_id,
436 )
438 # Calculate totals
439 total_examples = len(task_queue)
440 completed_examples = 0
442 # Initialize task tracking
443 task_id = f"benchmark_{benchmark_run_id}_{int(datetime.now(UTC).timestamp())}"
444 username = benchmark_data.get("username", "benchmark_user")
445 self.queue_tracker.add_task(task_id, username, "benchmark")
446 self.queue_tracker.update_task_status(
447 task_id, BenchmarkTaskStatus.PROCESSING
448 )
450 # Track progress in memory
451 progress_info = {
452 "total_examples": total_examples,
453 "completed_examples": completed_examples,
454 "failed_examples": 0,
455 "start_time": datetime.now(UTC),
456 }
458 # Process tasks
459 logger.info(
460 f"Benchmark {benchmark_run_id} starting to process {len(task_queue)} tasks"
461 )
462 for i, task in enumerate(task_queue):
463 # Check if benchmark has been cancelled
464 if (
465 benchmark_run_id in self.active_runs
466 and self.active_runs[benchmark_run_id].get("status")
467 == "cancelled"
468 ):
469 logger.info(
470 f"Benchmark {benchmark_run_id} was cancelled, stopping processing"
471 )
472 break
474 logger.info(
475 f"Benchmark {benchmark_run_id} processing task {i + 1}/{len(task_queue)}"
476 )
477 try:
478 # Add username and password to task for metrics tracking
479 task["username"] = benchmark_data.get("username")
480 task["user_password"] = benchmark_data.get("user_password")
482 # Acquire the global research semaphore so benchmark
483 # tasks count against the server-wide concurrency limit
484 _global_research_semaphore.acquire()
485 try:
486 # Process single task
487 result = self._process_benchmark_task(
488 task,
489 search_config,
490 evaluation_config,
491 )
492 finally:
493 _global_research_semaphore.release()
495 # Store result in memory for now (will be saved later)
496 if "results" not in self.active_runs[benchmark_run_id]: 496 ↛ 497line 496 didn't jump to line 497 because the condition on line 496 was never true
497 self.active_runs[benchmark_run_id]["results"] = []
498 self.active_runs[benchmark_run_id]["results"].append(result)
500 # Update progress
501 progress_info["completed_examples"] += 1
503 logger.info(
504 f"Benchmark {benchmark_run_id} task {i + 1}/{len(task_queue)} completed successfully. "
505 f"Progress: {progress_info['completed_examples']}/{progress_info['total_examples']} total examples"
506 )
508 # Send real-time update
509 self._send_progress_update(
510 benchmark_run_id,
511 progress_info["completed_examples"],
512 progress_info["total_examples"],
513 )
515 except Exception as e:
516 logger.exception(f"Error processing task {i}")
517 progress_info["failed_examples"] += 1
518 logger.info(
519 f"Benchmark {benchmark_run_id} task {i + 1}/{len(task_queue)} failed. "
520 f"Total failed: {progress_info['failed_examples']}"
521 )
523 # Check if this is a rate limiting error
524 error_str = str(e).lower()
525 if ( 525 ↛ 462line 525 didn't jump to line 462 because the condition on line 525 was always true
526 "403" in error_str
527 or "rate limit" in error_str
528 or "forbidden" in error_str
529 ):
530 self.rate_limit_detected[benchmark_run_id] = True
531 # Send rate limit warning via WebSocket
532 self.socket_service.emit_to_subscribers(
533 "research_progress",
534 benchmark_run_id,
535 {
536 "rate_limit_detected": True,
537 "message": "SearXNG rate limiting detected",
538 },
539 )
541 # Mark as completed in memory tracker
542 progress_info["end_time"] = datetime.now(UTC)
544 # Check if benchmark was cancelled
545 was_cancelled = (
546 benchmark_run_id in self.active_runs
547 and self.active_runs[benchmark_run_id].get("status")
548 == "cancelled"
549 )
551 if was_cancelled:
552 status = BenchmarkStatus.CANCELLED
553 message = "Benchmark cancelled by user"
554 if task_id: 554 ↛ 567line 554 didn't jump to line 567 because the condition on line 554 was always true
555 self.queue_tracker.update_task_status(
556 task_id, BenchmarkTaskStatus.CANCELLED
557 )
558 else:
559 status = BenchmarkStatus.COMPLETED
560 message = "Benchmark completed successfully"
561 if task_id: 561 ↛ 567line 561 didn't jump to line 567 because the condition on line 561 was always true
562 self.queue_tracker.update_task_status(
563 task_id, BenchmarkTaskStatus.COMPLETED
564 )
566 # Store completion info for later database update
567 self.active_runs[benchmark_run_id]["completion_info"] = {
568 "status": status,
569 "end_time": progress_info["end_time"],
570 "completed_examples": progress_info["completed_examples"],
571 "failed_examples": progress_info["failed_examples"],
572 }
574 # Send completion notification
575 self.socket_service.emit_to_subscribers(
576 "research_progress",
577 benchmark_run_id,
578 {
579 "status": "cancelled" if was_cancelled else "completed",
580 "message": message,
581 "progress": (
582 progress_info["completed_examples"]
583 / progress_info["total_examples"]
584 * 100
585 )
586 if progress_info["total_examples"] > 0
587 else 0,
588 "benchmark_run_id": benchmark_run_id,
589 },
590 )
592 except Exception as e:
593 logger.exception(f"Benchmark run {benchmark_run_id} failed")
594 # Update task status if we have a task_id
595 if task_id: 595 ↛ 596line 595 didn't jump to line 596 because the condition on line 595 was never true
596 self.queue_tracker.update_task_status(
597 task_id, BenchmarkTaskStatus.FAILED
598 )
599 # Store failure info for later database update
600 if benchmark_run_id in self.active_runs: 600 ↛ 607line 600 didn't jump to line 607 because the condition on line 600 was always true
601 self.active_runs[benchmark_run_id]["completion_info"] = {
602 "status": BenchmarkStatus.FAILED,
603 "error_message": str(e),
604 }
605 finally:
606 # Clean up active run tracking
607 if benchmark_run_id in self.active_runs: 607 ↛ exitline 607 didn't return from function '_run_benchmark_thread' because the condition on line 607 was always true
608 # Mark that thread is done but keep data for database update
609 self.active_runs[benchmark_run_id]["thread_complete"] = True
611 # Try to save results to database immediately if possible
612 self._sync_results_to_database(benchmark_run_id)
614 def _create_task_queue(
615 self,
616 datasets_config: Dict,
617 benchmark_run_id: int,
618 ) -> List[Dict]:
619 """Create list of tasks to process.
621 Each dataset config may carry a "seed" for reproducible sampling:
622 the same seed and count select the same questions on every run,
623 keeping accuracy comparable across runs. Without a seed the sample
624 is random each time.
625 """
626 tasks: List[Dict[str, Any]] = []
628 for dataset_name, config in datasets_config.items():
629 if config.get("count", 0) > 0:
630 dataset = load_dataset(
631 dataset_type=dataset_name,
632 num_examples=config["count"],
633 seed=config.get("seed"),
634 )
636 for i, example in enumerate(dataset):
637 # All registered datasets expose "problem"/"answer"
638 question = example.get("problem", "")
639 correct_answer = example.get("answer", "")
640 example_id = example.get("id", f"example_{i}")
642 # Generate query hash (keyed on the example so distinct
643 # examples with identical question text don't collide)
644 query_hash = self.generate_query_hash(
645 question, dataset_name, example_id
646 )
648 tasks.append(
649 {
650 "benchmark_run_id": benchmark_run_id,
651 "example_id": example_id,
652 "dataset_type": dataset_name,
653 "question": question,
654 "correct_answer": correct_answer,
655 "query_hash": query_hash,
656 "task_index": len(tasks),
657 }
658 )
660 return tasks
662 def _process_benchmark_task(
663 self, task: Dict, search_config: Dict, evaluation_config: Dict
664 ) -> Dict:
665 """Process a single benchmark task."""
666 try:
667 logger.info(
668 f"Starting benchmark task {task['task_index'] + 1}: "
669 f"example_id={task['example_id']}, dataset={task['dataset_type']}, "
670 f"question_preview='{task['question'][:100]}...'"
671 )
673 # Get settings context from thread-local storage
674 from ...config.thread_settings import get_settings_context
676 settings_context = get_settings_context()
678 # Generate a unique tracking ID for this benchmark task
679 import uuid
681 tracking_id = str(uuid.uuid4())
682 logger.info(
683 f"Task {task['example_id']} assigned tracking_id: {tracking_id}"
684 )
686 # Format query
687 formatted_query = format_query(
688 task["question"], task["dataset_type"]
689 )
690 logger.info(
691 f"Task {task['example_id']} formatted query: '{formatted_query[:150]}...'"
692 )
694 # Run research with progress callback for WebSocket updates
695 start_time = time.time()
696 logger.info(f"Task {task['example_id']} starting research phase...")
698 def benchmark_progress_callback(
699 status: str, progress: int, data: dict
700 ):
701 """Progress callback to emit detailed research progress via WebSocket"""
702 try:
703 timestamp = datetime.now(UTC).isoformat()
705 # Create research-compatible log entry
706 log_entry = {
707 "time": timestamp,
708 "message": f"Example {task['example_id']}: {status}",
709 "progress": progress,
710 "metadata": {
711 "phase": data.get("phase", "benchmark_processing"),
712 "type": data.get("type", "info"),
713 "example_id": task["example_id"],
714 "benchmark_run_id": task["benchmark_run_id"],
715 **data, # Include all other data
716 },
717 }
719 # Determine log type based on status/message content
720 if (
721 "complete" in status.lower()
722 or "finished" in status.lower()
723 ):
724 log_entry["metadata"]["type"] = "milestone"
725 elif (
726 "error" in status.lower() or "failed" in status.lower()
727 ):
728 log_entry["metadata"]["type"] = "error"
729 elif (
730 "starting" in status.lower()
731 or "begin" in status.lower()
732 ):
733 log_entry["metadata"]["type"] = "milestone"
735 # Create progress data in research format
736 progress_data = {
737 "progress": progress,
738 "message": status,
739 "status": "in_progress",
740 "log_entry": log_entry,
741 "progress_log": json.dumps(
742 [log_entry]
743 ), # Array format expected by socket.js
744 }
746 # Emit using research_progress format that the UI expects
747 self.socket_service.emit_to_subscribers(
748 "research_progress",
749 task["benchmark_run_id"],
750 progress_data,
751 )
753 except Exception:
754 logger.exception("Error sending benchmark progress update")
756 # Get user password from task data
757 user_password = task.get("user_password")
759 search_result = quick_summary(
760 query=formatted_query,
761 research_id=tracking_id, # Pass the tracking ID
762 iterations=search_config.get("iterations", 8),
763 questions_per_iteration=search_config.get(
764 "questions_per_iteration", 5
765 ),
766 search_tool=search_config.get("search_tool", "searxng"),
767 search_strategy=search_config.get(
768 "search_strategy", "focused_iteration"
769 ),
770 progress_callback=benchmark_progress_callback,
771 model_name=search_config.get("model_name"),
772 provider=search_config.get("provider"),
773 temperature=search_config.get("temperature", 0.7),
774 openai_endpoint_url=search_config.get("openai_endpoint_url"),
775 settings_snapshot=settings_context.snapshot, # Pass settings snapshot for thread safety
776 username=task.get("username"), # Pass username
777 user_password=user_password, # Pass password for metrics tracking
778 # The web benchmark runs against the user's encrypted DB
779 # (it has username/password and wants search metrics
780 # persisted). quick_summary's default is programmatic_mode=True
781 # for true library callers; override here so the engine's
782 # metrics path stays active.
783 programmatic_mode=False,
784 )
785 processing_time = time.time() - start_time
786 logger.info(
787 f"Task {task['example_id']} research completed in {processing_time:.2f}s, "
788 f"model={search_config.get('model_name')}, provider={search_config.get('provider')}"
789 )
791 # Extract answer
792 response = search_result.get("summary", "")
793 logger.info(
794 f"Task {task['example_id']} response length: {len(response)} chars"
795 )
797 extracted_data = extract_answer_from_response(
798 response, task["dataset_type"]
799 )
800 extracted_answer = (
801 extracted_data.get("extracted_answer", "")
802 if isinstance(extracted_data, dict)
803 else str(extracted_data)
804 )
805 logger.info(
806 f"Task {task['example_id']} extracted answer: '{extracted_answer[:100]}...'"
807 )
809 # Extract sources - handle both direct sources and all_links_of_system
810 sources = search_result.get("sources", [])
811 if not sources and "all_links_of_system" in search_result:
812 sources = search_result.get("all_links_of_system", [])
814 # Log for debugging
815 logger.debug(f"Search result keys: {list(search_result.keys())}")
816 logger.debug(f"Sources found: {len(sources)} items")
818 # Prepare result
819 result = {
820 **task,
821 "response": response,
822 "extracted_answer": extracted_answer,
823 "confidence": str(
824 extracted_data.get("confidence", "100")
825 if isinstance(extracted_data, dict)
826 else "100"
827 ),
828 "processing_time": processing_time,
829 "sources": json.dumps(sources), # Convert to JSON string
830 "completed_at": datetime.now(UTC),
831 "research_id": tracking_id, # Store the UUID in the research_id field
832 }
834 # Evaluate result - requires proper grading model
835 try:
836 logger.info(f"Task {task['example_id']} starting evaluation...")
837 eval_start_time = time.time()
839 # Always attempt evaluation, regardless of provider
840 # Modern local models like Ollama are capable of grading
841 # Try to evaluate with proper model
842 result_data = {
843 "id": task["example_id"],
844 "problem": task["question"],
845 "correct_answer": task["correct_answer"],
846 "response": response,
847 "extracted_answer": extracted_answer,
848 }
850 eval_result = grade_single_result(
851 result_data,
852 task["dataset_type"],
853 evaluation_config,
854 settings_context.snapshot,
855 )
856 eval_time = time.time() - eval_start_time
857 logger.info(
858 f"Task {task['example_id']} evaluation completed in {eval_time:.2f}s"
859 )
860 if eval_result and not eval_result.get("grading_error"):
861 result.update(
862 {
863 "is_correct": eval_result.get("is_correct", False),
864 "graded_confidence": eval_result.get(
865 "graded_confidence", "0"
866 ),
867 "grader_response": eval_result.get(
868 "grader_response", ""
869 ),
870 }
871 )
872 else:
873 error_msg = (
874 eval_result.get(
875 "grading_error", "Unknown evaluation error"
876 )
877 if eval_result
878 else "No evaluation results returned"
879 )
880 result.update(
881 {
882 "is_correct": None,
883 "graded_confidence": "0",
884 "grader_response": f"Evaluation failed: {error_msg}",
885 "evaluation_error": error_msg,
886 }
887 )
889 except Exception as e:
890 logger.exception("Evaluation error")
891 result.update(
892 {
893 "is_correct": None,
894 "graded_confidence": "0",
895 "grader_response": f"Evaluation failed: {e!s}",
896 "evaluation_error": str(e),
897 }
898 )
900 return result
902 except Exception as e:
903 logger.exception("Research error")
904 return {
905 **task,
906 "research_error": str(e),
907 "completed_at": datetime.now(UTC),
908 }
910 def _persist_unsaved_results(
911 self, session, benchmark_run_id: int, run_data: dict
912 ) -> list[int]:
913 """Stage INSERTs for this run's not-yet-persisted results; return the
914 result indices that were staged.
916 Idempotent and rollback-safe. A result is skipped when its index is
917 already known-saved (``saved_indices``), its ``query_hash`` is already
918 committed for this run, or it repeats a ``query_hash`` staged earlier
919 in this same batch. Since ``query_hash`` covers ``example_id`` as well
920 as the question text (#5080), a repeat no longer means "a dataset
921 legitimately repeats a question": it is the same result entry reaching
922 this method twice, so the skip is logged at WARNING rather than passed
923 over in silence (#4860).
924 Dedup correctness rests on the DB-backed ``seen_hashes``, which
925 survives a rollback (the next sync re-reads it), NOT on any in-memory
926 flag.
928 This method deliberately does NOT mark ``saved_indices``: the caller
929 must do that ONLY after its commit succeeds. Marking before the commit
930 would, on a commit failure (disk full, lock timeout, ...), leave rows
931 flagged saved that were actually rolled back — silently dropped
932 forever. Combined with ``self._results_sync_lock`` held by the caller
933 ACROSS the commit, this also stops the worker thread and the request
934 thread from both inserting the same ``(benchmark_run_id, query_hash)``
935 row and tripping the ``uix_run_query`` unique constraint.
936 """
937 results = run_data.get("results", [])
938 # Read-only here — marking indices saved is the caller's job, and only
939 # after the commit lands (see the method docstring).
940 saved_indices = run_data.get("saved_indices", set())
942 # Every query_hash already committed for this run, fetched once. This
943 # is the rollback-safe source of truth for dedup; it is extended below
944 # so a question repeated within this batch is staged only once.
945 seen_hashes = {
946 row[0]
947 for row in session.query(BenchmarkResult.query_hash)
948 .filter(BenchmarkResult.benchmark_run_id == benchmark_run_id)
949 .all()
950 }
952 staged_indices = []
953 for idx, result in enumerate(results):
954 if idx in saved_indices:
955 continue
956 query_hash = result["query_hash"]
957 if query_hash in seen_hashes:
958 logger.warning(
959 f"Benchmark run {benchmark_run_id}: skipping result index "
960 f"{idx} (example_id={result['example_id']!r}), its "
961 f"query_hash is already persisted or staged for this run"
962 )
963 continue
964 session.add(
965 BenchmarkResult(
966 benchmark_run_id=benchmark_run_id,
967 example_id=result["example_id"],
968 query_hash=query_hash,
969 dataset_type=DatasetType(result["dataset_type"]),
970 research_id=result.get("research_id"),
971 question=result["question"],
972 correct_answer=result["correct_answer"],
973 response=result.get("response"),
974 extracted_answer=result.get("extracted_answer"),
975 confidence=result.get("confidence"),
976 processing_time=result.get("processing_time"),
977 sources=result.get("sources"),
978 is_correct=result.get("is_correct"),
979 graded_confidence=result.get("graded_confidence"),
980 grader_response=result.get("grader_response"),
981 completed_at=result.get("completed_at"),
982 research_error=result.get("research_error"),
983 evaluation_error=result.get("evaluation_error"),
984 task_index=result.get("task_index"),
985 )
986 )
987 seen_hashes.add(query_hash)
988 staged_indices.append(idx)
990 return staged_indices
992 def sync_pending_results(
993 self, benchmark_run_id: int, username: Optional[str] = None
994 ):
995 """Sync any pending results to database. Can be called from main thread."""
996 # Read outside _results_sync_lock, so the worker thread's
997 # ``del self.active_runs[benchmark_run_id]`` (in _sync_results_to_database,
998 # also unlocked) can land between the membership check and the subscript.
999 # A plain ``active_runs[id]`` would then raise KeyError, which escapes
1000 # before the try below and surfaces as a spurious 500 on /api/results.
1001 # Fetch once so the check and the read see the same state. See #4859.
1002 run_data = self.active_runs.get(benchmark_run_id)
1003 if run_data is None:
1004 return 0
1006 if not username:
1007 username = run_data.get("data", {}).get("username")
1008 user_password = run_data.get("data", {}).get("user_password")
1010 saved_count = 0
1011 session = None
1012 from ...database.session_context import (
1013 get_user_db_session,
1014 safe_rollback,
1015 )
1017 try:
1018 # Serialize with the worker thread's _sync_results_to_database so
1019 # the two can't insert the same (benchmark_run_id, query_hash) row.
1020 with self._results_sync_lock:
1021 with get_user_db_session(username, user_password) as session:
1022 staged = self._persist_unsaved_results(
1023 session, benchmark_run_id, run_data
1024 )
1025 if staged:
1026 session.commit()
1027 # Mark saved ONLY after the commit succeeds, so a
1028 # failed commit leaves these for the next sync to retry
1029 # instead of dropping them silently.
1030 run_data.setdefault("saved_indices", set()).update(
1031 staged
1032 )
1033 saved_count = len(staged)
1034 logger.info(
1035 f"Saved {saved_count} new results for benchmark "
1036 f"{benchmark_run_id}"
1037 )
1038 except Exception:
1039 logger.exception(
1040 f"Error syncing pending results for benchmark {benchmark_run_id}"
1041 )
1042 # Thread-local sessions are reused, so a failed flush/commit must be
1043 # rolled back explicitly or the next use raises PendingRollbackError.
1044 if session is not None: 1044 ↛ 1047line 1044 didn't jump to line 1047 because the condition on line 1044 was always true
1045 safe_rollback(session, "sync_pending_results")
1047 return saved_count
1049 def _sync_results_to_database(self, benchmark_run_id: int):
1050 """Sync benchmark results from memory to database after thread completes."""
1051 if benchmark_run_id not in self.active_runs:
1052 return
1054 run_data = self.active_runs[benchmark_run_id]
1055 if not run_data.get("thread_complete"):
1056 return
1058 username = run_data.get("data", {}).get("username")
1059 user_password = run_data.get("data", {}).get("user_password")
1060 session = None
1061 from ...database.session_context import (
1062 get_user_db_session,
1063 safe_rollback,
1064 )
1066 try:
1067 # Serialize with sync_pending_results (request thread) so the two
1068 # can't insert the same (benchmark_run_id, query_hash) row.
1069 with (
1070 self._results_sync_lock,
1071 get_user_db_session(username, user_password) as session,
1072 ):
1073 # Update benchmark run status
1074 benchmark_run = (
1075 session.query(BenchmarkRun)
1076 .filter(BenchmarkRun.id == benchmark_run_id)
1077 .first()
1078 )
1080 if benchmark_run and "completion_info" in run_data: 1080 ↛ 1141line 1080 didn't jump to line 1141
1081 info = run_data["completion_info"]
1082 benchmark_run.status = info["status"]
1083 benchmark_run.end_time = info.get(
1084 "end_time", datetime.now(UTC)
1085 )
1086 benchmark_run.completed_examples = info.get(
1087 "completed_examples", 0
1088 )
1089 benchmark_run.failed_examples = info.get(
1090 "failed_examples", 0
1091 )
1092 benchmark_run.error_message = info.get("error_message")
1094 # Stage any results not yet persisted (idempotent; safe
1095 # against a concurrent sync_pending_results on the request
1096 # thread — see _persist_unsaved_results).
1097 staged = self._persist_unsaved_results(
1098 session, benchmark_run_id, run_data
1099 )
1101 # Calculate final accuracy
1102 if benchmark_run.status == BenchmarkStatus.COMPLETED:
1103 correct_results = [
1104 r
1105 for r in run_data.get("results", [])
1106 if r.get("is_correct")
1107 ]
1108 evaluated_results = [
1109 r
1110 for r in run_data.get("results", [])
1111 if r.get("is_correct") is not None
1112 ]
1114 if evaluated_results:
1115 benchmark_run.overall_accuracy = (
1116 len(correct_results) / len(evaluated_results)
1117 ) * 100
1119 # Calculate processing rate
1120 total_time = sum(
1121 r.get("processing_time", 0)
1122 for r in evaluated_results
1123 )
1124 if total_time > 0: 1124 ↛ 1129line 1124 didn't jump to line 1129 because the condition on line 1124 was always true
1125 benchmark_run.processing_rate = len(
1126 evaluated_results
1127 ) / (total_time / 60)
1129 session.commit()
1130 # Mark saved only after the commit lands (see
1131 # _persist_unsaved_results) so a failed commit retries.
1132 if staged:
1133 run_data.setdefault("saved_indices", set()).update(
1134 staged
1135 )
1136 logger.info(
1137 f"Successfully synced results for benchmark {benchmark_run_id}"
1138 )
1140 # Clean up memory
1141 del self.active_runs[benchmark_run_id]
1143 except Exception:
1144 logger.exception("Error syncing benchmark results to database")
1145 # Thread-local session is reused — roll back a failed flush/commit
1146 # so the next use doesn't raise PendingRollbackError.
1147 if session is not None:
1148 safe_rollback(session, "_sync_results_to_database")
1150 def _send_progress_update(
1151 self, benchmark_run_id: int, completed: int, total: int
1152 ):
1153 """Send real-time progress update via websocket."""
1154 try:
1155 percentage = (completed / total * 100) if total > 0 else 0
1157 # Create log entry for milestone progress
1158 log_entry = {
1159 "time": datetime.now(UTC).isoformat(),
1160 "message": f"Completed {completed}/{total} examples ({percentage:.1f}%)",
1161 "progress": percentage,
1162 "metadata": {
1163 "phase": "benchmark_progress",
1164 "type": "milestone",
1165 "completed": completed,
1166 "total": total,
1167 "benchmark_run_id": benchmark_run_id,
1168 },
1169 }
1171 progress_data = {
1172 "status": "in_progress",
1173 "message": f"Processing examples: {completed}/{total}",
1174 "progress": percentage,
1175 "completed": completed,
1176 "total": total,
1177 "benchmark_run_id": benchmark_run_id,
1178 "log_entry": log_entry,
1179 "progress_log": json.dumps([log_entry]),
1180 }
1182 self.socket_service.emit_to_subscribers(
1183 "research_progress", benchmark_run_id, progress_data
1184 )
1186 except Exception:
1187 logger.exception("Error sending progress update")
1189 def _calculate_final_accuracy(
1190 self, benchmark_run_id: int, username: Optional[str] = None
1191 ):
1192 """Calculate and save final accuracy metrics."""
1193 from ...database.session_context import get_user_db_session
1195 with get_user_db_session(username) as session:
1196 try:
1197 # Get all results for this run
1198 results = (
1199 session.query(BenchmarkResult)
1200 .filter(
1201 BenchmarkResult.benchmark_run_id == benchmark_run_id
1202 )
1203 .filter(BenchmarkResult.is_correct.isnot(None))
1204 .all()
1205 )
1207 if results:
1208 correct_count = sum(1 for r in results if r.is_correct)
1209 overall_accuracy = (correct_count / len(results)) * 100
1211 # Calculate processing rate
1212 total_time = sum(r.processing_time or 0 for r in results)
1213 processing_rate = (
1214 (len(results) / (total_time / 60))
1215 if total_time > 0
1216 else 0
1217 )
1219 # Update benchmark run
1220 benchmark_run = (
1221 session.query(BenchmarkRun)
1222 .filter(BenchmarkRun.id == benchmark_run_id)
1223 .first()
1224 )
1225 if benchmark_run: 1225 ↛ exitline 1225 didn't jump to the function exit
1226 benchmark_run.overall_accuracy = overall_accuracy
1227 benchmark_run.processing_rate = processing_rate
1228 session.commit()
1230 except Exception:
1231 logger.exception("Error calculating final accuracy")
1233 def update_benchmark_status(
1234 self,
1235 benchmark_run_id: int,
1236 status: BenchmarkStatus,
1237 error_message: Optional[str] = None,
1238 username: Optional[str] = None,
1239 ):
1240 """Update benchmark run status."""
1241 from ...database.session_context import get_user_db_session
1243 with get_user_db_session(username) as session:
1244 try:
1245 benchmark_run = (
1246 session.query(BenchmarkRun)
1247 .filter(BenchmarkRun.id == benchmark_run_id)
1248 .first()
1249 )
1250 if benchmark_run:
1251 benchmark_run.status = status
1252 benchmark_run.updated_at = datetime.now(UTC)
1254 if error_message:
1255 benchmark_run.error_message = error_message
1257 if (
1258 status == BenchmarkStatus.IN_PROGRESS
1259 and not benchmark_run.start_time
1260 ):
1261 benchmark_run.start_time = datetime.now(UTC)
1262 elif ( 1262 ↛ 1269line 1262 didn't jump to line 1269 because the condition on line 1262 was always true
1263 status
1264 in [BenchmarkStatus.COMPLETED, BenchmarkStatus.FAILED]
1265 and not benchmark_run.end_time
1266 ):
1267 benchmark_run.end_time = datetime.now(UTC)
1269 session.commit()
1271 except Exception:
1272 session.rollback()
1273 logger.exception("Error updating benchmark status")
1275 def get_benchmark_status(
1276 self, benchmark_run_id: int, username: str = None
1277 ) -> Optional[Dict]:
1278 """Get current status of a benchmark run."""
1279 from ...database.session_context import get_user_db_session
1281 with get_user_db_session(username) as session:
1282 try:
1283 benchmark_run = (
1284 session.query(BenchmarkRun)
1285 .filter(BenchmarkRun.id == benchmark_run_id)
1286 .first()
1287 )
1288 if not benchmark_run:
1289 return None
1291 # Calculate running accuracy from this run's evaluated results
1292 results = (
1293 session.query(BenchmarkResult)
1294 .filter(
1295 BenchmarkResult.benchmark_run_id == benchmark_run_id
1296 )
1297 .filter(BenchmarkResult.is_correct.isnot(None))
1298 .all()
1299 )
1301 running_accuracy = None
1302 # Dynamic per-dataset accuracy tracking
1303 dataset_accuracies = {}
1305 if results:
1306 # Overall running accuracy
1307 correct_count = sum(1 for r in results if r.is_correct)
1308 running_accuracy = (correct_count / len(results)) * 100
1310 # Calculate accuracy for each dataset type dynamically
1311 from collections import defaultdict
1313 dataset_results = defaultdict(list)
1315 # Group results by dataset type
1316 for r in results:
1317 dataset_results[r.dataset_type.value].append(r)
1319 # Calculate accuracy for each dataset
1320 for (
1321 dataset_type,
1322 dataset_result_list,
1323 ) in dataset_results.items():
1324 if dataset_result_list: 1324 ↛ 1320line 1324 didn't jump to line 1320 because the condition on line 1324 was always true
1325 correct = sum(
1326 1 for r in dataset_result_list if r.is_correct
1327 )
1328 accuracy = (
1329 correct / len(dataset_result_list)
1330 ) * 100
1331 # Store with _accuracy suffix for consistency
1332 dataset_accuracies[f"{dataset_type}_accuracy"] = (
1333 accuracy
1334 )
1336 # Calculate time estimates and reliability metrics
1337 estimated_time_remaining = None
1338 total_elapsed_time = None
1339 avg_time_per_example = None
1340 accuracy_confidence = None
1342 # Get ALL results for timing calculation (including those pending evaluation)
1343 all_results_for_timing = (
1344 session.query(BenchmarkResult)
1345 .filter(
1346 BenchmarkResult.benchmark_run_id == benchmark_run_id
1347 )
1348 .all()
1349 )
1351 if benchmark_run.start_time and all_results_for_timing:
1352 # Calculate elapsed time
1353 current_time = datetime.now(UTC)
1354 total_elapsed_time = (
1355 current_time - benchmark_run.start_time
1356 ).total_seconds()
1358 # Calculate average processing time per example using actual count
1359 avg_time_per_example = total_elapsed_time / len(
1360 all_results_for_timing
1361 )
1363 logger.info(
1364 f"Time calculation - elapsed: {total_elapsed_time:.2f}s, "
1365 f"results_count: {len(all_results_for_timing)}, "
1366 f"avg_per_example: {avg_time_per_example:.2f}s"
1367 )
1369 # Estimate remaining time
1370 remaining_examples = benchmark_run.total_examples - len(
1371 all_results_for_timing
1372 )
1373 if remaining_examples > 0: 1373 ↛ 1385line 1373 didn't jump to line 1385 because the condition on line 1373 was always true
1374 estimated_time_remaining = (
1375 avg_time_per_example * remaining_examples
1376 )
1377 logger.info(
1378 f"Time estimation - total: {benchmark_run.total_examples}, "
1379 f"completed: {len(all_results_for_timing)}, remaining: {remaining_examples}, "
1380 f"avg_time: {avg_time_per_example:.2f}s, "
1381 f"estimated_remaining: {estimated_time_remaining:.2f}s"
1382 )
1384 # Calculate accuracy confidence interval (Wilson score, 95%)
1385 if results and len(results) >= 3: 1385 ↛ 1386line 1385 didn't jump to line 1386 because the condition on line 1385 was never true
1386 from local_deep_research.benchmarks.metrics.statistics import (
1387 wilson_score_interval,
1388 )
1390 n = len(results)
1391 correct = sum(1 for r in results if r.is_correct)
1392 ci = wilson_score_interval(correct, n)
1393 accuracy_confidence = {
1394 "lower_bound": ci["lower"] * 100,
1395 "upper_bound": ci["upper"] * 100,
1396 "margin_of_error": ci["margin_of_error"] * 100,
1397 "sample_size": n,
1398 }
1400 status_data = {
1401 "id": benchmark_run.id,
1402 "run_name": benchmark_run.run_name,
1403 "status": benchmark_run.status.value,
1404 "completed_examples": len(
1405 all_results_for_timing
1406 ), # Use actual count from DB
1407 "total_examples": benchmark_run.total_examples,
1408 "failed_examples": benchmark_run.failed_examples,
1409 "overall_accuracy": benchmark_run.overall_accuracy
1410 or running_accuracy, # Use running accuracy if final not calculated
1411 "running_accuracy": running_accuracy, # Current running accuracy
1412 "processing_rate": benchmark_run.processing_rate,
1413 "estimated_time_remaining": estimated_time_remaining, # seconds
1414 "total_elapsed_time": total_elapsed_time, # seconds
1415 "avg_time_per_example": avg_time_per_example, # seconds
1416 "accuracy_confidence": accuracy_confidence, # confidence interval
1417 "created_at": benchmark_run.created_at.isoformat()
1418 if benchmark_run.created_at
1419 else None,
1420 "start_time": benchmark_run.start_time.isoformat()
1421 if benchmark_run.start_time
1422 else None,
1423 "end_time": benchmark_run.end_time.isoformat()
1424 if benchmark_run.end_time
1425 else None,
1426 # Genericized at this boundary (CWE-209) — see
1427 # _GENERIC_BENCHMARK_ERROR. Returns None when there is no
1428 # failure so non-error states are unaffected.
1429 "error_message": (
1430 _GENERIC_BENCHMARK_ERROR
1431 if benchmark_run.error_message
1432 else None
1433 ),
1434 # Add all per-dataset accuracies dynamically
1435 **dataset_accuracies,
1436 }
1438 logger.info(
1439 f"Benchmark {benchmark_run_id} status - completed: {benchmark_run.completed_examples}, "
1440 f"running_acc: {running_accuracy}, dataset_accuracies: {dataset_accuracies}, "
1441 f"avg_time: {avg_time_per_example}"
1442 )
1444 return status_data
1446 except Exception:
1447 logger.exception("Error getting benchmark status")
1448 return None
1450 def cancel_benchmark(
1451 self, benchmark_run_id: int, username: Optional[str] = None
1452 ) -> bool:
1453 """Cancel a running benchmark."""
1454 try:
1455 if benchmark_run_id in self.active_runs:
1456 self.active_runs[benchmark_run_id]["status"] = "cancelled"
1458 self.update_benchmark_status(
1459 benchmark_run_id, BenchmarkStatus.CANCELLED, username=username
1460 )
1461 logger.info(f"Cancelled benchmark run {benchmark_run_id}")
1462 return True
1464 except Exception:
1465 logger.exception(f"Error cancelling benchmark {benchmark_run_id}")
1466 return False
1469# Global service instance
1470benchmark_service = BenchmarkService()