Sync from GitHub main

adapters/db/base.py

@@ -12,3 +12,6 @@ class DBAdapter(Protocol):
 
     def execute(self, sql: str) -> Tuple[List[Tuple[Any, ...]], List[str]]:
         """Execute a SELECT query and return (rows, columns)."""
+
+    def explain_query_plan(self, sql: str) -> None:
+        """Validate SQL by asking the DB to plan it (must be read-only). Raise on failure."""

adapters/db/postgres_adapter.py

@@ -68,3 +68,16 @@ class PostgresAdapter(DBAdapter):
                 desc = cur.description or ()
                 cols: List[str] = [d[0] for d in desc if d]
                 return rows, cols
+
+    def explain_query_plan(self, sql: str) -> None:
+        sql_stripped = (sql or "").strip().rstrip(";")
+        if not sql_stripped.lower().startswith("select"):
+            raise ValueError("Only SELECT statements are allowed.")
+
+        with psycopg.connect(self.dsn) as conn:
+            # Make it explicitly read-only at the session level
+            with conn.cursor() as cur:
+                cur.execute("SET TRANSACTION READ ONLY;")
+                cur.execute(f"EXPLAIN {sql_stripped}")
+                # We don't need the output; if planning fails, it raises.
+                _ = cur.fetchall()

adapters/db/sqlite_adapter.py

@@ -44,3 +44,20 @@ class SQLiteAdapter(DBAdapter):
             cols = [desc[0] for desc in cur.description]
             log.info("Query executed successfully. Returned %d rows.", len(rows))
             return rows, cols
+
+    def explain_query_plan(self, sql: str) -> None:
+        if not self.path.exists():
+            raise FileNotFoundError(f"SQLite DB does not exist: {self.path}")
+
+        sql_stripped = (sql or "").strip().rstrip(";")
+        if not sql_stripped.lower().startswith("select"):
+            raise ValueError("Only SELECT statements are allowed.")
+
+        uri = f"file:{self.path}?mode=ro"
+        with sqlite3.connect(uri, uri=True, timeout=3) as conn:
+            # Extra safety: enforce query-only mode if available
+            try:
+                conn.execute("PRAGMA query_only = ON;")
+            except Exception:
+                pass
+            conn.execute(f"EXPLAIN QUERY PLAN {sql_stripped}")

app/errors.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional, List
 
 
 @dataclass
@@ -8,38 +9,60 @@ class AppError(Exception):
     """Base class for domain-level errors."""
 
     message: str
+    http_status: int = 500
+    code: str = "internal_error"
+    retryable: bool = False
+    extra: Dict[str, Any] = field(default_factory=dict)
+    details: Optional[List[str]] = None
 
     def __str__(self) -> str:
         return self.message
 
 
-# 4xx-ish
+# 4xx
 @dataclass
-class DbNotFound(AppError):
-    """Requested DB (or db_id) does not exist."""
+class BadRequestError(AppError):
+    http_status: int = 400
+    code: str = "bad_request"
 
 
 @dataclass
-class InvalidRequest(AppError):
-    """User input is invalid or cannot be processed."""
+class SafetyViolationError(AppError):
+    http_status: int = 422
+    code: str = "safety_violation"
 
 
 @dataclass
-class SchemaRequired(AppError):
-    """Caller must provide schema_preview (e.g. postgres mode)."""
+class SchemaDeriveError(AppError):
+    http_status: int = 400
+    code: str = "schema_derive_error"
 
 
+# 5xx-ish
 @dataclass
-class SchemaDeriveError(AppError):
-    """Failed to derive schema preview from DB."""
+class DependencyError(AppError):
+    http_status: int = 503
+    code: str = "dependency_error"
+    retryable: bool = True
 
 
-# 5xx-ish
 @dataclass
 class PipelineConfigError(AppError):
-    """Pipeline/YAML/config is missing or malformed."""
+    http_status: int = 500
+    code: str = "pipeline_config_error"
 
 
 @dataclass
 class PipelineRunError(AppError):
