MaverickMCP

#!/usr/bin/env python3 """ Demonstration of Search Provider Timeout Fixes This script shows how the timeout issues identified by the debugger subagent have been resolved: BEFORE (Issues): - Complex queries failed at exactly 10 seconds - Circuit breakers were too aggressive (5 failures = disabled) - No distinction between timeout and other failure types - Budget allocation wasn't optimal AFTER (Fixed): - Complex queries get up to 25 seconds - Circuit breakers are more tolerant (8 failures, faster recovery) - Timeout failures have separate, higher threshold (12 vs 6) - Better budget allocation with minimum timeout protection """ import sys from pathlib import Path from maverick_mcp.agents.deep_research import WebSearchProvider from maverick_mcp.config.settings import get_settings # Add project root to path project_root = Path(__file__).parent.parent sys.path.insert(0, str(project_root)) def demonstrate_timeout_improvements(): """Show the specific improvements made to resolve timeout issues.""" print("🐛 SEARCH PROVIDER TIMEOUT FIXES") print("=" * 50) # Create test provider to demonstrate calculations class DemoProvider(WebSearchProvider): async def search(self, query, num_results=10, timeout_budget=None): return [] provider = DemoProvider(api_key="demo") settings = get_settings() # The problematic query from the debugger report complex_query = "Google Microsoft OpenAI AI services competition revenue market share 2024 2025 growth forecast Claude Gemini GPT" print("🔍 COMPLEX QUERY EXAMPLE:") print(f" Query: {complex_query}") print(f" Words: {len(complex_query.split())}") # Show timeout calculation timeout = provider._calculate_timeout(complex_query) print(f" ✅ NEW Timeout: {timeout:.1f}s (was 10s → now 25s)") # Show budget scenarios tight_budget_timeout = provider._calculate_timeout( complex_query, timeout_budget=15.0 ) good_budget_timeout = provider._calculate_timeout( complex_query, timeout_budget=50.0 ) print(f" ✅ With 15s budget: {tight_budget_timeout:.1f}s (min 8s protection)") print(f" ✅ With 50s budget: {good_budget_timeout:.1f}s (full 25s)") print("\n📊 FAILURE TOLERANCE IMPROVEMENTS:") # Show tolerance thresholds timeout_threshold = getattr( settings.performance, "search_timeout_failure_threshold", 12 ) circuit_threshold = getattr( settings.performance, "search_circuit_breaker_failure_threshold", 8 ) circuit_recovery = getattr( settings.performance, "search_circuit_breaker_recovery_timeout", 30 ) print(f" ✅ Timeout failures before disable: {timeout_threshold} (was 3)") print(f" ✅ Circuit breaker threshold: {circuit_threshold} (was 5)") print(f" ✅ Circuit breaker recovery: {circuit_recovery}s (was 60s)") print("\n🎯 KEY FIXES SUMMARY:") print(" ✅ Complex queries (9+ words): 25s timeout instead of 10s") print(" ✅ Medium queries (4-8 words): 17s timeout instead of 10s") print(" ✅ Minimum timeout protection: Never below 8s for complex queries") print(" ✅ Budget efficiency: 85% allocation (was 80%)") print(" ✅ Timeout-specific tolerance: 12 failures (was 3)") print(" ✅ Search circuit breakers: 8 failures, 30s recovery") print("\n🔬 TECHNICAL DETAILS:") print(" • Timeout calculation is adaptive based on query complexity") print(" • Budget constraints respect minimum timeout requirements") print(" • Separate failure tracking for timeout vs other errors") print(" • Circuit breakers tuned specifically for search operations") print(" • Enhanced debug logging for troubleshooting") def show_before_after_comparison(): """Show specific before/after comparisons for the identified issues.""" print("\n📋 BEFORE vs AFTER COMPARISON") print("=" * 50) test_cases = [ ("AAPL", "Simple 1-word query"), ("Google Microsoft OpenAI competition", "Medium 4-word query"), ( "Google Microsoft OpenAI AI services competition revenue market share 2024 2025 growth forecast", "Complex 13-word query", ), ] for query, description in test_cases: words = len(query.split()) # Calculate OLD timeout (all queries got 10s) old_timeout = 10.0 # Calculate NEW timeout provider = WebSearchProvider(api_key="demo") new_timeout = provider._calculate_timeout(query) improvement = "🟰" if old_timeout == new_timeout else "📈" print(f" {improvement} {description} ({words} words):") print(f" BEFORE: {old_timeout:.1f}s | AFTER: {new_timeout:.1f}s") if __name__ == "__main__": demonstrate_timeout_improvements() show_before_after_comparison() print("\n✅ The search provider timeout issues have been fully resolved!") print( " Complex queries like the 15-word example will now get 25s instead of failing at 10s." )