Build a Schema-Aware SQL Gate

Introduction

30 Second Summary

A confident answer can look trustworthy even when it was built from a made-up detail. The risk grows when that answer can trigger costly work before anyone reviews it.

In this project, you will add a reliability gate between Groq's generated SQL and PostgreSQL. The gate blocks unreliable queries before PostgreSQL can plan or run them.

What You'll Build

When you finish, a business question in Windows PowerShell produces a reliability report with visible cache reuse, local validation, estimated plan warnings, and a clear execution choice.

By the end of this project, you'll have:

• A visible schema cache where you can see MISS, REFRESH, or HIT beside a content fingerprint.
• A grounded SQL flow where What is total revenue? produces the discounted-revenue formula while a hallucinated column is blocked before PostgreSQL.
• A plan preflight that shows the root node, estimated rows, total cost, and warnings before you choose whether to run the query in a read-only transaction.
• An optional Secret Mission: build a three-case TPC-H evaluation harness that reports result-equivalence accuracy and categorized failures.

Are there any prerequisites?

You need an existing Windows-based single-agent text-to-SQL repository with working Groq and PostgreSQL access. The connected database must contain TPC-H SF10 data.

Before We Start

Generated SQL can look convincing while still exposing your database to avoidable failures. This step lets you choose one concrete failure that your reliability gate must prevent.

Set Up the Reliability Upgrade

Your existing text-to-SQL agent already reaches Groq and PostgreSQL. This upgrade must preserve that working path.

Pinned package versions make the reliability layer reproducible. A separate configuration template documents the new settings without exposing live credentials.

In this step, get ready to:

• Protect the working agent with a normal file-system backup.
• Install the pinned reliability packages in the existing virtual environment.
• Add the shared configuration names while keeping live credentials in .env.

Back up the working agent and add the project files

An additive upgrade starts with a normal file-system backup. The two new files sit beside the existing agent entry point.

• Press the Windows key to open the search bar.
• Type Visual Studio Code in the search bar.
• Press Enter to start Visual Studio Code.
• Open the existing text-to-SQL repository from the recent projects list.
• Select the existing agent entry point in the Explorer sidebar.
• Press Ctrl+C to copy the entry point.
• Press Ctrl+V to create a file-system copy beside it.
• Rename the copied file so its filename clearly identifies it as a backup.

You now have a known-good copy beside the working entry point. That safety copy gives you a clean way back if a later edit goes wrong.

• Create requirements-reliability.txt beside the existing entry point using the New File control in the Explorer sidebar.
• Paste the pinned dependency list into requirements-reliability.txt:

groq==1.7.0
psycopg[binary]==3.3.4
python-dotenv==1.2.3
sqlglot==30.17.0

What do these pins control?

• The groq pin keeps the model client on version 1.7.0.
• The psycopg[binary] pin provides the PostgreSQL driver at version 3.3.4.
• The python-dotenv pin loads the existing environment file with version 1.2.3.
• The sqlglot pin provides SQL parsing at version 30.17.0.

• Save requirements-reliability.txt.
• Confirm the Explorer sidebar lists requirements-reliability.txt beside the existing entry point.

Can't find the requirements file?

Check that requirements-reliability.txt appears in the same Explorer folder as the existing agent entry point. Move the file into that folder if it was created elsewhere.

Ask for help locating the new dependency file.

✔️ Awesome, I've got everything!

Great. Your pinned dependency list is saved beside the working agent.

ⓧ I'd like to double check the full code

groq==1.7.0
psycopg[binary]==3.3.4
python-dotenv==1.2.3
sqlglot==30.17.0

The configuration template records every setting the reliability gate expects. Its credential fields contain placeholders that are safe to share.

• Create .env.example beside the existing entry point using the New File control in the Explorer sidebar.
• Paste the configuration template into .env.example:

GROQ_API_KEY=your-api-key-here
GROQ_MODEL=openai/gpt-oss-20b
DATABASE_URL=your-existing-postgresql-connection-string
DB_SCHEMA=public
SCHEMA_CACHE_TTL_SECONDS=3600
PLAN_ROWS_WARNING=100000
PLAN_COST_WARNING=1000000

What does this template document?

• The GROQ_API_KEY field shows where the existing Groq credential belongs.
• The GROQ_MODEL field selects openai/gpt-oss-20b.
• The DATABASE_URL field represents the existing PostgreSQL connection string.
• The remaining settings define the schema name plus the cache and plan thresholds.

• Save .env.example.
• Confirm the Explorer sidebar lists .env.example beside requirements-reliability.txt.

Does the template contain a live credential?

Replace any live credential in .env.example with the placeholder from the reference below. Keep the working values only in the existing .env file.

Ask for help separating the template from the live environment file.

✔️ Awesome, I've got everything!

Your shareable configuration template is saved with placeholder credentials.

ⓧ I'd like to double check the full code

GROQ_API_KEY=your-api-key-here
GROQ_MODEL=openai/gpt-oss-20b
DATABASE_URL=your-existing-postgresql-connection-string
DB_SCHEMA=public
SCHEMA_CACHE_TTL_SECONDS=3600
PLAN_ROWS_WARNING=100000
PLAN_COST_WARNING=1000000

Install the pinned reliability packages

The existing virtual environment isolates these pins from other Python projects. Installing from requirements-reliability.txt makes that file the repeatable source for the upgrade.

• Open the integrated terminal in Visual Studio Code.
• Choose Windows PowerShell for the terminal session.

You should see PowerShell open at the path of the existing text-to-SQL repository.

• Activate the repository's existing virtual environment with its usual PowerShell activation command.

Your PowerShell prompt should show the active virtual environment before you install anything.

• Install the pinned reliability packages by running this command:

python -m pip install --upgrade -r requirements-reliability.txt

What does this command do?

Python runs pip inside the active virtual environment. The requirements flag tells it to install every exact pin from requirements-reliability.txt.

The upgrade option aligns existing packages with the versions recorded in the file. This gives every later step the same dependency set.

You should see installation output for the four packages. PowerShell returns to the prompt when the environment is ready.

Did the package installation stop?

Confirm that the virtual environment is active. Check that PowerShell is in the folder containing requirements-reliability.txt.

Ask for help with the package installation.

Merge the reliability settings and verify the setup

The existing .env file is the only file that holds the live Groq key and database connection string. The template stays safe because it contains placeholder credentials.

Credential edits can feel risky. You are only adding non-secret settings, so the working values stay untouched.

• Select the existing .env file in the Explorer sidebar.
• Keep the existing GROQ_API_KEY value unchanged.
• Keep the existing DATABASE_URL value unchanged.
• Add the non-secret reliability settings by pasting these lines into .env:

GROQ_MODEL=openai/gpt-oss-20b
DB_SCHEMA=public
SCHEMA_CACHE_TTL_SECONDS=3600
PLAN_ROWS_WARNING=100000
PLAN_COST_WARNING=1000000

What do these settings control?

• The GROQ_MODEL value selects the model used by the reliability gate.
• The DB_SCHEMA value limits schema discovery to public.
• The SCHEMA_CACHE_TTL_SECONDS value gives cached metadata a one-hour lifetime.
• The PLAN_ROWS_WARNING value defines the estimated-row warning threshold.
• The PLAN_COST_WARNING value defines the estimated-cost warning threshold.

• Save .env.
• Leave .env uncommitted.
• Review .env.example to confirm its credential fields still contain placeholders.

Before you check the versions, which four package numbers should confirm that the active environment matches the pinned file?

• Verify the installed dependency versions by running this command:

python -c "from importlib.metadata import version; print({p: version(p) for p in ['groq','psycopg','python-dotenv','sqlglot']})"

What does this check prove?

The command reads the installed metadata for each required package. It prints all four versions in one mapping.

You should see groq mapped to 1.7.0. You should see psycopg mapped to 3.3.4.

You should also see python-dotenv mapped to 1.2.3. You should see sqlglot mapped to 30.17.0.

Strong start. PowerShell now confirms the exact dependency set that supports every reliability check ahead.

Do the versions differ?

Confirm that PowerShell still shows the repository's virtual environment as active. Run the installation command again from the folder containing requirements-reliability.txt.

Ask for help aligning the installed package versions.

Your working agent is backed up. Next, you will build a visible cache for its live PostgreSQL schema.

Cache the Live PostgreSQL Schema

Your pinned Python environment is ready for the reliability upgrade. The agent still wastes database calls whenever it rediscovers metadata that has not changed.

A schema cache stores the visible PostgreSQL structure in a local JSON file. A TTL controls when that snapshot expires.

In this step, get ready to:

• Discover the configured PostgreSQL schema in a read-only transaction.
• Persist a fingerprinted schema snapshot.
• Reuse fresh snapshots through TTL checks or explicit refresh.

Prepare schema discovery

Schema discovery needs the configured database connection plus a metadata query scoped to DB_SCHEMA. The query orders each table's columns so repeated discoveries produce stable input for the fingerprint.

• In Visual Studio Code from earlier, create reliability_gate.py beside requirements-reliability.txt.
• Paste the environment loading and schema query into reliability_gate.py using this code:

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from pathlib import Path
from typing import Any

import psycopg
from dotenv import load_dotenv

load_dotenv()

CACHE_PATH = Path(".schema_cache.json")

SCHEMA_SQL = """
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = %s
ORDER BY table_name, ordinal_position
"""

What does this setup do?

• The environment loader makes your existing .env values available to Python.
• CACHE_PATH points to the local schema snapshot.
• SCHEMA_SQL requests visible table names from the configured schema.
• SCHEMA_SQL also requests column names plus PostgreSQL data types.

