Sync from GitHub main @ 8f40ad2807fc87dbdaae076316a949ce2aa8d865

README.md

@@ -10,7 +10,7 @@ pinned: false
 # NL2SQL Copilot — Safety-First, Production-Grade Text-to-SQL
 
 [![CI](https://github.com/melika-kheirieh/nl2sql-copilot/actions/workflows/ci.yml/badge.svg)](https://github.com/melika-kheirieh/nl2sql-copilot/actions/workflows/ci.yml)
-[![Docker](https://img.shields.io/badge/docker-ready-blue?logo=docker)](#)
+[![Docker](https://img.shields.io/badge/docker--compose-demo-blue?logo=docker)](docs/runbook.md)
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
 
 A **production-oriented, multi-stage Natural Language → SQL system** built around
@@ -103,7 +103,7 @@ make demo-metrics
 ```
 
 >For a complete end-to-end setup (API, infra, metrics, dashboards, UIs),
-see [docs//runbook.md](docs//runbook.md).
+see [docs/runbook.md](docs/runbook.md).
 
 ---
 

app/routers/nl2sql.py

@@ -213,7 +213,7 @@ def health():
     return {"status": "ok", "version": settings.app_version}
 
 
-def _ck(db_id: Optional[str], query: str, schema_preview: str) -> str:
+def _ck(db_id: Optional[str], query: str) -> str:
     db_part = db_id or "__default__"
     seed = f"{db_part}\n{query.strip()}"
     return hashlib.sha1(seed.encode("utf-8")).hexdigest()
@@ -232,20 +232,6 @@ def nl2sql_handler(
 ) -> NL2SQLResponse | ClarifyResponse:
     db_id = getattr(request, "db_id", None)
 
-    # # ---- deterministic SELECT-only guard ----
-    # # Block DML/DDL intents early (before schema derivation, cache, or LLM calls).
-    # if _is_unsafe_intent(getattr(request, "query", "")):
-    #     raise HTTPException(
-    #         status_code=400,
-    #         detail={
-    #             "error": {
-    #                 "code": "BAD_REQUEST",
-    #                 "retryable": False,
-    #                 "details": ["non_select_query"],
-    #             }
-    #         },
-    #     )
-
     # ---- schema preview ----
     try:
         final_preview = svc.get_schema_preview(
@@ -266,7 +252,7 @@ def nl2sql_handler(
         ) from exc
 
     # ---- cache lookup ----
-    cache_key = _ck(db_id, request.query, final_preview)
+    cache_key = _ck(db_id, request.query)
     cached_payload = cache.get(cache_key)
     if cached_payload is not None:
         # Cache stores dicts; convert back to response models for type safety.

scripts/demo_cache_showcase.py

@@ -0,0 +1,351 @@
+#!/usr/bin/env python3
+"""Generate a reproducible cache/metrics screenshot workload.
+
+What it does:
+1) Waits for API readiness (healthz + readyz + router health).
+2) Uploads a demo SQLite DB to the API (upload_db) and captures db_id.
+3) Sends a burst of unique queries (mostly misses).
+4) Sends repeated queries over ~70–90s (hits), with jitter so charts look natural.
+5) Triggers a safety violation once (should be blocked) WITHOUT failing the whole demo.
+6) Sends a final "recovery" query (OK).
+7) (Optional) Prints a Prometheus instant-query sanity check for cache metrics.
+
+Expected API:
+- POST {API_BASE}/api/v1/nl2sql/upload_db  (multipart form: file=@db.sqlite) -> {db_id: "..."}
+- POST {API_BASE}/api/v1/nl2sql           (json: {db_id, query, schema_preview?}) -> 200 or 4xx/5xx
+- GET  {API_BASE}/healthz
+- GET  {API_BASE}/readyz
+- GET  {API_BASE}/api/v1/nl2sql/health
+
+Env:
+- API_BASE  (default http://127.0.0.1:8000)
+- API_KEY   (default dev-key)
+- DB_PATH   (default /tmp/nl2sql_dbs/smoke_demo.sqlite)
+- PROM_BASE (default http://127.0.0.1:9090) (optional; set empty to skip)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import random
+import subprocess
+import time
+from dataclasses import dataclass
+from typing import Any
+
+
+def sh(args: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
+    """Run a command and return the completed process (text mode)."""
+    return subprocess.run(
+        args,
+        check=check,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+
+@dataclass(frozen=True)
+class Cfg:
+    api_base: str
+    api_key: str
+    db_path: str
+    prom_base: str | None
+
+
+def load_cfg() -> Cfg:
+    api_base = os.getenv("API_BASE", "http://127.0.0.1:8000").rstrip("/")
+    api_key = os.getenv("API_KEY", "dev-key")
+    db_path = os.getenv("DB_PATH", "/tmp/nl2sql_dbs/smoke_demo.sqlite")
+    prom_base_env = os.getenv("PROM_BASE", "http://127.0.0.1:9090").rstrip("/")
+    prom_base: str | None = prom_base_env if prom_base_env else None
+    return Cfg(api_base=api_base, api_key=api_key, db_path=db_path, prom_base=prom_base)
+
+
+def wait_for_ready(cfg: Cfg, timeout_s: float = 60.0) -> None:
+    """Wait until API is responsive and ready.
+
+    We try multiple endpoints because on cold starts the container may accept TCP but reset early requests.
+    """
+    endpoints = [
+        f"{cfg.api_base}/healthz",
+        f"{cfg.api_base}/readyz",
+        f"{cfg.api_base}/api/v1/nl2sql/health",
+    ]
+
+    start = time.time()
+    last = ""
+    while time.time() - start < timeout_s:
+        ok = True
+        for url in endpoints:
+            cp = subprocess.run(
+                ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", url],
+                check=False,
+                text=True,
+                capture_output=True,
+            )
+            code = (cp.stdout or "").strip()
+            if code != "200":
+                ok = False
+                last = f"url={url} http={code} stderr={cp.stderr.strip()!r}"
+                break
+
+        if ok:
+            return
+
+        time.sleep(0.6)
+
+    raise RuntimeError(f"API not ready after {timeout_s:.0f}s. Last={last}")
+
+
+def upload_db(cfg: Cfg) -> str:
+    if not os.path.exists(cfg.db_path):
+        raise FileNotFoundError(f"DB_PATH not found: {cfg.db_path}")
+
+    url = f"{cfg.api_base}/api/v1/nl2sql/upload_db"
+
+    # Do NOT use -f here; on error we want the body.
+    cp = subprocess.run(
+        [
+            "curl",
+            "-sS",
+            "-D",
+            "-",
+            "-H",
+            f"X-API-Key: {cfg.api_key}",
+            "-F",
+            f"file=@{cfg.db_path}",
+            url,
+        ],
+        check=False,
+        text=True,
+        capture_output=True,
+    )
+
+    if cp.returncode != 0:
+        raise RuntimeError(
+            f"upload_db curl failed (rc={cp.returncode}). stderr={cp.stderr.strip()!r}\nstdout:\n{cp.stdout}"
+        )
+
+    # Split headers/body
+    raw = cp.stdout
+    parts = raw.split("\r\n\r\n", 1)
+    if len(parts) != 2:
+        parts = raw.split("\n\n", 1)
+    if len(parts) != 2:
+        raise RuntimeError(f"upload_db returned unexpected response:\n{raw}")
+
+    headers, body = parts[0], parts[1]
+    status_line = headers.splitlines()[0] if headers.splitlines() else ""
+    if " 200 " not in status_line:
+        raise RuntimeError(f"upload_db non-200.\n{headers}\n\n{body}")
+
+    try:
+        data = json.loads(body)
+    except json.JSONDecodeError as e:
+        raise RuntimeError(
+            f"upload_db returned non-JSON body.\n{headers}\n\n{body}"
+        ) from e
+
+    db_id = data.get("db_id")
+    if not isinstance(db_id, str) or not db_id:
+        raise RuntimeError(f"upload_db response missing db_id: {data}")
+    return db_id
+
+
+def post_query(
+    cfg: Cfg, *, db_id: str, query: str, fail_on_non_200: bool = True
+) -> int:
+    """POST a query. Returns HTTP status code. Optionally raises on non-200 with full response."""
+    url = f"{cfg.api_base}/api/v1/nl2sql"
+    payload = json.dumps({"db_id": db_id, "query": query})
+
+    cp = subprocess.run(
+        [
+            "curl",
+            "-sS",
+            "-D",
+            "-",
+            "-H",
+            f"X-API-Key: {cfg.api_key}",
+            "-H",
+            "Content-Type: application/json",
+            "-d",
+            payload,
+            url,
+        ],
+        check=False,
+        text=True,
+        capture_output=True,
+    )
+
+    if cp.returncode != 0:
+        raise RuntimeError(
+            f"query curl failed (rc={cp.returncode}). query={query!r}\n"
+            f"stderr={cp.stderr.strip()!r}\nstdout:\n{cp.stdout}"
+        )
+
+    raw = cp.stdout
+    parts = raw.split("\r\n\r\n", 1)
+    if len(parts) != 2:
+        parts = raw.split("\n\n", 1)
+    if len(parts) != 2:
+        raise RuntimeError(
+            f"query returned unexpected response. query={query!r}\n{raw}"
+        )
+
+    headers, body = parts[0], parts[1]
+    status_line = headers.splitlines()[0] if headers.splitlines() else ""
+
+    # Parse HTTP status code from first line: HTTP/1.1 200 OK
+    status_code = 0
+    try:
+        status_code = int(status_line.split()[1])
+    except Exception:
+        status_code = 0
+
+    if fail_on_non_200 and status_code != 200:
+        raise RuntimeError(f"Non-200 response for query={query!r}\n{headers}\n\n{body}")
+
+    return status_code
+
+
+def prom_instant_query(cfg: Cfg, expr: str) -> Any | None:
+    if not cfg.prom_base:
+        return None
+    url = f"{cfg.prom_base}/api/v1/query"
+    cp = sh(["curl", "-fsS", url, "--data-urlencode", f"query={expr}"])
+    return json.loads(cp.stdout)
+
+
+def post_dev_safety(cfg: Cfg, sql: str) -> int:
+    """Trigger the Safety stage directly (dev endpoint) so OK-rate panels aren't affected."""
+    url = f"{cfg.api_base}/api/v1/_dev/safety"
+    payload = json.dumps({"sql": sql})
+    cp = subprocess.run(
+        [
+            "curl",
+            "-sS",
+            "-D",
+            "-",
+            "-H",
+            f"X-API-Key: {cfg.api_key}",
+            "-H",
+            "Content-Type: application/json",
+            "-d",
+            payload,
+            url,
+        ],
+        check=False,
+        text=True,
+        capture_output=True,
+    )
+    raw = cp.stdout
+    # Parse status code from HTTP status line.
+    header_block = raw.split("\r\n\r\n", 1)[0]
+    status_line = header_block.splitlines()[0] if header_block.splitlines() else ""
+    try:
+        return int(status_line.split()[1])
+    except Exception:
+        return 0
+
+
+def print_cache_sanity(cfg: Cfg) -> None:
+    if not cfg.prom_base:
+        return
+
+    candidates = [
+        "nl2sql:cache_hit_ratio",
+        'sum(rate(cache_events_total{hit="true"}[5m])) / sum(rate(cache_events_total[5m]))',
+    ]
+
+    for expr in candidates:
+        try:
+            data = prom_instant_query(cfg, expr)
+            if data is None:
+                continue
+        except Exception:
+            continue
+        try:
+            result = data["data"]["result"]
+        except Exception:
+            continue
+        if result:
+            value = result[0].get("value", [None, None])[1]
+            print(f"[prom] {expr} = {value}")
+            return
+
+    print("[prom] Could not find cache ratio metric yet (ok right after cold start).")
+
+
+def main() -> int:
+    cfg = load_cfg()
+
+    random.seed(7)  # deterministic-ish graphs
+
+    print("Waiting for API readiness...")
+    wait_for_ready(cfg, timeout_s=75)
+
+    print("Uploading DB...")
+    db_id = upload_db(cfg)
+    print(f"DB_ID={db_id}")
+
+    # Phase A: warm-up (mostly misses)
+    unique = [
+        "List the first 10 artists.",
+        "Which customer spent the most based on total invoice amount?",
+        "Top 5 tracks by duration.",
+    ]
+
+    print("Phase A: warmup (mostly misses)...")
+    for q in unique:
+        post_query(cfg, db_id=db_id, query=q)
+        time.sleep(0.7)
+
+    # Phase B: repeats (hits)
+    repeats = [
+        "Which customer spent the most based on total invoice amount?",
+        "List the first 10 artists.",
+        "Which customer spent the most based on total invoice amount?",
+        "Top 5 tracks by duration.",
+        "List the first 10 artists.",
+    ]
+
+    print("Phase B: repeated queries (hits)...")
+    # ~60 requests over ~1.5–2 minutes (enough signal for window-based panels)
+    for _ in range(60):
+        q = random.choice(repeats)
+        post_query(cfg, db_id=db_id, query=q)
+        time.sleep(1.1 + random.random() * 0.5)
+
+    # Give Prometheus a moment to scrape after the last request.
+    time.sleep(10)
+
+    print("\nSanity check:")
+    print_cache_sanity(cfg)
+
+    print("\n>>> NOW TAKE SCREENSHOT <<<")
+    print(
+        "Grafana: set time range to Last 10 minutes (or Last 15 minutes), refresh 5s, wait ~10s."
+    )
+    print("Tip: if hit% looks low, wait one more scrape interval and refresh.")
+
+    # Phase C: safety check (expected block) — after screenshot so OK% stays high in-window.
+    print("\nPhase C: safety check (expected block, after screenshot)...")
+    code = post_dev_safety(cfg, "drop table users;")
+    print(f"Safety request status={code} (expected non-200)")
+
+    # Phase D: recovery
+    print("Phase D: recovery...")
+    post_query(cfg, db_id=db_id, query="List the first 10 artists.")
+
+    print("\nDone. Suggested screenshot steps:")
+    print("  1) In Grafana set time range: Last 10 minutes (or Last 15 minutes).")
+    print("  2) Set refresh to 5s–10s and wait 10–20s for panels to catch up.")
+    print("  3) Expect Requests-in-window > 10 and Cache Hit Ratio > 0.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/smoke_api.py

@@ -16,6 +16,7 @@ import json
 import os
 import time
 from pathlib import Path
+import re
 
 import requests
 
@@ -26,6 +27,19 @@ API_KEY = os.getenv("API_KEY", "dev-key")
 DB_DIR = Path("/tmp/nl2sql_dbs")
 DB_PATH = DB_DIR / "smoke_demo.sqlite"
 
+_DML_DDL_SQL_RE = re.compile(
+    r"\b(delete|update|insert|drop|alter|truncate|create|replace)\b", re.IGNORECASE
+)
+
+
+def _is_select_only_sql(sql: str | None) -> bool:
+    if not sql:
+        return False
+    s = sql.strip().lower()
+    if not s.startswith("select"):
+        return False
+    return _DML_DDL_SQL_RE.search(sql) is None
+
 
 def _ensure_demo_db(path: Path) -> None:
     """Delegate to scripts/smoke_run.py if available; otherwise fail."""
@@ -112,7 +126,7 @@ def main() -> int:
     checks = [
         ("List the first 10 artists.", True),
         ("Which customer spent the most based on total invoice amount?", True),
-        ("DELETE FROM users;", False),  # must be blocked
+        ("SELECT * FROM Invoice;", False),  # must be blocked (full scan without LIMIT)
     ]
 
     ok_all = True
@@ -130,12 +144,21 @@ def main() -> int:
         else:
             allowed = {
                 "LLM_BAD_OUTPUT",
-                "SQL_NOT_ALLOWED",
-                "INVALID_SQL",
-                "BAD_REQUEST",
+                "PIPELINE_CRASH",  # e.g. full_scan_without_limit guardrail
+                "SAFETY_NON_SELECT",
+                "SAFETY_MULTI_STATEMENT",
             }
-            if not _is_expected_block(status=status, body=body, allowed_codes=allowed):
-                ok_all = False
+
+            if status != 200:
+                if not _is_expected_block(
+                    status=status, body=body, allowed_codes=allowed
+                ):
+                    ok_all = False
+            else:
+                # Accept safe refusal: 200 but SQL must be SELECT-only.
+                sql = body.get("sql") if isinstance(body, dict) else None
+                if not _is_select_only_sql(sql):
+                    ok_all = False
 
     if ok_all:
         print("\n✅ demo-smoke passed")