fix(backend): serve the current Ofsted status from fact_ofsted_latest

The list query and the batch Ofsted fetch each picked 'the latest row' of fact_ofsted_inspection themselves, and _ofsted_block carried an older ungraded grade forward when the latest graded inspection gave none. All three now read marts.fact_ofsted_latest. List rows: ofsted_grade is the grade still in force, ofsted_grade_date when it was awarded or confirmed, ofsted_date the latest visit. The ofsted block gains current_grade and latest_visit and loses grade_source; overall_effectiveness is the graded inspection's own result. A school with only an inspection stays publishable in the sitemap.

Requires fact_ofsted_latest (PR #183's pipeline run) on the database.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/app.py

@@ -97,8 +97,9 @@ STATIC_SITEMAP_PATHS = ("/", "/rankings", "/compare", "/admissions")
 
 # A page has something a search result could state if any of these is present
 # in any year. Shared by _has_publishable_data and the per-school check in
-# _school_sitemap_rows so the two can never drift.
-_PUBLISHABLE_FIELDS = ("rwm_expected_pct", "attainment_8_score", "ofsted_grade")
+# _school_sitemap_rows so the two can never drift. An inspection with no
+# overall grade (ofsted_date alone) still gives a page something to state.
+_PUBLISHABLE_FIELDS = ("rwm_expected_pct", "attainment_8_score", "ofsted_grade", "ofsted_date")
 
 
 def _has_publishable_data(row) -> bool:

backend/data_loader.py

@@ -18,7 +18,7 @@ from .config import settings
 from .database import SessionLocal, engine
 from .models import (
     DimSchool, DimLocation, KS2Performance,
-    FactOfstedInspection, FactAdmissions, FactAdmissionDistance,
+    FactOfstedLatest, FactAdmissions, FactAdmissionDistance,
     FactDeprivation, FactFinance, FactPupilCharacteristics,
     FactKs4Destinations, FactKs5Destinations,
 )
@@ -251,10 +251,11 @@ _MAIN_QUERY = text("""
         s.website,
         s.telephone,
         s.nursery_provision,
-        foi.ofsted_grade,
-        foi.ofsted_date,
-        foi.ofsted_framework,
-        foi.ofsted_rc_date,
+        foi.current_grade         AS ofsted_grade,
+        foi.current_grade_date    AS ofsted_grade_date,
+        foi.latest_visit_date     AS ofsted_date,
+        foi.framework             AS ofsted_framework,
+        foi.rc_inspection_date    AS ofsted_rc_date,
         l.local_authority_name AS local_authority,
         l.local_authority_code,
         l.address_line1        AS address1,
@@ -335,21 +336,10 @@ _MAIN_QUERY = text("""
     FROM marts.dim_school s
     JOIN marts.dim_location l ON s.urn = l.urn
     LEFT JOIN marts.fact_performance p ON s.urn = p.urn
-    LEFT JOIN (
-        SELECT DISTINCT ON (urn)
-            urn,
-            -- Fall back to the ungraded-inspection grade when no graded grade exists.
-            COALESCE(overall_effectiveness, ungraded_grade) AS ofsted_grade,
-            inspection_date       AS ofsted_date,
-            framework             AS ofsted_framework,
-            -- Report-card signal for list/map badges: non-null only when the
-            -- latest inspection carries report-card grades. framework is the
-            -- raw event grouping ("Schools - S5"), never "ReportCard", so it
-            -- can't be used to detect report cards.
-            rc_inspection_date    AS ofsted_rc_date
-        FROM marts.fact_ofsted_inspection
-        ORDER BY urn, inspection_date DESC NULLS LAST
-    ) foi ON s.urn = foi.urn
+    -- One current Ofsted status per school (pipeline: int_ofsted_latest): the
+    -- grade still in force, dated by the inspection that awarded or confirmed
+    -- it, and the latest visit of any kind.
+    LEFT JOIN marts.fact_ofsted_latest foi ON s.urn = foi.urn
     ORDER BY s.school_name, p.year
 """)
 
@@ -731,39 +721,36 @@ def compute_benchmarks(df: pd.DataFrame, census_benchmarks: dict | None = None)
     }
 
 
+def _iso(d):
+    return d.isoformat() if d else None
+
+
 def _ofsted_block(o, urn: int) -> dict:
-    """Serialize the latest Ofsted inspection row for API responses.
+    """Serialize a fact_ofsted_latest row for API responses.
 
-    `grade_source` records where the effective overall grade came from:
-    a graded (Section 5) inspection, or carried forward from an ungraded
-    (Section 8) outcome — materially different claims a UI must be able
-    to distinguish. `report_card` holds coded+labelled renewed-framework
-    (Nov 2025) area judgements; safeguarding is a separate boolean and
-    never appears among the graded areas.
+    `current_grade` is the overall grade still in force, dated by the
+    inspection that awarded or confirmed it; `latest_visit` is the school's
+    most recent inspection of any kind. The rule lives in int_ofsted_latest
+    (docs/superpowers/specs/2026-10-05-ofsted-current-status-design.md).
+    `overall_effectiveness` and `inspection_date` describe the graded
+    inspection itself and label its area judgements. `report_card` holds the
+    renewed-framework (Nov 2025) area judgements; safeguarding is a separate
+    boolean and never appears among the graded areas.
     """
-    if o.overall_effectiveness is not None:
-        grade_source = "graded"
-        overall = o.overall_effectiveness
-    elif o.ungraded_grade is not None:
-        # Fall back to the grade parsed from an ungraded (Section 8) outcome
-        # (e.g. "School remains Good") so the detail page matches the list badge.
-        grade_source = "ungraded_carried_forward"
-        overall = o.ungraded_grade
-    else:
-        grade_source = None
-        overall = None
-
     block = {
         "framework": o.framework,
-        "inspection_date": o.inspection_date.isoformat() if o.inspection_date else None,
-        "rc_inspection_date": (
-            o.rc_inspection_date.isoformat()
-            if getattr(o, "rc_inspection_date", None)
-            else None
-        ),
+        "inspection_date": _iso(o.graded_inspection_date),
+        "rc_inspection_date": _iso(o.rc_inspection_date),
         "inspection_type": o.inspection_type,
-        "overall_effectiveness": overall,
-        "grade_source": grade_source,
+        "overall_effectiveness": o.overall_effectiveness if o.overall_effectiveness in (1, 2, 3, 4) else None,
+        "current_grade": (
+            {"grade": o.current_grade, "date": _iso(o.current_grade_date), "basis": o.current_grade_basis}
+            if o.current_grade is not None else None
+        ),
+        "latest_visit": (
+            {"date": _iso(o.latest_visit_date), "kind": o.latest_visit_kind, "outcome": o.latest_visit_outcome}
+            if o.latest_visit_date else None
+        ),
         "quality_of_education": o.quality_of_education,
         "behaviour_attitudes": o.behaviour_attitudes,
         "personal_development": o.personal_development,
@@ -1097,20 +1084,14 @@ def get_supplementary_data_batch(db: Session, urns: list[int]) -> dict:
             logging.getLogger(__name__).error("batch supplementary query failed: %s", e)
             db.rollback()
 
-    # Ofsted — latest inspection per URN. Ordered so the first row seen per
-    # URN is the most recent.
+    # Ofsted — the mart already holds one current status per URN.
     def _ofsted():
         rows = (
-            db.query(FactOfstedInspection)
-            .filter(FactOfstedInspection.urn.in_(urns))
-            .order_by(FactOfstedInspection.urn, FactOfstedInspection.inspection_date.desc())
+            db.query(FactOfstedLatest)
+            .filter(FactOfstedLatest.urn.in_(urns))
             .all()
         )
-        seen = set()
         for o in rows:
-            if o.urn in seen:
-                continue
-            seen.add(o.urn)
             result[o.urn]["ofsted"] = _ofsted_block(o, o.urn)
     _safe(_ofsted)
 

backend/models.py

@@ -162,6 +162,49 @@ class FactOfstedInspection(Base):
     report_url = Column(Text)
 
 
+class FactOfstedLatest(Base):
+    """Current Ofsted status — one row per URN (pipeline: int_ofsted_latest).
+
+    `current_grade` is the overall grade still in force, dated by the
+    inspection that awarded or confirmed it; `latest_visit_*` is the school's
+    most recent inspection of any kind.
+    """
+    __tablename__ = "fact_ofsted_latest"
+    __table_args__ = MARTS
+
+    urn = Column(Integer, primary_key=True)
+    latest_visit_date = Column(Date)
+    latest_visit_kind = Column(String(20))
+    latest_visit_outcome = Column(String(100))
+    current_grade = Column(Integer)
+    current_grade_date = Column(Date)
+    current_grade_basis = Column(String(20))
+    graded_inspection_date = Column(Date)
+    ungraded_inspection_date = Column(Date)
+    rc_inspection_date = Column(Date)
+    inspection_type = Column(String(100))
+    framework = Column(String(20))
+    overall_effectiveness = Column(Integer)
+    quality_of_education = Column(Integer)
+    behaviour_attitudes = Column(Integer)
+    personal_development = Column(Integer)
+    leadership_management = Column(Integer)
+    early_years_provision = Column(Integer)
+    sixth_form_provision = Column(Integer)
+    ungraded_outcome = Column(String(100))
+    ungraded_grade = Column(Integer)
+    rc_safeguarding_met = Column(Boolean)
+    rc_inclusion = Column(Integer)
+    rc_curriculum_teaching = Column(Integer)
+    rc_achievement = Column(Integer)
+    rc_attendance_behaviour = Column(Integer)
+    rc_personal_development = Column(Integer)
+    rc_leadership_governance = Column(Integer)
+    rc_early_years = Column(Integer)
+    rc_sixth_form = Column(Integer)
+    report_url = Column(Text)
+
+
 class FactAdmissions(Base):
     """School admissions — one row per URN per year."""
     __tablename__ = "fact_admissions"

backend/schemas.py

@@ -574,6 +574,7 @@ SCHOOL_COLUMNS = [
     "gender",
     "admissions_policy",
     "ofsted_grade",
+    "ofsted_grade_date",
     "ofsted_date",
     "ofsted_framework",
     "ofsted_rc_date",

backend/tests/test_compare_enrichment.py

@@ -12,7 +12,8 @@ from fastapi.testclient import TestClient
 LATEST = 202425
 
 CANNED_SUPPLEMENTARY = {
-    "ofsted": {"overall_effectiveness": 2, "grade_source": "graded",
+    "ofsted": {"overall_effectiveness": 2,
+               "current_grade": {"grade": 2, "date": "2023-01-01", "basis": "graded"},
                "report_card": {}, "ofsted_page_url": "https://reports.ofsted.gov.uk/provider/21/100140"},
     "census": {"year": 202526, "fsm_pct": 29.8},
     "admissions": {"year": 202627, "second_preference_offers": 4},
@@ -86,7 +87,7 @@ def test_each_school_gains_supplementary_blocks(client):
     body = client.get("/api/compare?urns=100140,138690").json()
     for urn in ("100140", "138690"):
         school = body["comparison"][urn]
-        assert school["ofsted"]["grade_source"] == "graded"
+        assert school["ofsted"]["current_grade"]["basis"] == "graded"
         assert school["census"]["fsm_pct"] == 29.8
         assert school["admissions"]["second_preference_offers"] == 4
         assert school["admissions_history"][0]["year"] == 202627

backend/tests/test_ofsted_status_payload.py

@@ -0,0 +1,56 @@
+"""The search badge reads ofsted_grade, ofsted_grade_date and ofsted_date from
+list rows (nextjs-app/lib/utils.ts buildOfstedListBadge). A field the list
+never sends would leave the badge without its year, or worse, fall back to
+"Not yet inspected". The school page reads current_grade and latest_visit from
+the ofsted block (lib/ofstedStatus.ts)."""
+
+import numpy as np
+import pandas as pd
+import pytest
+from fastapi.testclient import TestClient
+
+from backend.schemas import SCHOOL_COLUMNS
+
+
+def test_list_columns_include_the_status_fields():
+    for field in ("ofsted_grade", "ofsted_grade_date", "ofsted_date", "ofsted_rc_date"):
+        assert field in SCHOOL_COLUMNS
+
+
+def _df() -> pd.DataFrame:
+    # Rabbsfarm (102408): latest inspection 17 June 2025 gave no overall grade.
+    return pd.DataFrame([{
+        "urn": 102408, "school_name": "Rabbsfarm Primary School", "phase": "Primary",
+        "school_type": "Community school", "local_authority": "Hillingdon",
+        "address": "Gordon Road, Yiewsley, UB7 8AH", "postcode": "UB7 8AH",
+        "latitude": 51.51, "longitude": -0.47, "year": 202425, "rwm_expected_pct": 58.0,
+        "total_pupils": 60, "gias_total_pupils": 616,
+        "ofsted_grade": np.nan, "ofsted_grade_date": None, "ofsted_date": "2025-06-17",
+        "ofsted_framework": "Schools - S5", "ofsted_rc_date": None,
+    }])
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    from backend import app as app_module
+
+    monkeypatch.setattr(app_module, "load_school_data", _df)
+    monkeypatch.setattr(app_module, "load_latest_school_data", _df)
+    monkeypatch.setattr(app_module, "_place_registry", None)
+    return TestClient(app_module.app, raise_server_exceptions=False)
+
+
+def test_search_rows_carry_the_status_fields(client):
+    resp = client.get("/api/schools")
+    assert resp.status_code == 200, resp.text
+    row = resp.json()["schools"][0]
+    assert row["ofsted_grade"] is None
+    assert row["ofsted_date"] == "2025-06-17"
+    assert "ofsted_grade_date" in row
+
+
+def test_a_school_with_only_an_inspection_is_publishable():
+    from backend.app import _has_publishable_data
+
+    assert _has_publishable_data({"rwm_expected_pct": None, "attainment_8_score": None,
+                                  "ofsted_grade": None, "ofsted_date": "2025-06-17"})

backend/tests/test_supplementary_batch.py

@@ -85,12 +85,15 @@ def _ofsted_row(urn, date, oe):
         "framework", "inspection_type", "quality_of_education", "behaviour_attitudes",
         "personal_development", "leadership_management", "early_years_provision",
         "sixth_form_provision", "ungraded_outcome", "ungraded_grade",
+        "ungraded_inspection_date", "rc_inspection_date", "latest_visit_outcome",
         "rc_safeguarding_met", "rc_inclusion", "rc_curriculum_teaching", "rc_achievement",
         "rc_attendance_behaviour", "rc_personal_development", "rc_leadership_governance",
         "rc_early_years", "rc_sixth_form", "report_url",
     )}
-    base.update(urn=urn, inspection_date=types.SimpleNamespace(isoformat=lambda: date),
-                overall_effectiveness=oe, grade_source=None)
+    when = types.SimpleNamespace(isoformat=lambda: date)
+    base.update(urn=urn, graded_inspection_date=when, latest_visit_date=when,
+                latest_visit_kind="graded", overall_effectiveness=oe,
+                current_grade=oe, current_grade_date=when, current_grade_basis="graded")
     return types.SimpleNamespace(**base)
 
 
@@ -114,10 +117,9 @@ def _dist_row(urn, year, distance_m, route_count=1):
 
 def test_one_query_per_table_and_latest_row_per_urn():
     rows = {
-        # URN 1 has two Ofsted rows; the batch must keep the most recent (2023).
-        "FactOfstedInspection": [
+        # The mart holds one current Ofsted status per URN.
+        "FactOfstedLatest": [
             _ofsted_row(1, "2023-01-01", 2),
-            _ofsted_row(1, "2019-01-01", 3),
             _ofsted_row(2, "2021-06-01", 1),
         ],
         "FactAdmissions": [_adm_row(1, 202526), _adm_row(1, 202627), _adm_row(2, 202627)],
@@ -142,14 +144,14 @@ def test_one_query_per_table_and_latest_row_per_urn():
     assert sorted(session.queries) == [
         "FactAdmissionDistance", "FactAdmissions", "FactDeprivation",
         "FactFinance", "FactKs4Destinations", "FactKs5Destinations",
-        "FactOfstedInspection", "FactPupilCharacteristics",
+        "FactOfstedLatest", "FactPupilCharacteristics",
     ]
 
     # A school with no destination rows gets null, not an empty shell — the
     # frontend renders the section from the block's presence.
     assert out[1]["destinations"] is None
 
-    # Latest Ofsted kept per URN
+    # Each URN's current Ofsted status
     assert out[1]["ofsted"]["overall_effectiveness"] == 2
     assert out[2]["ofsted"]["overall_effectiveness"] == 1
 
@@ -175,7 +177,7 @@ def test_one_query_per_table_and_latest_row_per_urn():
 
 
 def test_single_wrapper_matches_batch(monkeypatch):
-    session = _FakeSession({"FactOfstedInspection": [_ofsted_row(5, "2022-01-01", 2)]})
+    session = _FakeSession({"FactOfstedLatest": [_ofsted_row(5, "2022-01-01", 2)]})
     single = data_loader.get_supplementary_data(session, 5)
     assert single["ofsted"]["overall_effectiveness"] == 2
     assert single["admissions_history"] == []

backend/tests/test_supplementary_enrichment.py

@@ -10,7 +10,10 @@ from backend.data_loader import _admissions_row_dict, _ofsted_block
 
 def _row(**kw):
     base = dict(
-        framework="RC", inspection_date=None, inspection_type=None,
+        framework="RC", inspection_type=None,
+        graded_inspection_date=None, ungraded_inspection_date=None, rc_inspection_date=None,
+        latest_visit_date=None, latest_visit_kind=None, latest_visit_outcome=None,
+        current_grade=None, current_grade_date=None, current_grade_basis=None,
         overall_effectiveness=None, quality_of_education=None,
         behaviour_attitudes=None, personal_development=None,
         leadership_management=None, early_years_provision=None,
@@ -33,20 +36,41 @@ def test_report_card_block_and_provider_url():
     assert block["ofsted_page_url"] == "https://reports.ofsted.gov.uk/provider/21/100140"
 
 
-def test_grade_source_graded_vs_carried_forward():
-    assert _ofsted_block(_row(overall_effectiveness=1), urn=1)["grade_source"] == "graded"
-    carried = _ofsted_block(_row(ungraded_grade=2), urn=1)
-    assert carried["grade_source"] == "ungraded_carried_forward"
-    assert carried["overall_effectiveness"] == 2
-    assert _ofsted_block(_row(), urn=1)["grade_source"] is None
+def test_no_grade_is_carried_past_a_newer_inspection():
+    # Rabbsfarm (102408): the 2025 inspection gave no overall grade.
+    block = _ofsted_block(_row(
+        graded_inspection_date=date(2025, 6, 17), ungraded_inspection_date=date(2020, 2, 6),
+        latest_visit_date=date(2025, 6, 17), latest_visit_kind="graded",
+        ungraded_grade=2, ungraded_outcome="School remains Good", quality_of_education=3,
+    ), urn=102408)
+    assert block["current_grade"] is None
+    assert block["overall_effectiveness"] is None
+    assert block["latest_visit"] == {"date": "2025-06-17", "kind": "graded", "outcome": None}
+    assert block["inspection_date"] == "2025-06-17"
+    assert "grade_source" not in block
+
+
+def test_confirmed_grade_is_dated_by_the_confirming_visit():
+    block = _ofsted_block(_row(
+        graded_inspection_date=date(2020, 1, 7), ungraded_inspection_date=date(2024, 7, 18),
+        latest_visit_date=date(2024, 7, 18), latest_visit_kind="ungraded",
+        latest_visit_outcome="School remains Good", overall_effectiveness=2,
+        current_grade=2, current_grade_date=date(2024, 7, 18), current_grade_basis="confirmed",
+    ), urn=104762)
+    assert block["current_grade"] == {"grade": 2, "date": "2024-07-18", "basis": "confirmed"}
+    assert block["overall_effectiveness"] == 2
+    assert block["inspection_date"] == "2020-01-07"
+
+
+def test_overall_sentinel_is_not_served_as_a_grade():
+    assert _ofsted_block(_row(overall_effectiveness=9), urn=1)["overall_effectiveness"] is None
 
 
 def test_ofsted_block_carries_rc_inspection_date():
     o = _row(
-        ungraded_grade=2,
         rc_achievement=1,
         rc_inspection_date=date(2026, 2, 3),
-        inspection_date=date(2021, 10, 7),
+        graded_inspection_date=date(2021, 10, 7),
     )
     block = _ofsted_block(o, urn=138690)
     assert block["rc_inspection_date"] == "2026-02-03"
@@ -55,7 +79,7 @@ def test_ofsted_block_carries_rc_inspection_date():
 
 
 def test_ofsted_block_rc_inspection_date_none_when_absent():
-    o = _row(overall_effectiveness=1, inspection_date=date(2021, 10, 13))
+    o = _row(overall_effectiveness=1, graded_inspection_date=date(2021, 10, 13))
     block = _ofsted_block(o, urn=136276)
     assert block["rc_inspection_date"] is None
 
@@ -63,6 +87,7 @@ def test_ofsted_block_rc_inspection_date_none_when_absent():
 def test_ofsted_block_keeps_existing_keys():
     block = _ofsted_block(_row(overall_effectiveness=2, quality_of_education=2), urn=1)
     for key in ("framework", "inspection_date", "overall_effectiveness",
+                "current_grade", "latest_visit",
                 "quality_of_education", "rc_inclusion", "report_url"):
         assert key in block