-    """Unexpected failure while running the pipeline."""
+    http_status: int = 500
+    code: str = "pipeline_run_error"
+
+
+@dataclass
+class DbNotFound(BadRequestError):
+    code: str = "db_not_found"
+
+
+@dataclass
+class SchemaRequired(BadRequestError):
+    code: str = "schema_required"

app/exception_handlers.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from typing import Any, Dict
+import uuid
+from typing import Any, Dict, Optional, List
 
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
@@ -9,24 +10,32 @@ from app.errors import AppError
 
 
 def register_exception_handlers(app: FastAPI) -> None:
-    """
-    Register global exception handlers for the FastAPI application.
-    """
+    """Register global exception handlers for the FastAPI application."""
 
     @app.exception_handler(AppError)
     async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
-        """
-        Map domain-level AppError instances to HTTP responses.
-        This keeps routers thin and lets the domain raise AppError freely.
-        """
+        request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
+
         status = getattr(exc, "http_status", 500)
         code = getattr(exc, "code", "app_error")
         message = getattr(exc, "message", str(exc))
+        retryable = bool(getattr(exc, "retryable", False))
         extra: Dict[str, Any] = getattr(exc, "extra", {}) or {}
+        details: Optional[List[str]] = getattr(exc, "details", None)
 
         payload = {
-            "code": code,
-            "message": message,
-            "extra": extra,
+            "error": {
+                "code": code,
+                "message": message,
+                "details": details,
+                "retryable": retryable,
+                "request_id": request_id,
+                "extra": extra,
+            }
         }
-        return JSONResponse(status_code=status, content=payload)
+
+        headers = {"X-Request-ID": request_id}
+        if retryable:
+            headers["Retry-After"] = "2"
+
+        return JSONResponse(status_code=status, content=payload, headers=headers)

app/routers/nl2sql.py

@@ -23,6 +23,10 @@ from app.services.nl2sql_service import NL2SQLService
 from app.settings import get_settings
 from app.errors import (
     AppError,
+    BadRequestError,
+    SafetyViolationError,
+    DependencyError,
+    PipelineRunError,
 )
 
 logger = logging.getLogger(__name__)
