feat(nl2sql): EXPLAIN dry-run pre-flight for translate_nl_to_sql

Second of the two remaining security-hardening.md items. Before returning
or executing the generated SQL, run a non-executing EXPLAIN (reusing
mcpg.query.explain_query, so it's SafeSqlDriver-validated) to catch queries
that pass the structural allowlist but reference a non-existent
column/table or mistype something:

- Planner rejection -> structured error='query invalid: <message>' with the
  SQL returned for inspection; nothing executed.
- Pre-flight statement_timeout -> degrade to 'skip, proceed' (don't block a
  valid translation).
- New explain_preflight tool arg (default on); bypassable per call.

Surface snapshot regenerated (translate_nl_to_sql gains explain_preflight).
2 new unit tests (invalid-query surfaced; bypass when disabled) + an EXPLAIN
route in the nl2sql test schema so the default-on pre-flight passes.

Completes security-hardening.md — both NL->SQL items now shipped.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0122yLZLJ8t4W43sdN6BmTZc

Author: claude

CHANGELOG.md

@@ -8,6 +8,13 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 ### Added
 
+- **NL→SQL EXPLAIN dry-run pre-flight.** `translate_nl_to_sql` runs a
+  non-executing `EXPLAIN` on the generated SQL before returning/executing,
+  catching queries that pass the structural allowlist but reference a
+  non-existent column/table. A planner rejection surfaces as
+  `error="query invalid: …"` (SQL returned, nothing executed); a pre-flight
+  timeout degrades to "proceed". Toggle via the `explain_preflight` arg
+  (default on).
 - **NL→SQL prompt-injection hardening (boundary defence).**
   `translate_nl_to_sql` now fences the user question in
   `<user_request>` … `</user_request>` delimiters and instructs the model

docs/security-hardening.md

@@ -275,7 +275,15 @@ default).
 **Effort:** small (system-prompt edit + refusal-path handling + 3-5
 tests). Surfaced 2026-06-30 from a security-feature review.
 
-### ⬜ NL→SQL EXPLAIN dry-run pre-flight
+### ✅ NL→SQL EXPLAIN dry-run pre-flight
+**Shipped.** `translate_nl_to_sql` now runs a non-executing
+`EXPLAIN (FORMAT JSON)` (reusing `mcpg.query.explain_query`, so the SQL is
+`SafeSqlDriver`-validated first) before returning or executing. A planner
+rejection surfaces as a structured `error="query invalid: <message>"` with
+the SQL returned for inspection and nothing executed; a pre-flight
+`statement_timeout` degrades to "skip, proceed" rather than blocking a valid
+translation. Controlled by the `explain_preflight` tool arg (default on).
+
 **Problem.** Generated SQL is validated structurally (AST allowlist +
 single-statement assertion) but not **semantically** — a query that
 references a non-existent column/table or has a type mismatch is

src/mcpg/nl2sql.py

@@ -40,7 +40,7 @@
 
 from mcpg._vendor.sql import SqlDriver, obfuscate_password
 from mcpg.introspection import describe_table, list_foreign_keys, list_tables
-from mcpg.query import DEFAULT_MAX_ROWS, QueryError, run_select
+from mcpg.query import DEFAULT_MAX_ROWS, QueryError, explain_query, run_select
 
 if TYPE_CHECKING:
     from mcpg.config import Settings
@@ -1006,6 +1006,35 @@ def _detect_refusal(sql: str, raw: str) -> str | None:
     return None
 
 