• Save reliability_gate.py.

You will see reliability_gate.py in the repository file list. The cache file appears after schema discovery runs.

• Add the environment and connection helpers below SCHEMA_SQL using this code:

def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def connect_db() -> psycopg.Connection[Any]:
    return psycopg.connect(require_env("DATABASE_URL"))

How do these helpers protect the connection?

require_env() stops the program when a required setting is absent. This prevents the cache from attempting a connection with an empty database URL.

connect_db() keeps connection creation in one place. Later cache operations can reuse that path.

• Save reliability_gate.py.
• Return to the PowerShell terminal from earlier.
• Check that the module loads by running this command:

python reliability_gate.py

What does this check prove?

Python imports the installed dependency. It also loads .env without finding a syntax problem in the new module.

PowerShell returns to the prompt without a traceback. The database remains untouched because the connection helper has only been defined.

Module not loading?

Confirm that the active PowerShell location is the repository containing reliability_gate.py.

Check the indentation around require_env() if Python reports a syntax problem.

Help me diagnose the module startup problem.

Build the fingerprinted cache

The cache must distinguish a reusable snapshot from stale metadata. Its SHA-256 fingerprint gives the sorted table map a compact content identity.

• Add the SchemaCache class below connect_db() using this code:

class SchemaCache:
    def __init__(self, path: Path = CACHE_PATH) -> None:
        self.path = path
        self.schema_name = os.getenv("DB_SCHEMA", "public")
        self.ttl_seconds = int(os.getenv("SCHEMA_CACHE_TTL_SECONDS", "3600"))

What does the cache remember?

• path identifies the snapshot file.
• schema_name limits discovery to the configured PostgreSQL schema.
• ttl_seconds uses a one-hour default when the environment does not provide a value.

• Save reliability_gate.py.
• Check the class definition by running this command:

python reliability_gate.py

What does this class check prove?

Python loads the class definition before exiting. A clean return confirms that its constructor is indented correctly.

PowerShell returns to the prompt without a traceback. The class is ready for its cache lifecycle method.

Seeing a class syntax problem?

Keep each self assignment indented inside __init__().

Help me fix the SchemaCache constructor.

• Add the get() method below __init__() using this code:

    def get(self, force_refresh: bool = False) -> dict[str, Any]:
        if self.path.exists() and not force_refresh:
            try:
                payload = json.loads(self.path.read_text(encoding="utf-8"))
                age = time.time() - float(payload["cached_at"])
                same_schema = payload.get("schema_name") == self.schema_name
                if same_schema and age < self.ttl_seconds:
                    print(
                        "Schema cache: HIT "
                        f"fingerprint={payload['fingerprint']} age={int(age)}s"
                    )
                    return payload
            except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                pass

        label = "REFRESH" if force_refresh else "MISS"
        print(f"Schema cache: {label}")
        tables = self._discover_tables()
        fingerprint = hashlib.sha256(
            json.dumps(tables, sort_keys=True).encode("utf-8")
        ).hexdigest()[:12]
        payload = {
            "cached_at": time.time(),
            "schema_name": self.schema_name,
            "fingerprint": fingerprint,
            "tables": tables,
        }
        self.path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
        print(f"Schema cache: SAVED fingerprint={fingerprint}")
        return payload

How does the cache lifecycle work?

• A matching snapshot within the TTL returns immediately as a HIT.
• A forced request uses the REFRESH path.
• An absent or expired snapshot uses the MISS path.
• The fingerprint keeps the first 12 hexadecimal characters from the hash of the sorted table map.

• Save reliability_gate.py.
• Check the cache lifecycle syntax by running this command:

python reliability_gate.py

What does this lifecycle check prove?

Python parses the complete get() method before returning to PowerShell. This catches malformed indentation before schema discovery begins.

PowerShell returns to the prompt without a traceback. The remaining method supplies the table map that get() fingerprints.

Cache method not parsing?

Keep get() aligned with __init__() inside the class.

Check that every opening parenthesis has a matching closing parenthesis.

Help me find the syntax problem in get().

• Add _discover_tables() below get() using this code:

    def _discover_tables(self) -> dict[str, dict[str, str]]:
        tables: dict[str, dict[str, str]] = {}
        with connect_db() as conn:
            with conn.cursor() as cur:
                cur.execute("SET TRANSACTION READ ONLY")
                cur.execute(SCHEMA_SQL, (self.schema_name,))
                for table_name, column_name, data_type in cur.fetchall():
                    tables.setdefault(table_name, {})[column_name] = data_type

        if not tables:
            raise RuntimeError(
                f"No visible columns found in PostgreSQL schema {self.schema_name!r}."
            )
        return tables

Why is discovery read-only?

SET TRANSACTION READ ONLY prevents this transaction from changing non-temporary database objects. Schema discovery only reads the metadata visible to your current PostgreSQL user.

The nested table map preserves each column's data type. That structure becomes both the cache payload and the fingerprint input.

• Save reliability_gate.py.
• Check the discovery method syntax by running this command:

python reliability_gate.py

What does this discovery check prove?

Python now parses every schema-cache method before returning to the prompt. The command-line entry point activates those methods next.

PowerShell returns without a traceback. The module is ready to discover the live schema.

Discovery method not loading?

Confirm that _discover_tables() is aligned with the other class methods.

Help me fix the schema discovery method.

Add refresh and verify reuse

The command-line entry point makes cache invalidation visible. A forced refresh is useful after database DDL changes the available tables or columns.

• Add the command-line parser below the SchemaCache class using this code:

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Schema-aware SQL reliability gate")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--question", help="Generate SQL from a business question")
    source.add_argument("--sql", help="Inspect an existing SQL statement")
    parser.add_argument(
        "--refresh-schema",
        action="store_true",
        help="Ignore the TTL and rebuild the schema cache",
    )
    return parser

What does the parser add?

The --refresh-schema switch bypasses the TTL. The other source options reserve mutually exclusive inputs for the reliability stages that follow.

• Add the CLI entry point below build_parser() using this code:

def main() -> int:
    args = build_parser().parse_args()
    schema_payload = SchemaCache().get(force_refresh=args.refresh_schema)

    if not args.question and not args.sql:
        print(
            f"Cached {len(schema_payload['tables'])} tables from "
            f"schema {schema_payload['schema_name']}."
        )
        return 0


if __name__ == "__main__":
    raise SystemExit(main())

How does the entry point expose the cache?

main() passes the refresh choice into SchemaCache.get(). A cache-only run also reports how many tables were discovered.

• Save reliability_gate.py.

Before you run the refresh, which cache path do you expect the explicit flag to choose?

• Force live schema discovery by running this command:

python reliability_gate.py --refresh-schema

What does the refresh command do?

The explicit flag ignores any existing cache age. The CLI queries the configured PostgreSQL schema before writing a new snapshot.

You will see Schema cache: REFRESH followed by Schema cache: SAVED. The saved line includes a 12-character fingerprint.

Schema refresh not completing?

Confirm that the active .env contains a working DATABASE_URL value.

Confirm that DB_SCHEMA names the schema containing your TPC-H tables.

Help me diagnose the schema refresh safely.

• Return to Visual Studio Code from earlier.
• Select .schema_cache.json from the repository file list.

You will see cached_at plus schema_name in the saved payload. You will also see fingerprint plus the nested tables map.

Before the second run, do you expect another database discovery or immediate cache reuse?

• Return to the PowerShell terminal from earlier.
• Test cache reuse by running this command:

python reliability_gate.py

What does the second run test?

The second run leaves the refresh flag unset. SchemaCache.get() checks the saved schema name plus the cache age before deciding whether discovery is necessary.

You will see Schema cache: HIT with the same fingerprint as the refresh. That is the first reliability win: unchanged schema metadata now avoids another discovery query.

Seeing another cache refresh?

Check that .schema_cache.json remains beside reliability_gate.py.

Check that the cached schema_name matches the current DB_SCHEMA value.

Help me work out why the cache is not returning HIT.

✔️ Awesome, I've got everything!

Your schema cache refreshes on demand. It now reuses the same fingerprint while the snapshot remains fresh.

ⓧ I'd like to double check the full code

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from pathlib import Path
from typing import Any

import psycopg
from dotenv import load_dotenv

load_dotenv()

CACHE_PATH = Path(".schema_cache.json")

SCHEMA_SQL = """
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = %s
ORDER BY table_name, ordinal_position
"""


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def connect_db() -> psycopg.Connection[Any]:
    return psycopg.connect(require_env("DATABASE_URL"))


