Guard the portfolio audit against the 26m-row recommendation seq-scan

The `recommendation` table (~26m rows) has no index on `plan_id`, so any query
reaching it via `plan_id` — including the audit's own rollup — seq-scans the
whole table and saturates the shared DB (it blocked the DB during the
portfolio-796 audit). Make the audit safe-by-default:

- statement_timeout (120s) on the audit connection — a hard ceiling so a bad
  plan aborts instead of hammering the DB.
- The recommendation rollup (the two solar checks) is now opt-in via
  --with-recommendations, and EXPLAIN-gated: it refuses to run (raising
  RecommendationScanError) when the plan contains a Seq Scan on recommendation,
  which it does on any large portfolio until idx_recommendation_plan_id exists.
- SKILL.md documents the plan_id-no-index trap, the reach-via-property_id /
  EXPLAIN-first / confirm-with-user rules, and the index as the real fix.

Verified on 796/1268: default run is bounded and completes (2,952 anomalies over
31,919 properties); --with-recommendations aborts pre-scan portfolio-wide but is
allowed for a single property.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.claude/skills/audit-ara-portfolio/SKILL.md

@@ -16,6 +16,35 @@ Ask for **portfolio_id** and **scenario_id** if the user didn't give them.
 `scenario_id` is optional — without it, each Property's *default* plan (the one
 shown in the FE) is audited.
 