@@ -330,14 +334,12 @@ def nl2sql_handler(
         # Let the global handler convert it to an HTTP response.
         raise
     except Exception as exc:
-        logger.exception(
-            "Unexpected pipeline crash in NL2SQLService.run_query",
-            exc_info=exc,
+        logger.exception("Unexpected pipeline crash in NL2SQLService.run_query")
+        raise PipelineRunError(
+            message="Internal pipeline error.",
+            details=[str(exc)],
+            extra={"stage": "unknown"},
         )
-        raise HTTPException(
-            status_code=500,
-            detail="internal pipeline error",
-        ) from exc
 
     # ---- type sanity check ----
     if not isinstance(result, FinalResult):
@@ -345,9 +347,10 @@ def nl2sql_handler(
             "Pipeline returned unexpected type",
             extra={"type": type(result).__name__},
         )
-        raise HTTPException(
-            status_code=500,
-            detail="pipeline returned unexpected type",
+        raise PipelineRunError(
+            message="Pipeline returned unexpected type.",
+            details=[type(result).__name__],
+            extra={"stage": "unknown"},
         )
 
     # ---- ambiguity path → 200 with clarification questions ----
@@ -355,18 +358,66 @@ def nl2sql_handler(
         qs = result.questions or []
         return ClarifyResponse(ambiguous=True, questions=qs)
 
-    # ---- error path → 400 with joined details ----
+    # ---- error path: map pipeline failures to stable HTTP+JSON error contract ----
     if (not result.ok) or result.error:
         logger.debug(
             "Pipeline reported failure",
-            extra={
-                "ok": result.ok,
-                "error": result.error,
-                "details": result.details,
-            },
+            extra={"ok": result.ok, "error": result.error, "details": result.details},
+        )
+
+        details = list(result.details or [])
+        traces = list(result.traces or [])
+        last_stage = str(traces[-1].get("stage", "unknown")) if traces else "unknown"
+        details_l = " ".join(d.lower() for d in details)
+
+        # 1) Safety violations → 422
+        if last_stage == "safety":
+            raise SafetyViolationError(
+                message="Rejected by safety checks.",
+                details=details or None,
+                extra={"stage": last_stage},
+            )
+
+        # 2) Retryable dependency failures → 503
+        retry_hints = (
+            "timeout",
+            "timed out",
+            "rate limit",
+            "429",
+            "too many requests",
+            "locked",
+            "busy",
+            "unavailable",
+            "connection",
+        )
+        if any(h in details_l for h in retry_hints):
+            raise DependencyError(
+                message="Temporary dependency failure. Please retry.",
+                details=details or None,
+                extra={"stage": last_stage},
+            )
+
+        # 3) User-fixable parse/syntax-ish errors → 400
+        user_hints = (
+            "parse_error",
+            "non-select",
+            "explain not allowed",
+            "multiple statements",
+            "forbidden",
+        )
+        if any(h in details_l for h in user_hints):
+            raise BadRequestError(
+                message="Request could not be processed.",
+                details=details or None,
+                extra={"stage": last_stage},
+            )
+
+        # 4) Default → 500
+        raise PipelineRunError(
+            message="Pipeline failed unexpectedly.",
+            details=details or None,
+            extra={"stage": last_stage},
         )
-        message = "; ".join(result.details or []) or "Unknown error"
-        raise HTTPException(status_code=400, detail=message)
 
     # ---- success path → 200 (normalize traces and executor result) ----
     traces = [_round_trace(t) for t in (result.traces or [])]

nl2sql/pipeline.py

@@ -422,22 +422,41 @@ class Pipeline:
             if r_exec.ok and isinstance(r_exec.data, dict):
                 exec_result = dict(r_exec.data)
 
-            # --- 6) verifier ---
+            # --- 6) verifier (run with repair for consistency) ---
             t0 = time.perf_counter()
-            r_ver = self._safe_stage(
+            r_ver = self._run_with_repair(
+                "verifier",
                 self.verifier.run,
+                repair_input_builder=self._sql_repair_input_builder,
+                max_attempts=1,
                 sql=sql,
                 exec_result=(r_exec.data or {}),
-                adapter=getattr(
-                    self.executor, "adapter", None
-                ),  # let verifier use adapter
+                adapter=getattr(self.executor, "adapter", None),
+                traces=traces,
             )
             dt = (time.perf_counter() - t0) * 1000.0
             stage_duration_ms.labels("verifier").observe(dt)
+
+            # Traces
             traces.extend(self._trace_list(r_ver))
             if not getattr(r_ver, "trace", None):
                 _fallback_trace("verifier", dt, r_ver.ok)
-            verified = bool(r_ver.data and r_ver.data.get("verified")) or r_ver.ok
+
+            # If verifier (or its repair) produced a new SQL, consume it
+            if r_ver.data and isinstance(r_ver.data, dict):
+                repaired_sql = r_ver.data.get("sql")
+                if repaired_sql:
+                    sql = repaired_sql
+
+            # Verified flag
+            verified = (
+                bool(
+                    r_ver.data
+                    and isinstance(r_ver.data, dict)
+                    and r_ver.data.get("verified")
+                )
+                or r_ver.ok
+            )
 
             # consume repaired SQL from verifier if any
             if r_ver.data and "sql" in r_ver.data and r_ver.data["sql"]:

nl2sql/verifier.py

@@ -4,32 +4,33 @@ import re
 import time
 from typing import Any, Dict
 
+from nl2sql.metrics import verifier_checks_total, verifier_failures_total
 from nl2sql.types import StageResult, StageTrace
-from nl2sql.metrics import (
-    verifier_checks_total,
-    verifier_failures_total,
-)
 
+from adapters.db.base import DBAdapter
 
-class Verifier:
-    """Static verifier used by tests.
 
-    Provides verify(...) for tests and run(...) for pipeline.
+class Verifier:
+    """
+    Verifier stage:
+    - Lightweight sanity checks (lint-like; NOT safety policy)
+    - Optional DB-backed plan validation via adapter.explain_query_plan(sql)
+      (read-only, no query execution)
     """
 
     required = False
 
-    def verify(self, sql: str, *, adapter: Any | None = None) -> StageResult:
+    def verify(self, sql: str, *, adapter: DBAdapter | None = None) -> StageResult:
         t0 = time.perf_counter()
         notes: Dict[str, Any] = {}
-        reason = "ok"  # new field
+        reason = "ok"
 
         s = (sql or "").strip()
         sl = s.lower()
         notes["sql_length"] = len(s)
 
         try:
-            # --- quick parse sanity: require SELECT and FROM ---
+            # --- quick sanity: require SELECT and FROM (lint-like) ---
             has_select = bool(re.search(r"\bselect\b", sl))
             has_from = bool(re.search(r"\bfrom\b", sl))
             notes["has_select"] = has_select
@@ -45,6 +46,7 @@ class Verifier:
                 )
 
             # --- semantic sanity: aggregation without GROUP BY (unless allowed) ---
+            # This is NOT a safety rule; it is a quality check to catch common mistakes.
             has_over = " over (" in sl
             has_group_by = " group by " in sl
             has_distinct = sl.startswith("select distinct") or (
@@ -83,20 +85,29 @@ class Verifier:
                     reason=reason,
                 )
 
-            # --- execution-error sentinel for tests ---
-            if "imaginary_table" in sl:
-                reason = "exec-error"
-                return self._fail(
-                    t0,
-                    notes,
-                    error=["exec_error: no such table: imaginary_table"],
-                    reason=reason,
-                )
+            # --- DB-backed plan validation (read-only), if adapter provided ---
+            # Safety policy (SELECT-only, no multi-statement, etc.) must be enforced upstream.
+            if adapter is not None:
+                try:
+                    adapter.explain_query_plan(s)
+                    notes["plan_check"] = "ok"
+                except Exception as e:
+                    reason = "plan-error"
+                    notes["plan_check"] = "failed"
+                    return self._fail(
+                        t0,
+                        notes,
+                        error=[str(e)],
+                        reason=reason,
+                        exc_type=type(e).__name__,
+                    )
 
             # --- pass ---
             dt = int(round((time.perf_counter() - t0) * 1000.0))
             notes.update({"verified": True, "reason": reason})
+
             verifier_checks_total.labels(ok="true").inc()
+
             trace = StageTrace(
                 stage="verifier",
                 duration_ms=dt,
@@ -106,6 +117,7 @@ class Verifier:
             return StageResult(ok=True, data={"verified": True}, trace=trace)
 
         except Exception as e:
+            # Unexpected verifier crash (bug)
             reason = "exception"
             return self._fail(
                 t0,
@@ -115,6 +127,16 @@ class Verifier:
                 exc_type=type(e).__name__,
             )
 
+    def run(
+        self,
+        *,
+        sql: str,
+        exec_result: Dict[str, Any],
+        adapter: DBAdapter | None = None,
+    ) -> StageResult:
+        # exec_result kept for signature compatibility, not used here.
+        return self.verify(sql, adapter=adapter)
+
     def _fail(
         self,
         t0: float,
@@ -125,6 +147,7 @@ class Verifier:
         exc_type: str | None = None,
     ) -> StageResult:
         dt = int(round((time.perf_counter() - t0) * 1000.0))
+
         notes.update({"verified": False, "reason": reason})
         if exc_type:
             notes["exception_type"] = exc_type
@@ -144,8 +167,3 @@ class Verifier:
             trace=trace,
             error=error,
         )
-
-    def run(
-        self, *, sql: str, exec_result: Dict[str, Any], adapter: Any = None
-    ) -> StageResult:
-        return self.verify(sql, adapter=adapter)