class SchemaCache:
    def __init__(self, path: Path = CACHE_PATH) -> None:
        self.path = path
        self.schema_name = os.getenv("DB_SCHEMA", "public")
        self.ttl_seconds = int(os.getenv("SCHEMA_CACHE_TTL_SECONDS", "3600"))

    def get(self, force_refresh: bool = False) -> dict[str, Any]:
        if self.path.exists() and not force_refresh:
            try:
                payload = json.loads(self.path.read_text(encoding="utf-8"))
                age = time.time() - float(payload["cached_at"])
                same_schema = payload.get("schema_name") == self.schema_name
                if same_schema and age < self.ttl_seconds:
                    print(
                        "Schema cache: HIT "
                        f"fingerprint={payload['fingerprint']} age={int(age)}s"
                    )
                    return payload
            except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                pass

        label = "REFRESH" if force_refresh else "MISS"
        print(f"Schema cache: {label}")
        tables = self._discover_tables()
        fingerprint = hashlib.sha256(
            json.dumps(tables, sort_keys=True).encode("utf-8")
        ).hexdigest()[:12]
        payload = {
            "cached_at": time.time(),
            "schema_name": self.schema_name,
            "fingerprint": fingerprint,
            "tables": tables,
        }
        self.path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
        print(f"Schema cache: SAVED fingerprint={fingerprint}")
        return payload

    def _discover_tables(self) -> dict[str, dict[str, str]]:
        tables: dict[str, dict[str, str]] = {}
        with connect_db() as conn:
            with conn.cursor() as cur:
                cur.execute("SET TRANSACTION READ ONLY")
                cur.execute(SCHEMA_SQL, (self.schema_name,))
                for table_name, column_name, data_type in cur.fetchall():
                    tables.setdefault(table_name, {})[column_name] = data_type

        if not tables:
            raise RuntimeError(
                f"No visible columns found in PostgreSQL schema {self.schema_name!r}."
            )
        return tables


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Schema-aware SQL reliability gate")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--question", help="Generate SQL from a business question")
    source.add_argument("--sql", help="Inspect an existing SQL statement")
    parser.add_argument(
        "--refresh-schema",
        action="store_true",
        help="Ignore the TTL and rebuild the schema cache",
    )
    return parser


def main() -> int:
    args = build_parser().parse_args()
    schema_payload = SchemaCache().get(force_refresh=args.refresh_schema)

    if not args.question and not args.sql:
        print(
            f"Cached {len(schema_payload['tables'])} tables from "
            f"schema {schema_payload['schema_name']}."
        )
        return 0


if __name__ == "__main__":
    raise SystemExit(main())

Your agent now carries a reusable picture of the live database schema. Next up, you will pair that structure with business definitions that ground revenue questions in precise TPC-H formulas.

Ground Revenue with a Business Glossary

Your schema cache now gives the agent a stable map of your live database. That foundation keeps repeated metadata reads away from PostgreSQL.

Business questions introduce a different risk. A term such as revenue can map to several formulas.

A glossary pins each business term to one meaning. However, a hallucinated column can still travel to PostgreSQL because the current path sends supplied SQL straight to the database.

In this step, get ready to:

• Define revenue and active customer using precise TPC-H formulas.
• Ground Groq prompts with the cached schema and business glossary.
• Test how the current execution path handles a hallucinated column.

Create the business glossary

A business glossary connects everyday language to precise database logic. This JSON file gives revenue and active customer one consistent TPC-H meaning.

• Select the repository folder in the Visual Studio Code Explorer sidebar.
• Click the New File button in the Explorer toolbar.
• Enter glossary.json as the file name.
• Define the two business terms by pasting this content into glossary.json:

{
  "revenue": {
    "definition": "Discounted sales value before tax.",
    "sql": "SUM(l_extendedprice * (1 - l_discount))",
    "grain": "One contribution per lineitem row."
  },
  "active customer": {
    "definition": "A customer with at least one order in the requested date range.",
    "sql": "COUNT(DISTINCT o_custkey)",
    "grain": "Distinct orders.o_custkey after applying the requested orders.o_orderdate filter."
  }
}

How Does the Glossary Work?

• The definition field explains what the business term means.
• The sql field records the exact database expression that implements the definition.
• The grain field states what each contribution or count represents.

• Save glossary.json.
• Confirm glossary.json appears beside reliability_gate.py in the Explorer sidebar.

You should see both glossary entries in the saved file. Each entry contains a definition, SQL expression, and grain.

Connect the glossary to Groq

The Groq request needs two grounding sources. The cached tables constrain the available schema while the glossary constrains business meaning.

• Return to reliability_gate.py in the editor.
• Find the existing from dotenv import load_dotenv import.
• Add the Groq client import directly below it by using this line:

from groq import Groq

What Does This Import Add?

The Groq class creates the API client used by generate_sql(). The client reads the existing API key through require_env().

• Find CACHE_PATH = Path(".schema_cache.json") near the top of reliability_gate.py.
• Add the glossary path directly below it by using this line:

GLOSSARY_PATH = Path("glossary.json")

Why Store the Glossary Path?

The GLOSSARY_PATH constant gives every glossary read one shared location. This keeps file access consistent with CACHE_PATH.

• Save reliability_gate.py.
• Confirm the script still loads by running this command in the existing PowerShell terminal:

python reliability_gate.py

What Does This Check Prove?

This command loads the updated imports and constants before reading the existing cache. A successful cache report proves the additions did not stop the script from starting.

You should see Schema cache: HIT with the existing fingerprint. You should also see the cached table count.

• Add the glossary loading and model-output cleanup helpers below the SchemaCache class by pasting this code:

def load_glossary(path: Path = GLOSSARY_PATH) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def clean_model_sql(text: str) -> str:
    lines = text.strip().splitlines()
    if lines and lines[0].strip().startswith("```"):
        lines = lines[1:]
    if lines and lines[-1].strip() == "```":
        lines = lines[:-1]
    return "\n".join(lines).strip()

What Do These Helpers Do?

• The load_glossary() helper reads glossary.json into a Python dictionary.
• The clean_model_sql() helper removes Markdown code fences from a model response.
• The cleaned value leaves one SQL statement ready for the command-line path.

• Save reliability_gate.py.
• Check that the new helpers parse correctly by running this command:

python reliability_gate.py

Why Run the Cache Command Again?

Python parses the entire file before it runs the cache path. Seeing the normal cache report confirms both helper definitions have valid syntax.

You should see another Schema cache: HIT report with the same fingerprint.

• Start the generate_sql() function below clean_model_sql() by pasting this first section:

def generate_sql(
    question: str,
    schema_payload: dict[str, Any],
    glossary: dict[str, Any],
) -> str:
    client = Groq(api_key=require_env("GROQ_API_KEY"))
    model = os.getenv("GROQ_MODEL", "openai/gpt-oss-20b")
    schema_json = json.dumps(schema_payload["tables"], indent=2)
    glossary_json = json.dumps(glossary, indent=2)

    system_prompt = f"""
You translate business questions into PostgreSQL SQL for one known database.
Return exactly one SELECT statement and no markdown or explanation.
Do not use CTEs, SELECT INTO, write statements, or multiple statements.
Use only the tables and columns in SCHEMA.
Apply matching BUSINESS GLOSSARY definitions exactly.
Use explicit JOIN conditions and qualify ambiguous columns.

SCHEMA:
{schema_json}

BUSINESS GLOSSARY:
{glossary_json}
""".strip()

How Is the Prompt Grounded?

• The client uses the API key already stored outside your source code.
• The model value defaults to openai/gpt-oss-20b when GROQ_MODEL is absent.
• The prompt embeds the cached table map and glossary as separate JSON sections.
• The contract requests one PostgreSQL SELECT without CTEs or write operations.

