validation error resolve

chat/geocoding.py

@@ -6,10 +6,18 @@ Handles address validation with Google Maps API and smart mock fallback
 import os
 import googlemaps
 import logging
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
 from typing import Dict, Optional
 
 logger = logging.getLogger(__name__)
 
+# Thread pool for running blocking geocoding calls
+_geocoding_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="geocoding")
+
+# Timeout for geocoding operations (seconds)
+GEOCODING_TIMEOUT = 10
+
 # Common city coordinates for mock geocoding
 CITY_COORDINATES = {
     "san francisco": (37.7749, -122.4194),
@@ -47,8 +55,13 @@ class GeocodingService:
             self.gmaps_client = None
         else:
             try:
-                self.gmaps_client = googlemaps.Client(key=self.google_maps_key)
-                logger.info("Geocoding: Using Google Maps API")
+                # Configure with timeout to prevent hanging
+                self.gmaps_client = googlemaps.Client(
+                    key=self.google_maps_key,
+                    timeout=GEOCODING_TIMEOUT,  # 10 second timeout
+                    retry_timeout=GEOCODING_TIMEOUT  # Total retry timeout
+                )
+                logger.info("Geocoding: Using Google Maps API (timeout: 10s)")
             except Exception as e:
                 logger.error(f"Failed to initialize Google Maps client: {e}")
                 self.use_mock = True
@@ -56,7 +69,8 @@ class GeocodingService:
 
     def geocode(self, address: str) -> Dict:
         """
-        Geocode address, using mock if API unavailable
+        Geocode address, using mock if API unavailable.
+        This is a synchronous method with built-in timeout.
 
         Args:
             address: Street address to geocode
@@ -73,6 +87,35 @@ class GeocodingService:
                 logger.error(f"Google Maps API failed: {e}, falling back to mock")
                 return self._geocode_mock(address)
 
+    async def geocode_async(self, address: str) -> Dict:
+        """
+        Async-safe geocoding that runs blocking calls in a thread pool.
+        Use this in async contexts to prevent blocking the event loop.
+
+        Args:
+            address: Street address to geocode
+
+        Returns:
+            Dict with keys: lat, lng, formatted_address, confidence
+        """
+        if self.use_mock:
+            return self._geocode_mock(address)
+
+        loop = asyncio.get_event_loop()
+        try:
+            # Run blocking geocode in thread pool with timeout
+            result = await asyncio.wait_for(
+                loop.run_in_executor(_geocoding_executor, self._geocode_google, address),
+                timeout=GEOCODING_TIMEOUT + 2  # Extra buffer beyond client timeout
+            )
+            return result
+        except asyncio.TimeoutError:
+            logger.error(f"Geocoding timed out for address: {address}, falling back to mock")
+            return self._geocode_mock(address)
+        except Exception as e:
+            logger.error(f"Async geocoding failed: {e}, falling back to mock")
+            return self._geocode_mock(address)
+
     def _geocode_google(self, address: str) -> Dict:
         """Real Google Maps API geocoding"""
         try:
@@ -128,7 +171,8 @@ class GeocodingService:
 
     def reverse_geocode(self, lat: float, lng: float) -> Dict:
         """
-        Reverse geocode coordinates to address
+        Reverse geocode coordinates to address.
+        This is a synchronous method with built-in timeout.
 
         Args:
             lat: Latitude
@@ -146,6 +190,41 @@ class GeocodingService:
                 logger.error(f"Google Maps reverse geocoding failed: {e}, falling back to mock")
                 return self._reverse_geocode_mock(lat, lng)
 
+    async def reverse_geocode_async(self, lat: float, lng: float) -> Dict:
+        """
+        Async-safe reverse geocoding that runs blocking calls in a thread pool.
+        Use this in async contexts to prevent blocking the event loop.
+
+        Args:
+            lat: Latitude
+            lng: Longitude
+
+        Returns:
+            Dict with keys: address, city, state, country, formatted_address
+        """
+        if self.use_mock:
+            return self._reverse_geocode_mock(lat, lng)
+
+        loop = asyncio.get_event_loop()
+        try:
+            # Run blocking reverse_geocode in thread pool with timeout
+            result = await asyncio.wait_for(
+                loop.run_in_executor(
+                    _geocoding_executor,
+                    self._reverse_geocode_google,
+                    lat,
+                    lng
+                ),
+                timeout=GEOCODING_TIMEOUT + 2
+            )
+            return result
+        except asyncio.TimeoutError:
+            logger.error(f"Reverse geocoding timed out for ({lat}, {lng}), falling back to mock")
+            return self._reverse_geocode_mock(lat, lng)
+        except Exception as e:
+            logger.error(f"Async reverse geocoding failed: {e}, falling back to mock")
+            return self._reverse_geocode_mock(lat, lng)
+
     def _reverse_geocode_google(self, lat: float, lng: float) -> Dict:
         """Real Google Maps API reverse geocoding"""
         try:

chat/tools.py

@@ -8,19 +8,70 @@ import sys
 from pathlib import Path
 from datetime import datetime, timedelta
 import logging
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
 
 # Add parent directory to path
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
 from database.connection import execute_write, execute_query, get_db_connection
-from chat.geocoding import GeocodingService
+from chat.geocoding import GeocodingService, GEOCODING_TIMEOUT
 from psycopg2.extras import RealDictCursor
 
 logger = logging.getLogger(__name__)
 
+# Thread pool for running blocking operations (geocoding, external APIs)
+_blocking_executor = ThreadPoolExecutor(max_workers=5, thread_name_prefix="blocking_ops")
+
 # Initialize geocoding service
 geocoding_service = GeocodingService()
 
+
+def safe_geocode(address: str) -> dict:
+    """
+    Thread-safe geocoding with timeout protection.
+    Runs geocoding in a thread pool to prevent blocking the main event loop.
+
+    Args:
+        address: Address to geocode
+
+    Returns:
+        Geocoding result dict
+    """
+    try:
+        future = _blocking_executor.submit(geocoding_service.geocode, address)
+        result = future.result(timeout=GEOCODING_TIMEOUT + 2)
+        return result
+    except FuturesTimeoutError:
+        logger.error(f"Geocoding timed out for: {address}, using mock fallback")
+        return geocoding_service._geocode_mock(address)
+    except Exception as e:
+        logger.error(f"Geocoding failed for {address}: {e}, using mock fallback")
+        return geocoding_service._geocode_mock(address)
+
+
+def safe_reverse_geocode(lat: float, lng: float) -> dict:
+    """
+    Thread-safe reverse geocoding with timeout protection.
+    Runs reverse geocoding in a thread pool to prevent blocking.
+
+    Args:
+        lat: Latitude
+        lng: Longitude
+
+    Returns:
+        Reverse geocoding result dict
+    """
+    try:
+        future = _blocking_executor.submit(geocoding_service.reverse_geocode, lat, lng)
+        result = future.result(timeout=GEOCODING_TIMEOUT + 2)
+        return result
+    except FuturesTimeoutError:
+        logger.error(f"Reverse geocoding timed out for ({lat}, {lng}), using mock fallback")
+        return geocoding_service._reverse_geocode_mock(lat, lng)
+    except Exception as e:
+        logger.error(f"Reverse geocoding failed for ({lat}, {lng}): {e}, using mock fallback")
+        return geocoding_service._reverse_geocode_mock(lat, lng)
+
 # Tool schemas for Claude
 TOOLS_SCHEMA = [
     {
@@ -625,7 +676,8 @@ def handle_geocode_address(tool_input: dict) -> dict:
 
     logger.info(f"Geocoding address: {address}")
 
-    result = geocoding_service.geocode(address)
+    # Use safe_geocode with timeout protection
+    result = safe_geocode(address)
 
     return {
         "success": True,
@@ -828,8 +880,8 @@ def _location_to_latlng(location: str) -> dict:
             except ValueError:
                 pass  # Not valid coordinates, treat as address
 
-    # Geocode the address
-    geocoded = geocoding_service.geocode(location)
+    # Geocode the address using safe version with timeout
+    geocoded = safe_geocode(location)
     return {
         "latitude": geocoded["lat"],
         "longitude": geocoded["lng"]
@@ -1034,9 +1086,9 @@ def _calculate_route_routes_api(origin: str, destination: str, mode: str, altern
         speed_intervals = travel_advisory.get("speedReadingIntervals", [])
         has_traffic_data = len(speed_intervals) > 0
 
-        # Get origin and destination addresses (geocode if needed)
-        origin_geocoded = geocoding_service.geocode(origin)
-        dest_geocoded = geocoding_service.geocode(destination)
+        # Get origin and destination addresses (geocode if needed) - use safe version with timeout
+        origin_geocoded = safe_geocode(origin)
+        dest_geocoded = safe_geocode(destination)
 
         # Build enhanced response with vehicle-specific data
         response_data = {
@@ -1178,10 +1230,10 @@ def _calculate_route_mock(origin: str, destination: str, mode: str) -> dict:
     import math
     from datetime import datetime
 
-    # Try to geocode both locations to get coordinates
+    # Try to geocode both locations to get coordinates (using safe version)
     try:
-        origin_geocoded = geocoding_service.geocode(origin)
-        dest_geocoded = geocoding_service.geocode(destination)
+        origin_geocoded = safe_geocode(origin)
+        dest_geocoded = safe_geocode(destination)
 
         origin_lat = origin_geocoded["lat"]
         origin_lng = origin_geocoded["lng"]
@@ -1510,6 +1562,24 @@ def handle_create_driver(tool_input: dict, user_id: str = None) -> dict:
     skills_raw = tool_input.get("skills", [])
     skills = list(skills_raw) if skills_raw else []
 
+    # Define valid skills
+    VALID_SKILLS = [
+        "refrigerated",
+        "medical_certified",
+        "fragile_handler",
+        "overnight",
+        "express_delivery"
+    ]
+
+    # Validate skills
+    if skills:
+        invalid_skills = [s for s in skills if s not in VALID_SKILLS]
+        if invalid_skills:
+            return {
+                "success": False,
+                "error": f"Invalid skills: {invalid_skills}. Valid skills are: {VALID_SKILLS}"
+            }
+
     status = tool_input.get("status", "active")
 
     # Validate ALL required fields (name, vehicle_type, current_lat, current_lng)
@@ -1651,12 +1721,10 @@ def handle_update_order(tool_input: dict, user_id: str = None) -> dict:
     current_assigned_driver = existing[0].get("assigned_driver_id")
 
     # Auto-geocode if delivery address is updated without coordinates
+    # Use safe_geocode with timeout protection
     if "delivery_address" in tool_input and ("delivery_lat" not in tool_input or "delivery_lng" not in tool_input):
-        from chat.geocoding import GeocodingService
-        geocoding_service = GeocodingService()
-
         try:
-            geocode_result = geocoding_service.geocode(tool_input["delivery_address"])
+            geocode_result = safe_geocode(tool_input["delivery_address"])
             tool_input["delivery_lat"] = geocode_result["lat"]
             tool_input["delivery_lng"] = geocode_result["lng"]
             logger.info(f"Auto-geocoded delivery address: {geocode_result['formatted_address']}")
@@ -3263,12 +3331,11 @@ def handle_get_driver_details(tool_input: dict, user_id: str = None) -> dict:
                 skills = []
 
         # Reverse geocode location to get address
+        # Use safe_reverse_geocode with timeout protection
         location_address = None
         if row['current_lat'] and row['current_lng']:
             try:
-                from chat.geocoding import GeocodingService
-                geocoding_service = GeocodingService()
-                reverse_result = geocoding_service.reverse_geocode(
+                reverse_result = safe_reverse_geocode(
                     float(row['current_lat']),
                     float(row['current_lng'])
                 )
@@ -4273,7 +4340,16 @@ Analyze ALL parameters comprehensively:
 
 **IMPORTANT:** Return ONLY valid JSON. Do not include markdown formatting, code blocks, or explanatory text outside the JSON."""
 
-        response = model.generate_content(prompt)
+        # Call Gemini with timeout protection (30 seconds max)
+        try:
+            future = _blocking_executor.submit(model.generate_content, prompt)
+            response = future.result(timeout=30)
+        except FuturesTimeoutError:
+            logger.error("Gemini AI call timed out after 30 seconds")
+            return {
+                "success": False,
+                "error": "AI assignment timed out. Please try auto_assign_order instead."
+            }
         response_text = response.text.strip()
 
         # Clean response (remove markdown code blocks if present)

proxy.py

@@ -16,6 +16,7 @@ import asyncio
 import logging
 import os
 from aiohttp import web, ClientSession, ClientTimeout
+from aiohttp.client_exceptions import ClientConnectionResetError
 from urllib.parse import urlencode, parse_qs, urlparse, urlunparse
 import sys
 
@@ -169,6 +170,12 @@ async def proxy_handler(request):
                         headers=dict(resp.headers)
                     )
 
+        except (ClientConnectionResetError, ConnectionResetError) as e:
+            # Client disconnected - this is normal for SSE connections
+            # Log at DEBUG level to reduce noise
+            logger.debug(f"[SSE] Client disconnected: {e}")
+            return web.Response(text="Client disconnected", status=499)
+
         except Exception as e:
             import traceback
             error_details = traceback.format_exc()