Aug 27 - Add MySQL optimize and security check

CLAUDE.md

@@ -31,6 +31,8 @@
 21. [Change Philosophy](#21-change-philosophy)
 22. [Object Storage Migration (R2)](#22-object-storage-migration-r2)
 23. [Photo Capture-Time / Geo Overlay](#23-photo-capture-time--geo-overlay)
+24. [Enrollment Form](#24-enrollment-form-enrollment)
+25. [Database Health Check](#25-database-health-check-scriptsdb_healthpy)
 
 ---
 
@@ -168,6 +170,8 @@ part of the tree — see §7. Device registration on the API side lives in
 | `ENROLLMENT_DIR` | Optional. Directory for enrollment-form JSON submissions. Defaults to `<instance_path>/enrollments` (git-ignored). Created at boot. |
 | `DEFAULT_UI_THEME` | Optional, default `modern` (phase50). The design shown when a user has no stored preference — i.e. new accounts and unauthenticated pages. A stored `users.ui_theme` always wins. Set `classic` to revert the default **without** touching anyone's saved choice. |
 | `COMMENTS_VISIBLE_TO_ALL` | Optional, default `true`. **TEMPORARY (Aug 2026).** When true, customers see *every* comment on an issue, not only those ticked "Share with customer". Set `false` to restore the phase22 staff-only filtering — `is_customer_visible` is still written on every comment, so the revert needs no data repair. |
+| `DB_POOL_RECYCLE` | Optional, default `1800` (seconds). Retires a pooled connection after this long. **Must stay below the server's `wait_timeout`** or MySQL closes the socket first and the next request gets `OperationalError 2006`. `scripts/db_health.py` cross-checks the two. |
+| `DB_POOL_SIZE` / `DB_MAX_OVERFLOW` | Optional, default `5` / `5`. Per-**worker** pool. Gunicorn runs `cpu*2+1` sync workers and each holds its own pool, so the ceiling is `workers x (size + overflow)` — the library defaults (5+10) put a 9-worker box at 135 against a `max_connections` of 151. A sync worker serves one request at a time and needs one connection in steady state; the overflow is headroom for the background email/notification threads. |
 | `PHOTO_STAMP_ENABLED` | Optional, default `true`. Burns a capture-time + geo overlay into photos uploaded via `POST /api/v1/photos/upload`. Set `false` to store raw uploads. |
 
 ### Email SSL Auto-Detection
@@ -1974,3 +1978,47 @@ Login-free, so: CSRF-protected form, `@limiter.limit('5 per hour')` on POST only
 ### Admin
 
 `/enrollment/admin` (admin-only, linked from the **Admin** nav dropdown in both layouts). List → detail → office-use fields (Receive Date / Program By / Date email invitation) + status (new / in_progress / completed). `GET /admin/<id>.json` downloads the raw file; `GET /admin/export.csv` emits **one row per person, not per submission** — that is the unit of work when actually creating the accounts. Task cells a person's role cannot have export as `n/a`, distinct from an unticked `''`.
+---
+
+## 25. Database Health Check (`scripts/db_health.py`)
+
+A standalone operations tool for the MySQL side. It imports the app factory for
+config and nothing else — no request layer, no uploads tree — and the plain
+invocation is **strictly read-only** (INFORMATION_SCHEMA / SHOW / EXPLAIN only).
+
+```bash
+python scripts/db_health.py                     # read-only report
+python scripts/db_health.py --json /tmp/db.json # + machine-readable
+python scripts/db_health.py --apply-indexes     # create the missing indexes
+python scripts/db_health.py --analyze           # refresh optimizer stats (safe)
+python scripts/db_health.py --optimize --yes    # rebuild tables (LOCKS — window only)
+python scripts/db_health.py --emit-migration migrations/versions/phase54_perf_indexes.py                            --revision phase54_perf_indexes
+```
+
+**It never DROPs anything.** Redundant indexes are reported with the SQL to run
+by hand, because "unused" is a judgement the tool should not make for you. It
+also refuses to offer a **FK-backed** index as a drop candidate — dropping one
+fails with errno 150, since MySQL needs it for the constraint.
+
+Checks, in order: SQLAlchemy pool options (`pool_pre_ping` / `pool_recycle` —
+the two that decide whether an idle overnight produces `MySQL server has gone
+away`); server settings cross-checked against the app (`max_connections` vs the
+worst-case Gunicorn pool, `wait_timeout` vs `pool_recycle`,
+`innodb_buffer_pool_size` vs the live data size, slow-query log, STRICT mode);
+schema hygiene (non-InnoDB, non-utf8mb4, mixed collations — a collation mismatch
+on a join column silently disables the index); table footprint; missing indexes
+against a curated list; redundant indexes; unindexed foreign keys; and EXPLAIN
+over the dashboard, both list pages, the SLA cron and the mobile notification
+poll, flagging full scans / filesorts / temp tables.
+
+`RECOMMENDED_INDEXES` in the script is the **single place** the index wish-list
+lives, and every entry names the query that justifies it. An index nothing runs
+is pure write-amplification, so keep speculative entries out — and when a new
+hot query lands, add its index there rather than to an ad-hoc migration, so the
+checker keeps agreeing with the schema.
+
+`--emit-migration` writes a re-runnable Alembic migration (INFORMATION_SCHEMA
+guards per rule 16) rather than applying DDL out of band. It guesses
+`down_revision` from the versions directory — confirm against `flask db heads`
+before committing.
+

app/api/inspections.py

@@ -267,8 +267,8 @@ def list_inspections():
     if user.role not in _ALLOWED_ROLES:
         return api_error('Access denied', 403)
 
-    limit  = min(int(request.args.get('limit', 50)), 200)
+    limit  = min(request.args.get('limit',  50, type=int) or 50, 200)
-    offset = max(int(request.args.get('offset', 0)), 0)
+    offset = max(request.args.get('offset', 0,  type=int) or 0,  0)
 
     query = Inspection.query
 

app/api/issues.py

@@ -149,8 +149,8 @@ def list_issues():
     if user.role not in _ALLOWED_ROLES:
         return api_error('Access denied', 403)
 
-    limit  = min(int(request.args.get('limit',  100)), 200)
+    limit  = min(request.args.get('limit',  100, type=int) or 100, 200)
-    offset = max(int(request.args.get('offset', 0)),   0)
+    offset = max(request.args.get('offset', 0,   type=int) or 0,   0)
 
     query = Issue.query
 

app/api/notifications.py

@@ -73,7 +73,7 @@ def list_notifications():
     """
     user  = g.api_user
     since = _parse_since(request.args.get('since'))
-    limit = min(int(request.args.get('limit', 50)), 50)
+    limit = min(request.args.get('limit', 50, type=int) or 50, 50)
 
     def _run_orm():
         q = Notification.query.filter_by(user_id=user.id, is_read=False)

app/api/scheduled.py

@@ -103,8 +103,8 @@ def list_scheduled():
     if user.role not in _ALLOWED_ROLES:
         return api_error('Access denied', 403)
 
-    limit  = min(int(request.args.get('limit',  100)), 200)
+    limit  = min(request.args.get('limit',  100, type=int) or 100, 200)
-    offset = max(int(request.args.get('offset', 0)),   0)
+    offset = max(request.args.get('offset', 0,   type=int) or 0,   0)
 
     query = ScheduledInspection.query.filter(ScheduledInspection.active.is_(True))
 

app/api/stats.py

@@ -111,7 +111,14 @@ def dashboard_stats():
                 )
             )
 
-    open_issues_all = open_q.all()
+    # Counts and buckets only — never a hydrated Issue. For an admin this is
+    # every open issue in the system, fetched on every iPad dashboard refresh;
+    # the full entity would drag the description TEXT and the JSON photo
+    # columns along with it. A Row exposes the same attribute names, so
+    # sla_status() below works unchanged.
+    open_issues_all = open_q.with_entities(
+        Issue.id, Issue.severity, Issue.status, Issue.reported_at
+    ).all()
     open_issues     = len(open_issues_all)
 
     # ── Severity breakdown (derived from the same open_issues_all list) ───

app/routes/dashboard.py

@@ -17,6 +17,16 @@ bp = Blueprint('dashboard', __name__)
 logger = logging.getLogger(__name__)
 
 
+# Columns the dashboard actually reads off an issue row. The cards below need
+# counts and buckets, never a hydrated Issue — loading the full entity pulls the
+# description TEXT and three JSON photo columns for every open issue in scope,
+# on every dashboard load, and registers each one in the identity map.
+# A Row exposes the same attribute names, so _handler_split() and sla_status()
+# work against these unchanged.
+_ISSUE_CARD_COLS = (Issue.id, Issue.severity, Issue.status,
+                    Issue.reported_at, Issue.handler_type)
+
+
 def _handler_split(issues):
     """Count a list of Issues by handler_type (phase35). Rows default to
     'internal' when unset. Returns a dict keyed internal/facility/vendor."""
@@ -102,7 +112,7 @@ def index():
             ))
 
     # Single query — derive count from the list to avoid hitting the DB twice
-    open_issues_all = open_issues_q.all()
+    open_issues_all = open_issues_q.with_entities(*_ISSUE_CARD_COLS).all()
     open_issues     = len(open_issues_all)
     severity_breakdown = {
         'critical': sum(1 for i in open_issues_all if i.severity == 'critical'),
@@ -224,7 +234,7 @@ def index():
     elif is_customer and not customer_facility_ids:
         all_open_issues = []
     else:
-        all_open_issues = sla_q.all()
+        all_open_issues = sla_q.with_entities(*_ISSUE_CARD_COLS).all()
     sla_breached = sum(1 for i in all_open_issues if sla_status(i) == 'breached')
     sla_at_risk  = sum(1 for i in all_open_issues if sla_status(i) == 'at_risk')
 
@@ -250,7 +260,7 @@ def index():
                 Issue.facility_id.in_(customer_facility_ids),
                 _AreaT.facility_id.in_(customer_facility_ids),
             ))
-    opened_today_all    = opened_today_q.all()
+    opened_today_all    = opened_today_q.with_entities(*_ISSUE_CARD_COLS).all()
     issues_opened_today = len(opened_today_all)
     opened_today_handler = _handler_split(opened_today_all)
 
@@ -331,7 +341,7 @@ def index():
             ))
     elif is_customer:
         unassigned_q = unassigned_q.filter(False)  # not relevant for customers
-    unassigned_all     = unassigned_q.all()
+    unassigned_all     = unassigned_q.with_entities(*_ISSUE_CARD_COLS).all()
     unassigned_open    = len(unassigned_all)
     unassigned_handler = _handler_split(unassigned_all)
 

app/routes/inspections.py

@@ -253,7 +253,8 @@ def index():
         q = q.filter(Inspection.status == status_filter)
     if contract_filter.isdigit():
         _contract_fids = [
-            f.id for f in Facility.query.filter_by(project_id=int(contract_filter)).all()
+            fid for (fid,) in db.session.query(Facility.id)
+            .filter(Facility.project_id == int(contract_filter)).all()
         ]
         q = q.filter(Inspection.facility_id.in_(_contract_fids)) if _contract_fids else q.filter(False)
     if facility_filter.isdigit():
@@ -1243,7 +1244,8 @@ def export_list_pdf():
         q = q.filter(Inspection.status == status_filter)
     if contract_filter.isdigit():
         _contract_fids = [
-            f.id for f in Facility.query.filter_by(project_id=int(contract_filter)).all()
+            fid for (fid,) in db.session.query(Facility.id)
+            .filter(Facility.project_id == int(contract_filter)).all()
         ]
         q = q.filter(Inspection.facility_id.in_(_contract_fids)) if _contract_fids else q.filter(False)
     if facility_filter.isdigit():

app/routes/issues.py

@@ -166,13 +166,14 @@ def export_list_pdf():
             date_to_filter = ''
     if contract_filter.isdigit():
         _contract_fids = [
-            f.id for f in Facility.query.filter_by(project_id=int(contract_filter)).all()
+            fid for (fid,) in db.session.query(Facility.id)
+            .filter(Facility.project_id == int(contract_filter)).all()
         ]
         q = q.filter(db.or_(
             Issue.facility_id.in_(_contract_fids),
             Area.facility_id.in_(_contract_fids),
         )) if _contract_fids else q.filter(False)
-    if facility_filter:
+    if facility_filter.isdigit():
         fid = int(facility_filter)
         q = q.filter(db.or_(Issue.facility_id == fid, Area.facility_id == fid))
     if reporter_filter.isdigit():
@@ -201,7 +202,7 @@ def export_list_pdf():
         p = db.session.get(Project, int(contract_filter))
         if p:
             filter_parts.append(f'Contract: {p.name}')
-    if facility_filter:
+    if facility_filter.isdigit():
         f = db.session.get(Facility, int(facility_filter))
         if f:
             filter_parts.append(f'Facility: {f.name}')
@@ -303,7 +304,8 @@ def index():
             date_to_filter = ''
     if contract_filter.isdigit():
         _contract_fids = [
-            f.id for f in Facility.query.filter_by(project_id=int(contract_filter)).all()
+            fid for (fid,) in db.session.query(Facility.id)
+            .filter(Facility.project_id == int(contract_filter)).all()
         ]
         q = q.filter(db.or_(
             Issue.facility_id.in_(_contract_fids),
@@ -311,7 +313,7 @@ def index():
         )) if _contract_fids else q.filter(False)
     if reporter_filter.isdigit():
         q = q.filter(Issue.reported_by == int(reporter_filter))
-    if facility_filter:
+    if facility_filter.isdigit():
         fid = int(facility_filter)
         q = q.filter(
             db.or_(
@@ -338,8 +340,9 @@ def index():
     # can render the following badge and inline unfollow button without an
     # additional query per row.
     followed_ids = {
-        f.issue_id
+        iid for (iid,) in
-        for f in IssueFollower.query.filter_by(user_id=current_user.id).all()
+        db.session.query(IssueFollower.issue_id)
+        .filter(IssueFollower.user_id == current_user.id).all()
     }
 
     # Facilities for the filter dropdown — scoped for inspectors/customers,

app/routes/reports.py

@@ -74,23 +74,29 @@ def index():
     if current_user.role in ('admin', 'director', 'project_manager'):
         inspector_filter = request.args.get('inspector_id', type=int) or None
 
-    # Pre-compute inspection ID sets used by _scope_issue to avoid join conflicts.
+    # Scope issues by the relevant inspector's inspections, as a SUBQUERY rather
-    inspector_inspection_ids = []   # own inspections (inspector role)
+    # than a materialised id list. The previous form pulled every inspection id
-    filter_inspection_ids    = None # filtered inspector's inspections (admin/dir/PM)
+    # that inspector had ever performed into Python and sent them straight back
+    # as a literal IN (1, 2, 3, ... N): the round trip is wasted, the statement
+    # grows without bound with the inspector's history, and a long enough list
+    # eventually trips max_allowed_packet. A subquery is also still a single
+    # statement, so the "avoid join conflicts" reason for pre-computing holds.
+    #
+    # IN (empty subquery) already matches nothing, so the explicit empty-list
+    # guards the old code needed are gone rather than merely moved.
+    inspector_insp_subq = None
     if is_inspector:
-        inspector_inspection_ids = [
+        inspector_insp_subq = (
-            row[0] for row in
             db.session.query(Inspection.id)
             .filter(Inspection.inspector_id == current_user.id)
-            .all()
+            .scalar_subquery()
-        ]
+        )
     elif inspector_filter:
-        filter_inspection_ids = [
+        inspector_insp_subq = (
-            row[0] for row in
             db.session.query(Inspection.id)
             .filter(Inspection.inspector_id == inspector_filter)
-            .all()
+            .scalar_subquery()
-        ]
+        )
 
     def _scope_insp(q):
         if is_inspector:
@@ -104,14 +110,8 @@ def index():
         return q
 
     def _scope_issue(q):
-        if is_inspector:
+        if inspector_insp_subq is not None:
-            if not inspector_inspection_ids:
+            return q.filter(Issue.inspection_id.in_(inspector_insp_subq))
-                return q.filter(False)
-            return q.filter(Issue.inspection_id.in_(inspector_inspection_ids))
-        if filter_inspection_ids is not None:
-            if not filter_inspection_ids:
-                return q.filter(False)
-            return q.filter(Issue.inspection_id.in_(filter_inspection_ids))
         if customer_facility_ids is not None:
             if not customer_facility_ids:
                 return q.filter(False)

app/utils/scope.py

@@ -18,6 +18,7 @@ that no facility-level scoping is required (full access applies).
 """
 
 import logging
+from app import db
 from app.models.project import CustomerAssignment
 from app.models.facility import Facility
 
@@ -43,30 +44,34 @@ def get_customer_scope(user) -> list[int] | None:
     if user.role != 'customer':
         return None  # no scoping needed for internal staff
 
-    assignments = CustomerAssignment.query.filter_by(user_id=user.id).all()
+    # Select only the two columns needed. The previous .all() built full
+    # CustomerAssignment ORM objects (and their identity-map entries) purely to
+    # read two integers off each one; this function runs on nearly every
+    # request for a customer, sometimes more than once.
+    assignments = db.session.query(
+        CustomerAssignment.project_id,
+        CustomerAssignment.facility_id,
+    ).filter(CustomerAssignment.user_id == user.id).all()
 
     if not assignments:
         return []
 
     # Separate direct facility assignments from project-level assignments
-    direct_facility_ids = {a.facility_id for a in assignments if a.facility_id}
+    direct_facility_ids = {fac_id for _, fac_id in assignments if fac_id}
-    project_ids         = {a.project_id  for a in assignments if not a.facility_id}
+    project_ids         = {proj_id for proj_id, fac_id in assignments if not fac_id}
 
     facility_ids = set(direct_facility_ids)
 
     # Single bulk query for all project-scoped facilities — replaces the
-    # previous per-assignment Facility.query loop (N+1 pattern).
+    # previous per-assignment Facility.query loop (N+1 pattern). Only the id
+    # column is read; nothing here needs a hydrated Facility.
     if project_ids:
-        project_facilities = (
+        facility_ids.update(
-            Facility.query
+            fid for (fid,) in db.session.query(Facility.id).filter(
-            .filter(
                 Facility.project_id.in_(project_ids),
                 Facility.active == True,
-            )
+            ).all()
-            .all()
         )
-        for f in project_facilities:
-            facility_ids.add(f.id)
 
     logger.debug(
         'SCOPE | customer_scope | user_id=%s username=%s facility_ids=%s',
@@ -102,16 +107,19 @@ def get_inspector_scope(user) -> list[int] | None:
 
     from app.models.inspector_assignment import InspectorAssignment
 
+    # Column-only selects — see the note in get_customer_scope(). This runs on
+    # every scoped request for both inspector roles.
     project_ids = [
-        a.project_id
+        pid for (pid,) in
-        for a in InspectorAssignment.query.filter_by(user_id=user.id).all()
+        db.session.query(InspectorAssignment.project_id)
+        .filter(InspectorAssignment.user_id == user.id).all()
     ]
 
     if not project_ids:
         return []  # strict: no assignments = no access
 
     facility_ids = [
-        f.id for f in Facility.query.filter(
+        fid for (fid,) in db.session.query(Facility.id).filter(
             Facility.project_id.in_(project_ids),
             Facility.active == True,
         ).all()

app/utils/sla.py

@@ -110,11 +110,38 @@ def send_sla_alerts():
 
     logger = logging.getLogger(__name__)
 
-    # yield_per streams rows in batches of 100 rather than loading all open
+    # Narrow to actual CANDIDATES in SQL rather than reading every open issue
-    # issues into memory at once.  At current scale this is a no-op difference,
+    # and deciding in Python. This runs every 30 minutes forever, so the old
-    # but it prevents a memory spike if the issue count grows large.
+    # form's cost grew with the whole open-issue backlog even on a quiet night
+    # where nothing was due. Three filters, each mirroring a `continue` below:
+    #
+    #   1. reported_at IS NOT NULL — the column is nullable, and sla_status()
+    #      raises TypeError on a NULL (datetime + timedelta). One such row
+    #      would abort the entire cron run, so exclude it in SQL.
+    #   2. sla_notified <> 'breached' — the highest level is already sent; the
+    #      loop skips these unconditionally.
+    #   3. old enough to be at least at-risk for its OWN severity, i.e.
+    #      reported_at <= now - (window * 0.75). A critical issue qualifies
+    #      after 3h, a low one after 90h.
+    #
+    # Anything this excludes would have hit a `continue` anyway, so the set of
+    # notifications sent is unchanged — only the rows read are.
+    now = now_eastern()
+    age_clauses = [
+        db.and_(
+            Issue.severity == severity,
+            Issue.reported_at <= now - timedelta(hours=hours * AT_RISK_THRESHOLD),
+        )
+        for severity, hours in SLA_HOURS.items()
+    ]
+
+    # yield_per streams the survivors in batches rather than materialising them
+    # all at once.
     open_issues = Issue.query.filter(
-        Issue.status.in_(['open', 'in_progress', 'pending_verification'])
+        Issue.status.in_(['open', 'in_progress', 'pending_verification']),
+        Issue.reported_at.isnot(None),
+        db.or_(Issue.sla_notified.is_(None), Issue.sla_notified != 'breached'),
+        db.or_(*age_clauses),
     ).yield_per(100)
 
     total_sent = 0

config.py

@@ -32,6 +32,34 @@ class Config:
     SQLALCHEMY_TRACK_MODIFICATIONS = False
     SQLALCHEMY_ECHO = False
 
+    # ── Connection pool ─────────────────────────────────────────────────────
+    # Without these the pool runs on library defaults, which is where the
+    # intermittent OperationalError 2006 ("MySQL server has gone away") comes
+    # from: MySQL closes a connection after wait_timeout (8h by default) and
+    # SQLAlchemy hands the dead socket to the next request.
+    #
+    #   pool_pre_ping  — cheap liveness check before a connection is handed out;
+    #                    a dead one is discarded and replaced transparently.
+    #   pool_recycle   — retire connections after 30 min, well under any sane
+    #                    wait_timeout, so they are never the stale ones.
+    #   pool_size /    — Gunicorn runs sync workers (cpu*2+1), and each worker
+    #   max_overflow     holds its OWN pool. Library defaults (5 + 10) mean a
+    #                    9-worker box can open 135 connections against a
+    #                    max_connections of 151. A sync worker serves one
+    #                    request at a time, so it needs one connection in
+    #                    steady state; the small overflow is headroom for
+    #                    background email/notification threads.
+    #
+    # Tune with scripts/db_health.py, which cross-checks these against the
+    # server's live max_connections and wait_timeout.
+    SQLALCHEMY_ENGINE_OPTIONS = {
+        'pool_pre_ping': True,
+        'pool_recycle':  int(os.environ.get('DB_POOL_RECYCLE', '1800')),
+        'pool_size':     int(os.environ.get('DB_POOL_SIZE', '5')),
+        'max_overflow':  int(os.environ.get('DB_MAX_OVERFLOW', '5')),
+        'pool_timeout':  30,
+    }
+
     # ── File uploads ────────────────────────────────────────────────────────
     UPLOAD_FOLDER = os.path.join(basedir, 'app/static/uploads')
     MAX_CONTENT_LENGTH = 50 * 1024 * 1024   # 50 MB

scripts/db_health.py

@@ -0,0 +1,963 @@
+#!/usr/bin/env python3
+"""
+scripts/db_health.py  --  MySQL health check + index optimizer
+==============================================================
+
+A standalone operations tool. It does NOT import the request layer, does not
+touch the uploads tree, and by DEFAULT writes nothing at all -- the plain
+invocation is strictly read-only (SELECTs against INFORMATION_SCHEMA / SHOW /
+EXPLAIN).
+
+What it checks
+--------------
+  1. SQLAlchemy engine options      pool_pre_ping / pool_recycle -- the two that
+                                    decide whether you get "MySQL server has
+                                    gone away" after an idle overnight.
+  2. Server + connection settings   max_connections vs the Gunicorn pool,
+                                    wait_timeout vs pool_recycle,
+                                    innodb_buffer_pool_size vs the live data
+                                    size, slow-query-log state, sql_mode.
+  3. Schema hygiene                 non-InnoDB tables, non-utf8mb4 charsets,
+                                    mixed collations (a collation mismatch on a
+                                    join column silently disables the index).
+  4. Table + index footprint        row estimates, data/index bytes.
+  5. Missing indexes                a curated, individually justified list
+                                    checked against what is really there.
+  6. Redundant indexes              leftmost-prefix duplicates. FK-backed
+                                    indexes are excluded from drop candidates --
+                                    dropping one fails with errno 150.
+  7. Unindexed foreign keys         should be empty on InnoDB; a gap here means
+                                    every parent DELETE scans the child table.
+  8. EXPLAIN on hot queries         the dashboard, both list pages, the SLA
+                                    cron, the mobile notification poll. Flags
+                                    full table scans, filesorts, temp tables.
+
+What it can change (only when explicitly asked)
+-----------------------------------------------
+    --apply-indexes   CREATE the missing recommended indexes (idempotent --
+                      re-checks INFORMATION_SCHEMA first, per CLAUDE.md rule 16;
+                      CREATE INDEX IF NOT EXISTS needs MySQL >= 8.0.12).
+    --analyze         ANALYZE TABLE on every table. Cheap, non-locking on
+                      InnoDB, refreshes optimizer cardinality statistics.
+                      Safe to run in production.
+    --optimize        OPTIMIZE TABLE -- rebuilds each table to reclaim space
+                      after large deletes. Requires --yes. This LOCKS each
+                      table for the rebuild: maintenance window only.
+
+It never DROPs anything. Redundant indexes are reported with the exact SQL to
+run by hand, because "unused" is a judgement call the tool should not make for
+you.
+
+Usage
+-----
+    cd /home/jqc/janitorial_qc
+    source venv/bin/activate
+
+    python scripts/db_health.py                          # read-only report
+    python scripts/db_health.py --json /tmp/db.json      # + machine-readable
+    python scripts/db_health.py --no-explain             # skip the EXPLAIN pass
+    python scripts/db_health.py --emit-migration migrations/versions/phase54_perf_indexes.py \
+                               --revision phase54_perf_indexes
+    python scripts/db_health.py --apply-indexes          # create missing indexes
+    python scripts/db_health.py --analyze                # refresh statistics
+    python scripts/db_health.py --optimize --yes         # rebuild (locks!)
+
+Exit codes
+----------
+    0  no findings at or above --fail-on
+    1  findings at or above --fail-on  (default 'none' -- never fails)
+    2  could not connect / not a MySQL database / refused a destructive flag
+"""
+
+import os
+import re
+import sys
+import json
+import argparse
+from datetime import datetime
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from sqlalchemy import text                      # noqa: E402
+from app import create_app, db                   # noqa: E402
+
+
+# -- Output helpers ----------------------------------------------------------
+
+class C:
+    OK   = '\033[92m'
+    WARN = '\033[93m'
+    ERR  = '\033[91m'
+    DIM  = '\033[90m'
+    BOLD = '\033[1m'
+    END  = '\033[0m'
+
+
+_NO_COLOR = (not sys.stdout.isatty()) or bool(os.environ.get('NO_COLOR'))
+
+
+def _c(code, s):
+    return s if _NO_COLOR else f'{code}{s}{C.END}'
+
+
+def header(title):
+    print()
+    print(_c(C.BOLD, f'-- {title} ' + '-' * max(0, 72 - len(title))))
+
+
+SEVERITY_ORDER = {'info': 0, 'low': 1, 'medium': 2, 'high': 3}
+
+FINDINGS = []
+
+
+def finding(severity, area, message, fix=None):
+    """Record a finding and print it. severity: info|low|medium|high."""
+    FINDINGS.append({'severity': severity, 'area': area,
+                     'message': message, 'fix': fix})
+    tag = {
+        'info':   _c(C.DIM,  'INFO '),
+        'low':    _c(C.DIM,  'LOW  '),
+        'medium': _c(C.WARN, 'MED  '),
+        'high':   _c(C.ERR,  'HIGH '),
+    }[severity]
+    print(f'  {tag} {message}')
+    if fix:
+        for line in fix.strip().splitlines():
+            print(_c(C.DIM, f'        {line}'))
+
+
+def ok(message):
+    print(f'  {_c(C.OK, "OK   ")} {message}')
+
+
+def human_bytes(n):
+    n = float(n or 0)
+    for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
+        if n < 1024:
+            return f'{n:.1f} {unit}'
+        n /= 1024
+    return f'{n:.1f} PB'
+
+
+# -- The recommended index set -----------------------------------------------
+#
+# Each entry is justified by a specific query in the application. An index that
+# nothing runs is pure write-amplification, so nothing speculative belongs here.
+#
+#   (table, index_name, 'col, col', priority, why)
+#
+# priority: 'high'   -- a hot path scans without it (every page load / every cron)
+#           'medium' -- a heavy but less frequent report or filter
+#           'low'    -- worth having as the table grows
+
+RECOMMENDED_INDEXES = [
+    # --- notifications: polled by the bell on every page AND by every iPad ---
+    ('notifications', 'ix_notif_user_read_created', 'user_id, is_read, created_at',
+     'high',
+     "GET /api/v1/notifications and the in-app bell both run "
+     "WHERE user_id=? AND is_read=0 [AND created_at > ?] ORDER BY created_at. "
+     "Only user_id is indexed today, so MySQL reads every notification ever "
+     "sent to that user and then sorts them."),
+
+    ('notifications', 'ix_notif_digest_user', 'digest_pending, user_id',
+     'medium',
+     "send_pending_digests() runs SELECT DISTINCT user_id WHERE "
+     "digest_pending=1. The existing single-column digest_pending index cannot "
+     "satisfy the DISTINCT from the index alone."),
+
+    # --- issues: the SLA cron scans these every 30 minutes, forever ----------
+    ('issues', 'ix_issues_status_reported', 'status, reported_at',
+     'high',
+     "send_sla_alerts() selects every open/in_progress/pending_verification "
+     "issue every 30 min, and the issues list orders by reported_at DESC after "
+     "filtering on status. One composite serves both; ix_issues_status then "
+     "becomes a redundant prefix of it."),
+
+    ('issues', 'ix_issues_resolved_at', 'resolved_at',
+     'medium',
+     "The dashboard 'Resolved Today' card and the SLA Compliance report both "
+     "filter on resolved_at ranges; the column is unindexed."),
+
+    ('issues', 'ix_issues_area_status', 'area_id, status',
+     'medium',
+     "Every facility-scoped issue query outer-joins areas and ORs on "
+     "Area.facility_id. area_id carries only the plain FK index; pairing it "
+     "with status lets the legacy area-routed half of the OR resolve from the "
+     "index instead of reading rows."),
+
+    ('issues', 'ix_issues_handler_status', 'handler_type, status',
+     'low',
+     "Dashboard handler-split cards and the ?handler_type= list filter."),
+
+    # --- inspections --------------------------------------------------------
+    ('inspections', 'ix_insp_followup_status', 'follow_up_required, status',
+     'medium',
+     "The Pending Follow-ups dashboard card and the ?status=follow_up filter "
+     "run WHERE follow_up_required=1 AND status='completed' AND NOT EXISTS("
+     "child re-inspection)."),
+
+    ('inspections', 'ix_insp_followup_assignee', 'follow_up_assigned_to',
+     'low',
+     "phase53 follow-up ownership (Inspection.follow_up_owned_by) filters on "
+     "this column. It is a late-added nullable FK -- confirm it is indexed."),
+
+    # --- audit_logs: append-only, and the fastest-growing table here ---------
+    ('audit_logs', 'ix_audit_entity', 'entity_type, entity_id',
+     'medium',
+     "The audit trail is filtered by entity. entity_type alone is very low "
+     "cardinality (about a dozen distinct values), so it is close to useless "
+     "on its own."),
+
+    ('audit_logs', 'ix_audit_user_created', 'user_id, created_at',
+     'low',
+     "Per-user audit filtering with the default created_at DESC ordering."),
+
+    # --- issue_comments -----------------------------------------------------
+    ('issue_comments', 'ix_comments_issue_created', 'issue_id, created_at',
+     'medium',
+     "Both the web issue detail and GET /api/v1/issues/<id>/comments read the "
+     "thread ordered by created_at. The plain FK index on issue_id leaves the "
+     "sort to a filesort."),
+
+    # --- scheduled_inspections: cron every 30 min ---------------------------
+    ('scheduled_inspections', 'ix_sched_active_due', 'active, next_due_date',
+     'medium',
+     "The reminder cron and the dashboard panel both read "
+     "WHERE active=1 ORDER BY next_due_date."),
+
+    # --- support ------------------------------------------------------------
+    ('support_chat_messages', 'ix_scm_session_created', 'session_id, created_at',
+     'low',
+     "Transcript rendering reads a session's messages in created_at order."),
+]
+
+
+# -- Representative hot queries for the EXPLAIN pass -------------------------
+#
+# Literal-free: every value is a bound parameter, so this pass reflects what the
+# app actually sends and cannot itself be a vector for anything.
+
+EXPLAIN_QUERIES = [
+    ('dashboard: open issues (scoped)',
+     "SELECT i.id FROM issues i LEFT JOIN areas a ON i.area_id = a.id "
+     "WHERE i.status IN ('open','in_progress')",
+     {}),
+
+    ('dashboard: resolved today',
+     "SELECT COUNT(*) FROM issues WHERE status='resolved' "
+     "AND resolved_at >= :d0 AND resolved_at < :d1",
+     {'d0': '2026-01-01 00:00:00', 'd1': '2026-01-02 00:00:00'}),
+
+    ('dashboard: pending follow-ups',
+     "SELECT COUNT(*) FROM inspections i WHERE i.follow_up_required = 1 "
+     "AND i.status = 'completed' AND NOT EXISTS "
+     "(SELECT 1 FROM inspections c WHERE c.parent_inspection_id = i.id)",
+     {}),
+
+    ('issues list: page 1, newest first',
+     "SELECT i.id FROM issues i LEFT JOIN areas a ON i.area_id = a.id "
+     "ORDER BY i.reported_at DESC LIMIT 25",
+     {}),
+
+    ('inspections list: page 1, newest first',
+     "SELECT i.id FROM inspections i ORDER BY i.inspection_date DESC LIMIT 25",
+     {}),
+
+    ('SLA cron: candidate scan (every 30 min)',
+     "SELECT id FROM issues WHERE status IN "
+     "('open','in_progress','pending_verification')",
+     {}),
+
+    ('mobile: unread notification poll',
+     "SELECT id FROM notifications WHERE user_id = :uid AND is_read = 0 "
+     "AND created_at > :since ORDER BY created_at ASC LIMIT 50",
+     {'uid': 1, 'since': '2026-01-01 00:00:00'}),
+
+    ('digest cron: pending recipients',
+     "SELECT DISTINCT user_id FROM notifications WHERE digest_pending = 1",
+     {}),
+
+    ('reports: avg score by facility (30d)',
+     "SELECT f.id, AVG(i.overall_score) FROM facilities f "
+     "JOIN inspections i ON f.id = i.facility_id "
+     "WHERE i.inspection_date >= :d0 AND i.status='completed' "
+     "AND i.overall_score IS NOT NULL GROUP BY f.id",
+     {'d0': '2026-01-01 00:00:00'}),
+
+    ('audit trail: newest first',
+     "SELECT id FROM audit_logs ORDER BY created_at DESC LIMIT 50",
+     {}),
+]
+
+
+# -- Introspection -----------------------------------------------------------
+
+def db_name(conn):
+    return conn.execute(text('SELECT DATABASE()')).scalar()
+
+
+def is_mysql(conn):
+    return conn.dialect.name in ('mysql', 'mariadb')
+
+
+def server_version(conn):
+    return conn.execute(text('SELECT VERSION()')).scalar()
+
+
+def global_vars(conn, names):
+    out = {}
+    for n in names:
+        try:
+            row = conn.execute(
+                text('SHOW GLOBAL VARIABLES LIKE :n'), {'n': n}).fetchone()
+            if row:
+                out[row[0]] = row[1]
+        except Exception:
+            pass
+    return out
+
+
+def fetch_tables(conn, schema):
+    rows = conn.execute(text("""
+        SELECT TABLE_NAME, ENGINE, TABLE_ROWS, DATA_LENGTH, INDEX_LENGTH,
+               TABLE_COLLATION
+        FROM INFORMATION_SCHEMA.TABLES
+        WHERE TABLE_SCHEMA = :s AND TABLE_TYPE = 'BASE TABLE'
+        ORDER BY (DATA_LENGTH + INDEX_LENGTH) DESC
+    """), {'s': schema}).fetchall()
+    return [{'name': r[0], 'engine': r[1], 'rows': r[2] or 0,
+             'data': r[3] or 0, 'index': r[4] or 0, 'collation': r[5]}
+            for r in rows]
+
+
+def fetch_indexes(conn, schema):
+    """Return {table: {index_name: {'cols': [...], 'unique': bool}}}."""
+    rows = conn.execute(text("""
+        SELECT TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX, COLUMN_NAME, NON_UNIQUE
+        FROM INFORMATION_SCHEMA.STATISTICS
+        WHERE TABLE_SCHEMA = :s
+        ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX
+    """), {'s': schema}).fetchall()
+    out = {}
+    for tbl, idx, seq, col, non_unique in rows:
+        entry = out.setdefault(tbl, {}).setdefault(
+            idx, {'cols': [], 'unique': not non_unique})
+        entry['cols'].append(col)
+    return out
+
+
+def fetch_fk_protection(conn, schema):
+    """Index names and columns MySQL relies on for FOREIGN KEY constraints.
+
+    Dropping such an index fails with errno 150, so these must never be
+    offered as drop candidates no matter how redundant they look.
+    """
+    rows = conn.execute(text("""
+        SELECT TABLE_NAME, CONSTRAINT_NAME
+        FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
+        WHERE TABLE_SCHEMA = :s AND CONSTRAINT_TYPE = 'FOREIGN KEY'
+    """), {'s': schema}).fetchall()
+    fk_names = {}
+    for tbl, name in rows:
+        fk_names.setdefault(tbl, set()).add(name)
+
+    rows = conn.execute(text("""
+        SELECT TABLE_NAME, COLUMN_NAME
+        FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
+        WHERE TABLE_SCHEMA = :s AND REFERENCED_TABLE_NAME IS NOT NULL
+    """), {'s': schema}).fetchall()
+    fk_cols = {}
+    for tbl, col in rows:
+        fk_cols.setdefault(tbl, set()).add(col)
+    return fk_names, fk_cols
+
+
+def index_exists(conn, schema, table, index_name):
+    return conn.execute(text("""
+        SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
+        WHERE TABLE_SCHEMA = :s AND TABLE_NAME = :t AND INDEX_NAME = :i
+    """), {'s': schema, 't': table, 'i': index_name}).scalar() > 0
+
+
+def table_exists(conn, schema, table):
+    return conn.execute(text("""
+        SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
+        WHERE TABLE_SCHEMA = :s AND TABLE_NAME = :t
+    """), {'s': schema, 't': table}).scalar() > 0
+
+
+def column_exists(conn, schema, table, column):
+    return conn.execute(text("""
+        SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS
+        WHERE TABLE_SCHEMA = :s AND TABLE_NAME = :t AND COLUMN_NAME = :c
+    """), {'s': schema, 't': table, 'c': column}).scalar() > 0
+
+
+# -- Checks ------------------------------------------------------------------
+
+def check_engine_options(app):
+    header('SQLAlchemy engine options')
+    opts = app.config.get('SQLALCHEMY_ENGINE_OPTIONS') or {}
+    uri = app.config.get('SQLALCHEMY_DATABASE_URI', '')
+
+    if not opts:
+        finding('high', 'engine',
+                'SQLALCHEMY_ENGINE_OPTIONS is not set -- the pool runs on '
+                'library defaults with no liveness check and no recycle.',
+                "Add to config.Config:\n"
+                "  SQLALCHEMY_ENGINE_OPTIONS = {\n"
+                "      'pool_pre_ping': True,   # kills 'MySQL server has gone away'\n"
+                "      'pool_recycle': 1800,    # must be < MySQL wait_timeout\n"
+                "      'pool_size': 5, 'max_overflow': 5,\n"
+                "      'pool_timeout': 30,\n"
+                "  }")
+    else:
+        if not opts.get('pool_pre_ping'):
+            finding('high', 'engine',
+                    'pool_pre_ping is off -- an idle connection killed by MySQL '
+                    'surfaces as OperationalError 2006 on the next request.',
+                    "SQLALCHEMY_ENGINE_OPTIONS['pool_pre_ping'] = True")
+        else:
+            ok('pool_pre_ping enabled')
+
+        recycle = opts.get('pool_recycle')
+        if not recycle or recycle < 0:
+            finding('medium', 'engine',
+                    'pool_recycle is unset -- connections are held indefinitely '
+                    'and can outlive the server-side wait_timeout.',
+                    "SQLALCHEMY_ENGINE_OPTIONS['pool_recycle'] = 1800")
+        else:
+            ok(f'pool_recycle = {recycle}s')
+
+    if 'mysql' in uri and 'charset=' not in uri:
+        finding('low', 'engine',
+                'DATABASE_URL has no ?charset= -- PyMySQL negotiates the server '
+                'default, which on older installs is latin1.',
+                'Append ?charset=utf8mb4 to DATABASE_URL.')
+    return opts
+
+
+def check_server(conn, opts, tables):
+    header('MySQL server settings')
+    print(f'  {_c(C.DIM, "version")} {server_version(conn)}')
+
+    g = global_vars(conn, [
+        'max_connections', 'wait_timeout', 'interactive_timeout',
+        'innodb_buffer_pool_size', 'slow_query_log', 'long_query_time',
+        'sql_mode', 'character_set_server', 'collation_server',
+        'innodb_flush_log_at_trx_commit', 'table_open_cache',
+        'performance_schema',
+    ])
+
+    # --- pool sizing vs max_connections -------------------------------------
+    try:
+        maxc = int(g.get('max_connections', 0))
+    except (TypeError, ValueError):
+        maxc = 0
+    pool_size = (opts or {}).get('pool_size', 5)
+    overflow = (opts or {}).get('max_overflow', 10)
+    try:
+        import multiprocessing
+        workers = multiprocessing.cpu_count() * 2 + 1   # gunicorn_config.py
+    except Exception:
+        workers = 9
+    worst_case = workers * (pool_size + overflow)
+    print(f'  {_c(C.DIM, "max_connections")} {maxc}   '
+          f'{_c(C.DIM, "gunicorn workers")} ~{workers}   '
+          f'{_c(C.DIM, "worst-case pool")} {worst_case}')
+    if maxc and worst_case > maxc * 0.8:
+        finding('high', 'server',
+                f'{workers} sync workers x (pool_size {pool_size} + overflow '
+                f'{overflow}) = {worst_case} possible connections against '
+                f'max_connections={maxc}. Under load that returns "Too many '
+                f'connections" instead of queueing.',
+                'Either cap the pool (pool_size 5 / max_overflow 5) or raise '
+                'max_connections. A sync worker only ever needs one connection '
+                'in steady state.')
+    elif maxc:
+        ok(f'connection ceiling has headroom ({worst_case} worst case / {maxc})')
+
+    # --- wait_timeout vs pool_recycle ---------------------------------------
+    try:
+        wait = int(g.get('wait_timeout', 0))
+    except (TypeError, ValueError):
+        wait = 0
+    recycle = (opts or {}).get('pool_recycle')
+    if wait and recycle and recycle >= wait:
+        finding('high', 'server',
+                f'pool_recycle ({recycle}s) is not below wait_timeout ({wait}s) '
+                f'-- the server closes the connection before SQLAlchemy retires '
+                f'it, which is exactly the case pool_recycle exists to prevent.',
+                f'Set pool_recycle to roughly {int(wait * 0.6)}.')
+    elif wait:
+        ok(f'wait_timeout = {wait}s')
+
+    # --- buffer pool vs live data -------------------------------------------
+    try:
+        pool_bytes = int(g.get('innodb_buffer_pool_size', 0))
+    except (TypeError, ValueError):
+        pool_bytes = 0
+    total = sum(t['data'] + t['index'] for t in tables)
+    print(f'  {_c(C.DIM, "innodb_buffer_pool_size")} {human_bytes(pool_bytes)}   '
+          f'{_c(C.DIM, "data+indexes")} {human_bytes(total)}')
+    if pool_bytes and total > pool_bytes:
+        finding('medium', 'server',
+                f'The working set ({human_bytes(total)}) exceeds the InnoDB '
+                f'buffer pool ({human_bytes(pool_bytes)}) -- hot pages are being '
+                f'evicted and re-read from disk.',
+                'On a dedicated box set innodb_buffer_pool_size to about '
+                '60-70% of RAM.')
+    elif pool_bytes:
+        ok('the entire dataset fits in the InnoDB buffer pool')
+
+    # --- slow query log ------------------------------------------------------
+    if str(g.get('slow_query_log', 'OFF')).upper() in ('OFF', '0'):
+        finding('medium', 'observability',
+                'slow_query_log is OFF -- there is no record of which queries '
+                'are actually slow in production, so tuning stays guesswork.',
+                "SET GLOBAL slow_query_log = 'ON';\n"
+                "SET GLOBAL long_query_time = 0.5;\n"
+                "(and persist both in my.cnf so they survive a restart)")
+    else:
+        ok(f"slow_query_log ON (long_query_time={g.get('long_query_time')})")
+
+    # --- sql_mode ------------------------------------------------------------
+    mode = g.get('sql_mode', '') or ''
+    if 'STRICT_TRANS_TABLES' not in mode and 'STRICT_ALL_TABLES' not in mode:
+        finding('medium', 'integrity',
+                'sql_mode has no STRICT mode -- an over-long string or an '
+                'invalid ENUM value is silently coerced instead of raising.',
+                'Add STRICT_TRANS_TABLES to sql_mode in my.cnf.')
+    else:
+        ok('STRICT mode enabled')
+    return g
+
+
+def check_schema_hygiene(conn, schema, tables):
+    header('Schema hygiene')
+    bad_engine = [t for t in tables if (t['engine'] or '').upper() != 'INNODB']
+    if bad_engine:
+        finding('high', 'schema',
+                'Non-InnoDB tables found: ' +
+                ', '.join(f"{t['name']} ({t['engine']})" for t in bad_engine),
+                'MyISAM has no transactions and locks the whole table on write.\n'
+                + '\n'.join(f'ALTER TABLE {t["name"]} ENGINE=InnoDB;'
+                            for t in bad_engine))
+    else:
+        ok('all tables are InnoDB')
+
+    collations = {}
+    for t in tables:
+        collations.setdefault(t['collation'] or 'unknown', []).append(t['name'])
+
+    non_utf8mb4 = {c: n for c, n in collations.items()
+                   if not str(c).startswith('utf8mb4')}
+    if non_utf8mb4:
+        for coll, names in non_utf8mb4.items():
+            finding('medium', 'schema',
+                    f'{len(names)} table(s) are {coll}, not utf8mb4: '
+                    + ', '.join(names[:8]) + ('...' if len(names) > 8 else ''),
+                    'utf8mb3/latin1 cannot store emoji or many non-Latin names, '
+                    'and a join between two different collations cannot use an '
+                    'index on that column.\n'
+                    + '\n'.join(
+                        f'ALTER TABLE {n} CONVERT TO CHARACTER SET utf8mb4 '
+                        f'COLLATE utf8mb4_unicode_ci;' for n in names[:8]))
+    else:
+        ok('all tables are utf8mb4')
+
+    if len(collations) > 1:
+        finding('low', 'schema',
+                'Mixed collations across tables: '
+                + ', '.join(sorted(str(c) for c in collations))
+                + '. Joining two columns with different collations silently '
+                  'disables index use on the join.')
+
+
+def check_table_footprint(tables):
+    header('Table footprint (largest first)')
+    print('  ' + _c(C.DIM, 'table'.ljust(32) + 'rows~'.rjust(12)
+                    + 'data'.rjust(12) + 'index'.rjust(12)))
+    for t in tables[:15]:
+        print(f'  {t["name"][:32].ljust(32)}'
+              f'{str(t["rows"]).rjust(12)}'
+              f'{human_bytes(t["data"]).rjust(12)}'
+              f'{human_bytes(t["index"]).rjust(12)}')
+    for t in tables:
+        if t['index'] > t['data'] * 2 and t['data'] > 10 * 1024 * 1024:
+            finding('low', 'indexes',
+                    f'{t["name"]}: index bytes ({human_bytes(t["index"])}) are '
+                    f'more than twice the data ({human_bytes(t["data"])}) -- '
+                    f'likely over-indexed, which slows every write.')
+
+
+def check_missing_indexes(conn, schema, existing):
+    header('Missing indexes (recommended set)')
+    missing = []
+    for table, name, cols, prio, why in RECOMMENDED_INDEXES:
+        if not table_exists(conn, schema, table):
+            continue
+        col_list = [c.strip() for c in cols.split(',')]
+        if any(not column_exists(conn, schema, table, c) for c in col_list):
+            continue   # column absent in this schema version -- skip quietly
+
+        tbl_idx = existing.get(table, {})
+        # Covered when some existing index STARTS with exactly these columns in
+        # this order: a longer index with the right prefix serves the same
+        # queries, so adding a duplicate would only cost write time.
+        covered = any(idx['cols'][:len(col_list)] == col_list
+                      for idx in tbl_idx.values())
+        if covered:
+            continue
+
+        missing.append((table, name, cols, prio, why))
+        finding(prio, 'indexes',
+                f'{table}.{name} ({cols}) is missing',
+                f'{why}\nCREATE INDEX {name} ON {table} ({cols});')
+
+    if not missing:
+        ok('every recommended index is present')
+    return missing
+
+
+def check_redundant_indexes(existing, fk_names, fk_cols):
+    header('Redundant indexes')
+    found = False
+    for table, idxs in sorted(existing.items()):
+        names = sorted(idxs.keys())
+        for a in names:
+            if a == 'PRIMARY':
+                continue
+            for b in names:
+                if a == b or b == 'PRIMARY':
+                    continue
+                ca, cb = idxs[a]['cols'], idxs[b]['cols']
+                # a is redundant when b starts with all of a's columns and b is
+                # strictly longer.
+                if len(ca) >= len(cb) or cb[:len(ca)] != ca:
+                    continue
+                if idxs[a]['unique']:
+                    continue   # a unique index is a constraint, not a hint
+
+                protected = (a in fk_names.get(table, set())
+                             or (len(ca) == 1
+                                 and ca[0] in fk_cols.get(table, set())))
+                if protected:
+                    finding('info', 'indexes',
+                            f'{table}.{a} ({", ".join(ca)}) is a prefix of '
+                            f'{b} ({", ".join(cb)}) but backs a FOREIGN KEY -- '
+                            f'leave it alone (dropping it fails, errno 150).')
+                else:
+                    found = True
+                    finding('low', 'indexes',
+                            f'{table}.{a} ({", ".join(ca)}) is fully covered by '
+                            f'{b} ({", ".join(cb)}) -- it costs write time and '
+                            f'buffer-pool space for nothing.',
+                            '-- confirm against your slow log first, then:\n'
+                            f'DROP INDEX {a} ON {table};')
+                break
+    if not found:
+        ok('no droppable duplicate indexes')
+
+
+def check_unindexed_fks(conn, schema, existing):
+    header('Foreign keys without a usable index')
+    rows = conn.execute(text("""
+        SELECT TABLE_NAME, COLUMN_NAME
+        FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
+        WHERE TABLE_SCHEMA = :s AND REFERENCED_TABLE_NAME IS NOT NULL
+    """), {'s': schema}).fetchall()
+    bad = []
+    for tbl, col in rows:
+        idxs = existing.get(tbl, {})
+        if not any(i['cols'] and i['cols'][0] == col for i in idxs.values()):
+            bad.append((tbl, col))
+    if bad:
+        for tbl, col in bad:
+            finding('high', 'indexes',
+                    f'{tbl}.{col} is a foreign key with no leading index -- '
+                    f'every DELETE on the parent table scans this one.',
+                    f'CREATE INDEX ix_{tbl}_{col} ON {tbl} ({col});')
+    else:
+        ok('every foreign key column has a leading index')
+
+
+def check_explain(conn):
+    header('EXPLAIN on representative hot queries')
+    for label, sql, params in EXPLAIN_QUERIES:
+        try:
+            rows = conn.execute(text('EXPLAIN ' + sql), params).mappings().all()
+        except Exception as exc:
+            print(f'  {_c(C.DIM, "skip ")} {label}: '
+                  f'{str(exc).splitlines()[0][:90]}')
+            continue
+
+        problems = []
+        for r in rows:
+            typ = (r.get('type') or '').upper()
+            extra = r.get('Extra') or ''
+            tbl = r.get('table') or '?'
+            nrows = r.get('rows') or 0
+            if typ == 'ALL':
+                problems.append(f'full scan of {tbl} (~{nrows} rows)')
+            if 'Using filesort' in extra:
+                problems.append(f'filesort on {tbl}')
+            if 'Using temporary' in extra:
+                problems.append(f'temp table on {tbl}')
+
+        if problems:
+            finding('medium', 'query-plan',
+                    f'{label}: ' + '; '.join(sorted(set(problems))), sql)
+        else:
+            ok(label)
+
+
+# -- Mutating actions (opt-in only) ------------------------------------------
+
+def apply_indexes(conn, schema, missing):
+    header('Applying missing indexes')
+    if not missing:
+        ok('nothing to apply')
+        return
+    for table, name, cols, prio, why in missing:
+        # Re-check rather than trusting the earlier pass: CREATE INDEX IF NOT
+        # EXISTS is unavailable before MySQL 8.0.12 (CLAUDE.md rule 16).
+        if index_exists(conn, schema, table, name):
+            print(f'  {_c(C.DIM, "skip ")} {name} already exists')
+            continue
+        sql = f'CREATE INDEX {name} ON {table} ({cols})'
+        print(f'  {_c(C.BOLD, "run  ")} {sql}')
+        conn.execute(text(sql))
+        print(f'  {_c(C.OK, "done ")} {name}')
+
+
+def run_analyze(conn, tables):
+    header('ANALYZE TABLE (refresh optimizer statistics)')
+    for t in tables:
+        conn.execute(text(f'ANALYZE TABLE `{t["name"]}`'))
+        print(f'  {_c(C.OK, "done ")} {t["name"]}')
+
+
+def run_optimize(conn, tables):
+    header('OPTIMIZE TABLE (rebuild -- locks each table)')
+    for t in tables:
+        before = t['data'] + t['index']
+        conn.execute(text(f'OPTIMIZE TABLE `{t["name"]}`'))
+        print(f'  {_c(C.OK, "done ")} {t["name"]} (was {human_bytes(before)})')
+
+
+# -- Alembic migration emitter -----------------------------------------------
+
+MIGRATION_TEMPLATE = '''"""Performance indexes -- generated by scripts/db_health.py
+
+Adds the indexes the health check found missing. Every statement is guarded by
+an INFORMATION_SCHEMA lookup, so this migration is safe to re-run (CLAUDE.md
+rule 16: CREATE INDEX IF NOT EXISTS needs MySQL >= 8.0.12).
+
+Generated: {generated}
+
+Revision ID: {revision}
+Revises: {down_revision}
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# Revision ids must be <= 32 chars -- alembic_version.version_num is VARCHAR(32).
+revision      = '{revision}'
+down_revision = '{down_revision}'
+branch_labels = None
+depends_on    = None
+
+
+INDEXES = [
+{index_rows}
+]
+
+
+def _index_exists(conn, table, name):
+    return conn.execute(sa.text("""
+        SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t AND INDEX_NAME = :i
+    """), {{'t': table, 'i': name}}).scalar() > 0
+
+
+def upgrade():
+    conn = op.get_bind()
+    for table, name, cols in INDEXES:
+        if not _index_exists(conn, table, name):
+            op.execute(sa.text(f'CREATE INDEX {{name}} ON {{table}} ({{cols}})'))
+
+
+def downgrade():
+    conn = op.get_bind()
+    for table, name, cols in INDEXES:
+        if _index_exists(conn, table, name):
+            op.execute(sa.text(f'DROP INDEX {{name}} ON {{table}}'))
+'''
+
+
+def emit_migration(path, missing, revision, down_revision):
+    rows = '\n'.join(
+        f'    ({table!r}, {name!r}, {cols!r}),'
+        for table, name, cols, prio, why in missing
+    ) or '    # nothing was missing at generation time'
+    body = MIGRATION_TEMPLATE.format(
+        generated=datetime.now().strftime('%Y-%m-%d %H:%M'),
+        revision=revision,
+        down_revision=down_revision,
+        index_rows=rows,
+    )
+    with open(path, 'w', encoding='utf-8') as fh:
+        fh.write(body)
+    print(f'\n  {_c(C.OK, "wrote")} {path}')
+    print(_c(C.DIM, '        Review it, confirm down_revision is the real head '
+                    '(flask db heads), then run flask db upgrade.'))
+
+
+def current_head():
+    """Best-effort read of the current Alembic head from the versions dir."""
+    vdir = os.path.join(os.path.dirname(__file__), '..', 'migrations', 'versions')
+    revs, downs = set(), set()
+    try:
+        for fn in os.listdir(vdir):
+            if not fn.endswith('.py'):
+                continue
+            with open(os.path.join(vdir, fn), encoding='utf-8') as fh:
+                src = fh.read()
+            m = re.search(r"^revision\s*=\s*['\"]([^'\"]+)", src, re.M)
+            d = re.search(r"^down_revision\s*=\s*['\"]([^'\"]+)", src, re.M)
+            if m:
+                revs.add(m.group(1))
+            if d:
+                downs.add(d.group(1))
+    except OSError:
+        return '<REPLACE_WITH_CURRENT_HEAD>'
+    heads = revs - downs
+    return sorted(heads)[0] if len(heads) == 1 else '<REPLACE_WITH_CURRENT_HEAD>'
+
+
+# -- Main --------------------------------------------------------------------
+
+def main():
+    p = argparse.ArgumentParser(
+        description='MySQL health check and index optimizer for JQC. '
+                    'Read-only unless an --apply/--analyze/--optimize flag is '
+                    'given.')
+    p.add_argument('--json', metavar='PATH', help='write the full report as JSON')
+    p.add_argument('--no-explain', action='store_true',
+                   help='skip the EXPLAIN pass')
+    p.add_argument('--apply-indexes', action='store_true',
+                   help='CREATE the missing recommended indexes (idempotent)')
+    p.add_argument('--analyze', action='store_true',
+                   help='ANALYZE TABLE on every table (safe, non-locking)')
+    p.add_argument('--optimize', action='store_true',
+                   help='OPTIMIZE TABLE on every table -- LOCKS each table; '
+                        'needs --yes')
+    p.add_argument('--emit-migration', metavar='PATH',
+                   help='write an Alembic migration for the missing indexes')
+    p.add_argument('--revision', default='phaseNN_perf_indexes',
+                   help='revision id for --emit-migration (max 32 chars)')
+    p.add_argument('--yes', action='store_true',
+                   help='confirm locking operations')
+    p.add_argument('--fail-on', choices=['none', 'low', 'medium', 'high'],
+                   default='none',
+                   help='exit 1 when a finding of this severity or above exists')
+    args = p.parse_args()
+
+    if args.optimize and not args.yes:
+        print('--optimize rebuilds every table and holds a lock for the '
+              'duration. Re-run with --yes inside a maintenance window.')
+        return 2
+    if len(args.revision) > 32:
+        print('--revision must be <= 32 characters (alembic_version.version_num '
+              'is VARCHAR(32)).')
+        return 2
+
+    app = create_app(os.getenv('FLASK_ENV') or 'production')
+
+    with app.app_context():
+        engine_opts = check_engine_options(app)
+
+        try:
+            conn = db.engine.connect()
+        except Exception as exc:
+            print(_c(C.ERR, f'\nCould not connect to the database: {exc}'))
+            return 2
+
+        with conn:
+            if not is_mysql(conn):
+                print(_c(C.ERR,
+                         f'\nThis script targets MySQL/MariaDB; the configured '
+                         f'database is "{conn.dialect.name}". Nothing to do.'))
+                return 2
+
+            schema = db_name(conn)
+            print(_c(C.BOLD, f'\nJQC database health check -- schema "{schema}" '
+                             f'@ {datetime.now():%Y-%m-%d %H:%M}'))
+
+            tables = fetch_tables(conn, schema)
+            existing = fetch_indexes(conn, schema)
+            fk_names, fk_cols = fetch_fk_protection(conn, schema)
+
+            server = check_server(conn, engine_opts, tables)
+            check_schema_hygiene(conn, schema, tables)
+            check_table_footprint(tables)
+            missing = check_missing_indexes(conn, schema, existing)
+            check_redundant_indexes(existing, fk_names, fk_cols)
+            check_unindexed_fks(conn, schema, existing)
+            if not args.no_explain:
+                check_explain(conn)
+
+            # -- opt-in mutations ---------------------------------------------
+            if args.apply_indexes:
+                with conn.begin():
+                    apply_indexes(conn, schema, missing)
+            if args.analyze:
+                run_analyze(conn, tables)
+            if args.optimize:
+                run_optimize(conn, tables)
+
+            if args.emit_migration:
+                emit_migration(args.emit_migration, missing,
+                               args.revision, current_head())
+
+        # -- summary ---------------------------------------------------------
+        header('Summary')
+        counts = {'high': 0, 'medium': 0, 'low': 0, 'info': 0}
+        for f in FINDINGS:
+            counts[f['severity']] += 1
+        print(f'  {_c(C.ERR, "high")} {counts["high"]}   '
+              f'{_c(C.WARN, "medium")} {counts["medium"]}   '
+              f'{_c(C.DIM, "low")} {counts["low"]}   '
+              f'{_c(C.DIM, "info")} {counts["info"]}')
+        if not (args.apply_indexes or args.analyze or args.optimize):
+            print(_c(C.DIM, '  (read-only run -- nothing was modified)'))
+
+        if args.json:
+            report = {
+                'generated_at': datetime.now().isoformat(timespec='seconds'),
+                'schema': schema,
+                'server': server,
+                'engine_options': {k: str(v) for k, v in (engine_opts or {}).items()},
+                'tables': tables,
+                'missing_indexes': [
+                    {'table': t, 'name': n, 'columns': c,
+                     'priority': pr, 'why': w}
+                    for t, n, c, pr, w in missing
+                ],
+                'findings': FINDINGS,
+                'counts': counts,
+            }
+            with open(args.json, 'w', encoding='utf-8') as fh:
+                json.dump(report, fh, indent=2, default=str)
+            print(f'  report written to {args.json}')
+
+        if args.fail_on != 'none':
+            threshold = SEVERITY_ORDER[args.fail_on]
+            if any(SEVERITY_ORDER[f['severity']] >= threshold for f in FINDINGS):
+                return 1
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/test_scoped_pages_smoke.py

@@ -0,0 +1,205 @@
+"""
+Smoke coverage for the pages whose queries were rewritten during the database
+optimization pass: the dashboard, the issues list, and the reports overview.
+
+These routes had no request-level tests, so a query rewrite that produced
+invalid SQL or a broken scope would only have surfaced in production. Each test
+here drives the real route through the test client for a role that exercises a
+DIFFERENT branch of the scoping code:
+
+  admin      -- unscoped branch
+  inspector  -- InspectorAssignment / get_inspector_scope() branch
+  customer   -- CustomerAssignment / get_customer_scope() branch
+
+The assertions are deliberately about behaviour that must hold (status code,
+scope isolation), not about markup.
+"""
+from datetime import timedelta
+
+import pytest
+
+from app import db as _db
+from app.models.issue import Issue
+from app.models.project import Project, CustomerAssignment
+from app.models.inspector_assignment import InspectorAssignment
+from app.utils.time_utils import now_eastern
+
+
+@pytest.fixture
+def make_issue(db):
+    def _make(facility=None, area=None, severity='high', status='open',
+              hours_ago=1, **kw):
+        issue = Issue(
+            facility_id=facility.id if facility else None,
+            area_id=area.id if area else None,
+            severity=severity,
+            description=kw.pop('description', 'test issue'),
+            status=status,
+            reported_at=now_eastern() - timedelta(hours=hours_ago),
+        )
+        for k, v in kw.items():
+            setattr(issue, k, v)
+        db.session.add(issue)
+        db.session.commit()
+        return issue
+    return _make
+
+
+@pytest.fixture
+def project(db):
+    proj = Project(name='Contract A', active=True)
+    db.session.add(proj)
+    db.session.commit()
+    return proj
+
+
+@pytest.fixture
+def other_project(db):
+    proj = Project(name='Contract B', active=True)
+    db.session.add(proj)
+    db.session.commit()
+    return proj
+
+
+# ── Dashboard ────────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize('role', ['admin', 'director', 'project_manager', 'auditor'])
+def test_dashboard_renders_for_staff_roles(client, login, make_user,
+                                           make_facility, make_issue,
+                                           project, role):
+    """The card queries (severity / handler / SLA splits) build and run."""
+    facility = make_facility(project=project)
+    make_issue(facility=facility, severity='critical', hours_ago=100)
+    make_issue(facility=facility, severity='low', status='in_progress')
+    make_issue(facility=facility, status='resolved',
+               resolved_at=now_eastern())
+
+    login(make_user(role=role))
+    assert client.get('/').status_code == 200
+
+
+def test_dashboard_renders_for_scoped_inspector(client, login, make_user, db,
+                                                make_facility, make_area,
+                                                make_issue, project):
+    inspector = make_user(role='inspector')
+    db.session.add(InspectorAssignment(user_id=inspector.id,
+                                       project_id=project.id))
+    db.session.commit()
+
+    facility = make_facility(project=project)
+    area = make_area(facility)
+    make_issue(area=area, severity='high', hours_ago=50)   # reached via area
+    make_issue(facility=facility, severity='medium')       # reached directly
+
+    login(inspector)
+    assert client.get('/').status_code == 200
+
+
+def test_dashboard_renders_for_inspector_with_no_assignments(client, login,
+                                                             make_user):
+    """Strict scoping: no assignments must render an empty dashboard, not 500."""
+    login(make_user(role='inspector'))
+    assert client.get('/').status_code == 200
+
+
+def test_dashboard_renders_for_customer(client, login, make_user, db,
+                                        make_facility, make_issue, project):
+    customer = make_user(role='customer')
+    facility = make_facility(project=project)
+    db.session.add(CustomerAssignment(user_id=customer.id,
+                                      project_id=project.id))
+    db.session.commit()
+    make_issue(facility=facility)
+
+    login(customer)
+    assert client.get('/').status_code == 200
+
+
+# ── Issues list ──────────────────────────────────────────────────────────────
+
+def test_issues_list_renders(client, login, make_user, make_facility,
+                             make_issue, project):
+    facility = make_facility(project=project)
+    make_issue(facility=facility)
+    login(make_user(role='admin'))
+    assert client.get('/issues/').status_code == 200
+
+
+@pytest.mark.parametrize('query', [
+    '?facility_id=abc',            # non-numeric -- used to raise ValueError -> 500
+    '?facility_id=',
+    '?facility_id=999999',         # numeric but nonexistent
+    '?contract_id=abc',
+    '?issue_id=abc',
+    '?severity=high&status=open',
+    '?handler_type=vendor',
+    '?unassigned=1',
+    '?sla=breached',
+    '?date_from=notadate&date_to=alsonot',
+])
+def test_issues_list_survives_malformed_filters(client, login, make_user,
+                                                make_facility, make_issue,
+                                                project, query):
+    """A hand-edited or stale query string must never 500 the list."""
+    facility = make_facility(project=project)
+    make_issue(facility=facility)
+    login(make_user(role='admin'))
+    assert client.get('/issues/' + query).status_code == 200
+
+
+def test_issues_list_scopes_a_customer_to_their_own_facilities(
+        client, login, make_user, db, make_facility, make_issue,
+        project, other_project):
+    """The scope filter still isolates customers after the query rewrite."""
+    customer = make_user(role='customer')
+    mine = make_facility(project=project, name='Mine')
+    theirs = make_facility(project=other_project, name='Theirs')
+    db.session.add(CustomerAssignment(user_id=customer.id,
+                                      project_id=project.id))
+    db.session.commit()
+
+    make_issue(facility=mine, description='visible to me')
+    make_issue(facility=theirs, description='other customer only')
+
+    login(customer)
+    body = client.get('/issues/').get_data(as_text=True)
+    assert 'visible to me' in body
+    assert 'other customer only' not in body
+
+
+# ── Reports overview ─────────────────────────────────────────────────────────
+
+def test_reports_overview_renders_for_admin(client, login, make_user,
+                                            make_facility, make_template,
+                                            make_inspection, project):
+    admin = make_user(role='admin')
+    inspector = make_user(role='inspector')
+    facility = make_facility(project=project)
+    make_inspection(facility, inspector, make_template())
+
+    login(admin)
+    assert client.get('/reports/').status_code == 200
+
+
+def test_reports_overview_inspector_scope_uses_subquery(
+        client, login, make_user, db, make_facility, make_template,
+        make_inspection, make_issue, project):
+    """_scope_issue() now filters via a subquery instead of an id list."""
+    inspector = make_user(role='inspector')
+    db.session.add(InspectorAssignment(user_id=inspector.id,
+                                       project_id=project.id))
+    db.session.commit()
+
+    facility = make_facility(project=project)
+    insp = make_inspection(facility, inspector, make_template())
+    make_issue(facility=facility, inspection_id=insp.id)
+
+    login(inspector)
+    assert client.get('/reports/').status_code == 200
+
+
+def test_reports_overview_inspector_with_no_inspections(client, login,
+                                                        make_user):
+    """IN (empty subquery) must match nothing rather than error."""
+    login(make_user(role='inspector'))
+    assert client.get('/reports/').status_code == 200

tests/test_sla_candidate_query.py

@@ -0,0 +1,165 @@
+"""
+Equivalence test for the SQL prefilter in send_sla_alerts().
+
+The cron used to read EVERY open issue and decide in Python which ones deserved
+a notification. It now narrows to candidates in SQL first.
+
+The contract is a CONSERVATIVE SUPERSET, not equality:
+
+  * Nothing the old loop would have notified may be missed. This is the safety
+    property -- a miss is a silently unsent SLA alert.
+  * The SQL may return extra rows, because it deliberately does not replicate
+    the "already notified at_risk and still only at_risk" skip (that would mean
+    writing the per-severity deadline arithmetic a second time, in SQL). The
+    Python loop still applies that skip, so no extra notification is ever sent
+    -- only a handful of extra rows are read.
+
+The reference predicate below is the old loop's skip logic, written out.
+"""
+from datetime import timedelta
+
+import pytest
+
+from app import db
+from app.models.issue import Issue
+from app.utils.sla import SLA_HOURS, AT_RISK_THRESHOLD, sla_status
+from app.utils.time_utils import now_eastern
+
+
+def _candidate_query():
+    """The prefilter exactly as send_sla_alerts() builds it."""
+    now = now_eastern()
+    age_clauses = [
+        db.and_(
+            Issue.severity == severity,
+            Issue.reported_at <= now - timedelta(hours=hours * AT_RISK_THRESHOLD),
+        )
+        for severity, hours in SLA_HOURS.items()
+    ]
+    return Issue.query.filter(
+        Issue.status.in_(['open', 'in_progress', 'pending_verification']),
+        Issue.reported_at.isnot(None),
+        db.or_(Issue.sla_notified.is_(None), Issue.sla_notified != 'breached'),
+        db.or_(*age_clauses),
+    )
+
+
+def _old_loop_would_act(issue):
+    """The pre-change Python logic: which rows survived to send a notification."""
+    if issue.status not in ('open', 'in_progress', 'pending_verification'):
+        return False
+    if issue.reported_at is None:
+        return False          # would have raised TypeError -- see the NULL test
+    status = sla_status(issue)
+    if status not in ('at_risk', 'breached'):
+        return False
+    already = issue.sla_notified
+    if already == 'breached':
+        return False
+    if already == 'at_risk' and status == 'at_risk':
+        return False
+    return True
+
+
+def _add(severity, hours_ago, status='open', sla_notified=None):
+    issue = Issue(
+        severity=severity,
+        description='x',
+        status=status,
+        sla_notified=sla_notified,
+        reported_at=now_eastern() - timedelta(hours=hours_ago),
+    )
+    db.session.add(issue)
+    return issue
+
+
+@pytest.fixture
+def population(app):
+    """One issue per interesting combination of severity / age / state."""
+    rows = []
+    for severity, window in SLA_HOURS.items():
+        at_risk_at = window * AT_RISK_THRESHOLD
+        rows += [
+            _add(severity, 0),                       # fresh -> ok
+            _add(severity, at_risk_at * 0.5),        # halfway -> ok
+            _add(severity, at_risk_at + 1),          # at risk
+            _add(severity, window + 1),              # breached
+            # already-notified variants
+            _add(severity, at_risk_at + 1, sla_notified='at_risk'),
+            _add(severity, window + 1, sla_notified='at_risk'),
+            _add(severity, window + 1, sla_notified='breached'),
+            # non-actionable / other statuses
+            _add(severity, window + 1, status='resolved'),
+            _add(severity, window + 1, status='in_progress'),
+            _add(severity, window + 1, status='pending_verification'),
+        ]
+    db.session.commit()
+    return rows
+
+
+def test_prefilter_never_misses_an_actionable_issue(population):
+    """Safety property: every row the old loop notified is still selected."""
+    expected = {i.id for i in population if _old_loop_would_act(i)}
+    actual = {i.id for i in _candidate_query().all()}
+
+    assert expected, 'fixture built no actionable issues -- test is vacuous'
+    assert expected <= actual, (
+        'the prefilter drops issues the old loop would have alerted on: '
+        f'{sorted(expected - actual)}'
+    )
+
+
+def test_prefilter_extras_are_all_skipped_by_the_loop(population):
+    """The extra rows the SQL lets through produce no extra notifications.
+
+    Each must be a row the Python loop independently skips, so the set of alerts
+    actually sent is unchanged.
+    """
+    by_id = {i.id: i for i in population}
+    expected = {i.id for i in population if _old_loop_would_act(i)}
+    extras = {i.id for i in _candidate_query().all()} - expected
+
+    for iid in extras:
+        issue = by_id[iid]
+        assert not _old_loop_would_act(issue)
+        # and the only reason it is allowed through is the at_risk bookkeeping
+        assert issue.sla_notified == 'at_risk' and sla_status(issue) == 'at_risk', (
+            f'issue {iid} is an unexplained extra: status={issue.status} '
+            f'severity={issue.severity} sla_notified={issue.sla_notified} '
+            f'sla_status={sla_status(issue)}'
+        )
+
+
+def test_prefilter_excludes_the_bulk_of_open_issues(population):
+    """The point of the change: most open issues are never read."""
+    total_open = Issue.query.filter(
+        Issue.status.in_(['open', 'in_progress', 'pending_verification'])
+    ).count()
+    candidates = _candidate_query().count()
+    assert candidates < total_open
+
+
+def test_null_reported_at_is_excluded_not_crashed(app):
+    """reported_at is nullable and sla_status() raises TypeError on NULL.
+
+    Before the prefilter one such row aborted the whole cron run; it must now be
+    excluded in SQL instead.
+    """
+    issue = Issue(severity='high', description='x', status='open')
+    db.session.add(issue)
+    db.session.commit()
+
+    # The column carries default=now_eastern, so an ORM insert can never leave
+    # it NULL -- the row has to be forced, which is exactly how a legacy or
+    # raw-SQL-inserted row would look.
+    db.session.execute(
+        db.text('UPDATE issues SET reported_at = NULL WHERE id = :i'),
+        {'i': issue.id},
+    )
+    db.session.commit()
+    db.session.expire(issue)
+
+    assert issue.reported_at is None
+    assert issue.id not in {i.id for i in _candidate_query().all()}
+    with pytest.raises(TypeError):
+        sla_status(issue)      # confirms the crash the filter is avoiding