• Complete generate_sql() by adding this section directly below """.strip() inside the function:

    completion = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
    )
    content = completion.choices[0].message.content or ""
    sql_text = clean_model_sql(content)
    if not sql_text:
        raise RuntimeError("Groq returned an empty SQL response.")
    return sql_text

How Does Generation Finish?

• The request sends the grounded instructions as the system message.
• The business question becomes the user message.
• The response passes through clean_model_sql() before the function returns it.
• An empty response raises a clear error instead of continuing with blank SQL.

• Save reliability_gate.py.
• Verify the completed function has valid syntax by running this command:

python reliability_gate.py

What Does This Run Confirm?

The command parses the completed generation function while reusing the schema cache. The normal cache output confirms the new function is ready for the CLI.

You should see Schema cache: HIT with the same fingerprint again.

Does the Script Stop Before the Cache Report?

Check that the completion block remains indented inside generate_sql(). Check that the closing triple quotes appear immediately before .strip().

Ask for help with the function structure: Help me compare my generate_sql function with the expected indentation.

Generate grounded SQL and expose the gap

The command-line interface now needs two input paths. One path asks the model a business question while the other accepts supplied SQL for direct testing.

• Find the existing build_parser() definition in reliability_gate.py.
• Replace that definition with this version:

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Schema-aware SQL reliability gate")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--question", help="Generate SQL from a business question")
    source.add_argument("--sql", help="Inspect an existing SQL statement")
    parser.add_argument(
        "--refresh-schema",
        action="store_true",
        help="Ignore the TTL and rebuild the schema cache",
    )
    return parser

What Changes in the Parser?

The mutually exclusive group prevents --question and --sql from being supplied together. The existing --refresh-schema option remains available.

• Save reliability_gate.py.
• Confirm the updated parser still supports the cache-only path by running this command:

python reliability_gate.py

What Does the Parser Check Show?

The no-input path still reports the cached schema. This preserves the visible cache behavior while adding two SQL source options.

You should see the existing Schema cache: HIT output and cached table count.

• Find the existing main() definition near the bottom of reliability_gate.py.
• Replace it with this first section:

def main() -> int:
    args = build_parser().parse_args()
    schema_payload = SchemaCache().get(force_refresh=args.refresh_schema)

    if not args.question and not args.sql:
        print(
            f"Cached {len(schema_payload['tables'])} tables from "
            f"schema {schema_payload['schema_name']}."
        )
        return 0

    if args.question:
        candidate_sql = generate_sql(
            args.question,
            schema_payload,
            load_glossary(),
        )
        print("Generated SQL:")
        print(candidate_sql)
    else:
        candidate_sql = args.sql

How Does the CLI Choose SQL?

• A cache-only run keeps the existing table-count output.
• A question loads the glossary before calling generate_sql().
• A supplied statement becomes candidate_sql directly.

• Complete main() by adding this section directly below candidate_sql = args.sql inside the function:

    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(candidate_sql)
            columns = [description[0] for description in (cur.description or [])]
            rows = cur.fetchmany(51)

    print(" | ".join(columns))
    for row in rows[:50]:
        print(" | ".join(str(value) for value in row))
    if len(rows) > 50:
        print("More than 50 rows exist. Showing the first 50 explicitly.")
    return 0

What Does the Current Execution Path Do?

• The database transaction is placed in read-only mode before the candidate runs.
• The cursor requests one extra row beyond the display limit.
• The output states when more than 50 rows exist.
• The candidate currently reaches PostgreSQL without a local schema check.

• Save reliability_gate.py.
• Compare your cumulative file with the completed version for this step.

✔️ Awesome, I've got everything!

Your glossary-backed generation path is ready. Keep reliability_gate.py saved before running the checks below.

ⓧ I'd like to double check the full code

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from pathlib import Path
from typing import Any

import psycopg
from dotenv import load_dotenv
from groq import Groq

load_dotenv()

CACHE_PATH = Path(".schema_cache.json")
GLOSSARY_PATH = Path("glossary.json")

SCHEMA_SQL = """
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = %s
ORDER BY table_name, ordinal_position
"""


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def connect_db() -> psycopg.Connection[Any]:
    return psycopg.connect(require_env("DATABASE_URL"))


class SchemaCache:
    def __init__(self, path: Path = CACHE_PATH) -> None:
        self.path = path
        self.schema_name = os.getenv("DB_SCHEMA", "public")
        self.ttl_seconds = int(os.getenv("SCHEMA_CACHE_TTL_SECONDS", "3600"))

    def get(self, force_refresh: bool = False) -> dict[str, Any]:
        if self.path.exists() and not force_refresh:
            try:
                payload = json.loads(self.path.read_text(encoding="utf-8"))
                age = time.time() - float(payload["cached_at"])
                same_schema = payload.get("schema_name") == self.schema_name
                if same_schema and age < self.ttl_seconds:
                    print(
                        "Schema cache: HIT "
                        f"fingerprint={payload['fingerprint']} age={int(age)}s"
                    )
                    return payload
            except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                pass

        label = "REFRESH" if force_refresh else "MISS"
        print(f"Schema cache: {label}")
        tables = self._discover_tables()
        fingerprint = hashlib.sha256(
            json.dumps(tables, sort_keys=True).encode("utf-8")
        ).hexdigest()[:12]
        payload = {
            "cached_at": time.time(),
            "schema_name": self.schema_name,
            "fingerprint": fingerprint,
            "tables": tables,
        }
        self.path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
        print(f"Schema cache: SAVED fingerprint={fingerprint}")
        return payload

    def _discover_tables(self) -> dict[str, dict[str, str]]:
        tables: dict[str, dict[str, str]] = {}
        with connect_db() as conn:
            with conn.cursor() as cur:
                cur.execute("SET TRANSACTION READ ONLY")
                cur.execute(SCHEMA_SQL, (self.schema_name,))
                for table_name, column_name, data_type in cur.fetchall():
                    tables.setdefault(table_name, {})[column_name] = data_type

        if not tables:
            raise RuntimeError(
                f"No visible columns found in PostgreSQL schema {self.schema_name!r}."
            )
        return tables


def load_glossary(path: Path = GLOSSARY_PATH) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def clean_model_sql(text: str) -> str:
    lines = text.strip().splitlines()
    if lines and lines[0].strip().startswith("```"):
        lines = lines[1:]
    if lines and lines[-1].strip() == "```":
        lines = lines[:-1]
    return "\n".join(lines).strip()


def generate_sql(
    question: str,
    schema_payload: dict[str, Any],
    glossary: dict[str, Any],
) -> str:
    client = Groq(api_key=require_env("GROQ_API_KEY"))
    model = os.getenv("GROQ_MODEL", "openai/gpt-oss-20b")
    schema_json = json.dumps(schema_payload["tables"], indent=2)
    glossary_json = json.dumps(glossary, indent=2)

    system_prompt = f"""
You translate business questions into PostgreSQL SQL for one known database.
Return exactly one SELECT statement and no markdown or explanation.
Do not use CTEs, SELECT INTO, write statements, or multiple statements.
Use only the tables and columns in SCHEMA.
Apply matching BUSINESS GLOSSARY definitions exactly.
Use explicit JOIN conditions and qualify ambiguous columns.

SCHEMA:
{schema_json}

