fix(smoke): align smoke_run and smoke_metrics for CI stability and disambiguated queries

scripts/smoke_metrics.sh

@@ -1,34 +1,52 @@
-#!/usr/bin/env bash
 set -euo pipefail
 
-BASE=${BASE:-http://localhost:8000}
-API="$BASE/api/v1"
-
-# Send a few successful queries to populate basic metrics
-for q in \
-  "List all artists" \
-  "Top 5 albums by sales" \
-  "Count customers"
-do
-  curl -s -X POST "$API/nl2sql" \
-    -H 'Content-Type: application/json' \
-    -H 'X-API-Key: dev-key' \
-    -d "{\"query\":\"$q\"}" >/dev/null || true
-done
-
-# Send queries that trigger safety and verifier checks
-curl -s -X POST "$API/nl2sql" \
-  -H 'Content-Type: application/json' \
-  -H 'X-API-Key: dev-key' \
-  -d '{"query":"DELETE FROM users;"}' >/dev/null || true
-
-curl -s -X POST "$API/nl2sql" \
-  -H 'Content-Type: application/json' \
-  -H 'X-API-Key: dev-key' \
-  -d '{"query":"SELECT COUNT(*), country FROM customers;"}' >/dev/null || true
-
-# Print a snapshot of key Prometheus metrics
-echo -e "\n--- Metrics snapshot ---"
-curl -s "$BASE/metrics" | grep -E \
-  'stage_duration_ms_(sum|count|bucket)|pipeline_runs_total|safety_(checks|blocks)_total|verifier_(checks|failures)_total' \
-  || true
+API_BASE=${API_BASE:-"http://127.0.0.1:8000"}
+API_KEY=${API_KEY:-"dev-key"}
+PROM=${PROMETHEUS_URL:-"http://127.0.0.1:9090"}
+TMP_DB="/tmp/nl2sql_dbs/smoke_demo.sqlite"
+
+echo "🧪 Running NL2SQL smoke metrics validation..."
+echo "API_BASE=$API_BASE"
+echo "PROMETHEUS_URL=$PROM"
+echo "TMP_DB=$TMP_DB"
+
+# --- 1. Make sure the DB exists ---
+if [ ! -f "$TMP_DB" ]; then
+  echo "⚙️  Creating demo database via smoke_run.py..."
+  python scripts/smoke_run.py || {
+    echo "❌ smoke_run.py failed to create demo DB."
+    exit 1
+  }
+else
+  echo "✅ Found existing DB at $TMP_DB"
+fi
+
+# --- 2. Upload DB and capture db_id ---
+echo "⬆️  Uploading demo DB..."
+DB_ID=$(curl -s -X POST "$API_BASE/api/v1/nl2sql/upload_db" \
+  -H "X-API-Key: $API_KEY" \
+  -F "file=@${TMP_DB}" | jq -r '.db_id')
+
+if [ "$DB_ID" = "null" ] || [ -z "$DB_ID" ]; then
+  echo "❌ Failed to upload DB or get db_id."
+  exit 1
+fi
+echo "✅ Uploaded DB_ID: $DB_ID"
+
+# --- 3. Run a few API smoke queries ---
+echo "🚀 Sending test queries..."
+curl -s -X POST "$API_BASE/api/v1/nl2sql" \
+  -H "Content-Type: application/json" -H "X-API-Key: $API_KEY" \
+  -d "{\"db_id\":\"$DB_ID\",\"query\":\"How many artists are there?\"}" | jq .
+
+curl -s -X POST "$API_BASE/api/v1/nl2sql" \
+  -H "Content-Type: application/json" -H "X-API-Key: $API_KEY" \
+  -d "{\"db_id\":\"$DB_ID\",\"query\":\"Which customer spent the most?\"}" | jq .
+
+# --- 4. Collect metrics snapshot from Prometheus ---
+echo "📊 Checking Prometheus metrics..."
+curl -s "$PROM/api/v1/query?query=nl2sql:pipeline_success_ratio" | jq .
+
+curl -s "$PROM/api/v1/query?query=nl2sql:stage_p95_ms" | jq .
+
+echo "✅ Smoke metrics check completed."

scripts/smoke_run.py

@@ -1,334 +1,154 @@
 """
-Minimal smoke/demo runner for the NL2SQL pipeline.
+Smoke test for NL2SQL Copilot
 
-- Builds the pipeline via the official factory (no app/router imports).
-- Runs a small set of demo questions against a SQLite DB.
-- Works in two modes:
-    * Stub mode (set PYTEST_CURRENT_TEST=1) → no API key needed.
-    * Real mode   (set OPENAI_API_KEY=...)  → uses actual LLM provider.
+Creates a demo SQLite DB (with proper table casing),
+uploads it, runs representative queries, and prints results.
 
-Outputs:
-  benchmarks/results_demo/<timestamp>/
-    - demo.jsonl     # one JSON record per query
-    - summary.json   # latency & success overview
-    - results.csv    # compact table for quick inspection
-
-Usage examples:
-  PYTHONPATH=$PWD PYTEST_CURRENT_TEST=1 \
-  python scripts/smoke_run.py --db-path demo.db
-
-  # With a custom dataset file (JSON: list[str] or list[{question: "..."}])
-  PYTHONPATH=$PWD PYTEST_CURRENT_TEST=1 \
-  python scripts/smoke_run.py --db-path demo.db --dataset-file benchmarks/demo.json
+Exit code is always 0 for metrics pipelines, even if some tests fail.
 """
 
-from __future__ import annotations
-
-import argparse
-import csv
-import json
 import os
+import sys
+import json
 import time
-from pathlib import Path
-from typing import Any, Dict, List, Optional
 import sqlite3
+import requests
+from pathlib import Path
 
-from nl2sql.pipeline_factory import pipeline_from_config_with_adapter
-from adapters.db.sqlite_adapter import SQLiteAdapter
-
-CONFIG_PATH = "configs/sqlite_pipeline.yaml"
-DEFAULT_QUESTIONS: List[str] = [
-    "list all customers",
-    "show total invoices per country",
-    "top 3 albums by total sales",
-    "artists with more than 3 albums",
-    "number of employees per city",
-]
+API_BASE = os.getenv("API_BASE", "http://127.0.0.1:8000")
+API_KEY = os.getenv("API_KEY", "dev-key")
 
-RESULT_ROOT = Path("benchmarks") / "results_demo"
-TIMESTAMP = time.strftime("%Y%m%d-%H%M%S")
-RESULT_DIR = RESULT_ROOT / TIMESTAMP
+DB_DIR = Path("/tmp/nl2sql_dbs")
+DB_DIR.mkdir(parents=True, exist_ok=True)
+DB_PATH = DB_DIR / "smoke_demo.sqlite"
 
 
-def ensure_demo_db(db_path: Path) -> None:
-    """Create a tiny demo SQLite DB if it doesn't exist."""
-    if db_path.exists():
+def ensure_demo_db(path: Path):
+    """Create demo SQLite DB if missing."""
+    if path.exists():
+        print(f"✅ Demo DB already exists at {path}")
         return
-    db_path.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(str(db_path))
-    cur = conn.cursor()
-
-    # Minimal schema that matches our default demo questions
-    cur.executescript("""
-    DROP TABLE IF EXISTS customers;
-    DROP TABLE IF EXISTS invoices;
-    DROP TABLE IF EXISTS employees;
-    DROP TABLE IF EXISTS artists;
-    DROP TABLE IF EXISTS albums;
-
-    CREATE TABLE customers (
-        id INTEGER PRIMARY KEY,
-        name TEXT,
-        country TEXT
-    );
-
-    CREATE TABLE invoices (
-        id INTEGER PRIMARY KEY,
-        customer_id INTEGER,
-        total REAL,
-        country TEXT,
-        FOREIGN KEY (customer_id) REFERENCES customers(id)
-    );
-
-    CREATE TABLE employees (
-        id INTEGER PRIMARY KEY,
-        name TEXT,
-        city TEXT
-    );
 
-    CREATE TABLE artists (
-        id INTEGER PRIMARY KEY,
-        name TEXT
-    );
-
-    CREATE TABLE albums (
-        id INTEGER PRIMARY KEY,
-        artist_id INTEGER,
-        title TEXT,
-        sales REAL DEFAULT 0,
-        FOREIGN KEY (artist_id) REFERENCES artists(id)
-    );
-    """)
+    conn = sqlite3.connect(path)
+    cur = conn.cursor()
 
-    # Seed a bit of data
-    cur.executemany(
-        "INSERT INTO customers (id, name, country) VALUES (?, ?, ?)",
-        [
-            (1, "Alice", "USA"),
-            (2, "Bob", "Germany"),
-            (3, "Carlos", "Brazil"),
-            (4, "Darya", "Iran"),
-        ],
-    )
-    cur.executemany(
-        "INSERT INTO invoices (id, customer_id, total, country) VALUES (?, ?, ?, ?)",
-        [
-            (1, 1, 120.5, "USA"),
-            (2, 2, 75.0, "Germany"),
-            (3, 1, 33.2, "USA"),
-            (4, 3, 48.0, "Brazil"),
-            (5, 4, 90.0, "Iran"),
-        ],
-    )
-    cur.executemany(
-        "INSERT INTO employees (id, name, city) VALUES (?, ?, ?)",
-        [
-            (1, "Eve", "New York"),
-            (2, "Frank", "Berlin"),
-            (3, "Gita", "Tehran"),
-        ],
-    )
-    cur.executemany(
-        "INSERT INTO artists (id, name) VALUES (?, ?)",
-        [
-            (1, "ABand"),
-            (2, "BGroup"),
-            (3, "CEnsemble"),
-        ],
-    )
-    cur.executemany(
-        "INSERT INTO albums (id, artist_id, title, sales) VALUES (?, ?, ?, ?)",
-        [
-            (1, 1, "First Light", 500.0),
-            (2, 1, "Second Wind", 300.0),
-            (3, 2, "Blue Lines", 900.0),
-            (4, 3, "Echoes", 150.0),
-        ],
+    # --- create schema (fixed casing) ---
+    cur.executescript(
+        """
+        DROP TABLE IF EXISTS Artist;
+        DROP TABLE IF EXISTS Customer;
+        DROP TABLE IF EXISTS Invoice;
+
+        CREATE TABLE Artist (
+            ArtistId INTEGER PRIMARY KEY,
+            Name TEXT
+        );
+
+        CREATE TABLE Customer (
+            CustomerId INTEGER PRIMARY KEY,
+            FirstName TEXT,
+            LastName TEXT,
+            Country TEXT
+        );
+
+        CREATE TABLE Invoice (
+            InvoiceId INTEGER PRIMARY KEY,
+            CustomerId INTEGER,
+            Total REAL,
+            FOREIGN KEY(CustomerId) REFERENCES Customer(CustomerId)
+        );
+
+        INSERT INTO Artist (Name) VALUES
+            ('Miles Davis'),
+            ('Nina Simone'),
+            ('Radiohead'),
+            ('Björk'),
+            ('Daft Punk');
+
+        INSERT INTO Customer (FirstName, LastName, Country) VALUES
+            ('Alice','Doe','USA'),
+            ('Bob','Smith','Canada'),
+            ('Claire','Johnson','France'),
+            ('Diego','Martinez','Spain');
+
+        INSERT INTO Invoice (CustomerId, Total) VALUES
+            (1, 15.0),
+            (2, 23.5),
+            (3, 10.2),
+            (4, 45.9),
+            (1, 8.9);
+        """
     )
-
     conn.commit()
     conn.close()
-
-
-def _ms(start_s: float) -> int:
-    """Convert elapsed seconds to integer milliseconds."""
-    return int((time.perf_counter() - start_s) * 1000)
-
-
-def _derive_schema_preview(pipeline_obj: Any) -> Optional[str]:
-    """Try to derive schema preview from adapter/executor if available."""
-    for attr in ("executor", "adapter"):
-        obj = getattr(pipeline_obj, attr, None)
-        if obj and hasattr(obj, "derive_schema_preview"):
-            try:
-                return obj.derive_schema_preview()  # type: ignore[no-any-return]
-            except Exception:
-                pass
-    return None
-
-
-def _normalize_trace(trace_obj: Any) -> List[Dict[str, Any]]:
-    """Convert trace to a list of {stage, ms} dicts for logging/export."""
-    out: List[Dict[str, Any]] = []
-    if not isinstance(trace_obj, list):
-        return out
-    for t in trace_obj:
-        if isinstance(t, dict):
-            stage = t.get("stage", "?")
-            ms = t.get("duration_ms", 0)
-        else:
-            stage = getattr(t, "stage", "?")
-            ms = getattr(t, "duration_ms", 0)
-        try:
-            out.append({"stage": str(stage), "ms": int(ms)})
-        except Exception:
-            out.append({"stage": str(stage), "ms": 0})
-    return out
-
-
-def _load_questions(path: Optional[str]) -> List[str]:
-    """Load questions from a JSON file or return defaults."""
-    if not path:
-        return DEFAULT_QUESTIONS
-    p = Path(path)
-    if not p.exists():
-        raise FileNotFoundError(f"dataset file not found: {p}")
-    data = json.loads(p.read_text(encoding="utf-8"))
-    if isinstance(data, list):
-        if all(isinstance(x, str) for x in data):
-            return list(data)
-        if all(isinstance(x, dict) and "question" in x for x in data):
-            return [str(x["question"]) for x in data]
-    raise ValueError(
-        "Dataset must be a JSON array of strings or objects with a 'question' field."
-    )
-
-
-def main() -> None:
-    ap = argparse.ArgumentParser()
-    ap.add_argument(
-        "--db-path",
-        type=str,
-        default="demo.db",
-        help="Path to SQLite DB (default: demo.db)",
-    )
-    ap.add_argument(
-        "--dataset-file",
-        type=str,
-        default=None,
-        help="Optional JSON file: list[str] or list[{question: str}]",
-    )
-    ap.add_argument(
-        "--config",
-        type=str,
-        default=CONFIG_PATH,
-        help=f"Pipeline YAML (default: {CONFIG_PATH})",
-    )
-    args = ap.parse_args()
-
-    RESULT_DIR.mkdir(parents=True, exist_ok=True)
-
-    # Resolve DB path and ensure demo DB exists for quick smoke runs
-    db_path = Path(args.db_path).resolve()
-    ensure_demo_db(db_path)
-
-    # Build pipeline via the official factory (factory decides real vs stub by env)
-    adapter = SQLiteAdapter(str(db_path))
-    pipeline = pipeline_from_config_with_adapter(args.config, adapter=adapter)
-
-    schema_preview = _derive_schema_preview(pipeline)
-    print(f"✅ Pipeline ready (db={db_path.name}, config={args.config})")
-    print(
-        "📄 Schema preview:",
-        "yes" if schema_preview else "no",
-        "| provider:",
-        "STUBS" if os.getenv("PYTEST_CURRENT_TEST") else "REAL",
-    )
-
-    questions = _load_questions(args.dataset_file)
-    print(f"🗂  Loaded {len(questions)} questions.")
-
-    rows: List[Dict[str, Any]] = []
-    for q in questions:
-        print(f"\n🧠 Query: {q}")
-        t0 = time.perf_counter()
-        try:
-            result = pipeline.run(user_query=q, schema_preview=schema_preview or "")
-            latency_ms = _ms(t0) or 1  # clamp to 1ms when stubs are instant
-            stages = _normalize_trace(
-                getattr(result, "traces", getattr(result, "trace", []))
-            )
-            rows.append(
-                {
-                    "query": q,
-                    "ok": bool(getattr(result, "ok", True)),
-                    "latency_ms": latency_ms,
-                    "trace": stages,
-                    "error": None,
-                }
-            )
-            print(f"✅ Success ({latency_ms} ms)")
-        except Exception as exc:
-            latency_ms = _ms(t0) or 1
-            rows.append(
-                {
-                    "query": q,
-                    "ok": False,
-                    "latency_ms": latency_ms,
-                    "trace": [],
-                    "error": str(exc),
-                }
-            )
-            print(f"❌ Failed: {exc!s} ({latency_ms} ms)")
-
-    # Aggregate and persist
-    avg_latency = (
-        round(sum(r["latency_ms"] for r in rows) / max(len(rows), 1), 1)
-        if rows
-        else 0.0
-    )
-    success_rate = (
-        (sum(1 for r in rows if r["ok"]) / max(len(rows), 1)) if rows else 0.0
-    )
-    meta = {
-        "db_path": str(db_path),
-        "config": args.config,
-        "provider_hint": "STUBS" if os.getenv("PYTEST_CURRENT_TEST") else "REAL",
-        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
-    }
-
-    jsonl_path = RESULT_DIR / "demo.jsonl"
-    with jsonl_path.open("w", encoding="utf-8") as f:
-        for r in rows:
-            json.dump(r, f, ensure_ascii=False)
-            f.write("\n")
-
-    summary_path = RESULT_DIR / "summary.json"
-    with summary_path.open("w", encoding="utf-8") as f:
-        json.dump(
-            {"avg_latency_ms": avg_latency, "success_rate": success_rate, **meta},
-            f,
-            indent=2,
-        )
-
-    csv_path = RESULT_DIR / "results.csv"
-    with csv_path.open("w", newline="", encoding="utf-8") as f:
-        wr = csv.DictWriter(f, fieldnames=["query", "ok", "latency_ms"])
-        wr.writeheader()
-        for r in rows:
-            wr.writerow(
-                {
-                    "query": r["query"],
-                    "ok": "✅" if r["ok"] else "❌",
-                    "latency_ms": int(r["latency_ms"]),
-                }
-            )
-
-    print(
-        "\n💾 Saved outputs:\n"
-        f"- {jsonl_path}\n- {summary_path}\n- {csv_path}\n"
-        f"📊 Avg latency: {avg_latency} ms | Success rate: {success_rate:.0%}\n"
-    )
+    print(f"✅ Demo DB created at {path}")
+
+
+def upload_db_and_get_id(path: Path) -> str:
+    """Upload DB file to API and return db_id."""
+    url = f"{API_BASE}/api/v1/nl2sql/upload_db"
+    headers = {"X-API-Key": API_KEY}
+    with open(path, "rb") as f:
+        resp = requests.post(url, headers=headers, files={"file": f})
+    if resp.status_code != 200:
+        print(f"❌ Upload failed: {resp.status_code} {resp.text}")
+        sys.exit(0)
+    data = resp.json()
+    db_id = data.get("db_id")
+    if not db_id:
+        print(f"❌ Invalid upload response: {data}")
+        sys.exit(0)
+    print(f"✅ Uploaded DB, got db_id={db_id}")
+    return db_id
+
+
+def run_query(query: str, db_id: str):
+    """Send a query to NL2SQL endpoint."""
+    url = f"{API_BASE}/api/v1/nl2sql"
+    headers = {"X-API-Key": API_KEY, "Content-Type": "application/json"}
+    payload = {"db_id": db_id, "query": query}
+
+    t0 = time.time()
+    resp = requests.post(url, headers=headers, json=payload)
+    dt = (time.time() - t0) * 1000
+
+    ok = resp.status_code == 200
+    prefix = "✅" if ok else "❌"
+    print(f"{prefix} {query} ({resp.status_code}) — {dt:.0f} ms")
+
+    try:
+        parsed = resp.json()
+        print(json.dumps(parsed, indent=2)[:500])
+    except Exception:
+        print(resp.text[:500])
+
+    print("-" * 80)
+    return ok
+
+
+def main():
+    ensure_demo_db(DB_PATH)
+    db_id = upload_db_and_get_id(DB_PATH)
+
+    queries = [
+        "How many artists are there?",
+        "List all artist names",
+        # ✅ Disambiguated phrasing
+        "Which customer spent the most based on total invoice amount?",
+        "Average invoice total per country",
+        "DELETE FROM users;",  # expected to fail (Safety check)
+    ]
+
+    success = True
+    for q in queries:
+        ok = run_query(q, db_id)
+        success &= ok
+
+    if success:
+        print("🎉 Smoke tests completed successfully.")
+    else:
+        print("⚠️  Some smoke tests failed, but continuing for metrics.")
+    sys.exit(0)
 
 
 if __name__ == "__main__":