+## Query safety (READ FIRST — a bad query can take the shared DB down)
+
+The `recommendation` table is **~26m rows and has NO index on `plan_id`** (only
+`id` and `property_id`). Any query that reaches `recommendation` through
+`plan_id` — a `JOIN ... ON r.plan_id = pl.id`, or a correlated
+`EXISTS (SELECT 1 FROM recommendation WHERE plan_id = pl.id ...)` — forces a full
+seq-scan of all 26m rows. On a large portfolio (e.g. 796/1268 has ~32k default
+plans) the planner picks that seq-scan **even when the query also joins
+`property_id`**. This is what blocked the DB during the portfolio-796 audit.
+
+Rules for EVERY query you write or run here (the runner already follows them):
+
+- **Confirm with the user before running any ad-hoc SQL** that touches
+  `recommendation`, and show them the `EXPLAIN` plan. Never improvise a
+  `recommendation` query against the live DB unprompted.
+- **`EXPLAIN` first** (no `ANALYZE` — it executes nothing). If the plan contains
+  `Seq Scan on recommendation`, do NOT run it. Rewrite, or add the index below.
+- **Never reach `recommendation` via `plan_id` or a correlated subquery.** Scope
+  to the portfolio's properties and reach `recommendation` via the indexed
+  `property_id`, and prefer ONE scoped aggregate over per-property subqueries.
+- The runner sets a **`statement_timeout` (120s)** on its connection as a hard
+  ceiling, and the `recommendation` rollup is **opt-in** (see Phase 1). Keep any
+  ad-hoc connection equally bounded.
+- **The real fix is an index:** `CREATE INDEX CONCURRENTLY
+  idx_recommendation_plan_id ON recommendation (plan_id)` (non-blocking build).
+  Until it exists, the solar checks can't run cheaply on a large portfolio — that
+  is a deliberate trade-off, not a bug. Propose it as a migration if the solar
+  checks are needed portfolio-wide.
+
 ## Phase 1 — Build the dataset
 
 ```
@@ -26,6 +55,16 @@ Writes `modelling_audit.md` (grouped, ranked by severity) and
 `modelling_audit.csv` (every flagged row: property_id, uprn, severity, check,
 detail). Read the printed summary and `modelling_audit.md`.
 
+**Recommendation checks are OFF by default.** The two solar checks
+(`excessive-solar-sap`, `low-solar-bill-savings`) read the `recommendation`
+rollup, which is the only query that scans the 26m-row table. They are gated
+behind `--with-recommendations`, which is EXPLAIN-guarded and will **abort** if
+the plan seq-scans `recommendation` (raising `RecommendationScanError`). Without
+the flag those two checks are inert and every other check is bounded by the
+portfolio — this is the safe default. Only pass `--with-recommendations` once
+`idx_recommendation_plan_id` exists (or on a small portfolio where EXPLAIN is
+clean), and confirm with the user first.
+
 ## Phase 2 — Review the high-level results
 
 For each check group, HIGH severity first: note the count, read a few example
@@ -53,6 +92,13 @@ group, cluster the flagged properties by a distinguishing trait — EPC source
 (lodged / predicted), `rebaseline_reason`, property type, the dominant measure,
 fuel — using SQL against the DB. Report the sub-classes and their sizes.
 
+Traits from `property` / `property_baseline_performance` / `plan` are cheap to
+cluster on. **Traits that need `recommendation` (dominant measure, measure mix)
+are NOT** — obey the Query-safety rules above: scope to the *flagged* property
+ids only (a few hundred, so the `property_id` index is used), `EXPLAIN` first,
+and confirm the query with the user. Never cluster the whole portfolio's
+measures in one `recommendation` query.
+
 ## Phase 5 — Cross-reference open work
 
 Before proposing fixes, check whether one is already in flight:

scripts/audit/anomalies.py

@@ -34,7 +34,7 @@ from dataclasses import dataclass
 from enum import IntEnum
 from typing import Callable, Optional
 
-from sqlalchemy import text
+from sqlalchemy import Connection, text
 
 from datatypes.epc.domain.epc import Epc
 from scripts.e2e_common import build_engine, load_env
@@ -261,8 +261,22 @@ _QUERY = text(
 )
 
 
-_ROLLUP_QUERY = text(
-    """
+# Bound every audit query so a bad plan aborts instead of saturating the shared
+# DB. The `recommendation` table (~26m rows) has no index on `plan_id`, so any
+# query reaching it via `plan_id` seq-scans the whole table — exactly what
+# blocked the DB during the portfolio-796 audit. The timeout is the hard ceiling;
+# `_guard_recommendation_scan` is the smarter gate that refuses such a plan up
+# front rather than running it for the full timeout window.
+_STATEMENT_TIMEOUT_MS = 120_000
+
+
+# The rollup is kept as a string (not a pre-built ``text()``) so we can prepend
+# ``EXPLAIN`` and inspect the plan before executing it. It reaches
+# ``recommendation`` via the indexed ``property_id`` (never ``plan_id``); on a
+# large portfolio the planner may still pick a seq-scan, which the EXPLAIN gate
+# catches. Only the ``solar_pv`` rollups + ``n_measures`` come from here, feeding
+# the two opt-in recommendation checks (excessive-solar-sap, low-solar-bill).
+_ROLLUP_SQL = """
     SELECT r.property_id,
            MAX(r.sap_points) FILTER (WHERE r.type = 'solar_pv') AS solar_sap,
            MAX(r.energy_cost_savings) FILTER (WHERE r.type = 'solar_pv') AS solar_bill,
@@ -275,28 +289,69 @@ _ROLLUP_QUERY = text(
       AND (:property_id IS NULL OR p.id = :property_id)
     GROUP BY r.property_id
     """
-)
+_ROLLUP_QUERY = text(_ROLLUP_SQL)
+
+
+class RecommendationScanError(RuntimeError):
+    """Raised when a recommendation query would seq-scan the 26m-row table.
+
+    The fix is operational, not a retry: add ``idx_recommendation_plan_id`` (see
+    the audit skill's query-safety notes) or run without ``--with-recommendations``
+    so the two solar checks are simply skipped.
+    """
+
+
+def _guard_recommendation_scan(
+    conn: Connection, params: dict[str, Optional[int]]
+) -> None:
+    """Refuse to run the rollup if its plan seq-scans ``recommendation``.
+
+    EXPLAIN (no ANALYZE) executes nothing, so this is free. A full seq-scan of
+    the 26m-row table is the pattern that blocked the DB; aborting here is far
+    cheaper than discovering it via the statement timeout.
+    """
+    plan = "\n".join(
+        str(row[0]) for row in conn.execute(text("EXPLAIN " + _ROLLUP_SQL), params)
+    )
+    if "Seq Scan on recommendation" in plan:
+        raise RecommendationScanError(
+            "rollup would seq-scan the 26m-row `recommendation` table "
+            f"(no index on plan_id; portfolio likely too large). Plan:\n{plan}"
+        )
 
 
 def _load(
     portfolio_id: Optional[int],
     property_id: Optional[int],
     scenario_id: Optional[int],
+    with_recommendations: bool = False,
 ) -> list[PropertyAudit]:
+    """Join one row per Property for the checks to run over.
+
+    ``with_recommendations`` is OFF by default: the ``recommendation`` rollup is
+    the only query that touches the 26m-row table and is the one that blocked the
+    DB, so it is opt-in and EXPLAIN-gated. With it off, the two solar checks see
+    ``None`` and are inert — every other check is bounded by the portfolio.
+    """
     engine = build_engine()
     out: list[PropertyAudit] = []
-    params = {
+    params: dict[str, Optional[int]] = {
         "portfolio_id": portfolio_id,
         "property_id": property_id,
         "scenario_id": scenario_id,
     }
     with engine.connect() as conn:
-        rollups: dict[int, tuple[Optional[float], Optional[float], int]] = {
-            m["property_id"]: (m["solar_sap"], m["solar_bill"], m["n_measures"])
-            for m in (
-                row._mapping for row in conn.execute(_ROLLUP_QUERY, params)
-            )
-        }
+        # Hard ceiling: a runaway plan aborts instead of hammering the shared DB.
+        conn.execute(text(f"SET statement_timeout = {_STATEMENT_TIMEOUT_MS}"))
+        rollups: dict[int, tuple[Optional[float], Optional[float], int]] = {}
+        if with_recommendations:
+            _guard_recommendation_scan(conn, params)
+            rollups = {
+                m["property_id"]: (m["solar_sap"], m["solar_bill"], m["n_measures"])
+                for m in (
+                    row._mapping for row in conn.execute(_ROLLUP_QUERY, params)
+                )
+            }
         for r in conn.execute(_QUERY, params):
             m = r._mapping
             solar_sap, solar_bill, n_measures = rollups.get(m["id"], (None, None, 0))
@@ -382,13 +437,27 @@ def main() -> None:
         default="low",
         help="minimum severity to report (default: low — all)",
     )
+    parser.add_argument(
+        "--with-recommendations",
+        action="store_true",
+        help="run the recommendation rollup (the solar checks). OFF by default: "
+        "it scans the 26m-row `recommendation` table and needs "
+        "idx_recommendation_plan_id to be bounded — EXPLAIN-gated when on.",
+    )
     args = parser.parse_args()
     min_severity = Severity[args.severity.upper()]
 
-    audits = _load(args.portfolio, args.property, args.scenario)
+    audits = _load(
+        args.portfolio, args.property, args.scenario, args.with_recommendations
+    )
     anomalies = run(audits, min_severity)
     _write_reports(anomalies, len(audits))
 
+    if not args.with_recommendations:
+        print(
+            "NOTE: recommendation rollup skipped (--with-recommendations off) — "
+            "the solar checks (excessive-solar-sap, low-solar-bill-savings) are inert."
+        )
     print(f"scanned {len(audits)} properties · {len(anomalies)} anomalies "
           f"(>= {min_severity.name})")
     counts: dict[str, int] = {}