BUSINESS GLOSSARY:
{glossary_json}
""".strip()

    completion = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
    )
    content = completion.choices[0].message.content or ""
    sql_text = clean_model_sql(content)
    if not sql_text:
        raise RuntimeError("Groq returned an empty SQL response.")
    return sql_text


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Schema-aware SQL reliability gate")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--question", help="Generate SQL from a business question")
    source.add_argument("--sql", help="Inspect an existing SQL statement")
    parser.add_argument(
        "--refresh-schema",
        action="store_true",
        help="Ignore the TTL and rebuild the schema cache",
    )
    return parser


def main() -> int:
    args = build_parser().parse_args()
    schema_payload = SchemaCache().get(force_refresh=args.refresh_schema)

    if not args.question and not args.sql:
        print(
            f"Cached {len(schema_payload['tables'])} tables from "
            f"schema {schema_payload['schema_name']}."
        )
        return 0

    if args.question:
        candidate_sql = generate_sql(
            args.question,
            schema_payload,
            load_glossary(),
        )
        print("Generated SQL:")
        print(candidate_sql)
    else:
        candidate_sql = args.sql

    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(candidate_sql)
            columns = [description[0] for description in (cur.description or [])]
            rows = cur.fetchmany(51)

    print(" | ".join(columns))
    for row in rows[:50]:
        print(" | ".join(str(value) for value in row))
    if len(rows) > 50:
        print("More than 50 rows exist. Showing the first 50 explicitly.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

What Should Match?

Compare the imports, glossary helpers, generation function, parser, and current direct databaseottery path. Keep your existing schema-cache implementation identical to this reference.

• Generate SQL for the revenue question by running this command in PowerShell:

python reliability_gate.py --question "What is total revenue?"

What Does This Command Test?

The command combines the cached schema with glossary.json before asking the configured model for SQL. The generated statement then runs through the current read-only database path.

You should see Generated SQL: followed by a statement containing SUM(l_extendedprice * (1 - l_discount)).

That is the grounding layer working: revenue now maps to the discounted-sales formula stored in your glossary.

Is the Revenue Formula Missing?

Check that glossary.json is saved beside reliability_gate.py. Check that your configured model is openai/gpt-oss-20b.

Ask for help tracing the prompt inputs: Help me find why the revenue glossary formula is missing from my generated SQL.

Before you run the final check, do you think the glossary can stop a supplied statement from using a column that does not exist?

• Test the current path with a hallucinated column by running this command:

python reliability_gate.py --sql "SELECT l_revenue FROM lineitem"

What Does This Test Bypass?

The --sql path bypasses model generation and uses the supplied statement as candidate_sql. The glossary has no opportunity to rewrite that statement.

You will see PostgreSQL reject the statement and identify l_revenue as the problem. This failure is intentional because the candidate reached the database before its schema references were checked locally.

You have now seen both sides of the reliability problem. Next, you will turn the cached schema into a local gate that blocks this hallucinated column before PostgreSQL receives it.

Block Hallucinated SQL Locally

Your glossary now gives revenue a precise TPC-H meaning. The l_revenue test exposed the remaining gap because a hallucinated column still reached PostgreSQL.

This step adds a local gate with SQLGlot before PostgreSQL receives the candidate SQL. The gate checks the statement structure against your cached schema.

In this step, get ready to:

• Add SQL parsing types for local validation.
• Reject unsafe statements and unknown schema references.
• Block the hallucinated column before any PostgreSQL request.

Add the validation foundations

A local validator needs an abstract syntax tree to understand the candidate as SQL structure. It also needs a consistent result object so the CLI can handle every failure through one path.

• Return to reliability_gate.py in Visual Studio Code.
• Replace the existing import section with this complete import block:

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import psycopg
import sqlglot
from dotenv import load_dotenv
from groq import Groq
from sqlglot import exp
from sqlglot.errors import OptimizeError, ParseError, SchemaError, TokenError
from sqlglot.optimizer.qualify import qualify

What do these imports provide?

• The sqlglot module parses each candidate with the PostgreSQL dialect.
• The exp module exposes SQL expression types such as exp.Select and exp.Table.
• The imported error classes let the gate convert parsing and schema failures into local validation messages.
• The qualify function resolves columns against your cached table map.

• Save reliability_gate.py.
• Confirm the editor shows no unresolved import warning for sqlglot.

Are the SQLGlot imports unresolved?

Confirm your existing virtual environment is still active. Check that requirements-reliability.txt contains the pinned SQLGlot dependency from Step 1.

Use this if the import remains unresolved: Help me diagnose an unresolved SQLGlot import in my active Python virtual environment.

• Find the closing line of SCHEMA_SQL.
• Add the validation result structure directly below it by pasting this code:

@dataclass
class ValidationResult:
    valid: bool
    errors: list[str]
    normalized_sql: str | None = None

What does this result hold?

• The valid field records whether the candidate passed every local check.
• The errors field collects messages the CLI can show without exposing a database error.
• The normalized_sql field stores qualified PostgreSQL SQL only after validation succeeds.

• Save reliability_gate.py.
• Confirm the editor recognizes ValidationResult as a class.

Is ValidationResult flagged?

Check that @dataclass sits directly above the class declaration. Confirm the import block still includes from dataclasses import dataclass.

Use this if the class remains flagged: Help me fix the ValidationResult dataclass declaration.

Build the local validation pipeline

The validator applies checks from cheapest to most detailed. Parsing and root checks run first because malformed input should stop before schema resolution begins.

• Find the end of generate_sql() in reliability_gate.py.
• Start validate_sql() directly below that function by pasting this first section:

def validate_sql(sql_text: str, schema_payload: dict[str, Any]) -> ValidationResult:
    errors: list[str] = []
    try:
        statements = [
            statement
            for statement in sqlglot.parse(sql_text, read="postgres")
            if statement is not None
        ]
    except (ParseError, TokenError) as exc:
        return ValidationResult(False, [f"SQL parse error: {exc}"])

    if len(statements) != 1:
        return ValidationResult(False, ["Exactly one SQL statement is required."])

    expression = statements[0]
    if not isinstance(expression, exp.Select):
        return ValidationResult(False, ["Only a SELECT statement is allowed."])

What does the first validation layer do?

• The parser reads the candidate with the PostgreSQL dialect.
• An empty candidate produces zero statements and fails the single-statement requirement.
• Multiple statements also fail because the gate requires exactly one parsed expression.
• The exp.Select check blocks write statements before any database request.

• Save reliability_gate.py.
• Confirm every line beneath validate_sql() remains indented inside the function.

Seeing an indentation warning?

Keep the body of validate_sql() indented by four spaces. Align each return with the condition or exception block that owns it.

Use this if the function structure is unclear: Help me check the indentation in the first section of validate_sql().

A structurally valid SELECT can still name a table that your database does not contain. The next layer compares every table node with the table names in schema_payload["tables"].

• Place this table-checking section at the end of validate_sql() while keeping the four-space indentation:

    known_tables = {name.lower() for name in schema_payload["tables"]}
    referenced_tables = {
        table.name.lower() for table in expression.find_all(exp.Table)
    }
    if not referenced_tables:
        errors.append("The query does not reference a table.")

    unknown_tables = sorted(referenced_tables - known_tables)
    if unknown_tables:
        errors.append(f"Unknown tables: {', '.join(unknown_tables)}")

    if errors:
        return ValidationResult(False, errors)

How are table references checked?

• The cached table names become a lowercase set for consistent comparison.
• Every exp.Table node contributes its referenced table name.
• A query with no table reference receives a local error.
• Any name missing from the cache is reported before PostgreSQL sees the query.

• Save reliability_gate.py.
• Confirm known_tables and referenced_tables remain inside validate_sql().

Are table names failing unexpectedly?

Confirm the code reads from schema_payload["tables"]. Check that both cached names and referenced names call lower() before comparison.

Use this if a cached table is reported as unknown: Help me debug the table-name comparison in validate_sql().

Known tables can still contain invented columns. Schema-aware qualification resolves each column against the cached table-column map and returns normalized SQL only when resolution succeeds.

• Complete validate_sql() by adding this qualification section at the end of the function:

    try:
        qualified = qualify(
            expression,
            dialect="postgres",
            schema=schema_payload["tables"],
            validate_qualify_columns=True,
            quote_identifiers=False,
            identify=False,
            sql=sql_text,
        )
    except (OptimizeError, SchemaError, ParseError, TokenError) as exc:
        return ValidationResult(False, [f"Schema validation error: {exc}"])

    return ValidationResult(
        True,
        [],
        normalized_sql=qualified.sql(dialect="postgres"),
    )

Why add schema-aware qualification?

Parsing confirms that the candidate has recognizable SQL structure. Qualification checks whether its columns can resolve against the cached schema.

SQLGlot can still accept SQL that later fails during execution. This gate focuses on the structural and schema errors it can prove locally.

• Save reliability_gate.py.
• Confirm the final return ValidationResult( aligns with the try: block inside the function.

Is qualification failing for valid columns?

Confirm the schema argument receives schema_payload["tables"]. Check that the cached map contains table names with nested column-name-to-data-type maps.

Use this if valid TPC-H columns remain unresolved: Help me diagnose SQLGlot qualification against my cached schema.

Wire validation into the CLI

The validation function protects PostgreSQL only when every candidate passes through it. The CLI now needs to stop immediately when validation returns an error.

• Find the else: branch in main() that ends with candidate_sql = args.sql.
• Replace the database handoff beneath that branch with this validation flow:

    validation = validate_sql(candidate_sql, schema_payload)
    if not validation.valid or validation.normalized_sql is None:
        print("Validation blocked before EXPLAIN:")
        for error in validation.errors:
            print(f"- {error}")
        return 2

    print("Validation: PASSED")
    print("Normalized SQL:")
    print(validation.normalized_sql)

How does the CLI block a candidate?

• Every generated or supplied candidate now enters validate_sql().
• A failed result prints Validation blocked before EXPLAIN with the local error details.
• The exit code 2 ends the CLI before a PostgreSQL preflight can begin.
• A successful result prints the normalized PostgreSQL SQL for the next reliability layer.

• Save reliability_gate.py.

✔️ Awesome, I've got everything!

Your local validation path is complete. Double-check that reliability_gate.py is saved.

ⓧ I'd like to double check the full code

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import psycopg
import sqlglot
from dotenv import load_dotenv
from groq import Groq
from sqlglot import exp
from sqlglot.errors import OptimizeError, ParseError, SchemaError, TokenError
from sqlglot.optimizer.qualify import qualify

load_dotenv()

CACHE_PATH = Path(".schema_cache.json")
GLOSSARY_PATH = Path("glossary.json")

SCHEMA_SQL = """
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = %s
ORDER BY table_name, ordinal_position
"""


@dataclass
class ValidationResult:
    valid: bool
    errors: list[str]
    normalized_sql: str | None = None


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def connect_db() -> psycopg.Connection[Any]:
    return psycopg.connect(require_env("DATABASE_URL"))


class SchemaCache:
    def __init__(self, path: Path = CACHE_PATH) -> None:
        self.path = path
        self.schema_name = os.getenv("DB_SCHEMA", "public")
        self.ttl_seconds = int(os.getenv("SCHEMA_CACHE_TTL_SECONDS", "3600"))

    def get(self, force_refresh: bool = False) -> dict[str, Any]:
        if self.path.exists() and not force_refresh:
            try:
                payload = json.loads(self.path.read_text(encoding="utf-8"))
                age = time.time() - float(payload["cached_at"])
                same_schema = payload.get("schema_name") == self.schema_name
                if same_schema and age < self.ttl_seconds:
                    print(
                        "Schema cache: HIT "
                        f"fingerprint={payload['fingerprint']} age={int(age)}s"
                    )
                    return payload
            except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                pass

        label = "REFRESH" if force_refresh else "MISS"
        print(f"Schema cache: {label}")
        tables = self._discover_tables()
        fingerprint = hashlib.sha256(
            json.dumps(tables, sort_keys=True).encode("utf-8")
        ).hexdigest()[:12]
        payload = {
            "cached_at": time.time(),
            "schema_name": self.schema_name,
            "fingerprint": fingerprint,
            "tables": tables,
        }
        self.path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
        print(f"Schema cache: SAVED fingerprint={fingerprint}")
        return payload

    def _discover_tables(self) -> dict[str, dict[str, str]]:
        tables: dict[str, dict[str, str]] = {}
        with connect_db() as conn:
            with conn.cursor() as cur:
                cur.execute("SET TRANSACTION READ ONLY")
                cur.execute(SCHEMA_SQL, (self.schema_name,))
                for table_name, column_name, data_type in cur.fetchall():
                    tables.setdefault(table_name, {})[column_name] = data_type

        if not tables:
            raise RuntimeError(
                f"No visible columns found in PostgreSQL schema {self.schema_name!r}."
            )
        return tables


def load_glossary(path: Path = GLOSSARY_PATH) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def clean_model_sql(text: str) -> str:
    lines = text.strip().splitlines()
    if lines and lines[0].strip().startswith("```"):
        lines = lines[1:]
    if lines and lines[-1].strip() == "```":
        lines = lines[:-1]
    return "\n".join(lines).strip()


def generate_sql(
    question: str,
    schema_payload: dict[str, Any],
    glossary: dict[str, Any],
) -> str:
    client = Groq(api_key=require_env("GROQ_API_KEY"))
    model = os.getenv("GROQ_MODEL", "openai/gpt-oss-20b")
    schema_json = json.dumps(schema_payload["tables"], indent=2)
    glossary_json = json.dumps(glossary, indent=2)

    system_prompt = f"""
You translate business questions into PostgreSQL SQL for one known database.
Return exactly one SELECT statement and no markdown or explanation.
Do not use CTEs, SELECT INTO, write statements, or multiple statements.
Use only the tables and columns in SCHEMA.
Apply matching BUSINESS GLOSSARY definitions exactly.
Use explicit JOIN conditions and qualify ambiguous columns.

SCHEMA:
{schema_json}

