fix(api): match missing-column fallbacks on the DBAPI error, not the statement

str(ProgrammingError) embeds the full SQL, which contains every column
name — the substring check matched any error and could take the wrong
retry branch. Parse the missing column from exc.orig instead.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/app.py

@@ -604,6 +604,7 @@ async def get_school_details(request: Request, urn: int):
             "religious_denomination": latest.get("religious_denomination", ""),
             "age_range": latest.get("age_range", ""),
             "has_sixth_form": latest.get("has_sixth_form"),
+            "status": latest.get("status"),
             "latitude": latest.get("latitude"),
             "longitude": latest.get("longitude"),
             "phase": latest.get("phase"),

backend/data_loader.py

@@ -4,6 +4,7 @@ Provides efficient queries with caching.
 """
 
 import logging
+import re
 
 import pandas as pd
 import numpy as np
@@ -21,6 +22,38 @@ from .models import (
     FactDeprivation, FactFinance, FactPupilCharacteristics,
 )
 from .schemas import SCHOOL_TYPE_MAP
+from .gias_codes import (
+    ADMISSIONS_POLICY,
+    ESTABLISHMENT_STATUS,
+    PHASE_OF_EDUCATION,
+    RELIGIOUS_CHARACTER,
+    SCHOOL_TYPE,
+    translate,
+)
+
+# mart code column -> (API name column, dictionary)
+_GIAS_CODE_COLUMNS = {
+    "phase_code": ("phase", PHASE_OF_EDUCATION),
+    "school_type_code": ("school_type", SCHOOL_TYPE),
+    "status_code": ("status", ESTABLISHMENT_STATUS),
+    "religious_character_code": ("religious_denomination", RELIGIOUS_CHARACTER),
+    "admissions_policy_code": ("admissions_policy", ADMISSIONS_POLICY),
+}
+
+
+def translate_gias_code_columns(df: pd.DataFrame) -> pd.DataFrame:
+    """Map GIAS code columns to today's name columns (API contract).
+
+    Runs immediately after pd.read_sql so every downstream consumer —
+    filters, PHASE_GROUPS, payloads, /api/filters — keeps seeing names.
+    DataFrames without the code columns (old schema, test fixtures) pass
+    through unchanged.
+    """
+    for code_col, (name_col, mapping) in _GIAS_CODE_COLUMNS.items():
+        if code_col in df.columns:
+            df[name_col] = df[code_col].map(lambda c: translate(c, mapping))
+    return df
+
 
 _postcode_cache: Dict[str, Tuple[float, float]] = {}
 _typesense_client = None
@@ -121,15 +154,16 @@ _MAIN_QUERY = text("""
     SELECT
         s.urn,
         s.school_name,
-        s.phase,
-        s.school_type,
+        s.phase_code,
+        s.school_type_code,
         s.academy_trust_name   AS trust_name,
         s.academy_trust_uid    AS trust_uid,
-        s.religious_character  AS religious_denomination,
+        s.religious_character_code,
         s.gender,
         s.age_range,
         s.has_sixth_form,
-        s.admissions_policy,
+        s.status_code,
+        s.admissions_policy_code,
         s.capacity,
         s.total_pupils         AS gias_total_pupils,
         s.headteacher_name,
@@ -229,25 +263,81 @@ assert "NULL AS has_sixth_form" in str(_MAIN_QUERY_NO_SIXTH_FORM), (
     "expected replacement of 's.has_sixth_form,' to have taken effect"
 )
 
+# Fallback used when marts.dim_school predates the GIAS code-dictionary
+# migration (i.e. the nightly dbt pipeline hasn't rebuilt the mart yet on
+# this DB, so it still has the old name columns instead of *_code columns).
+_MAIN_QUERY_LEGACY_NAMES = str(_MAIN_QUERY)
+_LEGACY_NAME_REPLACEMENTS = [
+    ("s.phase_code,", "s.phase,"),
+    ("s.school_type_code,", "s.school_type,"),
+    (
+        "s.religious_character_code,",
+        "s.religious_character AS religious_denomination,",
+    ),
+    ("s.status_code,", "s.status,"),
+    ("s.admissions_policy_code,", "s.admissions_policy,"),
+]
+for _old, _new in _LEGACY_NAME_REPLACEMENTS:
+    assert _old in _MAIN_QUERY_LEGACY_NAMES, (
+        f"expected {_old!r} to be present in _MAIN_QUERY before replacement"
+    )
+    _MAIN_QUERY_LEGACY_NAMES = _MAIN_QUERY_LEGACY_NAMES.replace(_old, _new)
+_MAIN_QUERY_LEGACY_NAMES = text(_MAIN_QUERY_LEGACY_NAMES)
+
+_GIAS_CODE_COLUMN_NAMES = (
+    "phase_code",
+    "school_type_code",
+    "religious_character_code",
+    "status_code",
+    "admissions_policy_code",
+)
+
+_MISSING_COLUMN_RE = re.compile(r'column "?(?:s\.)?(\w+)"? does not exist')
+
+
+def _missing_column_name(exc: Exception) -> Optional[str]:
+    """Name of the missing column from a psycopg2 UndefinedColumn error.
+
+    Inspects exc.orig (the DBAPI error), whose message names only the
+    offending column — str(exc) also embeds the full SQL statement, which
+    contains every column name and therefore must not be matched against.
+    """
+    orig = getattr(exc, "orig", None)
+    match = _MISSING_COLUMN_RE.search(str(orig) if orig is not None else str(exc))
+    return match.group(1) if match else None
+
 
 def load_school_data_as_dataframe() -> pd.DataFrame:
     """Load all school + KS2 data as a pandas DataFrame."""
     try:
         df = pd.read_sql(_MAIN_QUERY, engine)
     except sqlalchemy.exc.ProgrammingError as exc:
-        if "has_sixth_form" not in str(exc):
+        missing = _missing_column_name(exc)
+        if missing in _GIAS_CODE_COLUMN_NAMES:
+            logging.getLogger(__name__).warning(
+                "marts predate the GIAS code migration — falling back to "
+                "legacy name-column query: %s",
+                exc,
+            )
+            try:
+                df = pd.read_sql(_MAIN_QUERY_LEGACY_NAMES, engine)
+            except Exception as exc2:
+                print(f"Warning: Could not load school data from marts: {exc2}")
+                return pd.DataFrame()
+        elif missing == "has_sixth_form":
+            logging.getLogger(__name__).warning(
+                "marts.dim_school is missing has_sixth_form (pipeline hasn't "
+                "rebuilt the mart yet on this DB) — retrying without it: %s",
+                exc,
+            )
+            try:
+                df = pd.read_sql(_MAIN_QUERY_NO_SIXTH_FORM, engine)
+            except Exception as exc2:
+                print(f"Warning: Could not load school data from marts: {exc2}")
+                return pd.DataFrame()
+        else:
             print(f"Warning: Could not load school data from marts: {exc}")
             return pd.DataFrame()
-        logging.getLogger(__name__).warning(
-            "marts.dim_school is missing has_sixth_form (pipeline hasn't "
-            "rebuilt the mart yet on this DB) — retrying without it: %s",
-            exc,
-        )
-        try:
-            df = pd.read_sql(_MAIN_QUERY_NO_SIXTH_FORM, engine)
-        except Exception as exc2:
-            print(f"Warning: Could not load school data from marts: {exc2}")
-            return pd.DataFrame()
     except Exception as exc:
         print(f"Warning: Could not load school data from marts: {exc}")
         return pd.DataFrame()
@@ -255,6 +345,8 @@ def load_school_data_as_dataframe() -> pd.DataFrame:
     if df.empty:
         return df
 