+async def _explain_preflight(driver: SqlDriver, sql: str) -> str | None:
+    """Non-executing ``EXPLAIN`` of the generated SQL — semantic pre-flight.
+
+    Returns a ``"query invalid: …"`` message when the planner rejects the
+    query (a non-existent column/table or type mismatch passes the structural
+    allowlist but fails here), or ``None`` to proceed. ``EXPLAIN`` without
+    ``ANALYZE`` does not run the query, so this is cheap and side-effect free
+    (it reuses :func:`mcpg.query.explain_query`, which validates through
+    ``SafeSqlDriver`` first). A pre-flight ``statement_timeout`` — possible on
+    a query over many partitions or a deep view — degrades to "skip, proceed"
+    rather than blocking a valid translation.
+    """
+    try:
+        _assert_single_statement(sql)
+        await explain_query(driver, sql)
+    except NL2SQLError as exc:
+        return f"query invalid: {exc}"
+    except QueryError as exc:
+        msg = str(exc)
+        lowered = msg.lower()
+        if "statement timeout" in lowered or "canceling statement" in lowered:
+            logger.warning("NL→SQL EXPLAIN pre-flight timed out; proceeding: %s", obfuscate_password(msg))
+            return None
+        return f"query invalid: {obfuscate_password(msg)}"
+    except Exception as exc:  # pragma: no cover - unexpected; degrade to skip
+        logger.warning("NL→SQL EXPLAIN pre-flight skipped: %s", obfuscate_password(str(exc)))
+    return None
+
+
 async def translate_nl_to_sql(
     driver: SqlDriver,
     *,
@@ -1014,6 +1043,7 @@ async def translate_nl_to_sql(
     question: str,
     schema: str,
     execute: bool = False,
+    explain_preflight: bool = True,
     table_filter: tuple[str, ...] | None = None,
     max_tokens: int = DEFAULT_MAX_TOKENS,
     max_rows: int = DEFAULT_MAX_ROWS,
@@ -1118,6 +1148,15 @@ async def translate_nl_to_sql(
     sql, explanation = _parse_response(raw)
 
     refusal_reason = _detect_refusal(sql, raw)
+
+    # Semantic pre-flight: a non-executing EXPLAIN catches queries that pass
+    # the structural allowlist but reference a non-existent column/table, so
+    # the caller gets a clean "query invalid" instead of a raw runtime error.
+    # Runs before both the generation-only return and the execute path.
+    preflight_error = None
+    if refusal_reason is None and explain_preflight and sql:
+        preflight_error = await _explain_preflight(driver, sql)
+
     if refusal_reason is not None:
         # The model declined (embedded instructions / out-of-scope). Surface
         # a structured refusal and never forward the sentinel to execution.
@@ -1134,6 +1173,20 @@ async def translate_nl_to_sql(
             refused=True,
             refusal_reason=refusal_reason,
         )
+    elif preflight_error is not None:
+        # EXPLAIN rejected the generated SQL — return it with the planner
+        # message so the agent can see what's wrong, but don't execute.
+        translation = TranslationResult(
+            sql=sql,
+            explanation=explanation,
+            model=model,
+            provider=provider.name,
+            executed=False,
+            rows=[],
+            columns=[],
+            row_count=0,
+            error=preflight_error,
+        )
     elif not execute or not sql:
         translation = TranslationResult(
             sql=sql,

src/mcpg/tools.py

@@ -3161,6 +3161,7 @@ async def translate_nl_to_sql(
         schema: str,
         provider: str | None = None,
         execute: bool = False,
+        explain_preflight: bool = True,
         table_filter: list[str] | None = None,
         max_rows: int = query.DEFAULT_MAX_ROWS,
         database: _DatabaseArg = None,
@@ -3179,6 +3180,7 @@ async def translate_nl_to_sql(
             question=question,
             schema=schema,
             execute=execute,
+            explain_preflight=explain_preflight,
             table_filter=tuple(table_filter) if table_filter else None,
             max_tokens=settings.nl2sql_max_tokens,
             max_rows=max_rows,

tests/contract/tool_surface.snapshot.json

@@ -8325,6 +8325,11 @@
             "title": "Execute",
             "type": "boolean"
           },
+          "explain_preflight": {
+            "default": true,
+            "title": "Explain Preflight",
+            "type": "boolean"
+          },
           "max_rows": {
             "default": 1000,
             "title": "Max Rows",

tests/unit/test_nl2sql.py

@@ -95,6 +95,9 @@ def _routes_for_simple_schema() -> dict[str, list[dict[str, object]]]:
         ],
         # list_foreign_keys — empty for the simple schema.
         "WHERE con.contype = 'f'": [],
+        # EXPLAIN pre-flight — return a minimal valid plan so the default-on
+        # semantic pre-flight passes for generated SELECTs in these tests.
+        "EXPLAIN (FORMAT JSON)": [{"QUERY PLAN": [{"Plan": {"Node Type": "Result"}}]}],
     }
 
 
@@ -460,6 +463,53 @@ async def test_translate_nl_to_sql_detects_bare_refusal_sentinel() -> None:
     assert result.sql == ""
 
 
+def _routes_with_failing_explain() -> dict[str, list[dict[str, object]]]:
+    """Simple-schema routes but with EXPLAIN returning no plan — simulates a
+    query that passes the structural allowlist yet the planner rejects."""
+    routes = _routes_for_simple_schema()
+    routes["EXPLAIN (FORMAT JSON)"] = []  # explain_query raises "produced no plan"
+    return routes
+
+
+async def test_translate_nl_to_sql_preflight_reports_invalid_query() -> None:
+    # The generated SQL is well-formed structurally but the EXPLAIN pre-flight
+    # rejects it — surface a structured "query invalid" instead of executing.
+    provider = _StubProvider(response='{"sql": "SELECT nope FROM public.widget", "explanation": "x"}')
+
+    result = await translate_nl_to_sql(
+        FakeRoutingDriver(_routes_with_failing_explain()),  # type: ignore[arg-type]
+        provider=provider,  # type: ignore[arg-type]
+        model="m",
+        question="give me the nope column",
+        schema="public",
+        execute=True,
+    )
+
+    assert result.executed is False
+    assert result.error is not None
+    assert result.error.startswith("query invalid")
+    assert result.sql == "SELECT nope FROM public.widget"  # returned for inspection
+
+
+async def test_translate_nl_to_sql_preflight_can_be_disabled() -> None:
+    # explain_preflight=False skips the EXPLAIN round-trip entirely, so even a
+    # driver whose EXPLAIN would fail yields a plain generation result.
+    provider = _StubProvider(response='{"sql": "SELECT nope FROM public.widget", "explanation": "x"}')
+
+    result = await translate_nl_to_sql(
+        FakeRoutingDriver(_routes_with_failing_explain()),  # type: ignore[arg-type]
+        provider=provider,  # type: ignore[arg-type]
+        model="m",
+        question="give me the nope column",
+        schema="public",
+        execute=False,
+        explain_preflight=False,
+    )
+
+    assert result.error is None
+    assert result.sql == "SELECT nope FROM public.widget"
+
+
 async def test_translate_nl_to_sql_records_provider_and_model_on_the_result() -> None:
     provider = _StubProvider(response='{"sql": "SELECT 1", "explanation": "smoke"}')