BUSINESS GLOSSARY:
{glossary_json}
""".strip()

    completion = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
    )
    content = completion.choices[0].message.content or ""
    sql_text = clean_model_sql(content)
    if not sql_text:
        raise RuntimeError("Groq returned an empty SQL response.")
    return sql_text


def validate_sql(sql_text: str, schema_payload: dict[str, Any]) -> ValidationResult:
    errors: list[str] = []
    try:
        statements = [
            statement
            for statement in sqlglot.parse(sql_text, read="postgres")
            if statement is not None
        ]
    except (ParseError, TokenError) as exc:
        return ValidationResult(False, [f"SQL parse error: {exc}"])

    if len(statements) != 1:
        return ValidationResult(False, ["Exactly one SQL statement is required."])

    expression = statements[0]
    if not isinstance(expression, exp.Select):
        return ValidationResult(False, ["Only a SELECT statement is allowed."])

    known_tables = {name.lower() for name in schema_payload["tables"]}
    referenced_tables = {
        table.name.lower() for table in expression.find_all(exp.Table)
    }
    if not referenced_tables:
        errors.append("The query does not reference a table.")

    unknown_tables = sorted(referenced_tables - known_tables)
    if unknown_tables:
        errors.append(f"Unknown tables: {', '.join(unknown_tables)}")

    if errors:
        return ValidationResult(False, errors)

    try:
        qualified = qualify(
            expression,
            dialect="postgres",
            schema=schema_payload["tables"],
            validate_qualify_columns=True,
            quote_identifiers=False,
            identify=False,
            sql=sql_text,
        )
    except (OptimizeError, SchemaError, ParseError, TokenError) as exc:
        return ValidationResult(False, [f"Schema validation error: {exc}"])

    return ValidationResult(
        True,
        [],
        normalized_sql=qualified.sql(dialect="postgres"),
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Schema-aware SQL reliability gate")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--question", help="Generate SQL from a business question")
    source.add_argument("--sql", help="Inspect an existing SQL statement")
    parser.add_argument(
        "--refresh-schema",
        action="store_true",
        help="Ignore the TTL and rebuild the schema cache",
    )
    return parser


def main() -> int:
    args = build_parser().parse_args()
    schema_payload = SchemaCache().get(force_refresh=args.refresh_schema)

    if not args.question and not args.sql:
        print(
            f"Cached {len(schema_payload['tables'])} tables from "
            f"schema {schema_payload['schema_name']}."
        )
        return 0

    if args.question:
        candidate_sql = generate_sql(
            args.question,
            schema_payload,
            load_glossary(),
        )
        print("Generated SQL:")
        print(candidate_sql)
    else:
        candidate_sql = args.sql

    validation = validate_sql(candidate_sql, schema_payload)
    if not validation.valid or validation.normalized_sql is None:
        print("Validation blocked before EXPLAIN:")
        for error in validation.errors:
            print(f"- {error}")
        return 2

    print("Validation: PASSED")
    print("Normalized SQL:")
    print(validation.normalized_sql)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

How should you use this reference?

Compare the imports, ValidationResult, validate_sql(), and the validation branch in main(). Keep your existing schema cache and glossary generation code unchanged.

Before you run the final check, do you expect the hallucinated column to reach PostgreSQL again?

• Test the same hallucinated column from the PowerShell terminal by running:

python reliability_gate.py --sql "SELECT l_revenue FROM lineitem"

What should you see?

You should see Validation blocked before EXPLAIN. The local schema error names l_revenue without showing a PostgreSQL error.

That result proves the cached schema now stops the hallucination before a database preflight.

Did the SQL still reach PostgreSQL?

Confirm the call to validate_sql(candidate_sql, schema_payload) appears before any database handoff in main(). Check that the failure branch returns 2 immediately.

Use this if the candidate still reaches the database: Help me trace why invalid SQL bypasses validate_sql().

That database round-trip is gone. Next, you will inspect valid SQL with a plain EXPLAIN preflight before deciding whether it is safe to run.

Add an EXPLAIN Preflight Gate

Your local SQL validator now catches hallucinated tables or columns before they reach the database. The next risk comes from valid queries that demand more work than you expect.

A schema-valid query can still ask PostgreSQL to scan millions of rows. This step adds a plain EXPLAIN preflight before optional read-only execution.

In this step, get ready to:

• Inspect JSON query plans for estimated rows or cost.
• Warn about large outputs or sequential scans.
• Require explicit approval before read-only execution.

Inspect the JSON plan before execution

Plain EXPLAIN returns planner estimates without executing the candidate query. A JSON plan gives your gate stable fields to inspect.

• In reliability_gate.py, place this dataclass directly below the existing ValidationResult definition:

@dataclass
class PlanReport:
    root_node: str
    plan_rows: int
    total_cost: float
    risks: list[str]
    plan: dict[str, Any]

What Does This Report Hold?

• The root_node field identifies the top operation in the plan.
• The plan_rows field records the estimated output size.
• The total_cost field records PostgreSQL's planner estimate.
• The risks field keeps the warnings separate from the original plan.

• Save reliability_gate.py.
• Confirm the new dataclass contains all five report fields.

Seeing a Dataclass Error?

Keep the @dataclass decorator directly above class PlanReport. Match the indentation used by ValidationResult.

Ask for help checking the new report definition:

Nested plans place child operations inside Plans. A recursive walk lets one warning rule inspect every level.

• Add the recursive plan walker directly below validate_sql() by copying this code:

def walk_plan(node: dict[str, Any]) -> list[dict[str, Any]]:
    nodes = [node]
    for child in node.get("Plans", []):
        nodes.extend(walk_plan(child))
    return nodes

How Does the Plan Walk Work?

• Each call starts with the current plan node.
• The loop visits every child inside Plans.
• Recursion returns one flat list for the warning checks.

• Save reliability_gate.py.
• Confirm walk_plan() ends by returning the collected nodes.

Is the Recursion Misaligned?

Keep nodes.extend(...) inside the loop. Keep return nodes outside the loop.

Ask for help checking the recursive structure:

The preflight sends only normalized SQL to EXPLAIN (FORMAT JSON). It also starts the transaction in read-only mode before the database sees the plan request.

• Add the plan preflight directly below walk_plan() by copying this function:

def explain_sql(sql_text: str) -> PlanReport:
    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(f"EXPLAIN (FORMAT JSON) {sql_text}")
            row = cur.fetchone()

    if row is None:
        raise RuntimeError("PostgreSQL returned no EXPLAIN result.")

    raw_plan = row[0]
    if isinstance(raw_plan, str):
        raw_plan = json.loads(raw_plan)
    document = raw_plan[0] if isinstance(raw_plan, list) else raw_plan
    root = document["Plan"]

    row_warning = int(os.getenv("PLAN_ROWS_WARNING", "100000"))
    cost_warning = float(os.getenv("PLAN_COST_WARNING", "1000000"))
    root_rows = int(root.get("Plan Rows", 0))
    total_cost = float(root.get("Total Cost", 0.0))
    risks: list[str] = []

    if root_rows >= row_warning:
        risks.append(
            f"Estimated output is {root_rows:,} rows, above {row_warning:,}."
        )
    if total_cost >= cost_warning:
        risks.append(
            f"Estimated total cost is {total_cost:,.2f}, above {cost_warning:,.2f}."
        )

What Does This Preflight Do?

• The read-only transaction protects the database before the plan request.
• The JSON response exposes Plan Rows plus Total Cost as typed values.
• The configured thresholds create estimate-only warnings.
• Planner costs use PostgreSQL's internal cost units. Treat them as review signals.

• Complete explain_sql() by adding the sequential-scan checks below risks.

    for node in walk_plan(root):
        if node.get("Node Type") == "Seq Scan":
            estimated_rows = int(node.get("Plan Rows", 0))
            if estimated_rows >= row_warning:
                relation = node.get("Relation Name", "unknown relation")
                risks.append(
                    f"Large sequential scan estimated on {relation}: "
                    f"{estimated_rows:,} rows."
                )

    return PlanReport(
        root_node=str(root.get("Node Type", "Unknown")),
        plan_rows=root_rows,
        total_cost=total_cost,
        risks=risks,
        plan=document,
    )

How Are Scan Risks Found?

• The walker checks every node for a Seq Scan operation.
• Large scans receive a warning when their estimated rows reach the configured threshold.
• The warning names the affected relation when PostgreSQL includes it.
• A sequential-scan warning is a prompt for review. Statistics or query shape can make the scan reasonable.

• Save reliability_gate.py.
• Confirm explain_sql() returns one PlanReport after inspecting every plan node.

Seeing an Incomplete Function?

The second snippet continues the same explain_sql() function. Keep its first line indented by four spaces.

Ask for help joining the two function chunks:

A compact terminal report makes the estimates visible before execution. Each warning remains clearly labelled as a plan warning.

• Add the report printer directly below explain_sql() by copying this function:

def print_plan_report(report: PlanReport) -> None:
    print(f"Plan root: {report.root_node}")
    print(f"Estimated output rows: {report.plan_rows:,}")
    print(f"Estimated total cost: {report.total_cost:,.2f}")
    if report.risks:
        for risk in report.risks:
            print(f"Plan warning: {risk}")
    else:
        print("Plan warning: none at the configured thresholds")

What Will the Report Show?

• The first line identifies the root plan operation.
• The next two lines expose estimated output rows or total cost.
• The remaining lines show every risk found by the preflight.

• Save reliability_gate.py.
• Confirm the function contains labels for the root node or both estimates.

Missing a Plan Label?

Compare each output label with the snippet. Keep the risk loop inside the if report.risks branch.

Ask for help checking the report output:

Add read-only opt-in execution

The preflight remains the default path. Query execution requires --run plus an additional typed approval whenever the plan contains warnings.

Approved queries run after SET TRANSACTION READ ONLY. The display fetches one extra row so truncation becomes visible.

• Add the limited execution helper directly below print_plan_report() by copying this function:

def execute_readonly(
    sql_text: str,
    max_rows: int = 50,
) -> tuple[list[str], list[tuple[Any, ...]], bool]:
    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(sql_text)
            columns = [description[0] for description in (cur.description or [])]
            rows = cur.fetchmany(max_rows + 1)

    truncated = len(rows) > max_rows
    return columns, rows[:max_rows], truncated

Why Fetch 51 Rows?

• The helper displays at most 50 rows.
• Fetching one extra row reveals whether additional results exist.
• The returned Boolean lets the printer state that truncation occurred.

• Save reliability_gate.py.
• Confirm the returned row slice ends at max_rows.

Returning Too Many Rows?

Keep max_rows + 1 inside fetchmany(). Keep rows[:max_rows] in the returned tuple.

Ask for help checking the row boundary:

The later evaluation workflow needs complete result sets for equivalence checks. It uses a separate helper with the same read-only protection.

• Add the full-result helper directly below execute_readonly() by copying this function:

def execute_all_readonly(sql_text: str) -> list[tuple[Any, ...]]:
    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(sql_text)
            return cur.fetchall()

Why Use a Separate Helper?

Interactive output needs a visible 50-row boundary. Evaluation needs every row so it can compare complete result sets.

• Save reliability_gate.py.
• Confirm execute_all_readonly() starts its transaction in read-only mode.

Is Full Execution Missing Protection?

Keep SET TRANSACTION READ ONLY before the candidate query. The order protects every call through this helper.

Ask for help checking the transaction order:

• Add the result printer directly below execute_all_readonly() by copying this function:

def print_rows(
    columns: list[str],
    rows: list[tuple[Any, ...]],
    truncated: bool,
) -> None:
    print(" | ".join(columns))
    for row in rows:
        print(" | ".join(str(value) for value in row))
    if truncated:
        print("More than 50 rows exist. Showing the first 50 explicitly.")

How Is Truncation Made Visible?

The printer shows column names before the returned values. A final message states when more than 50 rows exist.

• Save reliability_gate.py.
• Confirm the truncation message appears inside the if truncated branch.

Is the Overflow Message Always Printing?

Keep the final print() indented beneath if truncated. This limits the message to queries with extra rows.

Ask for help checking the result printer:

The command-line parser now needs an explicit execution switch. Its default remains a plan-only preflight.

• In build_parser(), add this argument directly below the --refresh-schema argument:

    parser.add_argument(
        "--run",
        action="store_true",
        help="Execute validated SQL in a read-only transaction",
    )

What Does the Switch Control?

The store_true action leaves execution disabled unless the command includes --run.

• Save reliability_gate.py.
• Confirm build_parser() returns the parser after defining --run.

Is the New Flag Missing?

Place the new argument before return parser. Keep the argument inside build_parser().

Ask for help checking the parser change:

The final CLI path connects validation to preflight. It stops after the plan unless the learner approves execution.

• In main(), find the line print(validation.normalized_sql).
• Replace the current final return beneath that line with this code:

    report = explain_sql(validation.normalized_sql)
    print_plan_report(report)

    if not args.run:
        print("Execution: SKIPPED. Add --run to approve read-only execution.")
        return 0

    if report.risks:
        approval = input("Plan warnings exist. Type RUN to continue: ").strip()
        if approval != "RUN":
            print("Execution: CANCELLED")
            return 0

    columns, rows, truncated = execute_readonly(validation.normalized_sql)
    print_rows(columns, rows, truncated)
    return 0

How Does Approval Work?

• Every valid query receives a plan report first.
• Commands without --run stop before execution.
• Plans with warnings require the exact approval text RUN.
• Approved results use the limited read-only helper.

• Save reliability_gate.py.
• Confirm main() calls explain_sql() only after validation passes.

Seeing an Undefined Name?

Confirm every helper appears above build_parser(). Check the spellings of explain_sql or execute_readonly against their definitions.

Ask for help checking the CLI wiring:

Preflight a risky query

The configured row or cost thresholds are tutorial heuristics. They flag estimates that deserve review before execution.

Before you run this check, do you expect the query to execute or stop after showing its plan?

• Preflight the full lineitem table by running this command in PowerShell:

python reliability_gate.py --sql "SELECT * FROM lineitem"

You should see Validation: PASSED before the plan report. The report shows a root node plus estimated rows or total cost.

You should also see at least one Plan warning: line. The final line says Execution: SKIPPED. Add --run to approve read-only execution.

What Did This Prove?

The candidate passed local schema validation before PostgreSQL planned it. Plain EXPLAIN exposed estimated risk without running the query.

The warnings guide review through estimates. Actual performance can vary with data or statistics.

Missing the Plan Report?

Confirm the report = explain_sql(validation.normalized_sql) line appears directly after the normalized SQL output. Confirm your configured database contains the lineitem table.

Ask for help diagnosing the preflight path:

That is the safety boundary working. Your agent now validates locally before it requests a plan or seeks approval for read-only execution.

✔️ Awesome, I've got everything!

Great. Double-check that you saved reliability_gate.py before recording your result.

ⓧ I'd like to double check the full code

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import psycopg
import sqlglot
from dotenv import load_dotenv
from groq import Groq
from sqlglot import exp
from sqlglot.errors import OptimizeError, ParseError, SchemaError, TokenError
from sqlglot.optimizer.qualify import qualify

load_dotenv()

CACHE_PATH = Path(".schema_cache.json")
GLOSSARY_PATH = Path("glossary.json")

SCHEMA_SQL = """
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = %s
ORDER BY table_name, ordinal_position
"""


@dataclass
class ValidationResult:
    valid: bool
    errors: list[str]
    normalized_sql: str | None = None


@dataclass
class PlanReport:
    root_node: str
    plan_rows: int
    total_cost: float
    risks: list[str]
    plan: dict[str, Any]


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def connect_db() -> psycopg.Connection[Any]:
    return psycopg.connect(require_env("DATABASE_URL"))


class SchemaCache:
    def __init__(self, path: Path = CACHE_PATH) -> None:
        self.path = path
        self.schema_name = os.getenv("DB_SCHEMA", "public")
        self.ttl_seconds = int(os.getenv("SCHEMA_CACHE_TTL_SECONDS", "3600"))

    def get(self, force_refresh: bool = False) -> dict[str, Any]:
        if self.path.exists() and not force_refresh:
            try:
                payload = json.loads(self.path.read_text(encoding="utf-8"))
                age = time.time() - float(payload["cached_at"])
                same_schema = payload.get("schema_name") == self.schema_name
                if same_schema and age < self.ttl_seconds:
                    print(
                        "Schema cache: HIT "
                        f"fingerprint={payload['fingerprint']} age={int(age)}s"
                    )
                    return payload
            except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                pass

        label = "REFRESH" if force_refresh else "MISS"
        print(f"Schema cache: {label}")
        tables = self._discover_tables()
        fingerprint = hashlib.sha256(
            json.dumps(tables, sort_keys=True).encode("utf-8")
        ).hexdigest()[:12]
        payload = {
            "cached_at": time.time(),
            "schema_name": self.schema_name,
            "fingerprint": fingerprint,
            "tables": tables,
        }
        self.path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
        print(f"Schema cache: SAVED fingerprint={fingerprint}")
        return payload

    def _discover_tables(self) -> dict[str, dict[str, str]]:
        tables: dict[str, dict[str, str]] = {}
        with connect_db() as conn:
            with conn.cursor() as cur:
                cur.execute("SET TRANSACTION READ ONLY")
                cur.execute(SCHEMA_SQL, (self.schema_name,))
                for table_name, column_name, data_type in cur.fetchall():
                    tables.setdefault(table_name, {})[column_name] = data_type

        if not tables:
            raise RuntimeError(
                f"No visible columns found in PostgreSQL schema {self.schema_name!r}."
            )
        return tables


def load_glossary(path: Path = GLOSSARY_PATH) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def clean_model_sql(text: str) -> str:
    lines = text.strip().splitlines()
    if lines and lines[0].strip().startswith("```"):
        lines = lines[1:]
    if lines and lines[-1].strip() == "```":
        lines = lines[:-1]
    return "\n".join(lines).strip()


def generate_sql(
    question: str,
    schema_payload: dict[str, Any],
    glossary: dict[str, Any],
) -> str:
    client = Groq(api_key=require_env("GROQ_API_KEY"))
    model = os.getenv("GROQ_MODEL", "openai/gpt-oss-20b")
    schema_json = json.dumps(schema_payload["tables"], indent=2)
    glossary_json = json.dumps(glossary, indent=2)

    system_prompt = f"""