+    df = translate_gias_code_columns(df)
+
     # Build address string
     df["address"] = df.apply(
         lambda r: ", ".join(

backend/gias_codes.py

@@ -0,0 +1,152 @@
+"""GIAS code -> name dictionaries.
+
+GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
+— do not edit by hand; rerun the script when the dbt drift test warns.
+The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
+must be byte-identical (enforced by backend/tests/test_gias_codes.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+
+logger = logging.getLogger(__name__)
+
+
+SCHOOL_TYPE: dict[int, str] = {
+    1: "Community school",
+    2: "Voluntary aided school",
+    3: "Voluntary controlled school",
+    5: "Foundation school",
+    6: "City technology college",
+    7: "Community special school",
+    8: "Non-maintained special school",
+    10: "Other independent special school",
+    11: "Other independent school",
+    12: "Foundation special school",
+    14: "Pupil referral unit",
+    15: "Local authority nursery school",
+    18: "Further education",
+    24: "Secure units",
+    25: "Offshore schools",
+    26: "Service children's education",
+    27: "Miscellaneous",
+    28: "Academy sponsor led",
+    29: "Higher education institutions",
+    30: "Welsh establishment",
+    31: "Sixth form centres",
+    32: "Special post 16 institution",
+    33: "Academy special sponsor led",
+    34: "Academy converter",
+    35: "Free schools",
+    36: "Free schools special",
+    37: "British schools overseas",
+    38: "Free schools alternative provision",
+    39: "Free schools 16 to 19",
+    40: "University technical college",
+    41: "Studio schools",
+    42: "Academy alternative provision converter",
+    43: "Academy alternative provision sponsor led",
+    44: "Academy special converter",
+    45: "Academy 16-19 converter",
+    46: "Academy 16 to 19 sponsor led",
+    49: "Online provider",
+    56: "Institution funded by other government department",
+    57: "Academy secure 16 to 19",
+}
+
+ESTABLISHMENT_STATUS: dict[int, str] = {
+    1: "Open",
+    2: "Closed",
+    3: "Open, but proposed to close",
+    4: "Proposed to open",
+}
+
+PHASE_OF_EDUCATION: dict[int, str] = {
+    0: "Not applicable",
+    1: "Nursery",
+    2: "Primary",
+    3: "Middle deemed primary",
+    4: "Secondary",
+    5: "Middle deemed secondary",
+    6: "16 plus",
+    7: "All-through",
+}
+
+OFFICIAL_SIXTH_FORM: dict[int, str] = {
+    0: "Not applicable",
+    1: "Has a sixth form",
+    2: "Does not have a sixth form",
+}
+
+RELIGIOUS_CHARACTER: dict[int, str] = {
+    0: "Does not apply",
+    2: "Church of England",
+    3: "Roman Catholic",
+    4: "Methodist",
+    5: "Jewish",
+    6: "None",
+    7: "Muslim",
+    8: "Seventh Day Adventist",
+    9: "Church of England/Methodist",
+    10: "Methodist/Church of England",
+    11: "Church of England/Roman Catholic",
+    12: "Church of England/United Reformed Church",
+    13: "Roman Catholic/Church of England",
+    14: "Quaker",
+    15: "Christian",
+    16: "United Reformed Church",
+    17: "Congregational Church",
+    18: "Free Church",
+    19: "Church of England/Free Church",
+    20: "Church of England/Christian",
+    21: "Sikh",
+    22: "Greek Orthodox",
+    24: "Buddhist",
+    25: "Hindu",
+    26: "Moravian",
+    28: "Inter- / non- denominational",
+    29: "Multi-faith",
+    30: "Church of England/Methodist/United Reform Church/Baptist",
+    31: "Anglican",
+    32: "Anglican/Christian",
+    33: "Anglican/Evangelical",
+    34: "Anglican/Church of England",
+    35: "Catholic",
+    36: "Charadi Jewish",
+    37: "Christian/Evangelical",
+    38: "Christian Science",
+    39: "Christian/Methodist",
+    40: "Christian/non-denominational",
+    41: "Church of England/Evangelical",
+    42: "Islam",
+    43: "Orthodox Jewish",
+    44: "Plymouth Brethren Christian Church",
+    45: "Protestant",
+    46: "Protestant/Evangelical",
+    47: "Reformed Baptist",
+    48: "Roman Catholic/Anglican",
+    49: "Sunni Deobandi",
+}
+
+ADMISSIONS_POLICY: dict[int, str] = {
+    0: "Not applicable",
+    2: "Selective",
+    4: "Non-selective",
+}
+
+
+def translate(code, mapping: dict[int, str]) -> str | None:
+    """Translate a GIAS code to its display name.
+
+    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
+    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
+    """
+    if code is None or (isinstance(code, float) and math.isnan(code)):
+        return None
+    code = int(code)
+    if code not in mapping:
+        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
+        return f"Unknown ({code})"
+    return mapping[code]

backend/models.py

@@ -17,11 +17,11 @@ class DimSchool(Base):
 
     urn = Column(Integer, primary_key=True)
     school_name = Column(String(255), nullable=False)
-    phase = Column(String(100))
-    school_type = Column(String(100))
+    phase_code = Column(Integer)
+    school_type_code = Column(Integer)
     academy_trust_name = Column(String(255))
     academy_trust_uid = Column(String(20))
-    religious_character = Column(String(100))
+    religious_character_code = Column(Integer)
     gender = Column(String(20))
     age_range = Column(String(20))
     has_sixth_form = Column(Boolean)
@@ -30,9 +30,9 @@ class DimSchool(Base):
     headteacher_name = Column(String(200))
     website = Column(String(255))
     telephone = Column(String(30))
-    status = Column(String(50))
+    status_code = Column(Integer)
     nursery_provision = Column(Boolean)
-    admissions_policy = Column(String(50))
+    admissions_policy_code = Column(Integer)
     # Denormalised Ofsted summary (updated by monthly pipeline)
     ofsted_grade = Column(Integer)
     ofsted_date = Column(Date)

backend/schemas.py

@@ -544,6 +544,7 @@ SCHOOL_COLUMNS = [
     "religious_denomination",
     "age_range",
     "has_sixth_form",
+    "status",
     "gender",
     "admissions_policy",
     "ofsted_grade",

backend/tests/test_gias_codes.py

@@ -0,0 +1,83 @@
+"""Tests for the GIAS code->name dictionaries (spec 2026-07-09).
+
+The dictionaries are generated from the live GIAS bulk CSV by
+pipeline/scripts/generate_gias_codes.py — these tests assert the module's
+contract, key sentinel values the marts/UI depend on, and that the pipeline
+copy has not drifted from the canonical backend module.
+"""
+
+import math
+from pathlib import Path
+
+from backend.gias_codes import (
+    ADMISSIONS_POLICY,
+    ESTABLISHMENT_STATUS,
+    OFFICIAL_SIXTH_FORM,
+    PHASE_OF_EDUCATION,
+    RELIGIOUS_CHARACTER,
+    SCHOOL_TYPE,
+    translate,
+)
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def test_translate_known_code():
+    open_code = next(c for c, n in ESTABLISHMENT_STATUS.items() if n == "Open")
+    assert translate(open_code, ESTABLISHMENT_STATUS) == "Open"
+
+
+def test_translate_unknown_code_degrades_gracefully():
+    assert translate(9999, ESTABLISHMENT_STATUS) == "Unknown (9999)"
+
+
+def test_translate_none_and_nan_return_none():
+    assert translate(None, ESTABLISHMENT_STATUS) is None
+    assert translate(float("nan"), ESTABLISHMENT_STATUS) is None
+
+
+def test_translate_accepts_float_codes():
+    # pd.read_sql yields float columns when NULLs are present
+    open_code = next(c for c, n in ESTABLISHMENT_STATUS.items() if n == "Open")
+    assert translate(float(open_code), ESTABLISHMENT_STATUS) == "Open"
+
+
+def test_sentinel_names_present():
+    """Names the marts/UI compare against must exist verbatim."""
+    assert "Open" in ESTABLISHMENT_STATUS.values()
+    assert "Open, but proposed to close" in ESTABLISHMENT_STATUS.values()
+    assert "Has a sixth form" in OFFICIAL_SIXTH_FORM.values()
+    assert "Primary" in PHASE_OF_EDUCATION.values()
+    assert "Secondary" in PHASE_OF_EDUCATION.values()
+    assert "Does not apply" in RELIGIOUS_CHARACTER.values()
+    assert all(len(d) > 0 for d in (
+        SCHOOL_TYPE, ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION,
+        OFFICIAL_SIXTH_FORM, RELIGIOUS_CHARACTER, ADMISSIONS_POLICY,
+    ))
+
+
+def test_pipeline_copy_is_identical():
+    canonical = (REPO / "backend" / "gias_codes.py").read_text()
+    copy = (REPO / "pipeline" / "scripts" / "gias_codes.py").read_text()
+    assert canonical == copy, (
+        "pipeline/scripts/gias_codes.py has drifted from backend/gias_codes.py — "
+        "regenerate with pipeline/scripts/generate_gias_codes.py and copy the file"
+    )
+
+
+def test_seed_matches_dictionaries():
+    import csv
+    fields = {
+        "school_type": SCHOOL_TYPE,
+        "establishment_status": ESTABLISHMENT_STATUS,
+        "phase_of_education": PHASE_OF_EDUCATION,
+        "official_sixth_form": OFFICIAL_SIXTH_FORM,
+        "religious_character": RELIGIOUS_CHARACTER,
+        "admissions_policy": ADMISSIONS_POLICY,
+    }
+    seed_path = REPO / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
+    seed: dict[str, dict[int, str]] = {k: {} for k in fields}
+    with open(seed_path, newline="") as fh:
+        for row in csv.DictReader(fh):
+            seed[row["field"]][int(row["code"])] = row["name"]
+    assert seed == fields

backend/tests/test_gias_translation.py

@@ -0,0 +1,132 @@
+"""API-boundary translation: marts now carry GIAS codes; the DataFrame the
+rest of the backend sees must carry today's name strings."""
+
+import numpy as np
+import pandas as pd
+
+from backend.data_loader import _missing_column_name, translate_gias_code_columns
+from backend.gias_codes import ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION
+
+
+def _code_for(mapping, name):
+    return next(c for c, n in mapping.items() if n == name)
+
+
+def test_codes_become_todays_names():
+    df = pd.DataFrame([{
+        "urn": 1,
+        "phase_code": float(_code_for(PHASE_OF_EDUCATION, "Primary")),
+        "school_type_code": np.nan,
+        "status_code": float(_code_for(ESTABLISHMENT_STATUS, "Open, but proposed to close")),
+        "religious_character_code": np.nan,
+        "admissions_policy_code": np.nan,
+    }])
+    out = translate_gias_code_columns(df)
+    row = out.iloc[0]
+    assert row["phase"] == "Primary"
+    assert row["status"] == "Open, but proposed to close"
+    assert row["school_type"] is None
+    assert row["religious_denomination"] is None
+    assert row["admissions_policy"] is None
+
+
+def test_unknown_code_degrades_not_blanks():
+    df = pd.DataFrame([{"urn": 1, "phase_code": 9999.0}])
+    out = translate_gias_code_columns(df)
+    assert out.iloc[0]["phase"] == "Unknown (9999)"
+
+
+def test_missing_code_columns_are_a_noop():
+    """Old-schema DataFrames (tests, pre-pipeline DBs) pass through untouched."""
+    df = pd.DataFrame([{"urn": 1, "phase": "Primary", "status": "Open"}])
+    out = translate_gias_code_columns(df)
+    assert out.iloc[0]["phase"] == "Primary"
+    assert out.iloc[0]["status"] == "Open"
+
+
+def _fake_exc(orig_message):
+    """A stand-in for sqlalchemy.exc.ProgrammingError: str(exc) embeds the
+    full SQL statement (deliberately containing every column name below, to
+    prove the matcher doesn't fall back to it), while .orig carries the real
+    DBAPI error message naming only the offending column."""
+    exc = Exception(
+        "SELECT s.phase_code, s.school_type_code, s.religious_character_code, "
+        "s.status_code, s.admissions_policy_code, s.has_sixth_form FROM ... "
+        f"[SQL: ...] (Background on this error at: https://...)"
+    )
+    exc.orig = Exception(orig_message) if orig_message is not None else None
+    return exc
+
+
+def test_missing_column_name_quoted():
+    assert _missing_column_name(_fake_exc('column "phase_code" does not exist')) == "phase_code"
+
+
+def test_missing_column_name_unquoted():
+    assert _missing_column_name(_fake_exc("column phase_code does not exist")) == "phase_code"
+
+
+def test_missing_column_name_table_prefixed():
+    assert (
+        _missing_column_name(_fake_exc("column s.has_sixth_form does not exist"))
+        == "has_sixth_form"
+    )
+
+
+def test_missing_column_name_no_match_returns_none():
+    assert _missing_column_name(_fake_exc("relation \"marts.dim_school\" does not exist")) is None
+
+
+def test_load_school_data_survives_premigration_marts(monkeypatch):
+    """Real prod state until the nightly pipeline first rebuilds the mart with
+    the GIAS code columns: marts.dim_school still has the old name columns
+    (phase, school_type, religious_character, status, admissions_policy)
+    instead of the new *_code columns. The first query raises UndefinedColumn
+    on s.phase_code; load_school_data_as_dataframe must retry with the
+    legacy name-column query rather than swallow the error and return (and
+    then have load_school_data cache) an empty DataFrame."""
+    import sqlalchemy.exc
+    from backend import data_loader
+
+    data_loader._df_cache = None
+    data_loader._df_latest_cache = None
+
+    good_df = pd.DataFrame(
+        [
+            {
+                "urn": 1,
+                "school_name": "Legacy School",
+                "phase": "Primary",
+                "school_type": "Academy",
+                "status": "Open",
+            }
+        ]
+    )
+    calls = []
+
+    def fake_read_sql(query, con):
+        calls.append(query)
+        if len(calls) == 1:
+            raise sqlalchemy.exc.ProgrammingError(
+                statement=str(data_loader._MAIN_QUERY),
+                params=None,
+                orig=Exception(
+                    "(psycopg2.errors.UndefinedColumn) column s.phase_code "
+                    "does not exist\nLINE 5:         s.phase_code,"
+                ),
+            )
+        return good_df.copy()
+
+    monkeypatch.setattr(data_loader.pd, "read_sql", fake_read_sql)
+
+    try:
+        df = data_loader.load_school_data_as_dataframe()
+    finally:
+        data_loader._df_cache = None
+        data_loader._df_latest_cache = None
+
+    assert len(calls) == 2, "must retry with the legacy name-column query variant"
+    assert calls[1] is data_loader._MAIN_QUERY_LEGACY_NAMES
+    assert not df.empty
+    assert df["phase"].iloc[0] == "Primary"
+    assert df["status"].iloc[0] == "Open"

backend/tests/test_school_status.py

@@ -0,0 +1,70 @@
+"""Tests for GIAS establishment status exposure.
+
+"Open, but proposed to close" schools are now kept by the dims; the API must
+surface `status` on list items and school_info so the UI can render the
+proposed-to-close marker (listing tag) and notice strip (detail page).
+"""
+
+import numpy as np
+import pandas as pd
+import pytest
+from fastapi.testclient import TestClient
+
+PROPOSED = "Open, but proposed to close"
+
+
+def _schools_df() -> pd.DataFrame:
+    base = {
+        "local_authority": "Testshire",
+        "school_type": "Academy",
+        "phase": "Secondary",
+        "address": "1 Test Street",
+        "town": "Testtown",
+        "postcode": "TS1 1AA",
+        "religious_denomination": None,
+        "gender": "Mixed",
+        "age_range": "11-16",
+        "admissions_policy": None,
+        "has_sixth_form": False,
+        "ofsted_grade": np.nan,
+        "ofsted_date": None,
+        "ofsted_framework": None,
+        "latitude": 51.5,
+        "longitude": -0.1,
+        "year": 202425,
+        "total_pupils": 800,
+        "rwm_expected_pct": np.nan,
+        "attainment_8_score": 48.0,
+    }
+    return pd.DataFrame(
+        [
+            {**base, "urn": 200001, "school_name": "Alpha Academy",
+             "status": "Open"},
+            {**base, "urn": 200002, "school_name": "Sarson High School",
+             "status": PROPOSED},
+        ]
+    )
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    from backend import app as app_module
+
+    monkeypatch.setattr(app_module, "load_latest_school_data", _schools_df)
+    monkeypatch.setattr(app_module, "load_school_data", _schools_df)
+    monkeypatch.setattr(app_module, "get_supplementary_data", lambda db, urn: {})
+    return TestClient(app_module.app, raise_server_exceptions=False)
+
+
+def test_list_payload_includes_status(client):
+    resp = client.get("/api/schools")
+    assert resp.status_code == 200, resp.text
+    by_urn = {s["urn"]: s for s in resp.json()["schools"]}
+    assert by_urn[200001]["status"] == "Open"
+    assert by_urn[200002]["status"] == PROPOSED
+
+
+def test_detail_payload_includes_status(client):
+    resp = client.get("/api/schools/200002")
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["school_info"]["status"] == PROPOSED

backend/tests/test_sixth_form_flag.py

@@ -148,10 +148,14 @@ def test_load_school_data_survives_missing_has_sixth_form_column(monkeypatch):
     def fake_read_sql(query, con):
         calls.append(query)
         if len(calls) == 1:
+            # The statement text still contains phase_code, school_type_code,
+            # etc. (it's the full _MAIN_QUERY SELECT list) — that's exactly
+            # the collision this test guards against: matching must be done
+            # against exc.orig (the DBAPI error), not str(exc)/the statement.
             raise sqlalchemy.exc.ProgrammingError(
-                "SELECT ...",
-                None,
-                Exception(
+                statement=str(data_loader._MAIN_QUERY),
+                params=None,
+                orig=Exception(
                     "(psycopg2.errors.UndefinedColumn) column s.has_sixth_form "
                     "does not exist"
                 ),

docs/superpowers/plans/2026-07-09-gias-code-dictionaries.md

@@ -0,0 +1,799 @@
+# GIAS Code Dictionaries Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Store the six GIAS classification fields as official DfE integer codes in the marts and translate code → name in application code, leaving the API contract (name strings) unchanged.
+
+**Architecture:** A generation script downloads the public GIAS bulk CSV and emits the dictionaries (Python dicts + a dbt seed) from real data. The tap ingests the `(code)` columns, staging casts them, `dim_school`/`dim_location` keep only codes, and translation happens in exactly two places: `backend/data_loader.py` right after `pd.read_sql`, and `pipeline/scripts/sync_typesense.py` before indexing. A dbt seed test warns when DfE adds/renames a value; a parity test keeps the backend and pipeline dictionary copies identical.
+
+**Tech Stack:** Singer SDK tap, dbt (Postgres), FastAPI + pandas, Typesense sync script, pytest.
+
+**Spec:** `docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md`
+
+## Global Constraints
+
+- **Numeric code values are never assumed.** Every literal code used in SQL or yml (status filter, sixth-form derivation, phase cascade) must be verified against `pipeline/transform/seeds/gias_code_names.csv` generated in Task 1 from the live CSV. The literals written in this plan are best-current-knowledge and each carries a verification step.
+- **Names served by the API must stay byte-identical** to today's strings (e.g. `Does not apply`, `Open, but proposed to close`) — UI heuristics compare exact strings.
+- The `(name)` columns stay declared in the tap and present in raw; staging stops exposing them.
+- `dim_school` and `dim_location` status filters must stay identical (API inner-joins them).
+- Backend tests run via: `uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests -v` (no local pytest exists).
+- dbt cannot run locally — dbt changes are verified statically (grep / yaml parse) + CI.
+- Never push to `main`. Work on branch `feat/gias-code-dictionaries` (branch off `docs/gias-code-dictionaries` so the spec is included, or off `main` if that has merged).
+- Commits end with: `Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>`
+- Deploy runbook (accepted window, spec §7): merge → deploy → trigger `school_data_daily` immediately. No code-level fallback for the old-schema window.
+
+---
+
+### Task 1: Dictionary generation script, canonical module, pipeline copy, seed
+
+**Files:**
+- Create: `pipeline/scripts/generate_gias_codes.py`
+- Create: `backend/gias_codes.py` (content generated by the script)
+- Create: `pipeline/scripts/gias_codes.py` (byte-identical copy)
+- Create: `pipeline/transform/seeds/gias_code_names.csv` (generated)
+- Test: `backend/tests/test_gias_codes.py`
+
+**Interfaces:**
+- Produces: `backend/gias_codes.py` exporting `SCHOOL_TYPE`, `ESTABLISHMENT_STATUS`, `PHASE_OF_EDUCATION`, `OFFICIAL_SIXTH_FORM`, `RELIGIOUS_CHARACTER`, `ADMISSIONS_POLICY` (each `dict[int, str]`) and `translate(code, mapping) -> str | None`. Task 4 imports these; Task 5 imports the pipeline copy; Task 3 reads code literals from the seed CSV.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `backend/tests/test_gias_codes.py`:
+
+```python
+"""Tests for the GIAS code->name dictionaries (spec 2026-07-09).
+
+The dictionaries are generated from the live GIAS bulk CSV by
+pipeline/scripts/generate_gias_codes.py — these tests assert the module's
+contract, key sentinel values the marts/UI depend on, and that the pipeline
+copy has not drifted from the canonical backend module.
+"""
+
+import math
+from pathlib import Path
+
+from backend.gias_codes import (
+    ADMISSIONS_POLICY,
+    ESTABLISHMENT_STATUS,
+    OFFICIAL_SIXTH_FORM,
+    PHASE_OF_EDUCATION,
+    RELIGIOUS_CHARACTER,
+    SCHOOL_TYPE,
+    translate,
+)
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def test_translate_known_code():
+    open_code = next(c for c, n in ESTABLISHMENT_STATUS.items() if n == "Open")
+    assert translate(open_code, ESTABLISHMENT_STATUS) == "Open"
+
+
+def test_translate_unknown_code_degrades_gracefully():
+    assert translate(9999, ESTABLISHMENT_STATUS) == "Unknown (9999)"
+
+
+def test_translate_none_and_nan_return_none():
+    assert translate(None, ESTABLISHMENT_STATUS) is None
+    assert translate(float("nan"), ESTABLISHMENT_STATUS) is None
+
+
+def test_translate_accepts_float_codes():
+    # pd.read_sql yields float columns when NULLs are present
+    open_code = next(c for c, n in ESTABLISHMENT_STATUS.items() if n == "Open")
+    assert translate(float(open_code), ESTABLISHMENT_STATUS) == "Open"
+
+
+def test_sentinel_names_present():
+    """Names the marts/UI compare against must exist verbatim."""
+    assert "Open" in ESTABLISHMENT_STATUS.values()
+    assert "Open, but proposed to close" in ESTABLISHMENT_STATUS.values()
+    assert "Has a sixth form" in OFFICIAL_SIXTH_FORM.values()
+    assert "Primary" in PHASE_OF_EDUCATION.values()
+    assert "Secondary" in PHASE_OF_EDUCATION.values()
+    assert "Does not apply" in RELIGIOUS_CHARACTER.values()
+    assert all(len(d) > 0 for d in (
+        SCHOOL_TYPE, ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION,
+        OFFICIAL_SIXTH_FORM, RELIGIOUS_CHARACTER, ADMISSIONS_POLICY,
+    ))
+
+
+def test_pipeline_copy_is_identical():
+    canonical = (REPO / "backend" / "gias_codes.py").read_text()
+    copy = (REPO / "pipeline" / "scripts" / "gias_codes.py").read_text()
+    assert canonical == copy, (
+        "pipeline/scripts/gias_codes.py has drifted from backend/gias_codes.py — "
+        "regenerate with pipeline/scripts/generate_gias_codes.py and copy the file"
+    )
+
+
+def test_seed_matches_dictionaries():
+    import csv
+    fields = {
+        "school_type": SCHOOL_TYPE,
+        "establishment_status": ESTABLISHMENT_STATUS,
+        "phase_of_education": PHASE_OF_EDUCATION,
+        "official_sixth_form": OFFICIAL_SIXTH_FORM,
+        "religious_character": RELIGIOUS_CHARACTER,
+        "admissions_policy": ADMISSIONS_POLICY,
+    }
+    seed_path = REPO / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
+    seed: dict[str, dict[int, str]] = {k: {} for k in fields}
+    with open(seed_path, newline="") as fh:
+        for row in csv.DictReader(fh):
+            seed[row["field"]][int(row["code"])] = row["name"]
+    assert seed == fields
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests/test_gias_codes.py -v`
+Expected: FAIL at import — `ModuleNotFoundError: No module named 'backend.gias_codes'`.
+
+- [ ] **Step 3: Write the generation script**
+
+Create `pipeline/scripts/generate_gias_codes.py`:
+
+```python
+"""Generate GIAS code->name dictionaries from the live bulk CSV.
+
+Writes:
+  - backend/gias_codes.py                        (canonical Python module)
+  - pipeline/scripts/gias_codes.py               (byte-identical copy)
+  - pipeline/transform/seeds/gias_code_names.csv (dbt seed for drift test)
+
+Run from the repo root whenever the dbt drift test warns that DfE
+added/renamed a value:  python pipeline/scripts/generate_gias_codes.py
+"""
+
+from __future__ import annotations
+
+import io
+import sys
+from datetime import date, timedelta
+from pathlib import Path
+
+import pandas as pd
+import requests
+
+GIAS_URL = (
+    "https://ea-edubase-api-prod.azurewebsites.net"
+    "/edubase/downloads/public/edubasealldata{date}.csv"
+)
+
+# (CSV code column, CSV name column, python dict name, seed field key)
+FIELDS = [
+    ("TypeOfEstablishment (code)", "TypeOfEstablishment (name)", "SCHOOL_TYPE", "school_type"),
+    ("EstablishmentStatus (code)", "EstablishmentStatus (name)", "ESTABLISHMENT_STATUS", "establishment_status"),
+    ("PhaseOfEducation (code)", "PhaseOfEducation (name)", "PHASE_OF_EDUCATION", "phase_of_education"),
+    ("OfficialSixthForm (code)", "OfficialSixthForm (name)", "OFFICIAL_SIXTH_FORM", "official_sixth_form"),
+    ("ReligiousCharacter (code)", "ReligiousCharacter (name)", "RELIGIOUS_CHARACTER", "religious_character"),
+    ("AdmissionsPolicy (code)", "AdmissionsPolicy (name)", "ADMISSIONS_POLICY", "admissions_policy"),
+]
+
+MODULE_HEADER = '''"""GIAS code -> name dictionaries.
+
+GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
+— do not edit by hand; rerun the script when the dbt drift test warns.
+The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
+must be byte-identical (enforced by backend/tests/test_gias_codes.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+
+logger = logging.getLogger(__name__)
+
+'''
+
+MODULE_FOOTER = '''
+
+def translate(code, mapping: dict[int, str]) -> str | None:
+    """Translate a GIAS code to its display name.
+
+    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
+    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
+    """
+    if code is None or (isinstance(code, float) and math.isnan(code)):
+        return None
+    code = int(code)
+    if code not in mapping:
+        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
+        return f"Unknown ({code})"
+    return mapping[code]
+'''
+
+
+def download_csv() -> pd.DataFrame:
+    for day in (date.today(), date.today() - timedelta(days=1)):
+        url = GIAS_URL.format(date=day.strftime("%Y%m%d"))
+        print(f"Downloading {url}")
+        resp = requests.get(url, timeout=300)
+        if resp.status_code == 404:
+            continue
+        resp.raise_for_status()
+        return pd.read_csv(
+            io.StringIO(resp.content.decode("latin-1")),
+            dtype=str, keep_default_na=False,
+        )
+    sys.exit("GIAS CSV not available for today or yesterday")
+
+
+def main() -> None:
+    repo = Path(__file__).resolve().parents[2]
+    df = download_csv()
+
+    module_parts = [MODULE_HEADER]
+    seed_rows: list[tuple[str, int, str]] = []
+
+    for code_col, name_col, dict_name, field_key in FIELDS:
+        pairs = (
+            df[[code_col, name_col]]
+            .loc[lambda d: (d[code_col] != "") & (d[name_col] != "")]
+            .drop_duplicates()
+        )
+        mapping = sorted((int(c), n) for c, n in pairs.itertuples(index=False))
+        dupes = len(mapping) - len({c for c, _ in mapping})
+        if dupes:
+            sys.exit(f"{code_col}: {dupes} codes map to multiple names — investigate before generating")
+        lines = [f"{dict_name}: dict[int, str] = {{"]
+        for code, name in mapping:
+            escaped = name.replace('"', '\\"')
+            lines.append(f'    {code}: "{escaped}",')
+        lines.append("}\n")
+        module_parts.append("\n".join(lines))
+        seed_rows += [(field_key, code, name) for code, name in mapping]
+
+    module = "\n".join(module_parts) + MODULE_FOOTER
+
+    (repo / "backend" / "gias_codes.py").write_text(module)
+    (repo / "pipeline" / "scripts" / "gias_codes.py").write_text(module)
+
+    seed_path = repo / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
+    with open(seed_path, "w", newline="") as fh:
+        import csv
+        w = csv.writer(fh)
+        w.writerow(["field", "code", "name"])
+        w.writerows(seed_rows)
+
+    print(f"Wrote backend/gias_codes.py, pipeline/scripts/gias_codes.py, {seed_path.name}")
+    print("\nKey codes for the dbt work (Task 3):")
+    for field in ("establishment_status", "phase_of_education", "official_sixth_form"):
+        print(f"  {field}:")
+        for f, code, name in seed_rows:
+            if f == field:
+                print(f"    {code} = {name}")
+
+
+if __name__ == "__main__":
+    main()
+```
+
+- [ ] **Step 4: Run the generator**
+
+Run: `cd /Users/tudor/projects/school_compare && uv run --with pandas --with requests python pipeline/scripts/generate_gias_codes.py`
+Expected: downloads the CSV (~100MB, may take a minute), writes the three files, and prints the status/phase/sixth-form code tables. **Record the printed code tables — Task 3 needs them.** If the download fails twice, report BLOCKED (no network or GIAS outage) rather than inventing dictionary content.
+
+- [ ] **Step 5: Run the tests again**
+
+Run: `cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests/test_gias_codes.py -v`
+Expected: 7 passed. If `test_sentinel_names_present` fails, the GIAS vocabulary differs from expectations — inspect the generated module and report DONE_WITH_CONCERNS naming the differing value; do not edit the generated names.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add pipeline/scripts/generate_gias_codes.py backend/gias_codes.py pipeline/scripts/gias_codes.py pipeline/transform/seeds/gias_code_names.csv backend/tests/test_gias_codes.py
+git commit -m "feat: GIAS code->name dictionaries generated from live bulk CSV
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 2: Tap ingests the (code) columns; staging exposes codes, drops names
+
+**Files:**
+- Modify: `pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py` (Singer schema)
+- Modify: `pipeline/transform/models/staging/stg_gias_establishments.sql`
+
+**Interfaces:**
+- Produces: staging columns `school_type_code`, `status_code`, `phase_code`, `official_sixth_form_code`, `religious_character_code`, `admissions_policy_code` (all int) consumed by Task 3. Staging **stops exposing** `school_type`, `status`, `phase`, `official_sixth_form`, `religious_character`, `admissions_policy` (names stay in raw only).
+
+- [ ] **Step 1: Add the six (code) properties to the Singer schema**
+
+In `tap.py`, `GIASEstablishmentsStream.schema`, add each `(code)` property directly above its existing `(name)` sibling:
+
+```python
+        th.Property("TypeOfEstablishment (code)", th.StringType),
+        th.Property("PhaseOfEducation (code)", th.StringType),
+        th.Property("EstablishmentStatus (code)", th.StringType),
+        th.Property("Gender (name)", ...)  # existing line — for placement reference only
+        th.Property("ReligiousCharacter (code)", th.StringType),
+        th.Property("AdmissionsPolicy (code)", th.StringType),
+        th.Property("OfficialSixthForm (code)", th.StringType),
+```
+
+(The exact insertion order doesn't matter — the schema is a dict — but keep each `(code)` adjacent to its `(name)` for readability. Do NOT remove any `(name)` property.)
+
+- [ ] **Step 2: Rewrite the six columns in staging**
+
+In `stg_gias_establishments.sql` `renamed` CTE, replace:
+
+```sql
+        "TypeOfEstablishment (name)"                    as school_type,
+        "PhaseOfEducation (name)"                       as phase,
+        nullif(trim("OfficialSixthForm (name)"), '')    as official_sixth_form,
+        "ReligiousCharacter (name)"                     as religious_character,
+        "AdmissionsPolicy (name)"                       as admissions_policy,
+        "EstablishmentStatus (name)"                    as status,
+```
+
+with:
+
+```sql
+        cast(nullif(trim("TypeOfEstablishment (code)"), '') as integer)  as school_type_code,
+        cast(nullif(trim("PhaseOfEducation (code)"), '') as integer)     as phase_code,
+        cast(nullif(trim("OfficialSixthForm (code)"), '') as integer)    as official_sixth_form_code,
+        cast(nullif(trim("ReligiousCharacter (code)"), '') as integer)   as religious_character_code,
+        cast(nullif(trim("AdmissionsPolicy (code)"), '') as integer)     as admissions_policy_code,
+        cast(nullif(trim("EstablishmentStatus (code)"), '') as integer)  as status_code,
+```
+
+(The name lines are scattered through the CTE — replace each in place; the six name aliases must no longer appear in the model.)
+
+- [ ] **Step 3: Verify statically**
+
+Run:
+```bash
+cd /Users/tudor/projects/school_compare && \
+python3 -c "import ast; ast.parse(open('pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py').read()); print('tap OK')" && \
+grep -c "(code)" pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py && \
+grep -E "as (school_type|status|phase|official_sixth_form|religious_character|admissions_policy)," pipeline/transform/models/staging/stg_gias_establishments.sql; echo "name-alias grep exit=$? (want 1 = none found)"
+```
+Expected: `tap OK`, code-column count `6`, and the final grep finds nothing (exit 1).
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py pipeline/transform/models/staging/stg_gias_establishments.sql
+git commit -m "feat(pipeline): ingest GIAS code columns; staging exposes codes not names
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 3: Marts store codes; dbt tests + drift test
+
+**Files:**
+- Modify: `pipeline/transform/models/marts/dim_school.sql`
+- Modify: `pipeline/transform/models/marts/dim_location.sql`
+- Modify: `pipeline/transform/models/marts/_marts_schema.yml`
+- Create: `pipeline/transform/tests/assert_gias_code_names_match_seed.sql`
+
+**Interfaces:**
+- Consumes: staging code columns from Task 2; code literals from `pipeline/transform/seeds/gias_code_names.csv` (Task 1).
+- Produces: `dim_school` columns `school_type_code`, `status_code`, `phase_code`, `religious_character_code`, `admissions_policy_code` (int) replacing their string columns; `has_sixth_form` unchanged (bool). Task 4's `_MAIN_QUERY` selects these.
+
+**Before writing SQL: open `pipeline/transform/seeds/gias_code_names.csv` and confirm the literals below.** Best-current-knowledge values (VERIFY EACH):
+`establishment_status`: 1 = Open, 3 = "Open, but proposed to close" (2 = Closed, 4 = Proposed to open).
+`phase_of_education`: 0 = Not applicable, 2 = Primary, 4 = Secondary, 7 = All-through.
+`official_sixth_form`: 1 = Has a sixth form, 2 = Does not have a sixth form, 0 = Not applicable.
+If any differ, use the seed's values everywhere below and say so in your report.
+
+- [ ] **Step 1: Rewrite dim_school.sql derivations in code space**
+
+Replace the phase cascade block (`case ... end as phase,`) with:
+
+```sql
+    -- Phase in GIAS code space (see seeds/gias_code_names.csv):
+    -- 2 = Primary, 4 = Secondary, 7 = All-through, 0 = Not applicable.
+    case
+        -- 1. Trust GIAS phase when it's a real value (0 = the catch-all "Not Applicable")
+        when s.phase_code is not null and s.phase_code != 0
+            then s.phase_code
+        -- 2. Infer from statutory age range (independent schools still publish these)
+        when s.statutory_high_age is not null and s.statutory_high_age <= 11 then 2
+        when s.statutory_low_age is not null and s.statutory_low_age >= 11 then 4
+        when s.statutory_low_age is not null and s.statutory_high_age is not null
+             and s.statutory_low_age < 11 and s.statutory_high_age > 11 then 7
+        -- 3. Fallback: infer from school name (covers independents with missing ages)
+        when s.school_name ilike '%primary%'
+             or s.school_name ilike '%infant%'
+             or s.school_name ilike '%junior%'
+             or s.school_name ilike '%preparatory%'
+             or s.school_name ilike '% prep school%'
+             or s.school_name ilike '% prep %'
+            then 2
+        when s.school_name ilike '%secondary%'
+             or s.school_name ilike '%high school%'
+             or s.school_name ilike '%grammar%'
+             or s.school_name ilike '%senior school%'
+             or s.school_name ilike '%upper school%'
+            then 4
+        -- 4. Give up — null renders no phase pill
+        else null
+    end as phase_code,
+```
+
+Replace `s.school_type,` with `s.school_type_code,`; `s.religious_character,` with `s.religious_character_code,`; `s.admissions_policy,` with `s.admissions_policy_code,`; `s.status,` with `s.status_code,`.
+
+Replace the has_sixth_form case with:
+
+```sql
+    -- GIAS OfficialSixthForm in code space: 1 = has, 2 = does not, 0 = N/A.
+    -- Null (rare, new establishments) falls back to the statutory age range.
+    case
+        when s.official_sixth_form_code = 1 then true
+        when s.official_sixth_form_code in (0, 2) then false
+        else coalesce(s.statutory_high_age >= 18, false)
+    end as has_sixth_form,
+```
+
+Replace the status filter with:
+
+```sql
+-- 1 = Open; 3 = Open, but proposed to close (still operating; drops out when
+-- GIAS flips to Closed — marts fully rebuild each run).
+where s.status_code in (1, 3)
+```
+
+- [ ] **Step 2: Same filter in dim_location.sql**
+
+Replace its `where s.status in ('Open', 'Open, but proposed to close')` (and the comment above it) with:
+
+```sql
+-- Must match dim_school's status filter exactly (the API inner-joins the two).
+where s.status_code in (1, 3)
+```
+
+- [ ] **Step 3: Update _marts_schema.yml**
+
+Under `dim_school` columns: rename `phase` → `phase_code` (keep the warn-severity not_null, reword description to mention codes); replace the `status` accepted_values block with:
+
+```yaml
+      - name: status_code
+        description: GIAS EstablishmentStatus code (1 = Open, 3 = Open but proposed to close)
+        tests:
+          - accepted_values:
+              values: [1, 3]
+```
+
+Add warn-severity accepted_values for the other codes, values copied from the seed (school_type/religious/admissions lists are long — paste the full code list from `gias_code_names.csv` for each):
+
+```yaml
+      - name: school_type_code
+        tests:
+          - accepted_values:
+              severity: warn
+              values: [<all school_type codes from the seed>]
+      - name: religious_character_code
+        tests:
+          - accepted_values:
+              severity: warn
+              values: [<all religious_character codes from the seed>]
+      - name: admissions_policy_code
+        tests:
+          - accepted_values:
+              severity: warn
+              values: [<all admissions_policy codes from the seed>]
+```
+
+(`<...>` here means: paste the actual comma-separated integers from the seed file — the lists exist by the time this task runs. Leaving a literal `<...>` in the yml is a task failure.)
+
+`has_sixth_form` tests stay unchanged.
+
+- [ ] **Step 4: Write the drift test**
+
+Create `pipeline/transform/tests/assert_gias_code_names_match_seed.sql`:
+
+```sql
+-- Warn when the live GIAS CSV carries a (code, name) pair we don't have in
+-- the dictionary seed — i.e. DfE added or renamed a value. Fix by rerunning
+-- pipeline/scripts/generate_gias_codes.py and committing the regenerated
+-- dictionaries + seed together.
+{{ config(severity='warn') }}
+
+with raw_pairs as (
+    {% for field_key, code_col, name_col in [
+        ('school_type',          'TypeOfEstablishment (code)',  'TypeOfEstablishment (name)'),
+        ('establishment_status', 'EstablishmentStatus (code)',  'EstablishmentStatus (name)'),
+        ('phase_of_education',   'PhaseOfEducation (code)',     'PhaseOfEducation (name)'),
+        ('official_sixth_form',  'OfficialSixthForm (code)',    'OfficialSixthForm (name)'),
+        ('religious_character',  'ReligiousCharacter (code)',   'ReligiousCharacter (name)'),
+        ('admissions_policy',    'AdmissionsPolicy (code)',     'AdmissionsPolicy (name)')
+    ] %}
+    select distinct
+        '{{ field_key }}' as field,
+        cast(nullif(trim("{{ code_col }}"), '') as integer) as code,
+        nullif(trim("{{ name_col }}"), '') as name
+    from {{ source('raw', 'gias_establishments') }}
+    where nullif(trim("{{ code_col }}"), '') is not null
+      and nullif(trim("{{ name_col }}"), '') is not null
+    {% if not loop.last %}union all{% endif %}
+    {% endfor %}
+)
+
+select r.*
+from raw_pairs r
+left join {{ ref('gias_code_names') }} s
+    on s.field = r.field
+   and s.code = r.code
+   and s.name = r.name
+where s.field is null
+```
+
+- [ ] **Step 5: Verify statically**
+
+Run:
+```bash
+cd /Users/tudor/projects/school_compare && \
+uv run --with pyyaml python -c "import yaml; yaml.safe_load(open('pipeline/transform/models/marts/_marts_schema.yml')); print('yml OK')" && \
+grep -c "_code" pipeline/transform/models/marts/dim_school.sql && \
+grep -n "status_code in (1, 3)" pipeline/transform/models/marts/dim_school.sql pipeline/transform/models/marts/dim_location.sql && \
+grep -rn "s\.status\b\|s\.phase\b\|s\.school_type\b\|s\.religious_character\b\|s\.admissions_policy\b\|official_sixth_form\b" pipeline/transform/models/marts/dim_school.sql | grep -v "_code"; echo "stale-name grep exit=$? (want 1)"
+```
+Expected: `yml OK`, both filters matched, and no stale name-column references (final grep exits 1).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add pipeline/transform/models/marts/dim_school.sql pipeline/transform/models/marts/dim_location.sql pipeline/transform/models/marts/_marts_schema.yml pipeline/transform/tests/assert_gias_code_names_match_seed.sql
+git commit -m "feat(pipeline): dim_school/dim_location store GIAS codes; seed drift test
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 4: Backend translates at the API boundary
+
+**Files:**
+- Modify: `backend/models.py` (DimSchool columns)
+- Modify: `backend/data_loader.py` (`_MAIN_QUERY` + translation)
+- Test: `backend/tests/test_gias_translation.py` (new)
+
+**Interfaces:**
+- Consumes: `backend/gias_codes.py` dictionaries + `translate` (Task 1); mart code columns (Task 3).
+- Produces: `translate_gias_code_columns(df) -> df` in `backend/data_loader.py`; after `load_school_data_as_dataframe()` the DataFrame carries today's name columns (`phase`, `school_type`, `status`, `religious_denomination`, `admissions_policy`) — every downstream consumer unchanged.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `backend/tests/test_gias_translation.py`:
+
+```python
+"""API-boundary translation: marts now carry GIAS codes; the DataFrame the
+rest of the backend sees must carry today's name strings."""
+
+import numpy as np
+import pandas as pd
+
+from backend.data_loader import translate_gias_code_columns
+from backend.gias_codes import ESTABLISHMENT_STATUS, PHASE_OF_EDUCATION
+
+
+def _code_for(mapping, name):
+    return next(c for c, n in mapping.items() if n == name)
+
+
+def test_codes_become_todays_names():
+    df = pd.DataFrame([{
+        "urn": 1,
+        "phase_code": float(_code_for(PHASE_OF_EDUCATION, "Primary")),
+        "school_type_code": np.nan,
+        "status_code": float(_code_for(ESTABLISHMENT_STATUS, "Open, but proposed to close")),
+        "religious_character_code": np.nan,
+        "admissions_policy_code": np.nan,
+    }])
+    out = translate_gias_code_columns(df)
+    row = out.iloc[0]
+    assert row["phase"] == "Primary"
+    assert row["status"] == "Open, but proposed to close"
+    assert row["school_type"] is None
+    assert row["religious_denomination"] is None
+    assert row["admissions_policy"] is None
+
+
+def test_unknown_code_degrades_not_blanks():
+    df = pd.DataFrame([{"urn": 1, "phase_code": 9999.0}])
+    out = translate_gias_code_columns(df)
+    assert out.iloc[0]["phase"] == "Unknown (9999)"
+
+
+def test_missing_code_columns_are_a_noop():
+    """Old-schema DataFrames (tests, pre-pipeline DBs) pass through untouched."""
+    df = pd.DataFrame([{"urn": 1, "phase": "Primary", "status": "Open"}])
+    out = translate_gias_code_columns(df)
+    assert out.iloc[0]["phase"] == "Primary"
+    assert out.iloc[0]["status"] == "Open"
+```
+
+- [ ] **Step 2: Run to verify failure**
+
+Run: `cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests/test_gias_translation.py -v`
+Expected: FAIL — `ImportError: cannot import name 'translate_gias_code_columns'`.
+
+- [ ] **Step 3: Implement translation in data_loader.py**
+
+Add near the top of `backend/data_loader.py` (after existing imports):
+
+```python
+from .gias_codes import (
+    ADMISSIONS_POLICY,
+    ESTABLISHMENT_STATUS,
+    PHASE_OF_EDUCATION,
+    RELIGIOUS_CHARACTER,
+    SCHOOL_TYPE,
+    translate,
+)
+
+# mart code column -> (API name column, dictionary)
+_GIAS_CODE_COLUMNS = {
+    "phase_code": ("phase", PHASE_OF_EDUCATION),
+    "school_type_code": ("school_type", SCHOOL_TYPE),
+    "status_code": ("status", ESTABLISHMENT_STATUS),
+    "religious_character_code": ("religious_denomination", RELIGIOUS_CHARACTER),
+    "admissions_policy_code": ("admissions_policy", ADMISSIONS_POLICY),
+}
+
+
+def translate_gias_code_columns(df: pd.DataFrame) -> pd.DataFrame:
+    """Map GIAS code columns to today's name columns (API contract).
+
+    Runs immediately after pd.read_sql so every downstream consumer —
+    filters, PHASE_GROUPS, payloads, /api/filters — keeps seeing names.
+    DataFrames without the code columns (old schema, test fixtures) pass
+    through unchanged.
+    """
+    for code_col, (name_col, mapping) in _GIAS_CODE_COLUMNS.items():
+        if code_col in df.columns:
+            df[name_col] = df[code_col].map(lambda c: translate(c, mapping))
+    return df
+```
+
+- [ ] **Step 4: Switch `_MAIN_QUERY` to code columns and call the translation**
+
+In `_MAIN_QUERY` replace:
+`s.phase,` → `s.phase_code,` · `s.school_type,` → `s.school_type_code,` · `s.religious_character AS religious_denomination,` → `s.religious_character_code,` · `s.admissions_policy,` → `s.admissions_policy_code,` · `s.status,` → `s.status_code,`
+
+In `load_school_data_as_dataframe()`, insert the call immediately after the empty-check and **before** the existing `normalize_school_type` line:
+
+```python
+    if df.empty:
+        return df
+
+    df = translate_gias_code_columns(df)
+
+    # Build address string
+    ...
+    # Normalize school type  (existing line — now normalises the translated name)
+    df["school_type"] = df["school_type"].apply(normalize_school_type)
+```
+
+- [ ] **Step 5: Update DimSchool in models.py**
+
+Replace `phase = Column(String(100))`, `school_type = Column(String(100))`, `religious_character = Column(String(100))`, `admissions_policy = Column(String(50))`, `status = Column(String(50))` with:
+
+```python
+    phase_code = Column(Integer)
+    school_type_code = Column(Integer)
+    religious_character_code = Column(Integer)
+    admissions_policy_code = Column(Integer)
+    status_code = Column(Integer)
+```
+
+Then check nothing else references the removed attributes:
+```bash
+grep -rn "\.phase\b\|\.school_type\b\|\.religious_character\b\|\.admissions_policy\b\|\.status\b" backend/*.py | grep -i "dimschool\|DimSchool" 
+```
+Expected: no hits (the backend reads via `_MAIN_QUERY`, not ORM attributes). If there are hits, update them to the `_code` columns + translation and note it in your report.
+
+- [ ] **Step 6: Run the new tests and the whole backend suite**
+
+Run: `cd /Users/tudor/projects/school_compare && uv run --with-requirements requirements.txt --with pytest --with "httpx==0.27.0" python -m pytest backend/tests -v`
+Expected: all pass — 3 new + all pre-existing (their fixtures carry name columns; translation is a no-op on them).
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add backend/models.py backend/data_loader.py backend/tests/test_gias_translation.py
+git commit -m "feat(api): translate GIAS codes to names at the query boundary
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 5: Typesense sync translates before indexing
+
+**Files:**
+- Modify: `pipeline/scripts/sync_typesense.py`
+
+**Interfaces:**
+- Consumes: `pipeline/scripts/gias_codes.py` (Task 1), mart code columns (Task 3).
+- Produces: identical Typesense documents to today (facet values are names).
+
+- [ ] **Step 1: Switch the SELECT and translate**
+
+In `sync_typesense.py`: add at the top (the DAG runs `python scripts/sync_typesense.py`, so `scripts/` is `sys.path[0]` and a plain import works):
+
+```python
+from gias_codes import PHASE_OF_EDUCATION, RELIGIOUS_CHARACTER, SCHOOL_TYPE, translate
+```
+
+In the SQL, replace `s.phase,` → `s.phase_code,`, `s.school_type,` → `s.school_type_code,`, `s.religious_character,` → `s.religious_character_code,`.
+
+In the document builder, replace:
+
+```python
+        "phase": row["phase"] or "",
+        "school_type": row["school_type"] or "",
+```
+with:
+```python
+        "phase": translate(row["phase_code"], PHASE_OF_EDUCATION) or "",
+        "school_type": translate(row["school_type_code"], SCHOOL_TYPE) or "",
+```
+and:
+```python
+    if row.get("religious_character"):
+        doc["religious_character"] = row["religious_character"]
+```
+with:
+```python
+    religious_character = translate(row.get("religious_character_code"), RELIGIOUS_CHARACTER)
+    if religious_character:
+        doc["religious_character"] = religious_character
+```
+
+- [ ] **Step 2: Verify statically**
+
+Run:
+```bash
+cd /Users/tudor/projects/school_compare && \
+python3 -c "import ast; ast.parse(open('pipeline/scripts/sync_typesense.py').read()); print('sync OK')" && \
+grep -n "row\[\"phase\"\]\|row\[\"school_type\"\]\|row\[\"religious_character\"\]" pipeline/scripts/sync_typesense.py; echo "stale grep exit=$? (want 1)"
+```
+Expected: `sync OK`, no stale name-column row accesses.
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add pipeline/scripts/sync_typesense.py
+git commit -m "feat(pipeline): typesense sync translates GIAS codes before indexing
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+### Task 6: Spec status, PR, deploy runbook
+
+**Files:**
+- Modify: `docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md` (status line)
+
+- [ ] **Step 1: Mark the spec implemented**
+
+Change `**Status:** Approved design` to `**Status:** Implemented 2026-07-09 — see docs/superpowers/plans/2026-07-09-gias-code-dictionaries.md`.
+
+- [ ] **Step 2: Commit and push**
+
+```bash
+git add docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md
+git commit -m "docs: mark GIAS code dictionaries spec implemented
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+git push -u origin feat/gias-code-dictionaries
+```
+
+- [ ] **Step 3: Open the PR (Gitea API via git credential fill — token-header auth 401s)**
+
+Title: `feat: GIAS classification fields stored as codes, translated in code`
+Body must include: (1) API contract unchanged — names still served, translation at the query boundary; (2) the **deploy runbook: merge → deploy → trigger `school_data_daily` immediately** (accepted empty-API window until the marts rebuild — spec §7); (3) dictionary maintenance loop (dbt drift test warns → rerun `generate_gias_codes.py` → commit regenerated files); (4) no frontend/e2e changes. End with the standard generation footer.
+
+- [ ] **Step 4: Watch CI**
+
+All PR checks must pass. Do not merge — merging triggers the deploy window; the human runs the runbook.

docs/superpowers/specs/2026-07-09-gias-code-dictionaries-design.md

@@ -0,0 +1,182 @@
+# GIAS Code Dictionaries — Codes in Marts, Names in Code
+
+**Date:** 2026-07-09
+**Status:** Implemented 2026-07-09 — see docs/superpowers/plans/2026-07-09-gias-code-dictionaries.md
+
+## Goal
+
+Six GIAS classification fields are stored in the marts as repeated name
+strings. Replace them with the official DfE integer codes and translate
+code → name in application code. After this change the marts carry only
+codes for:
+
+| GIAS field | Today (marts, string) | After (marts, int) |
+|---|---|---|
+| `TypeOfEstablishment (name)` | `dim_school.school_type` | `school_type_code` |
+| `EstablishmentStatus (name)` | `dim_school.status` | `status_code` |
+| `PhaseOfEducation (name)` | `dim_school.phase` | `phase_code` |
+| `OfficialSixthForm (name)` | (already reduced to `has_sixth_form` bool) | `official_sixth_form_code` in staging only; mart keeps the bool |
+| `ReligiousCharacter (name)` | `dim_school.religious_character` | `religious_character_code` |
+| `AdmissionsPolicy (name)` | `dim_school.admissions_policy` | `admissions_policy_code` |
+
+Motivation: smaller marts and stable enum values for filtering. (Honest
+sizing note: at ~25k open schools the raw performance win is modest; the
+durable benefits are storage, DfE-governed vocabulary, and filter values
+that can't drift with GIAS renames.)
+
+## Decisions (made during brainstorming)
+
+1. **GIAS native codes**, not custom enums. The GIAS bulk CSV publishes an
+   official `X (code)` column beside every `X (name)` column. We ingest the
+   DfE's own codes; no invented mapping to maintain.
+2. **Translation lives in the backend at the API boundary.** The API keeps
+   serving today's name strings; the frontend, e2e journeys, and API
+   consumers are untouched.
+
+## Design
+
+### 1. Tap (Singer schema)
+
+Add the six `(code)` columns to `GIASEstablishmentsStream.schema` in
+`pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py`:
+
+```
+"TypeOfEstablishment (code)", "EstablishmentStatus (code)",
+"PhaseOfEducation (code)", "OfficialSixthForm (code)",
+"ReligiousCharacter (code)", "AdmissionsPolicy (code)"
+```
+
+The `(name)` columns **stay declared** — raw keeps both so we can detect
+dictionary drift (§4) and regenerate dictionaries from live data.
+
+### 2. Staging (`stg_gias_establishments.sql`)
+
+- Add int casts: `school_type_code`, `status_code`, `phase_code`,
+  `official_sixth_form_code`, `religious_character_code`,
+  `admissions_policy_code` (all `cast(nullif(trim(...), '') as integer)`).
+- Remove the corresponding name columns from the staging select
+  (`school_type`, `status`, `phase`, `official_sixth_form`,
+  `religious_character`, `admissions_policy`). Names live only in raw.
+
+### 3. Marts
+
+**`dim_school`** stores codes only:
+
+- `school_type_code`, `status_code`, `phase_code`,
+  `religious_character_code`, `admissions_policy_code` replace their
+  string columns.
+- Status filter becomes `where status_code in (<open>, <proposed-to-close>)`.
+  The numeric values are read from live raw data at implementation time
+  (`select distinct "EstablishmentStatus (code)", "EstablishmentStatus (name)"`),
+  never assumed from memory. Same filter in `dim_location`.
+- `has_sixth_form` derives from `official_sixth_form_code`
+  (`<has-code>` → true, `<does-not>/<not-applicable>` → false, null →
+  `statutory_high_age >= 18` fallback). The `lower(trim(...))` string guard
+  becomes obsolete and is removed.
+- `phase_code` derivation keeps today's cascade but emits codes:
+  1. GIAS `phase_code` when it is a real value (not the not-applicable code);
+  2. statutory-age inference emits the matching GIAS code
+     (Primary / Secondary / All-through — numeric values confirmed from
+     live data at implementation);
+  3. school-name heuristics (unchanged — they match `school_name`, which is
+     not one of the six fields) emit the same codes;
+  4. else null.
+- dbt schema tests: `accepted_values` (severity **warn**) on every code
+  column, values taken from the dictionary; `not_null` warn on `phase_code`
+  (mirrors today's phase test); `has_sixth_form` tests unchanged.
+
+**`dim_location`**: only the status filter changes (must stay byte-identical
+to `dim_school`'s — the API inner-joins the two).
+
+### 4. Dictionaries
+
+**Canonical module: `backend/gias_codes.py`**
+
+```python
+ESTABLISHMENT_STATUS: dict[int, str]
+SCHOOL_TYPE: dict[int, str]
+PHASE_OF_EDUCATION: dict[int, str]
+OFFICIAL_SIXTH_FORM: dict[int, str]
+RELIGIOUS_CHARACTER: dict[int, str]
+ADMISSIONS_POLICY: dict[int, str]
+
+def translate(code: int | None, mapping: dict[int, str]) -> str | None:
+    """None -> None; unknown code -> 'Unknown (<code>)' + warning log."""
+```
+
+- Contents are generated from live raw data
+  (`SELECT DISTINCT code, name FROM raw.gias_establishments ...` per field)
+  and sanity-checked against the DfE GIAS registers. Names must be
+  byte-identical to what the API serves today.
+- Unknown codes never blank the UI: `translate` returns `"Unknown (<code>)"`
+  and logs, so a new DfE value degrades gracefully.
+
+**Pipeline copy: `pipeline/scripts/gias_codes.py`**
+
+The app and pipeline Docker images have disjoint build contexts
+(`Dockerfile` copies `backend/`; `pipeline/Dockerfile` copies `pipeline/`),
+so the Typesense sync cannot import the backend module. It gets a
+byte-identical copy, and a backend unit test asserts
+`backend/gias_codes.py` and `pipeline/scripts/gias_codes.py` have identical
+content — drift fails CI. (Deliberately chosen over codegen: six dicts do
+not justify build machinery.)
+
+**Seed for drift detection: `pipeline/transform/seeds/gias_code_names.csv`**
+
+Columns `field,code,name` mirroring the dictionary. A dbt test (severity
+warn) compares live raw `(code, name)` pairs against the seed; when DfE adds
+or renames a value the nightly run warns, prompting a dictionary + seed
+update in one PR.
+
+### 5. Backend translation (API contract unchanged)
+
+- `_MAIN_QUERY` selects the code columns instead of the name columns.
+- `load_school_data_as_dataframe()` translates immediately after
+  `pd.read_sql`, writing today's column names:
+
+```python
+df["phase"]                  = df["phase_code"].map(...)
+df["school_type"]            = df["school_type_code"].map(...)   # then normalize_school_type as today
+df["status"]                 = df["status_code"].map(...)
+df["religious_denomination"] = df["religious_character_code"].map(...)
+df["admissions_policy"]      = df["admissions_policy_code"].map(...)
+```
+
+Everything downstream — `PHASE_GROUPS`, filters, payload builders,
+`/api/filters`, frontend, e2e — sees exactly today's strings. No frontend
+changes.
+
+- `backend/models.py` `DimSchool`: string columns replaced by
+  `*_code = Column(Integer)`.
+
+### 6. Typesense sync
+
+`pipeline/scripts/sync_typesense.py` selects `phase`, `school_type`,
+`religious_character` today. It switches to the code columns and translates
+via `pipeline/scripts/gias_codes.py` before indexing, so facet values in
+search are unchanged.
+
+### 7. Rollout
+
+- No DB migration: marts are full-rebuild tables.
+- Deploy window: until the first post-merge pipeline run, the old marts
+  still carry string columns while the new backend queries code columns, so
+  the backend's query fails and it serves empty data (the one-column retry
+  built for `has_sixth_form` doesn't generalise to six columns, and a full
+  old-schema fallback query isn't worth it). **Decision: accept the window
+  and close it operationally — the runbook is merge → deploy → trigger
+  `school_data_daily` immediately.** The DAG's final step already calls
+  `/api/admin/reload`, so the backend recovers without a restart.
+- Tests: backend unit tests for `translate()` (known / unknown / None),
+  payload tests asserting names still served, the file-parity test, dbt
+  schema/seed tests. Frontend: no changes; existing Jest suite is the
+  regression net.
+
+## Out of scope
+
+- Recoding other string columns (`gender`, `urban_rural`,
+  `nursery_provision`, `local_authority_name` …) — same pattern can follow
+  later if this proves out.
+- Collapsing academy subtypes (today's `normalize_school_type`) — kept
+  as-is, applied after translation.
+- Serving codes through the API — the contract deliberately keeps names.

nextjs-app/__tests__/components/SecondarySchoolRow.test.tsx

@@ -44,3 +44,24 @@ describe('SecondarySchoolRow sixth-form tag', () => {
     expect(screen.queryByText('Sixth form')).not.toBeInTheDocument();
   });
 });
+
+describe('SecondarySchoolRow proposed-to-close tag', () => {
+  it('shows the tag when GIAS status is "Open, but proposed to close"', () => {
+    render(
+      <SecondarySchoolRow
+        school={{ ...base, status: 'Open, but proposed to close' }}
+      />,
+    );
+    expect(screen.getByText(/Proposed to close/)).toBeInTheDocument();
+  });
+
+  it('hides the tag for a plain open school', () => {
+    render(<SecondarySchoolRow school={{ ...base, status: 'Open' }} />);
+    expect(screen.queryByText(/Proposed to close/)).not.toBeInTheDocument();
+  });
+
+  it('hides the tag when status is missing', () => {
+    render(<SecondarySchoolRow school={base} />);
+    expect(screen.queryByText(/Proposed to close/)).not.toBeInTheDocument();
+  });
+});

nextjs-app/__tests__/lib/utils.test.ts

@@ -212,3 +212,14 @@ describe('computeYBounds', () => {
     expect(computeYBounds([], 'progress')).toEqual({});
   });
 });
+
+describe('isProposedToClose', () => {
+  const { isProposedToClose } = require('@/lib/utils');
+
+  it('is true only for the exact GIAS proposed-to-close status', () => {
+    expect(isProposedToClose({ status: 'Open, but proposed to close' })).toBe(true);
+    expect(isProposedToClose({ status: 'Open' })).toBe(false);
+    expect(isProposedToClose({ status: null })).toBe(false);
+    expect(isProposedToClose({})).toBe(false);
+  });
+});

nextjs-app/components/SchoolDetailView.module.css

@@ -1543,3 +1543,18 @@
 .historyDisclosure[open] > .historyToggle::before {
     transform: rotate(90deg);
 }
+
+/* GIAS "Open, but proposed to close" notice strip */
+.closingStrip {
+  background: #fdf6e3;
+  border-left: 4px solid #e2c96f;
+  border-radius: 0 6px 6px 0;
+  padding: 0.55rem 0.9rem;
+  margin: 0.5rem 0;
+  font-size: 0.88rem;
+  color: #6e5a00;
+  max-width: 68ch;
+}
+.closingStrip strong {
+  color: #8a6200;
+}

nextjs-app/components/SchoolDetailView.tsx

@@ -18,7 +18,7 @@ import type {
   SchoolDeprivation, SchoolFinance, NationalAverages,
 } from '@/lib/types';
 import {
-  formatPercentage, formatProgress, formatAcademicYear,
+  formatPercentage, formatProgress, formatAcademicYear, isProposedToClose,
 } from '@/lib/utils';
 import { DeltaChip } from './DeltaChip';
 
@@ -313,6 +313,12 @@ export function SchoolDetailView({
                 <span className={styles.metaItem}>{schoolInfo.gender}&apos;s school</span>
               )}
             </div>
+            {isProposedToClose(schoolInfo) && (
+              <div className={styles.closingStrip} role="note">
+                <strong>⚠ Proposed to close</strong> — this school is proposed for closure,
+                check with the local authority before applying.
+              </div>
+            )}
             {schoolInfo.address && (
               <p className={styles.address}>
                 {schoolInfo.address}{schoolInfo.postcode && `, ${schoolInfo.postcode}`}

nextjs-app/components/SchoolRow.module.css

@@ -254,3 +254,10 @@
     justify-content: center;
   }
 }
+
+/* GIAS "Open, but proposed to close" marker */
+.attrClosing {
+  background: #fdf6e3;
+  color: #8a6200;
+  border: 1px solid #e2c96f;
+}

nextjs-app/components/SchoolRow.tsx

@@ -9,7 +9,7 @@
  */
 
 import type { School } from '@/lib/types';
-import { formatPercentage, calculateTrend, getPhaseStyle, schoolUrl, buildOfstedListBadge, formatAgeRange } from '@/lib/utils';
+import { formatPercentage, calculateTrend, getPhaseStyle, schoolUrl, buildOfstedListBadge, formatAgeRange, isProposedToClose } from '@/lib/utils';
 import styles from './SchoolRow.module.css';
 
 interface SchoolRowProps {
@@ -78,6 +78,9 @@ export function SchoolRow({
           {school.age_range && <span className={styles.attr}>{formatAgeRange(school.age_range)}</span>}
           {showDenomination && <span className={styles.attr}>{school.religious_denomination}</span>}
           {showGender && <span className={styles.attr}>{school.gender}</span>}
+          {isProposedToClose(school) && (
+            <span className={`${styles.attr} ${styles.attrClosing}`}>⚠ Proposed to close</span>
+          )}
         </div>
 
         {/* Line 3: Key stats */}

nextjs-app/components/SecondarySchoolDetailView.module.css

@@ -1099,3 +1099,18 @@
         padding: 0.75rem;
     }
 }
+
+/* GIAS "Open, but proposed to close" notice strip */
+.closingStrip {
+  background: #fdf6e3;
+  border-left: 4px solid #e2c96f;
+  border-radius: 0 6px 6px 0;
+  padding: 0.55rem 0.9rem;
+  margin: 0.5rem 0;
+  font-size: 0.88rem;
+  color: #6e5a00;
+  max-width: 68ch;
+}
+.closingStrip strong {
+  color: #8a6200;
+}

nextjs-app/components/SecondarySchoolDetailView.tsx

@@ -23,7 +23,7 @@ import type {
   SchoolAdmissions, SenDetail, Phonics,
   SchoolDeprivation, SchoolFinance, NationalAverages,
 } from '@/lib/types';
-import { formatPercentage, formatProgress, formatAcademicYear, formatAgeRange } from '@/lib/utils';
+import { formatPercentage, formatProgress, formatAcademicYear, formatAgeRange, isProposedToClose } from '@/lib/utils';
 import { DeltaChip } from './DeltaChip';
 import { track, getNavigationSource } from '@/lib/analytics';
 import styles from './SecondarySchoolDetailView.module.css';
@@ -237,6 +237,12 @@ export function SecondarySchoolDetailView({
                 </span>
               )}
             </div>
+            {isProposedToClose(schoolInfo) && (
+              <div className={styles.closingStrip} role="note">
+                <strong>⚠ Proposed to close</strong> — this school is proposed for closure,
+                check with the local authority before applying.
+              </div>
+            )}
             {schoolInfo.address && (
               <p className={styles.address}>
                 {schoolInfo.address}{schoolInfo.postcode && `, ${schoolInfo.postcode}`}

nextjs-app/components/SecondarySchoolRow.module.css

@@ -266,3 +266,9 @@
     justify-content: center;
   }
 }
+
+.closingTag {
+  background: #fdf6e3;
+  color: #8a6200;
+  border: 1px solid #e2c96f;
+}

nextjs-app/components/SecondarySchoolRow.tsx

@@ -11,7 +11,7 @@
 'use client';
 
 import type { School } from '@/lib/types';
-import { buildOfstedListBadge, getPhaseStyle, schoolUrl, formatAgeRange } from '@/lib/utils';
+import { buildOfstedListBadge, getPhaseStyle, schoolUrl, formatAgeRange, isProposedToClose } from '@/lib/utils';
 import styles from './SecondarySchoolRow.module.css';
 
 function detectAdmissionsTag(school: School): string | null {
@@ -97,6 +97,9 @@ export function SecondarySchoolRow({
               {admissionsTag}
             </span>
           )}
+          {isProposedToClose(school) && (
+            <span className={`${styles.provisionTag} ${styles.closingTag}`}>⚠ Proposed to close</span>
+          )}
         </div>
 
         {/* Line 3: KS4 stats */}

nextjs-app/lib/types.ts

@@ -18,6 +18,7 @@ export interface School {
   religious_denomination: string | null;
   age_range: string | null;
   has_sixth_form?: boolean | null;
+  status?: string | null; // GIAS establishment status ("Open" / "Open, but proposed to close")
 
   // Address
   address1: string | null;

nextjs-app/lib/utils.ts

@@ -718,3 +718,18 @@ export function buildOfstedListBadge(school: {
 
   return { label: 'Not yet inspected', cssClass: 'ofstedPending' };
 }
+
+// ============================================================================
+// Establishment status
+// ============================================================================
+
+export const PROPOSED_TO_CLOSE_STATUS = 'Open, but proposed to close';
+
+/**
+ * GIAS lists some operating schools as "Open, but proposed to close".
+ * They remain open (and may stay open if the proposal is withdrawn), but the
+ * UI marks them so families check with the local authority before applying.
+ */
+export function isProposedToClose(school: { status?: string | null }): boolean {
+  return school.status === PROPOSED_TO_CLOSE_STATUS;
+}

pipeline/dags/school_data_pipeline.py

@@ -83,7 +83,7 @@ print(f'Validation passed: {{count}} GIAS rows')
 
     dbt_build = BashOperator(
         task_id="dbt_build",
-        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_gias_establishments+ stg_gias_links+ --exclude int_ks2_with_lineage+ int_ks4_with_lineage+",
+        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_gias_establishments+ stg_gias_links+ gias_code_names+ --exclude int_ks2_with_lineage+ int_ks4_with_lineage+",
     )
 
     sync_typesense = BashOperator(

pipeline/plugins/extractors/tap-uk-gias/tap_uk_gias/tap.py

@@ -31,16 +31,22 @@ class GIASEstablishmentsStream(Stream):
     schema = th.PropertiesList(
         th.Property("URN", th.IntegerType, required=True),
         th.Property("EstablishmentName", th.StringType),
+        th.Property("TypeOfEstablishment (code)", th.StringType),
         th.Property("TypeOfEstablishment (name)", th.StringType),
+        th.Property("PhaseOfEducation (code)", th.StringType),
         th.Property("PhaseOfEducation (name)", th.StringType),
+        th.Property("OfficialSixthForm (code)", th.StringType),
         th.Property("OfficialSixthForm (name)", th.StringType),
         th.Property("LA (code)", th.StringType),
         th.Property("LA (name)", th.StringType),
         th.Property("EstablishmentNumber", th.StringType),
+        th.Property("EstablishmentStatus (code)", th.StringType),
         th.Property("EstablishmentStatus (name)", th.StringType),
         th.Property("Postcode", th.StringType),
         th.Property("Gender (name)", th.StringType),
+        th.Property("ReligiousCharacter (code)", th.StringType),
         th.Property("ReligiousCharacter (name)", th.StringType),
+        th.Property("AdmissionsPolicy (code)", th.StringType),
         th.Property("AdmissionsPolicy (name)", th.StringType),
         th.Property("SchoolCapacity", th.StringType),
         th.Property("NumberOfPupils", th.StringType),

pipeline/scripts/generate_gias_codes.py

@@ -0,0 +1,134 @@
+"""Generate GIAS code->name dictionaries from the live bulk CSV.
+
+Writes:
+  - backend/gias_codes.py                        (canonical Python module)
+  - pipeline/scripts/gias_codes.py               (byte-identical copy)
+  - pipeline/transform/seeds/gias_code_names.csv (dbt seed for drift test)
+
+Run from the repo root whenever the dbt drift test warns that DfE
+added/renamed a value:  python pipeline/scripts/generate_gias_codes.py
+"""
+
+from __future__ import annotations
+
+import io
+import sys
+from datetime import date, timedelta
+from pathlib import Path
+
+import pandas as pd
+import requests
+
+GIAS_URL = (
+    "https://ea-edubase-api-prod.azurewebsites.net"
+    "/edubase/downloads/public/edubasealldata{date}.csv"
+)
+
+# (CSV code column, CSV name column, python dict name, seed field key)
+FIELDS = [
+    ("TypeOfEstablishment (code)", "TypeOfEstablishment (name)", "SCHOOL_TYPE", "school_type"),
+    ("EstablishmentStatus (code)", "EstablishmentStatus (name)", "ESTABLISHMENT_STATUS", "establishment_status"),
+    ("PhaseOfEducation (code)", "PhaseOfEducation (name)", "PHASE_OF_EDUCATION", "phase_of_education"),
+    ("OfficialSixthForm (code)", "OfficialSixthForm (name)", "OFFICIAL_SIXTH_FORM", "official_sixth_form"),
+    ("ReligiousCharacter (code)", "ReligiousCharacter (name)", "RELIGIOUS_CHARACTER", "religious_character"),
+    ("AdmissionsPolicy (code)", "AdmissionsPolicy (name)", "ADMISSIONS_POLICY", "admissions_policy"),
+]
+
+MODULE_HEADER = '''"""GIAS code -> name dictionaries.
+
+GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
+— do not edit by hand; rerun the script when the dbt drift test warns.
+The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
+must be byte-identical (enforced by backend/tests/test_gias_codes.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+
+logger = logging.getLogger(__name__)
+
+'''
+
+MODULE_FOOTER = '''
+
+def translate(code, mapping: dict[int, str]) -> str | None:
+    """Translate a GIAS code to its display name.
+
+    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
+    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
+    """
+    if code is None or (isinstance(code, float) and math.isnan(code)):
+        return None
+    code = int(code)
+    if code not in mapping:
+        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
+        return f"Unknown ({code})"
+    return mapping[code]
+'''
+
+
+def download_csv() -> pd.DataFrame:
+    for day in (date.today(), date.today() - timedelta(days=1)):
+        url = GIAS_URL.format(date=day.strftime("%Y%m%d"))
+        print(f"Downloading {url}")
+        resp = requests.get(url, timeout=300)
+        if resp.status_code == 404:
+            continue
+        resp.raise_for_status()
+        return pd.read_csv(
+            io.StringIO(resp.content.decode("latin-1")),
+            dtype=str, keep_default_na=False,
+        )
+    sys.exit("GIAS CSV not available for today or yesterday")
+
+
+def main() -> None:
+    repo = Path(__file__).resolve().parents[2]
+    df = download_csv()
+
+    module_parts = [MODULE_HEADER]
+    seed_rows: list[tuple[str, int, str]] = []
+
+    for code_col, name_col, dict_name, field_key in FIELDS:
+        pairs = (
+            df[[code_col, name_col]]
+            .loc[lambda d: (d[code_col] != "") & (d[name_col] != "")]
+            .drop_duplicates()
+        )
+        mapping = sorted((int(c), n) for c, n in pairs.itertuples(index=False))
+        dupes = len(mapping) - len({c for c, _ in mapping})
+        if dupes:
+            sys.exit(f"{code_col}: {dupes} codes map to multiple names — investigate before generating")
+        lines = [f"{dict_name}: dict[int, str] = {{"]
+        for code, name in mapping:
+            escaped = name.replace('"', '\\"')
+            lines.append(f'    {code}: "{escaped}",')
+        lines.append("}\n")
+        module_parts.append("\n".join(lines))
+        seed_rows += [(field_key, code, name) for code, name in mapping]
+
+    module = "\n".join(module_parts) + MODULE_FOOTER
+
+    (repo / "backend" / "gias_codes.py").write_text(module)
+    (repo / "pipeline" / "scripts" / "gias_codes.py").write_text(module)
+
+    seed_path = repo / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
+    with open(seed_path, "w", newline="") as fh:
+        import csv
+        w = csv.writer(fh)
+        w.writerow(["field", "code", "name"])
+        w.writerows(seed_rows)
+
+    print(f"Wrote backend/gias_codes.py, pipeline/scripts/gias_codes.py, {seed_path.name}")
+    print("\nKey codes for the dbt work (Task 3):")
+    for field in ("establishment_status", "phase_of_education", "official_sixth_form"):
+        print(f"  {field}:")
+        for f, code, name in seed_rows:
+            if f == field:
+                print(f"    {code} = {name}")
+
+
+if __name__ == "__main__":
+    main()

pipeline/scripts/gias_codes.py

@@ -0,0 +1,152 @@
+"""GIAS code -> name dictionaries.
+
+GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
+— do not edit by hand; rerun the script when the dbt drift test warns.
+The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
+must be byte-identical (enforced by backend/tests/test_gias_codes.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+
+logger = logging.getLogger(__name__)
+
+
+SCHOOL_TYPE: dict[int, str] = {
+    1: "Community school",
+    2: "Voluntary aided school",
+    3: "Voluntary controlled school",
+    5: "Foundation school",
+    6: "City technology college",
+    7: "Community special school",
+    8: "Non-maintained special school",
+    10: "Other independent special school",
+    11: "Other independent school",
+    12: "Foundation special school",
+    14: "Pupil referral unit",
+    15: "Local authority nursery school",
+    18: "Further education",
+    24: "Secure units",
+    25: "Offshore schools",
+    26: "Service children's education",
+    27: "Miscellaneous",
+    28: "Academy sponsor led",
+    29: "Higher education institutions",
+    30: "Welsh establishment",
+    31: "Sixth form centres",
+    32: "Special post 16 institution",
+    33: "Academy special sponsor led",
+    34: "Academy converter",
+    35: "Free schools",
+    36: "Free schools special",
+    37: "British schools overseas",
+    38: "Free schools alternative provision",
+    39: "Free schools 16 to 19",
+    40: "University technical college",
+    41: "Studio schools",
+    42: "Academy alternative provision converter",
+    43: "Academy alternative provision sponsor led",
+    44: "Academy special converter",
+    45: "Academy 16-19 converter",
+    46: "Academy 16 to 19 sponsor led",
+    49: "Online provider",
+    56: "Institution funded by other government department",
+    57: "Academy secure 16 to 19",
+}
+
+ESTABLISHMENT_STATUS: dict[int, str] = {
+    1: "Open",
+    2: "Closed",
+    3: "Open, but proposed to close",
+    4: "Proposed to open",
+}
+
+PHASE_OF_EDUCATION: dict[int, str] = {
+    0: "Not applicable",
+    1: "Nursery",
+    2: "Primary",
+    3: "Middle deemed primary",
+    4: "Secondary",
+    5: "Middle deemed secondary",
+    6: "16 plus",
+    7: "All-through",
+}
+
+OFFICIAL_SIXTH_FORM: dict[int, str] = {
+    0: "Not applicable",
+    1: "Has a sixth form",
+    2: "Does not have a sixth form",
+}
+
+RELIGIOUS_CHARACTER: dict[int, str] = {
+    0: "Does not apply",
+    2: "Church of England",
+    3: "Roman Catholic",
+    4: "Methodist",
+    5: "Jewish",
+    6: "None",
+    7: "Muslim",
+    8: "Seventh Day Adventist",
+    9: "Church of England/Methodist",
+    10: "Methodist/Church of England",
+    11: "Church of England/Roman Catholic",
+    12: "Church of England/United Reformed Church",
+    13: "Roman Catholic/Church of England",
+    14: "Quaker",
+    15: "Christian",
+    16: "United Reformed Church",
+    17: "Congregational Church",
+    18: "Free Church",
+    19: "Church of England/Free Church",
+    20: "Church of England/Christian",
+    21: "Sikh",
+    22: "Greek Orthodox",
+    24: "Buddhist",
+    25: "Hindu",
+    26: "Moravian",
+    28: "Inter- / non- denominational",
+    29: "Multi-faith",
+    30: "Church of England/Methodist/United Reform Church/Baptist",
+    31: "Anglican",
+    32: "Anglican/Christian",
+    33: "Anglican/Evangelical",
+    34: "Anglican/Church of England",
+    35: "Catholic",
+    36: "Charadi Jewish",
+    37: "Christian/Evangelical",
+    38: "Christian Science",
+    39: "Christian/Methodist",
+    40: "Christian/non-denominational",
+    41: "Church of England/Evangelical",
+    42: "Islam",
+    43: "Orthodox Jewish",
+    44: "Plymouth Brethren Christian Church",
+    45: "Protestant",
+    46: "Protestant/Evangelical",
+    47: "Reformed Baptist",
+    48: "Roman Catholic/Anglican",
+    49: "Sunni Deobandi",
+}
+
+ADMISSIONS_POLICY: dict[int, str] = {
+    0: "Not applicable",
+    2: "Selective",
+    4: "Non-selective",
+}
+
+
+def translate(code, mapping: dict[int, str]) -> str | None:
+    """Translate a GIAS code to its display name.
+
+    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
+    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
+    """
+    if code is None or (isinstance(code, float) and math.isnan(code)):
+        return None
+    code = int(code)
+    if code not in mapping:
+        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
+        return f"Unknown ({code})"
+    return mapping[code]

pipeline/scripts/sync_typesense.py

@@ -19,6 +19,8 @@ import psycopg2
 import psycopg2.extras
 import typesense
 
+from gias_codes import PHASE_OF_EDUCATION, RELIGIOUS_CHARACTER, SCHOOL_TYPE, translate
+
 COLLECTION_SCHEMA = {
     "fields": [
         {"name": "urn", "type": "int32"},
@@ -44,10 +46,10 @@ QUERY_BASE = """
     SELECT
         s.urn,
         s.school_name,
-        s.phase,
-        s.school_type,
+        s.phase_code,
+        s.school_type_code,
         l.local_authority_name as local_authority,
-        s.religious_character,
+        s.religious_character_code,
         s.ofsted_grade,
         l.postcode,
         s.headteacher_name,
@@ -85,14 +87,15 @@ def build_document(row: dict) -> dict:
         "id": str(row["urn"]),
         "urn": row["urn"],
         "school_name": row["school_name"] or "",
-        "phase": row["phase"] or "",
-        "school_type": row["school_type"] or "",
+        "phase": translate(row["phase_code"], PHASE_OF_EDUCATION) or "",
+        "school_type": translate(row["school_type_code"], SCHOOL_TYPE) or "",
         "local_authority": row["local_authority"] or "",
         "postcode": row["postcode"] or "",
     }
 
-    if row.get("religious_character"):
-        doc["religious_character"] = row["religious_character"]
+    religious_character = translate(row.get("religious_character_code"), RELIGIOUS_CHARACTER)
+    if religious_character:
+        doc["religious_character"] = religious_character
     if row.get("ofsted_grade"):
         doc["ofsted_rating"] = OFSTED_LABELS.get(row["ofsted_grade"], "")
     if row.get("headteacher_name"):

pipeline/transform/models/marts/_marts_schema.yml

@@ -8,9 +8,10 @@ models:
         tests: [not_null, unique]
       - name: school_name
         tests: [not_null]
-      - name: phase
+      - name: phase_code
         description: >
-          Primary / Secondary / All-through etc. May be null for a small number
+          GIAS PhaseOfEducation code (2 = Primary, 4 = Secondary, 7 = All-through,
+          etc. — see seeds/gias_code_names.csv). May be null for a small number
           of independent schools where GIAS publishes "Not Applicable", no
           statutory age range, and the school name gives no hint.
         tests:
@@ -27,10 +28,26 @@ models:
           - not_null
           - accepted_values:
               values: [true, false]
-      - name: status
+      - name: status_code
+        description: GIAS EstablishmentStatus code (1 = Open, 3 = Open but proposed to close)
         tests:
           - accepted_values:
-              values: ["Open"]
+              values: [1, 3]
+      - name: school_type_code
+        tests:
+          - accepted_values:
+              severity: warn
+              values: [1, 2, 3, 5, 6, 7, 8, 10, 11, 12, 14, 15, 18, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 49, 56, 57]
+      - name: religious_character_code
+        tests:
+          - accepted_values:
+              severity: warn
+              values: [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 24, 25, 26, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49]
+      - name: admissions_policy_code
+        tests:
+          - accepted_values:
+              severity: warn
+              values: [0, 2, 4]
 
   - name: dim_location
     description: School location dimension with PostGIS geometry

pipeline/transform/models/marts/dim_location.sql

@@ -31,4 +31,5 @@ select
         else null
     end as longitude
 from {{ ref('stg_gias_establishments') }} s
-where s.status = 'Open'
+-- Must match dim_school's status filter exactly (the API inner-joins the two).
+where s.status_code in (1, 3)

pipeline/transform/models/marts/dim_school.sql

@@ -19,16 +19,17 @@ select
     s.urn,
     s.local_authority_code * 1000 + s.establishment_number as laestab,
     s.school_name,
+    -- Phase in GIAS code space (see seeds/gias_code_names.csv):
+    -- 2 = Primary, 4 = Secondary, 7 = All-through, 0 = Not applicable.
     case
-        -- 1. Trust GIAS phase when it's a real value (not the catch-all "Not Applicable")
-        when s.phase is not null
-             and lower(trim(s.phase)) not in ('not applicable', '', 'unknown')
-            then s.phase
+        -- 1. Trust GIAS phase when it's a real value (0 = the catch-all "Not Applicable")
+        when s.phase_code is not null and s.phase_code != 0
+            then s.phase_code
         -- 2. Infer from statutory age range (independent schools still publish these)
-        when s.statutory_high_age is not null and s.statutory_high_age <= 11 then 'Primary'
-        when s.statutory_low_age is not null and s.statutory_low_age >= 11 then 'Secondary'
+        when s.statutory_high_age is not null and s.statutory_high_age <= 11 then 2
+        when s.statutory_low_age is not null and s.statutory_low_age >= 11 then 4
         when s.statutory_low_age is not null and s.statutory_high_age is not null
-             and s.statutory_low_age < 11 and s.statutory_high_age > 11 then 'All-through'
+             and s.statutory_low_age < 11 and s.statutory_high_age > 11 then 7
         -- 3. Fallback: infer from school name (covers independents with missing ages)
         when s.school_name ilike '%primary%'
              or s.school_name ilike '%infant%'
@@ -36,31 +37,27 @@ select
              or s.school_name ilike '%preparatory%'
              or s.school_name ilike '% prep school%'
              or s.school_name ilike '% prep %'
-            then 'Primary'
+            then 2
         when s.school_name ilike '%secondary%'
              or s.school_name ilike '%high school%'
              or s.school_name ilike '%grammar%'
              or s.school_name ilike '%senior school%'
              or s.school_name ilike '%upper school%'
-            then 'Secondary'
-        -- 4. Give up — leave phase null so the UI renders no pill
+            then 4
+        -- 4. Give up — null renders no phase pill
         else null
-    end as phase,
-    s.school_type,
+    end as phase_code,
+    s.school_type_code,
     s.academy_trust_name,
     s.academy_trust_uid,
-    s.religious_character,
+    s.religious_character_code,
     s.gender,
     s.statutory_low_age || '-' || s.statutory_high_age as age_range,
-    -- Authoritative sixth-form flag (spec §3): GIAS OfficialSixthForm.
-    -- "Not applicable" (nurseries, primaries, PRUs) => false. Blank GIAS
-    -- value (rare, new establishments) falls back to the statutory age range.
-    -- lower(trim()) guards against casing/whitespace variants in raw GIAS
-    -- data, same as the phase derivation above — an unmatched variant would
-    -- otherwise silently fall through to the age-range fallback.
+    -- GIAS OfficialSixthForm in code space: 1 = has, 2 = does not, 0 = N/A.
+    -- Null (rare, new establishments) falls back to the statutory age range.
     case
-        when lower(trim(s.official_sixth_form)) = 'has a sixth form' then true
-        when lower(trim(s.official_sixth_form)) in ('does not have a sixth form', 'not applicable') then false
+        when s.official_sixth_form_code = 1 then true
+        when s.official_sixth_form_code in (0, 2) then false
         else coalesce(s.statutory_high_age >= 18, false)
     end as has_sixth_form,
     s.capacity,
@@ -70,9 +67,9 @@ select
     s.telephone,
     s.open_date,
     s.close_date,
-    s.status,
+    s.status_code,
     s.nursery_provision,
-    s.admissions_policy,
+    s.admissions_policy_code,
 
     -- Latest Ofsted (populated after monthly Ofsted pipeline runs)
     {% if ofsted_relation is not none %}
@@ -91,4 +88,6 @@ from schools s
 {% if ofsted_relation is not none %}
 left join {{ ref('int_ofsted_latest') }} o on s.urn = o.urn
 {% endif %}
-where s.status = 'Open'
+-- 1 = Open; 3 = Open, but proposed to close (still operating; drops out when
+-- GIAS flips to Closed — marts fully rebuild each run).
+where s.status_code in (1, 3)

pipeline/transform/models/staging/stg_gias_establishments.sql

@@ -12,12 +12,12 @@ renamed as (
         "LA (name)"                                     as local_authority_name,
         cast(nullif("EstablishmentNumber", '') as integer) as establishment_number,
         "EstablishmentName"                             as school_name,
-        "TypeOfEstablishment (name)"                    as school_type,
-        "PhaseOfEducation (name)"                       as phase,
-        nullif(trim("OfficialSixthForm (name)"), '')    as official_sixth_form,
+        cast(nullif(trim("TypeOfEstablishment (code)"), '') as integer)  as school_type_code,
+        cast(nullif(trim("PhaseOfEducation (code)"), '') as integer)     as phase_code,
+        cast(nullif(trim("OfficialSixthForm (code)"), '') as integer)    as official_sixth_form_code,
         "Gender (name)"                                 as gender,
-        "ReligiousCharacter (name)"                     as religious_character,
-        "AdmissionsPolicy (name)"                       as admissions_policy,
+        cast(nullif(trim("ReligiousCharacter (code)"), '') as integer)   as religious_character_code,
+        cast(nullif(trim("AdmissionsPolicy (code)"), '') as integer)     as admissions_policy_code,
         "SchoolCapacity"                                as capacity,
         cast(nullif("NumberOfPupils", '') as integer)   as total_pupils,
         "HeadTitle (name)"                              as head_title,
@@ -30,7 +30,7 @@ renamed as (
         "Town"                                          as town,
         "County (name)"                                 as county,
         "Postcode"                                      as postcode,
-        "EstablishmentStatus (name)"                    as status,
+        cast(nullif(trim("EstablishmentStatus (code)"), '') as integer)  as status_code,
         case when "OpenDate" = '' then null else to_date("OpenDate", 'DD-MM-YYYY') end   as open_date,
         case when "CloseDate" = '' then null else to_date("CloseDate", 'DD-MM-YYYY') end as close_date,
         "Trusts (name)"                                 as academy_trust_name,

pipeline/transform/seeds/gias_code_names.csv

@@ -0,0 +1,105 @@
+field,code,name
+school_type,1,Community school
+school_type,2,Voluntary aided school
+school_type,3,Voluntary controlled school
+school_type,5,Foundation school
+school_type,6,City technology college
+school_type,7,Community special school
+school_type,8,Non-maintained special school
+school_type,10,Other independent special school
+school_type,11,Other independent school
+school_type,12,Foundation special school
+school_type,14,Pupil referral unit
+school_type,15,Local authority nursery school
+school_type,18,Further education
+school_type,24,Secure units
+school_type,25,Offshore schools
+school_type,26,Service children's education
+school_type,27,Miscellaneous
+school_type,28,Academy sponsor led
+school_type,29,Higher education institutions
+school_type,30,Welsh establishment
+school_type,31,Sixth form centres
+school_type,32,Special post 16 institution
+school_type,33,Academy special sponsor led
+school_type,34,Academy converter
+school_type,35,Free schools
+school_type,36,Free schools special
+school_type,37,British schools overseas
+school_type,38,Free schools alternative provision
+school_type,39,Free schools 16 to 19
+school_type,40,University technical college
+school_type,41,Studio schools
+school_type,42,Academy alternative provision converter
+school_type,43,Academy alternative provision sponsor led
+school_type,44,Academy special converter
+school_type,45,Academy 16-19 converter
+school_type,46,Academy 16 to 19 sponsor led
+school_type,49,Online provider
+school_type,56,Institution funded by other government department
+school_type,57,Academy secure 16 to 19
+establishment_status,1,Open
+establishment_status,2,Closed
+establishment_status,3,"Open, but proposed to close"
+establishment_status,4,Proposed to open
+phase_of_education,0,Not applicable
+phase_of_education,1,Nursery
+phase_of_education,2,Primary
+phase_of_education,3,Middle deemed primary
+phase_of_education,4,Secondary
+phase_of_education,5,Middle deemed secondary
+phase_of_education,6,16 plus
+phase_of_education,7,All-through
+official_sixth_form,0,Not applicable
+official_sixth_form,1,Has a sixth form
+official_sixth_form,2,Does not have a sixth form
+religious_character,0,Does not apply
+religious_character,2,Church of England
+religious_character,3,Roman Catholic
+religious_character,4,Methodist
+religious_character,5,Jewish
+religious_character,6,None
+religious_character,7,Muslim
+religious_character,8,Seventh Day Adventist
+religious_character,9,Church of England/Methodist
+religious_character,10,Methodist/Church of England
+religious_character,11,Church of England/Roman Catholic
+religious_character,12,Church of England/United Reformed Church
+religious_character,13,Roman Catholic/Church of England
+religious_character,14,Quaker
+religious_character,15,Christian
+religious_character,16,United Reformed Church
+religious_character,17,Congregational Church
+religious_character,18,Free Church
+religious_character,19,Church of England/Free Church
+religious_character,20,Church of England/Christian
+religious_character,21,Sikh
+religious_character,22,Greek Orthodox
+religious_character,24,Buddhist
+religious_character,25,Hindu
+religious_character,26,Moravian
+religious_character,28,Inter- / non- denominational
+religious_character,29,Multi-faith
+religious_character,30,Church of England/Methodist/United Reform Church/Baptist
+religious_character,31,Anglican
+religious_character,32,Anglican/Christian
+religious_character,33,Anglican/Evangelical
+religious_character,34,Anglican/Church of England
+religious_character,35,Catholic
+religious_character,36,Charadi Jewish
+religious_character,37,Christian/Evangelical
+religious_character,38,Christian Science
+religious_character,39,Christian/Methodist
+religious_character,40,Christian/non-denominational
+religious_character,41,Church of England/Evangelical
+religious_character,42,Islam
+religious_character,43,Orthodox Jewish
+religious_character,44,Plymouth Brethren Christian Church
+religious_character,45,Protestant
+religious_character,46,Protestant/Evangelical
+religious_character,47,Reformed Baptist
+religious_character,48,Roman Catholic/Anglican
+religious_character,49,Sunni Deobandi
+admissions_policy,0,Not applicable
+admissions_policy,2,Selective
+admissions_policy,4,Non-selective

pipeline/transform/tests/assert_gias_code_names_match_seed.sql

@@ -0,0 +1,33 @@
+-- Warn when the live GIAS CSV carries a (code, name) pair we don't have in
+-- the dictionary seed — i.e. DfE added or renamed a value. Fix by rerunning
+-- pipeline/scripts/generate_gias_codes.py and committing the regenerated
+-- dictionaries + seed together.
+{{ config(severity='warn') }}
+
+with raw_pairs as (
+    {% for field_key, code_col, name_col in [
+        ('school_type',          'TypeOfEstablishment (code)',  'TypeOfEstablishment (name)'),
+        ('establishment_status', 'EstablishmentStatus (code)',  'EstablishmentStatus (name)'),
+        ('phase_of_education',   'PhaseOfEducation (code)',     'PhaseOfEducation (name)'),
+        ('official_sixth_form',  'OfficialSixthForm (code)',    'OfficialSixthForm (name)'),
+        ('religious_character',  'ReligiousCharacter (code)',   'ReligiousCharacter (name)'),
+        ('admissions_policy',    'AdmissionsPolicy (code)',     'AdmissionsPolicy (name)')
+    ] %}
+    select distinct
+        '{{ field_key }}' as field,
+        cast(nullif(trim("{{ code_col }}"), '') as integer) as code,
+        nullif(trim("{{ name_col }}"), '') as name
+    from {{ source('raw', 'gias_establishments') }}
+    where nullif(trim("{{ code_col }}"), '') is not null
+      and nullif(trim("{{ name_col }}"), '') is not null
+    {% if not loop.last %}union all{% endif %}
+    {% endfor %}
+)
+
+select r.*
+from raw_pairs r
+left join {{ ref('gias_code_names') }} s
+    on s.field = r.field
+   and s.code = r.code
+   and s.name = r.name
+where s.field is null