Mapping Spatial API Errors to User-Friendly Prompts

import logging
from enum import Enum
from typing import Optional, Dict, Any, List
from pydantic import BaseModel, Field, ValidationError, field_validator
from shapely.validation import explain_validity
from shapely.geometry import shape
from shapely.geometry.base import BaseGeometry

logger = logging.getLogger("spatial_error_mapper")

class SeverityLevel(str, Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

class SpatialErrorPayload(BaseModel):
    error_code: str
    severity: SeverityLevel
    raw_exception: Optional[str] = None
    geometry_diagnostics: Optional[str] = None
    crs_status: str
    recommended_repair: str
    next_steps: List[str]
    retry_allowed: bool = Field(default=True)

    @field_validator("crs_status")
    @classmethod
    def validate_crs_format(cls, v: str) -> str:
        allowed = ["valid_epsg", "unknown", "mismatch", "missing"]
        if v not in allowed:
            raise ValueError(f"Invalid CRS status. Must be one of {allowed}")
        return v

class SpatialErrorInterceptor:
    def __init__(self, max_repair_attempts: int = 3):
        self.max_repair_attempts = max_repair_attempts

    def validate_coordinates(self, geom: BaseGeometry) -> Dict[str, Any]:
        """Explicit coordinate validation: bounds and validity checks."""
        diagnostics = {}
        try:
            validity_msg = explain_validity(geom)
            diagnostics["is_valid"] = validity_msg == "Valid Geometry"
            diagnostics["shapely_diagnostic"] = validity_msg

            # Coordinate bounds validation (WGS84 limits)
            minx, miny, maxx, maxy = geom.bounds
            if minx < -180 or maxx > 180 or miny < -90 or maxy > 90:
                diagnostics["bounds_violation"] = True
                diagnostics["bounds"] = (minx, miny, maxx, maxy)
            else:
                diagnostics["bounds_violation"] = False

        except Exception as e:
            logger.error(f"Coordinate validation failed: {e}")
            diagnostics["is_valid"] = False
            diagnostics["shapely_diagnostic"] = str(e)

        return diagnostics

    def map_error_to_prompt(
        self,
        exception: Exception,
        geom: Optional[BaseGeometry] = None,
        attempt: int = 0
    ) -> SpatialErrorPayload:
        """Maps raw spatial exceptions to structured LLM routing payloads."""
        error_str = str(exception)
        crs_status = "unknown"
        recommended_repair = "review_query_syntax"
        next_steps = []
        severity = SeverityLevel.WARNING
        retry_allowed = True

        # Explicit error handling for PostGIS/GEOS topology violations
        if "TopologyException" in error_str or "22023" in error_str:
            severity = SeverityLevel.CRITICAL
            recommended_repair = "ST_MakeValid"
            next_steps = [
                "Apply ST_MakeValid() to the target geometry before re-execution.",
                "If ST_MakeValid fails, run ST_SnapToGrid(geom, 1e-6) to resolve floating-point drift.",
                "Validate output with ST_IsValid() and route to human-in-the-loop if invalid."
            ]
        elif "CRS mismatch" in error_str or "transform" in error_str.lower():
            crs_status = "mismatch"
            recommended_repair = "CRS_Transformation"
            next_steps = [
                "Identify source EPSG code and target CRS using pyproj.Transformer.",
                "Apply gdf.to_crs(epsg=4326) or ST_Transform() before spatial joins.",
                "Re-run bounding box intersection to verify alignment."
            ]
        elif "Rate limit" in error_str or "429" in error_str:
            severity = SeverityLevel.INFO
            recommended_repair = "Exponential_Backoff"
            next_steps = [
                "Pause execution for 2^attempt seconds.",
                "Reduce batch size to 50% of original payload.",
                "Resume pipeline with cached successful geometries."
            ]
        else:
            severity = SeverityLevel.CRITICAL
            recommended_repair = "Manual_Review"
            next_steps = [
                "Inspect raw SQL/GeoJSON payload for malformed coordinates.",
                "Check spatial index existence on target tables.",
                "Escalate to platform engineering if retry threshold exceeded."
            ]

        # Coordinate validation integration
        geom_diagnostics = None
        if geom:
            geom_diagnostics = self.validate_coordinates(geom)
            if not geom_diagnostics.get("is_valid", False):
                crs_status = "unknown"
                next_steps.insert(0, f"Geometry invalid: {geom_diagnostics.get('shapely_diagnostic', 'Unknown')}")
                if attempt >= self.max_repair_attempts:
                    retry_allowed = False
                    next_steps.append("MAX_REPAIR_ATTEMPTS_EXCEEDED: Route to fallback LLM agent.")

        return SpatialErrorPayload(
            error_code=type(exception).__name__,
            severity=severity,
            raw_exception=error_str,
            geometry_diagnostics=str(geom_diagnostics) if geom_diagnostics else None,
            crs_status=crs_status,
            recommended_repair=recommended_repair,
            next_steps=next_steps,
            retry_allowed=retry_allowed
        )

import asyncio
from typing import Callable, Awaitable, Any

async def execute_with_spatial_interception(
    spatial_task: Callable[..., Awaitable[Any]],
    interceptor: SpatialErrorInterceptor,
    max_retries: int = 3
) -> Any:
    """Wraps async spatial execution with deterministic error mapping and routing."""
    attempt = 0
    while attempt <= max_retries:
        try:
            result = await spatial_task()
            return result
        except Exception as e:
            attempt += 1
            logger.warning(f"Spatial execution failed (attempt {attempt}/{max_retries}): {e}")

            payload = interceptor.map_error_to_prompt(e, attempt=attempt)

            if not payload.retry_allowed:
                logger.critical("Pipeline halted: Fallback routing triggered.")
                raise RuntimeError(f"Spatial pipeline exhausted retries. Payload: {payload.model_dump_json()}")

            logger.info(f"Routing repair prompt: {payload.recommended_repair}")
            logger.info(f"Next steps: {payload.next_steps}")

            # Exponential backoff before retry
            await asyncio.sleep(2 ** attempt)

            # In production, pass payload.next_steps to LLM system prompt here
            # to constrain the next tool invocation to the recommended repair.