You translate business questions into PostgreSQL SQL for one known database.
Return exactly one SELECT statement and no markdown or explanation.
Do not use CTEs, SELECT INTO, write statements, or multiple statements.
Use only the tables and columns in SCHEMA.
Apply matching BUSINESS GLOSSARY definitions exactly.
Use explicit JOIN conditions and qualify ambiguous columns.

SCHEMA:
{schema_json}

BUSINESS GLOSSARY:
{glossary_json}
""".strip()

    completion = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
    )
    content = completion.choices[0].message.content or ""
    sql_text = clean_model_sql(content)
    if not sql_text:
        raise RuntimeError("Groq returned an empty SQL response.")
    return sql_text


def validate_sql(sql_text: str, schema_payload: dict[str, Any]) -> ValidationResult:
    errors: list[str] = []
    try:
        statements = [
            statement
            for statement in sqlglot.parse(sql_text, read="postgres")
            if statement is not None
        ]
    except (ParseError, TokenError) as exc:
        return ValidationResult(False, [f"SQL parse error: {exc}"])

    if len(statements) != 1:
        return ValidationResult(False, ["Exactly one SQL statement is required."])

    expression = statements[0]
    if not isinstance(expression, exp.Select):
        return ValidationResult(False, ["Only a SELECT statement is allowed."])

    known_tables = {name.lower() for name in schema_payload["tables"]}
    referenced_tables = {
        table.name.lower() for table in expression.find_all(exp.Table)
    }
    if not referenced_tables:
        errors.append("The query does not reference a table.")

    unknown_tables = sorted(referenced_tables - known_tables)
    if unknown_tables:
        errors.append(f"Unknown tables: {', '.join(unknown_tables)}")

    if errors:
        return ValidationResult(False, errors)

    try:
        qualified = qualify(
            expression,
            dialect="postgres",
            schema=schema_payload["tables"],
            validate_qualify_columns=True,
            quote_identifiers=False,
            identify=False,
            sql=sql_text,
        )
    except (OptimizeError, SchemaError, ParseError, TokenError) as exc:
        return ValidationResult(False, [f"Schema validation error: {exc}"])

    return ValidationResult(
        True,
        [],
        normalized_sql=qualified.sql(dialect="postgres"),
    )


def walk_plan(node: dict[str, Any]) -> list[dict[str, Any]]:
    nodes = [node]
    for child in node.get("Plans", []):
        nodes.extend(walk_plan(child))
    return nodes


def explain_sql(sql_text: str) -> PlanReport:
    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(f"EXPLAIN (FORMAT JSON) {sql_text}")
            row = cur.fetchone()

    if row is None:
        raise RuntimeError("PostgreSQL returned no EXPLAIN result.")

    raw_plan = row[0]
    if isinstance(raw_plan, str):
        raw_plan = json.loads(raw_plan)
    document = raw_plan[0] if isinstance(raw_plan, list) else raw_plan
    root = document["Plan"]

    row_warning = int(os.getenv("PLAN_ROWS_WARNING", "100000"))
    cost_warning = float(os.getenv("PLAN_COST_WARNING", "1000000"))
    root_rows = int(root.get("Plan Rows", 0))
    total_cost = float(root.get("Total Cost", 0.0))
    risks: list[str] = []

    if root_rows >= row_warning:
        risks.append(
            f"Estimated output is {root_rows:,} rows, above {row_warning:,}."
        )
    if total_cost >= cost_warning:
        risks.append(
            f"Estimated total cost is {total_cost:,.2f}, above {cost_warning:,.2f}."
        )

    for node in walk_plan(root):
        if node.get("Node Type") == "Seq Scan":
            estimated_rows = int(node.get("Plan Rows", 0))
            if estimated_rows >= row_warning:
                relation = node.get("Relation Name", "unknown relation")
                risks.append(
                    f"Large sequential scan estimated on {relation}: "
                    f"{estimated_rows:,} rows."
                )

    return PlanReport(
        root_node=str(root.get("Node Type", "Unknown")),
        plan_rows=root_rows,
        total_cost=total_cost,
        risks=risks,
        plan=document,
    )


def print_plan_report(report: PlanReport) -> None:
    print(f"Plan root: {report.root_node}")
    print(f"Estimated output rows: {report.plan_rows:,}")
    print(f"Estimated total cost: {report.total_cost:,.2f}")
    if report.risks:
        for risk in report.risks:
            print(f"Plan warning: {risk}")
    else:
        print("Plan warning: none at the configured thresholds")


def execute_readonly(
    sql_text: str,
    max_rows: int = 50,
) -> tuple[list[str], list[tuple[Any, ...]], bool]:
    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(sql_text)
            columns = [description[0] for description in (cur.description or [])]
            rows = cur.fetchmany(max_rows + 1)

    truncated = len(rows) > max_rows
    return columns, rows[:max_rows], truncated


def execute_all_readonly(sql_text: str) -> list[tuple[Any, ...]]:
    with connect_db() as conn:
        with conn.cursor() as cur:
            cur.execute("SET TRANSACTION READ ONLY")
            cur.execute(sql_text)
            return cur.fetchall()


def print_rows(
    columns: list[str],
    rows: list[tuple[Any, ...]],
    truncated: bool,
) -> None:
    print(" | ".join(columns))
    for row in rows:
        print(" | ".join(str(value) for value in row))
    if truncated:
        print("More than 50 rows exist. Showing the first 50 explicitly.")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Schema-aware SQL reliability gate")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--question", help="Generate SQL from a business question")
    source.add_argument("--sql", help="Inspect an existing SQL statement")
    parser.add_argument(
        "--refresh-schema",
        action="store_true",
        help="Ignore the TTL and rebuild the schema cache",
    )
    parser.add_argument(
        "--run",
        action="store_true",
        help="Execute validated SQL in a read-only transaction",
    )
    return parser


def main() -> int:
    args = build_parser().parse_args()
    schema_payload = SchemaCache().get(force_refresh=args.refresh_schema)

    if not args.question and not args.sql:
        print(
            f"Cached {len(schema_payload['tables'])} tables from "
            f"schema {schema_payload['schema_name']}."
        )
        return 0

    if args.question:
        candidate_sql = generate_sql(
            args.question,
            schema_payload,
            load_glossary(),
        )
        print("Generated SQL:")
        print(candidate_sql)
    else:
        candidate_sql = args.sql

    validation = validate_sql(candidate_sql, schema_payload)
    if not validation.valid or validation.normalized_sql is None:
        print("Validation blocked before EXPLAIN:")
        for error in validation.errors:
            print(f"- {error}")
        return 2

    print("Validation: PASSED")
    print("Normalized SQL:")
    print(validation.normalized_sql)

    report = explain_sql(validation.normalized_sql)
    print_plan_report(report)

    if not args.run:
        print("Execution: SKIPPED. Add --run to approve read-only execution.")
        return 0

    if report.risks:
        approval = input("Plan warnings exist. Type RUN to continue: ").strip()
        if approval != "RUN":
            print("Execution: CANCELLED")
            return 0

    columns, rows, truncated = execute_readonly(validation.normalized_sql)
    print_rows(columns, rows, truncated)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Secret mission

Measure Accuracy with a TPC-H Eval Harness

A convincing demo can hide failures across other questions. Build a three-case TPC-H evaluation harness that compares database results with known-correct answers. You will finish with an accuracy score and a report that categorizes every outcome.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep the local reliability files, pause your work, or delete the additive upgrade. Keeping these files has no ongoing cost because Groq API usage occurs only when you send requests.

Resources you used:

• Core reliability files: reliability_gate.py, glossary.json, requirements-reliability.txt, and .env.example.
• Evaluation files: eval_cases.json and eval_harness.py.
• Generated files: .schema_cache.json and eval_report.json.

Keep everything running

No action is needed. Choose this if you plan to improve the gate or compare future evaluation scores.

• Keep .schema_cache.json so future runs can reuse the discovered schema within the configured TTL.
• Keep eval_report.json as a baseline for future accuracy comparisons.
• Keep the source files beside your existing agent so the reliability upgrade remains available.
• Keep .env out of source control because it contains your real credentials and connection details.

Pause - I'll come back to this later

Pausing ends your current work session while preserving every artifact. No background service keeps running after the CLI stops.

• Let any current command finish in PowerShell.
• Close the PowerShell window.
• Leave the project files in the existing text-to-SQL repository.
• Return later by activating the same existing virtual environment.

Your existing PostgreSQL database remains unchanged. Groq requests stop as soon as you stop using the CLI.

Delete - I don't want to use this again

Use a reset to clear generated state while keeping the upgrade. Use complete removal when you no longer want the additive reliability files.

Choose reset if unsure

Deleting these files can wait if you may revisit the project. Your normal backup is the recovery path after complete removal.

Your existing .env, original agent entry point, and TPC-H data remain untouched by the actions below.

For a reset:

• Switch back to Visual Studio Code with the existing text-to-SQL repository.
• Select .schema_cache.json and eval_report.json in the file sidebar.
• Press Delete to remove the selected files.
• Confirm that .schema_cache.json and eval_report.json no longer appear in the file sidebar.

The next schema run recreates the cache from the live database. The next evaluation run creates a fresh report.

For complete removal:

• Select reliability_gate.py, glossary.json, requirements-reliability.txt, .env.example, eval_cases.json, eval_harness.py, .schema_cache.json, and eval_report.json in the existing repository's file sidebar.
• Press Delete to remove the selected artifacts.
• Confirm that none of the eight filenames appear in the file sidebar.
• Leave the existing .env in place for the original text-to-SQL agent.

Nice Work!

Nice Work!

Excellent work. Your agent now has a reliability gate between generated SQL and database execution.

You've learned how to:

• Reuse a fingerprinted schema cache with a one-hour TTL. You can force a refresh after PostgreSQL schema changes.
• Ground business questions with a business glossary. Groq receives a precise TPC-H definition for discounted revenue. It also receives a precise definition for active customers.
• Apply a local SQL gate with SQLGlot. Inspect PostgreSQL JSON query plans before execution. Keep approved execution read-only. Require explicit opt-in for every run.
• Secret Mission: Built a three-case result-equivalence harness for TPC-H. It writes an accuracy score to eval_report.json. The report records one categorized outcome for every case.

Ready to quiz yourself?