Merge pull request 'feat: verify what we actually deployed, and publish data all-or-nothing' (#148) from feat/release-identity-and-reliability-gate into main

Reviewed-on: #148

.gitea/workflows/deploy.yml

@@ -5,6 +5,11 @@ on:
     branches:
       - main
 
+# Serialise the entire build/deploy/test cycle: no other run can move staging tags.
+concurrency:
+  group: staging-release
+  cancel-in-progress: false
+
 env:
   REGISTRY: privaterepo.sitaru.org
   BACKEND_IMAGE_NAME: ${{ gitea.repository }}-backend
@@ -12,7 +17,18 @@ env:
   PIPELINE_IMAGE_NAME: ${{ gitea.repository }}-pipeline
 
 jobs:
+  prepare:
+    runs-on: ubuntu-latest
+    outputs:
+      build_id: ${{ steps.identity.outputs.build_id }}
+    steps:
+      - id: identity
+        run: python3 -c 'import uuid; print("build_id=" + uuid.uuid4().hex)' >> "$GITHUB_OUTPUT"
+
   build-backend:
+    needs: [prepare]
+    outputs:
+      digest: ${{ steps.build.outputs.digest }}
     name: Build Backend (FastAPI)
     runs-on: ubuntu-latest
     steps:
@@ -46,17 +62,24 @@ jobs:
             type=raw,value=staging
 
       - name: Build and push Backend Docker image
+        id: build
         uses: docker/build-push-action@v5
         with:
           context: .
           file: ./Dockerfile
           push: true
+          build-args: |
+            BUILD_SHA=${{ gitea.sha }}
+            BUILD_ID=${{ needs.prepare.outputs.build_id }}
           tags: ${{ steps.meta-backend.outputs.tags }}
           labels: ${{ steps.meta-backend.outputs.labels }}
           cache-from: type=registry,ref=${{ env.REGISTRY }}/${{ env.BACKEND_IMAGE_NAME }}:buildcache
           cache-to: type=registry,ref=${{ env.REGISTRY }}/${{ env.BACKEND_IMAGE_NAME }}:buildcache,mode=max
 
   build-frontend:
+    needs: [prepare]
+    outputs:
+      digest: ${{ steps.build.outputs.digest }}
     name: Build Frontend (Next.js)
     runs-on: ubuntu-latest
     steps:
@@ -90,18 +113,23 @@ jobs:
             type=raw,value=staging
 
       - name: Build and push Frontend Docker image
+        id: build
         uses: docker/build-push-action@v5
         with:
           context: ./nextjs-app
           file: ./nextjs-app/Dockerfile
           push: true
+          build-args: |
+            BUILD_SHA=${{ gitea.sha }}
+            BUILD_ID=${{ needs.prepare.outputs.build_id }}
           tags: ${{ steps.meta-frontend.outputs.tags }}
           labels: ${{ steps.meta-frontend.outputs.labels }}
-          build-args: |
-            FASTAPI_URL=http://backend:80/api
           # Cache disabled due to registry size limits
 
   build-pipeline:
+    needs: [prepare]
+    outputs:
+      digest: ${{ steps.build.outputs.digest }}
     name: Build Pipeline (Meltano + dbt + Airflow)
     runs-on: ubuntu-latest
     steps:
@@ -135,11 +163,15 @@ jobs:
             type=raw,value=staging
 
       - name: Build and push Pipeline Docker image
+        id: build
         uses: docker/build-push-action@v5
         with:
           context: ./pipeline
           file: ./pipeline/Dockerfile
           push: true
+          build-args: |
+            BUILD_SHA=${{ gitea.sha }}
+            BUILD_ID=${{ needs.prepare.outputs.build_id }}
           tags: ${{ steps.meta-pipeline.outputs.tags }}
           labels: ${{ steps.meta-pipeline.outputs.labels }}
           cache-from: type=registry,ref=${{ env.REGISTRY }}/${{ env.PIPELINE_IMAGE_NAME }}:buildcache
@@ -148,30 +180,23 @@ jobs:
   deploy-staging:
     name: Deploy to Staging
     runs-on: ubuntu-latest
-    needs: [build-backend, build-frontend, build-pipeline]
+    needs: [prepare, build-backend, build-frontend, build-pipeline]
     steps:
       - name: Trigger staging stack update
         run: curl -fsSk -X POST "${{ secrets.PORTAINER_STAGING_WEBHOOK }}"
 
-      - name: Wait for staging to become healthy
-        run: |
-          echo "Polling ${STAGING_BASE_URL} for up to 5 minutes..."
-          for i in $(seq 1 60); do
-            if curl -fsS -o /dev/null --max-time 10 "${STAGING_BASE_URL}/"; then
-              echo "Staging is up (attempt $i)"
-              exit 0
-            fi
-            sleep 5
-          done
-          echo "Staging did not become healthy in time" >&2
-          exit 1
+      - uses: actions/checkout@v4
+      - name: Verify deployed release identity
+        run: python3 scripts/ci/release.py wait
         env:
-          STAGING_BASE_URL: ${{ secrets.STAGING_BASE_URL }}
+          BASE_URL: ${{ secrets.STAGING_BASE_URL }}
+          EXPECTED_SHA: ${{ gitea.sha }}
+          EXPECTED_BUILD_ID: ${{ needs.prepare.outputs.build_id }}
 
   e2e-staging:
     name: E2E Journeys against Staging
     runs-on: ubuntu-latest
-    needs: [deploy-staging]
+    needs: [prepare, deploy-staging, build-backend, build-frontend, build-pipeline]
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -187,11 +212,42 @@ jobs:
           npm ci
           npx playwright install --with-deps chromium
 
+      - name: Verify release before journeys
+        run: python3 scripts/ci/release.py wait --timeout 10
+        env:
+          BASE_URL: ${{ secrets.STAGING_BASE_URL }}
+          EXPECTED_SHA: ${{ gitea.sha }}
+          EXPECTED_BUILD_ID: ${{ needs.prepare.outputs.build_id }}
+
       - name: Run E2E journeys
         working-directory: e2e
         run: npx playwright test
         env:
           BASE_URL: ${{ secrets.STAGING_BASE_URL }}
+          EXPECTED_SHA: ${{ gitea.sha }}
+          EXPECTED_BUILD_ID: ${{ needs.prepare.outputs.build_id }}
+
+      - name: Verify release after journeys
+        run: python3 scripts/ci/release.py wait --timeout 10
+        env:
+          BASE_URL: ${{ secrets.STAGING_BASE_URL }}
+          EXPECTED_SHA: ${{ gitea.sha }}
+          EXPECTED_BUILD_ID: ${{ needs.prepare.outputs.build_id }}
+
+      - uses: docker/setup-buildx-action@v3
+      - uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ gitea.actor }}
+          password: ${{ secrets.REGISTRY_TOKEN }}
+      - name: Mark tested image digests as verified
+        run: python3 scripts/ci/release.py verify
+        env:
+          EXPECTED_SHA: ${{ gitea.sha }}
+          EXPECTED_BUILD_ID: ${{ needs.prepare.outputs.build_id }}
+          BACKEND_DIGEST: ${{ needs.build-backend.outputs.digest }}
+          FRONTEND_DIGEST: ${{ needs.build-frontend.outputs.digest }}
+          PIPELINE_DIGEST: ${{ needs.build-pipeline.outputs.digest }}
 
 # Production deployment is a second, manual approval: see promote.yml
 # ("Promote to Production (manual)") and docs/DEPLOY.md.

.gitea/workflows/pr-checks.yml

@@ -68,13 +68,13 @@ jobs:
           python-version: "3.12"
 
       - name: Install dependencies
-        run: pip install -r requirements.txt pytest "httpx<0.28"
+        run: pip install -r requirements.txt pytest "httpx<0.28" pyyaml
 
       - name: Import smoke test
         run: python -c "from backend.app import app; print('backend imports OK')"
 
       - name: Backend unit tests
-        run: python -m pytest backend/tests -q
+        run: python -m pytest backend/tests pipeline/tests scripts/ci/tests -q
 
   build-backend:
     name: Build Backend (no push)

.gitea/workflows/promote.yml

@@ -97,33 +97,15 @@ jobs:
           username: ${{ gitea.actor }}
           password: ${{ secrets.REGISTRY_TOKEN }}
 
-      - name: Retag approved images as prod (keeping rollback pointer)
-        run: |
-          SHORT_SHA="${{ steps.resolve.outputs.short }}"
-          for IMAGE in \
-            "${REGISTRY}/${BACKEND_IMAGE_NAME}" \
-            "${REGISTRY}/${FRONTEND_IMAGE_NAME}" \
-            "${REGISTRY}/${PIPELINE_IMAGE_NAME}"; do
-            # Keep a rollback pointer before moving :prod
-            docker buildx imagetools create -t "${IMAGE}:prod-previous" "${IMAGE}:prod" || true
-            docker buildx imagetools create -t "${IMAGE}:prod" "${IMAGE}:${SHORT_SHA}"
-            echo "Promoted ${IMAGE}:${SHORT_SHA} -> :prod"
-          done
+      - name: Resolve verified digests and promote the complete image set
+        run: python3 scripts/ci/release.py promote --output release.json
+        env:
+          EXPECTED_SHA: ${{ steps.resolve.outputs.full }}
 
       - name: Trigger production stack update
         run: curl -fsSk -X POST "${{ secrets.PORTAINER_PROD_WEBHOOK }}"
 
-      - name: Wait for production to become healthy
-        run: |
-          echo "Polling ${PROD_BASE_URL} for up to 5 minutes..."
-          for i in $(seq 1 60); do
-            if curl -fsS -o /dev/null --max-time 10 "${PROD_BASE_URL}/"; then
-              echo "Production is up (attempt $i)"
-              exit 0
-            fi
-            sleep 5
-          done
-          echo "Production did not become healthy in time" >&2
-          exit 1
+      - name: Verify production release identity
+        run: python3 scripts/ci/release.py wait --release release.json
         env:
-          PROD_BASE_URL: ${{ secrets.PROD_BASE_URL }}
+          BASE_URL: ${{ secrets.PROD_BASE_URL }}

Dockerfile

@@ -24,6 +24,12 @@ RUN pip install --no-cache-dir -r requirements.txt
 COPY backend/ ./backend/
 COPY scripts/ ./scripts/
 
+ARG BUILD_SHA=development
+ARG BUILD_ID=development
+LABEL io.schoolcompare.build-id=$BUILD_ID
+LABEL io.schoolcompare.commit=$BUILD_SHA
+RUN python -c 'import json,sys; open("backend/build-info.json", "w").write(json.dumps({"sha":sys.argv[1],"build_id":sys.argv[2]}))' "$BUILD_SHA" "$BUILD_ID"
+
 # Expose the application port
 EXPOSE 80
 

backend/app.py

@@ -26,7 +26,8 @@ from starlette.middleware.base import BaseHTTPMiddleware
 import asyncio
 from .config import settings
 from .data_loader import (
-    clear_cache,
+    build_latest_school_data,
+    load_school_data_as_dataframe,
     compute_benchmarks,
     load_school_data,
     load_latest_school_data,
@@ -272,7 +273,7 @@ def _places_payload(urn: int) -> list[dict]:
     return payload
 
 
-def _place_sitemap_rows(kinds: tuple[str, ...]) -> list[str]:
+def _place_sitemap_rows(kinds: tuple[str, ...], registry=None) -> list[str]:
     """A <url> per place, plus a phase variant wherever that phase clears the
     threshold on its own.
 
@@ -282,7 +283,9 @@ def _place_sitemap_rows(kinds: tuple[str, ...]) -> list[str]:
     linked from the place page either.
     """
     rows: list[str] = []
-    for p in sorted(get_place_registry().values(), key=lambda p: (p.kind, p.slug)):
+    if registry is None:
+        registry = get_place_registry()
+    for p in sorted(registry.values(), key=lambda p: (p.kind, p.slug)):
         if p.kind not in kinds:
             continue
         rows.append(_url_element(BASE_URL + _place_url(p)))
@@ -296,9 +299,10 @@ def _place_sitemap_rows(kinds: tuple[str, ...]) -> list[str]:
     return rows
 
 
-def build_sitemaps() -> dict[str, str]:
+def build_sitemaps(df=None, registry=None) -> dict[str, str]:
     """Build the sitemap index and every child, keyed by name."""
-    df = load_school_data()
+    if df is None:
+        df = load_school_data()
 
     children: dict[str, str] = {
         "static.xml": _urlset(
@@ -318,7 +322,7 @@ def build_sitemaps() -> dict[str, str]:
     # measured apart from the school pages'.
     for label, kinds in (("places", ("town", "locality", "authority")),
                          ("outcodes", ("outcode",))):
-        rows = _place_sitemap_rows(kinds)
+        rows = _place_sitemap_rows(kinds, registry)
         chunks = [rows[i:i + SITEMAP_CHUNK_SIZE]
                   for i in range(0, len(rows), SITEMAP_CHUNK_SIZE)] or [[]]
         for n, chunk in enumerate(chunks, start=1):
@@ -700,6 +704,15 @@ async def get_config():
     }
 
 
+@app.get("/api/release")
+async def release_identity():
+    import json
+    from pathlib import Path
+    path = Path(__file__).with_name("build-info.json")
+    identity = json.loads(path.read_text()) if path.exists() else {"sha": "development", "build_id": "development"}
+    return JSONResponse(identity, headers={"Cache-Control": "no-store"})
+
+
 @app.get("/api/schools")
 @limiter.limit(f"{settings.rate_limit_per_minute}/minute")
 async def get_schools(
@@ -736,7 +749,7 @@ async def get_schools(
     df_latest = load_latest_school_data()
 
     if df_latest.empty:
-        return {"schools": [], "total": 0, "page": page, "page_size": 0}
+        raise HTTPException(status_code=503, detail="School data temporarily unavailable")
 
     # Use configured default if not specified
     if page_size is None:
@@ -835,8 +848,8 @@ async def get_schools(
 
     # Apply filters
     if search:
-        ts_urns = search_schools_typesense(search)
-        if ts_urns:
+        ts_urns = await asyncio.to_thread(search_schools_typesense, search)
+        if ts_urns is not None:
             urn_order = {urn: i for i, urn in enumerate(ts_urns)}
             schools_df = schools_df[schools_df["urn"].isin(set(ts_urns))].copy()
             schools_df["_ts_rank"] = schools_df["urn"].map(urn_order)
@@ -844,9 +857,9 @@ async def get_schools(
         else:
             # Fallback: Typesense unavailable, use substring match
             search_lower = search.lower()
-            mask = schools_df["school_name"].str.lower().str.contains(search_lower, na=False)
+            mask = schools_df["school_name"].str.lower().str.contains(search_lower, na=False, regex=False)
             if "address" in schools_df.columns:
-                mask = mask | schools_df["address"].str.lower().str.contains(search_lower, na=False)
+                mask = mask | schools_df["address"].str.lower().str.contains(search_lower, na=False, regex=False)
             schools_df = schools_df[mask]
 
     if local_authority:
@@ -905,7 +918,7 @@ async def get_school_details(request: Request, urn: int):
     df = load_school_data()
 
     if df.empty:
-        raise HTTPException(status_code=404, detail="No data available")
+        raise HTTPException(status_code=503, detail="School data temporarily unavailable")
 
     school_data = df[df["urn"] == urn]
 
@@ -1542,20 +1555,51 @@ async def get_data_info(request: Request):
     }
 
 
+_publication_lock = asyncio.Lock()
+
+
+def _prepare_publication(df):
+    if df.empty:
+        raise ValueError("Refusing to publish an empty school dataset")
+    if not {"urn", "year", "school_name"}.issubset(df.columns):
+        raise ValueError("School dataset is missing required columns")
+    if df["urn"].isna().any() or df.duplicated(["urn", "year"]).any():
+        raise ValueError("School dataset has missing URNs or duplicate school years")
+    latest = build_latest_school_data(df)
+    registry = build_place_registry(df)
+    index = build_place_index(registry)
+    sitemaps = build_sitemaps(df, registry)
+    return df, latest, registry, index, sitemaps
+
+
+def _publish(prepared):
+    # Called on the event loop with no await: routes cannot observe half a swap.
+    # The application currently runs one worker; replicas require coordination.
+    from . import data_loader
+    global _place_registry, _place_index, _place_index_source, _sitemaps
+    df, latest, registry, index, sitemaps = prepared
+    data_loader._df_cache = df
+    data_loader._df_latest_cache = latest
+    _place_registry = registry
+    _place_index = index
+    _place_index_source = registry
+    _sitemaps = sitemaps
+
+
 @app.post("/api/admin/reload")
 @limiter.limit("5/minute")
-async def reload_data(
-    request: Request,
-    _: bool = Depends(verify_admin_api_key)
-):
-    """
-    Admin endpoint to force data reload (useful after data updates).
-    Requires X-API-Key header with valid admin API key.
-    """
-    clear_cache()
-    await asyncio.to_thread(load_school_data)
-    await asyncio.to_thread(load_latest_school_data)
-    return {"status": "reloaded"}
+async def reload_data(request: Request, _: bool = Depends(verify_admin_api_key)):
+    """Validate a complete replacement before publishing it; retain data on failure."""
+    async with _publication_lock:
+        try:
+            df = await asyncio.to_thread(load_school_data_as_dataframe)
+            prepared = await asyncio.to_thread(_prepare_publication, df)
+        except Exception as exc:
+            import logging
+            logging.getLogger(__name__).exception("Dataset reload failed")
+            raise HTTPException(status_code=503, detail="Dataset reload failed; previous data retained") from exc
+        _publish(prepared)
+    return {"status": "reloaded", "schools": len(prepared[1])}
 
 
 
@@ -1607,15 +1651,16 @@ async def regenerate_sitemap(
     request: Request,
     _: bool = Depends(verify_admin_api_key),
 ):
-    """Rebuild and cache the sitemap from current school data. Called by Airflow after data updates."""
-    global _sitemaps, _place_registry
-    # Places and sitemap are rebuilt together — they read the same marts, and
-    # letting them drift apart would submit URLs for places that no longer
-    # exist.
-    _place_registry = None
-    _sitemaps = build_sitemaps()
-    n = sum(x.count("<url>") for x in _sitemaps.values())
-    return {"status": "ok", "urls": n, "sitemaps": len(_sitemaps)}
+    """Rebuild derived publication data without clearing the live registry."""
+    async with _publication_lock:
+        try:
+            prepared = await asyncio.to_thread(_prepare_publication, load_school_data())
+        except Exception as exc:
+            raise HTTPException(status_code=503, detail="Sitemap rebuild failed; previous data retained") from exc
+        _publish(prepared)
+    n = sum(x.count("<url>") for x in prepared[4].values())
+    return {"status": "ok", "urls": n, "sitemaps": len(prepared[4])}
+
 
 
 # Mount static files directly (must be after all routes to avoid catching API calls)

backend/data_loader.py

@@ -84,21 +84,58 @@ def _get_typesense_client():
         return None
 
 
-def search_schools_typesense(query: str, limit: int = 250) -> List[int]:
-    """Search Typesense. Returns URNs in relevance order, or [] if unavailable."""
+SEARCH_PAGE_SIZE = 250
+# Search results are filtered again by the API (authority, phase, postcode,
+# etc.), so one page is too small for scoped searches. Keep the candidate set
+# bounded, though: a broad query must not turn into an unbounded sequence of
+# Typesense requests. Four pages is enough to preserve useful scoped matches
+# while putting a hard ceiling on latency and upstream load.
+SEARCH_MAX_CANDIDATES = 1_000
+
+
+def search_schools_typesense(query: str) -> Optional[List[int]]:
+    """Return a bounded set of matching URNs in relevance order.
+
+    ``None`` means Typesense is unavailable; ``[]`` is a valid zero-match
+    result. The API applies its remaining filters after this search, so the
+    first few pages are fetched rather than only the first page. Once the
+    candidate ceiling is reached, the relevance-ordered prefix is returned on
+    purpose; fetching every match would make common or adversarial queries
+    unbounded.
+    """
     client = _get_typesense_client()
     if client is None:
-        return []
+        return None
+    urns: list[int] = []
+    fetched = 0
     try:
-        result = client.collections["schools"].documents.search({
-            "q": query,
-            "query_by": "school_name,local_authority,postcode",
-            "per_page": min(limit, 250),
-            "typo_tokens_threshold": 1,
-        })
-        return [int(h["document"]["urn"]) for h in result.get("hits", [])]
+        page = 1
+        while fetched < SEARCH_MAX_CANDIDATES:
+            page_size = min(SEARCH_PAGE_SIZE, SEARCH_MAX_CANDIDATES - fetched)
+            result = client.collections["schools"].documents.search({
+                "q": query,
+                "query_by": "school_name,local_authority,postcode",
+                "per_page": page_size,
+                "page": page,
+                "typo_tokens_threshold": 1,
+            })
+            hits = result.get("hits", [])
+            urns.extend(int(h["document"]["urn"]) for h in hits)
+            fetched += len(hits)
+            if fetched >= result.get("found", fetched):
+                return list(dict.fromkeys(urns))
+            if not hits:
+                raise ValueError("Search pagination ended before all matches arrived")
+            page += 1
+        logging.getLogger(__name__).info(
+            "Typesense search capped at %d candidates for query %r",
+            SEARCH_MAX_CANDIDATES,
+            query,
+        )
+        return list(dict.fromkeys(urns))
     except Exception:
-        return []
+        logging.getLogger(__name__).exception("School search unavailable")
+        return None
 
 
 # The most a public endpoint will return in one response.
@@ -502,7 +539,12 @@ def load_latest_school_data() -> pd.DataFrame:
     if _df_latest_cache is not None:
         return _df_latest_cache
 
-    df = load_school_data()
+    _df_latest_cache = build_latest_school_data(load_school_data())
+    return _df_latest_cache
+
+
+def build_latest_school_data(df: pd.DataFrame) -> pd.DataFrame:
+    """Build a replacement snapshot without mutating the published caches."""
     if df.empty:
         return df
 
@@ -535,8 +577,7 @@ def load_latest_school_data() -> pd.DataFrame:
     df_latest = pd.concat([df_latest, df_no_perf], ignore_index=True)
 
     print(f"Latest-snapshot cache built: {len(df_latest)} schools")
-    _df_latest_cache = df_latest
-    return _df_latest_cache
+    return df_latest
 
 
 def clear_cache():

backend/tests/test_publication.py

@@ -0,0 +1,68 @@
+"""Publication must preserve the current dataset until every replacement is ready."""
+import asyncio
+import pandas as pd
+import pytest
+from fastapi.testclient import TestClient
+from backend import app as api, data_loader
+from backend.tests.test_sixth_form_flag import _schools_df
+
+
+@pytest.fixture
+def client(monkeypatch):
+    old = _schools_df()
+    monkeypatch.setattr(data_loader, '_df_cache', old)
+    monkeypatch.setattr(data_loader, '_df_latest_cache', old)
+    monkeypatch.setattr(api, '_place_registry', {'old': 'registry'})
+    monkeypatch.setattr(api, '_place_index', {'old': 'index'})
+    monkeypatch.setattr(api, '_place_index_source', api._place_registry)
+    monkeypatch.setattr(api, '_sitemaps', {'old.xml': 'old sitemap'})
+    monkeypatch.setattr(api, '_publication_lock', asyncio.Lock())
+    monkeypatch.setattr(api.limiter, 'enabled', False)
+    api.app.dependency_overrides[api.verify_admin_api_key] = lambda: True
+    yield TestClient(api.app, raise_server_exceptions=False)
+    api.app.dependency_overrides.clear()
+
+
+def state():
+    return (data_loader._df_cache, data_loader._df_latest_cache, api._place_registry,
+            api._place_index, api._place_index_source, api._sitemaps)
+
+
+@pytest.mark.parametrize('failure', ['empty', 'database', 'sitemap', 'duplicate'])
+def test_failed_reload_preserves_every_published_object(client, monkeypatch, failure):
+    before = state()
+    df = _schools_df()
+    if failure == 'empty':
+        df = pd.DataFrame()
+    if failure == 'duplicate':
+        df = pd.concat([df, df.iloc[:1]], ignore_index=True)
+    def load():
+        if failure == 'database':
+            raise RuntimeError('database unavailable')
+        return df
+    monkeypatch.setattr(api, 'load_school_data_as_dataframe', load)
+    if failure == 'sitemap':
+        monkeypatch.setattr(api, 'build_sitemaps', lambda *args: (_ for _ in ()).throw(RuntimeError('bad XML')))
+    response = client.post('/api/admin/reload')
+    assert response.status_code == 503
+    assert all(a is b for a, b in zip(before, state()))
+
+
+def test_success_publishes_school_data_places_and_sitemaps(client, monkeypatch):
+    df = _schools_df()
+    df.loc[0, 'school_name'] = 'Replacement School'
+    monkeypatch.setattr(api, 'load_school_data_as_dataframe', lambda: df)
+    response = client.post('/api/admin/reload')
+    assert response.status_code == 200
+    assert data_loader.load_school_data() is df
+    assert data_loader.load_latest_school_data().iloc[0].school_name == 'Replacement School'
+    assert api._place_index_source is api._place_registry
+    assert 'old.xml' not in api._sitemaps
+    assert 'replacement-school' in api._sitemaps['schools-1.xml']
+
+
+def test_failed_sitemap_regeneration_keeps_existing_publication(client, monkeypatch):
+    before = state()
+    monkeypatch.setattr(api, 'build_sitemaps', lambda *args: (_ for _ in ()).throw(RuntimeError('bad XML')))
+    assert client.post('/api/admin/regenerate-sitemap').status_code == 503
+    assert all(a is b for a, b in zip(before, state()))

backend/tests/test_search_completeness.py

@@ -0,0 +1,114 @@
+from types import SimpleNamespace
+import pytest
+from fastapi.testclient import TestClient
+from backend import app as api, data_loader
+from backend.tests.test_sixth_form_flag import _schools_df
+
+
+def client_for(monkeypatch, search):
+    client = SimpleNamespace(collections={'schools': SimpleNamespace(documents=SimpleNamespace(search=search))})
+    monkeypatch.setattr(data_loader, '_get_typesense_client', lambda: client)
+
+
+def test_search_returns_matches_beyond_first_page(monkeypatch):
+    pages = []
+    def search(params):
+        pages.append(params['page'])
+        urns = range(100000, 100250) if params['page'] == 1 else [100999]
+        return {'found': 251, 'hits': [{'document': {'urn': u}} for u in urns]}
+    client_for(monkeypatch, search)
+    result = data_loader.search_schools_typesense('academy')
+    assert len(result) == 251
+    assert result[-1] == 100999
+    assert pages == [1, 2]
+
+
+def test_search_caps_broad_queries_at_a_bounded_number_of_pages(monkeypatch):
+    requests = []
+
+    def search(params):
+        requests.append(params)
+        return {
+            'found': 10_000,
+            'hits': [
+                {'document': {'urn': 100000 + params['page'] * 1000 + i}}
+                for i in range(params['per_page'])
+            ],
+        }
+
+    client_for(monkeypatch, search)
+    result = data_loader.search_schools_typesense('school')
+
+    assert len(result) == data_loader.SEARCH_MAX_CANDIDATES
+    assert len(requests) == data_loader.SEARCH_MAX_CANDIDATES // data_loader.SEARCH_PAGE_SIZE
+    assert all(request['per_page'] == data_loader.SEARCH_PAGE_SIZE for request in requests)
+    assert requests[-1]['page'] == len(requests)
+
+
+def test_search_uses_a_smaller_final_page_when_the_cap_is_not_a_page_multiple(monkeypatch):
+    monkeypatch.setattr(data_loader, 'SEARCH_MAX_CANDIDATES', 251)
+    requests = []
+
+    def search(params):
+        requests.append(params)
+        return {
+            'found': 10_000,
+            'hits': [{'document': {'urn': 100000 + len(requests) * 1000 + i}}
+                     for i in range(params['per_page'])],
+        }
+
+    client_for(monkeypatch, search)
+    result = data_loader.search_schools_typesense('school')
+
+    assert len(result) == 251
+    assert [request['per_page'] for request in requests] == [250, 1]
+
+
+def test_later_page_failure_does_not_return_partial_results(monkeypatch):
+    def search(params):
+        if params['page'] == 2:
+            raise RuntimeError('timeout')
+        return {'found': 251, 'hits': [{'document': {'urn': u}} for u in range(100000, 100250)]}
+    client_for(monkeypatch, search)
+    assert data_loader.search_schools_typesense('academy') is None
+
+
+def test_zero_matches_are_distinct_from_unavailable(monkeypatch):
+    client_for(monkeypatch, lambda _: {'found': 0, 'hits': []})
+    assert data_loader.search_schools_typesense('academy') == []
+    monkeypatch.setattr(data_loader, '_get_typesense_client', lambda: None)
+    assert data_loader.search_schools_typesense('academy') is None
+
+
+@pytest.mark.parametrize('matches, expected', [([], []), (None, [100001])])
+def test_fallback_only_on_dependency_failure(monkeypatch, matches, expected):
+    monkeypatch.setattr(api.limiter, 'enabled', False)
+    monkeypatch.setattr(api, 'load_latest_school_data', _schools_df)
+    monkeypatch.setattr(api, 'search_schools_typesense', lambda _: matches)
+    response = TestClient(api.app).get('/api/schools?search=Alpha')
+    assert response.status_code == 200
+    assert [s['urn'] for s in response.json()['schools']] == expected
+
+
+def test_filtered_api_keeps_match_from_second_search_page(monkeypatch):
+    monkeypatch.setattr(api.limiter, 'enabled', False)
+    df = _schools_df()
+    monkeypatch.setattr(api, 'load_latest_school_data', lambda: df)
+    def search(params):
+        urns = range(200000, 200250) if params['page'] == 1 else [100001]
+        return {'found': 251, 'hits': [{'document': {'urn': u}} for u in urns]}
+    client_for(monkeypatch, search)
+    response = TestClient(api.app).get('/api/schools?search=Alpha&local_authority=Testshire')
+    assert response.status_code == 200
+    assert response.json()['total'] == 1
+    assert response.json()['schools'][0]['urn'] == 100001
+
+
+def test_unavailable_dataset_is_not_a_missing_school_or_empty_search(monkeypatch):
+    import pandas as pd
+    monkeypatch.setattr(api.limiter, 'enabled', False)
+    monkeypatch.setattr(api, 'load_school_data', lambda: pd.DataFrame())
+    monkeypatch.setattr(api, 'load_latest_school_data', lambda: pd.DataFrame())
+    client = TestClient(api.app)
+    assert client.get('/api/schools/100001').status_code == 503
+    assert client.get('/api/schools?search=school').status_code == 503

docs/ARCHITECTURE.md

@@ -80,9 +80,11 @@ API types. `payload-types.ts` and the Payload import map are generated artifacts
 
 1. Airflow DAGs extract and validate source data, then run selected dbt builds.
 2. Relevant DAGs rebuild Typesense and swap the `schools` alias.
-3. They call `POST /api/admin/reload` with `X-API-Key` to refresh school DataFrames.
-4. A separate weekly sitemap DAG calls `POST /api/admin/regenerate-sitemap`,
-   rebuilding places and sitemaps.
+3. They call `POST /api/admin/reload` with `X-API-Key`. It builds and validates
+   replacement DataFrames, places, reverse membership and sitemaps off the request
+   loop, then publishes them together. Failure returns 503 and preserves live data.
+4. A separate weekly sitemap DAG can regenerate the derived publication from the
+   current DataFrame without clearing the live registry first.
 
 GIAS is scheduled daily, Ofsted monthly, and annual datasets are manually
 triggered. The DAG definitions are authoritative for selectors and dependencies.
@@ -92,16 +94,27 @@ backend HTTP Cache-Control/ETags, Next.js fetch/page revalidation, and browser o
 shared HTTP caches where configured. Place fetches request a one-week revalidation
 interval. HTTP ETags are computed after route execution, not before database work.
 
-Known limitations: reload clears the old DataFrames before verifying replacement
-data; places/sitemaps refresh separately; Next.js caches are not explicitly purged
-by the pipeline; Typesense import results are not validated before alias publication.
-Do not describe this sequence as an atomic dataset release. These are follow-up
-reliability tasks, not changes implemented by the documentation cleanup.
+Typesense publication validates every import response and the final document
+count before switching aliases. A session-scoped PostgreSQL advisory lock
+serialises index reads/publication across DAGs. The previous collection remains
+available for rollback; old unaliased collections are pruned after success.
+Failed drafts are retained until a later successful cleanup, because an uncertain
+alias-update response must never cause deletion of a potentially live index.
+
+The backend snapshot swap is process-local and assumes the current single-worker
+deployment. It is not an atomic transaction spanning PostgreSQL marts, Typesense
+and Next.js caches. Next.js caches are not explicitly purged by the pipeline.
+School search retrieves a relevance-ordered candidate prefix (currently capped at
+1,000 URNs) before applying API filters. This keeps scoped searches useful while
+putting a hard ceiling on Typesense round trips; only a dependency failure invokes
+substring fallback, not a valid empty match set.
 
 ## Deployment references
 
 See [DEPLOY.md](DEPLOY.md). PR checks include frontend typechecking/tests, backend
 unit tests, image builds and AI review. Staging journeys run after merging.
-Production promotion retags a selected commit's images. Current health polling
-checks HTTP success, not the deployed commit identity; overlapping staging runs
-remain a release-verification concern.
+Staging runs are serialised across builds, deployment and E2E. Build-stamped
+frontend/backend identities are checked before and after journeys. Only then are
+the captured image digests marked verified. Promotion resolves and validates the
+complete verified image set before retagging production. See the runbook for
+first-rollout requirements and remaining integration checks.

docs/DEPLOY.md

@@ -19,8 +19,9 @@ PR checks (.gitea/workflows/pr-checks.yml)
    ▼
 Stage pipeline (.gitea/workflows/deploy.yml) — automatic
    1. build & push images        → tags sha-<sha>, staging
-   2. staging Portainer webhook  → wait for staging health
+   2. staging Portainer webhook  → verify frontend/backend SHA + build ID
    3. Playwright E2E journeys against staging      ← gate before human testing
+   4. verify identity again; tag tested digests verified-<full-sha>
    ▼
 Manual testing on staging (stx.schoolcompare.co.uk)
    │  Actions → "Promote to Production (manual)"   ← approval #2
@@ -28,14 +29,15 @@ Manual testing on staging (stx.schoolcompare.co.uk)
 Promote pipeline (.gitea/workflows/promote.yml) — manual dispatch
    1. resolve target sha (input, or latest main if empty)
    2. REFUSE unless that commit's "E2E Journeys against Staging" status is green
-   3. retag sha-<sha> → :prod    (same bytes — build once, promote the image)
+   3. resolve verified-<full-sha> digests, validate labels, retag digests → :prod
       previous :prod saved as :prod-previous
-   4. prod Portainer webhook     → wait for prod health
+   4. prod Portainer webhook     → verify expected SHA + build ID
 ```
 
 Key principle: **build once, promote the exact image**. Production pins `:prod`,
 which only moves when a human runs the promote workflow — and the workflow
-only accepts commits that passed the staging E2E gate. Nothing tags `:latest`
+only accepts commits that passed the staging E2E gate and have a complete verified
+image set. Nothing tags `:latest`
 anymore.
 
 ## Branch & PR workflow
@@ -256,3 +258,41 @@ how long any feature is exposed to this.
 If `UNLEASH_URL` is unset, every flag is `False` and no connection is
 attempted. That is the correct behaviour for local development and CI, and it
 means the test suites need no flag server.
+
+## Release identity and the P1 reliability gate
+
+Every staging run creates a random build ID before building its three images.
+Each image carries the commit and build ID as labels. Frontend/backend images
+also contain a build-time JSON file; environment overrides cannot rewrite it.
+`/release.json` returns both identities with `Cache-Control: no-store`. It fails
+with 503 when either identity cannot be read. FastAPI's internal endpoint is
+`/api/release`.
+
+The entire staging workflow shares one concurrency group, with cancellation
+disabled. This needs Gitea 1.26 or newer, where workflow concurrency is supported
+([release notes](https://blog.gitea.com/release-of-1.26.0/)); the configured server
+reported 1.27.3 during this change. Do not run the workflow on an older server
+that ignores the concurrency key. Manual deployments outside this workflow must
+also avoid changing staging during journeys.
+
+The gate checks both identities before and after Playwright. It then validates
+labels on the captured build output digests and tags them `verified-<full-sha>`.
+The manual promotion script resolves all three verified tags to immutable digests
+and confirms one matching commit/build ID before moving any `:prod` tag. It polls
+production for that same identity using a locally saved release manifest.
+A registry error can still interrupt the three tag writes; the Portainer webhook
+only runs after successful promotion, and rerunning promotion resolves the full
+verified set again. There is no cross-registry atomic tag transaction.
+
+**First rollout:** old green commits without verified tags/build identities are
+not promotable through this gate. Build and test a commit containing the new
+workflow first. The release route must be reachable through the configured
+`STAGING_BASE_URL`/`PROD_BASE_URL`; it deliberately avoids the public staging
+`/api` proxy limitation. No new deployment secret is required.
+
+`scripts/ci/release.py` implements identity polling and digest verification.
+Its mocked tests run in PR checks alongside backend and index-publication tests.
+The new Playwright journeys also check deployed identity and stale pagination.
+Local unit checks do not validate registry credentials, Portainer behaviour,
+proxy routing or a deployed image; those require the staging run. Production
+promotion remains a separate human action.

docs/DEVELOPMENT.md

@@ -42,8 +42,8 @@ From the repository root, using an available Python 3.11 or 3.12 interpreter:
 
 ```sh
 python3.11 -m venv /tmp/schoolcompare-backend-venv
-/tmp/schoolcompare-backend-venv/bin/python -m pip install -r requirements.txt pytest 'httpx<0.28'
-/tmp/schoolcompare-backend-venv/bin/python -m pytest backend/tests -q
+/tmp/schoolcompare-backend-venv/bin/python -m pip install -r requirements.txt pytest 'httpx<0.28' pyyaml
+/tmp/schoolcompare-backend-venv/bin/python -m pytest backend/tests pipeline/tests scripts/ci/tests -q
 ```
 
 Substitute `python3.12` if matching PR CI. The test dependencies above match the

e2e/tests/reliability.spec.ts

@@ -0,0 +1,41 @@
+import { test, expect, Route } from '@playwright/test';
+
+test('the deployed frontend and backend report the tested build', async ({ request }) => {
+  const response = await request.get('/release.json');
+  expect(response.ok()).toBeTruthy();
+  expect(response.headers()['cache-control']).toContain('no-store');
+  const identity = await response.json();
+  expect(identity.frontend).toEqual(identity.backend);
+  expect(identity.frontend.sha).toMatch(/^[a-f0-9]{40}$/);
+  expect(identity.frontend.build_id).toMatch(/^[a-f0-9]{32}$/);
+  if (process.env.EXPECTED_SHA) expect(identity.frontend.sha).toBe(process.env.EXPECTED_SHA);
+  if (process.env.EXPECTED_BUILD_ID) expect(identity.frontend.build_id).toBe(process.env.EXPECTED_BUILD_ID);
+});
+
+test('changing search while loading another page does not append old results', async ({ page }) => {
+  await page.goto('/?phase=primary');
+  await expect(page.getByRole('button', { name: 'Load more schools' })).toBeVisible();
+  let received!: (route: Route) => void;
+  const pending = new Promise<Route>(resolve => { received = resolve; });
+  await page.route('**/api/schools?**', async route => {
+    if (new URL(route.request().url()).searchParams.get('page') === '2') {
+      received(route);
+      return;
+    }
+    await route.continue();
+  });
+  await page.getByRole('button', { name: 'Load more schools' }).click();
+  const oldRequest = await pending;
+  const search = page.getByPlaceholder('School name or postcode').first();
+  await search.fill('secondary');
+  await search.press('Enter');
+  await page.waitForURL(/search=secondary/);
+  // A cancelled fetch may prevent route fulfilment altogether; either way,
+  // this deliberately late response must not become part of the new results.
+  await oldRequest.fulfill({ json: {
+    schools: [{ urn: 999998, school_name: 'P1 stale result sentinel', phase: 'Primary' }],
+    total: 2, page: 2, page_size: 1, total_pages: 2,
+  } }).catch(() => {});
+  await expect(page.getByText('P1 stale result sentinel')).toHaveCount(0);
+  await expect(page.getByRole('button', { name: 'Loading...' })).toHaveCount(0);
+});

nextjs-app/Dockerfile

@@ -28,6 +28,10 @@ ENV NODE_ENV=production
 ARG FASTAPI_URL=http://backend:80/api
 ENV FASTAPI_URL=${FASTAPI_URL}
 
+ARG BUILD_SHA=development
+ARG BUILD_ID=development
+RUN node -e 'require("fs").writeFileSync("build-info.json", JSON.stringify({sha:process.argv[1],build_id:process.argv[2]}))' "$BUILD_SHA" "$BUILD_ID"
+
 # Build application
 RUN npm run build
 
@@ -70,6 +74,12 @@ USER nextjs
 EXPOSE 3000
 
 # Set environment variables
+ARG BUILD_SHA=development
+ARG BUILD_ID=development
+LABEL io.schoolcompare.build-id=$BUILD_ID
+LABEL io.schoolcompare.commit=$BUILD_SHA
+COPY --from=builder /app/build-info.json ./build-info.json
+
 ENV PORT=3000
 ENV HOSTNAME="0.0.0.0"
 

nextjs-app/__tests__/app/dataFailures.test.tsx

@@ -0,0 +1,51 @@
+import { fireEvent, render, screen } from '@testing-library/react';
+import HomePage from '@/app/(frontend)/page';
+import SchoolPage from '@/app/(frontend)/school/[slug]/page';
+import ErrorPage from '@/app/(frontend)/error';
+import { APIFetchError, fetchSchools, fetchFilters, fetchSchoolDetails } from '@/lib/api';
+import { fetchPlace, fetchPlaces } from '@/lib/places';
+
+jest.mock('@/lib/api', () => ({
+  ...jest.requireActual('@/lib/api'),
+  fetchSchools: jest.fn(),
+  fetchSchoolDetails: jest.fn(),
+  fetchFilters: jest.fn(async () => ({})),
+  fetchDataInfo: jest.fn(async () => null),
+  fetchNationalAverages: jest.fn(async () => null),
+}));
+jest.mock('@/lib/flags', () => ({ getFlags: jest.fn(async () => ({})) }));
+jest.mock('next/navigation', () => ({
+  notFound: () => { throw new Error('NEXT_NOT_FOUND'); },
+  redirect: jest.fn(),
+}));
+const realFetch = global.fetch;
+afterEach(() => { global.fetch = realFetch; jest.clearAllMocks(); });
+
+test('school outages propagate; only a real 404 becomes not found', async () => {
+  const request = { params: Promise.resolve({ slug: '100001-school' }) };
+  const outage = new APIFetchError('unavailable', 503);
+  jest.mocked(fetchSchoolDetails).mockRejectedValueOnce(outage);
+  await expect(SchoolPage(request)).rejects.toBe(outage);
+  jest.mocked(fetchSchoolDetails).mockRejectedValueOnce(new APIFetchError('missing', 404));
+  await expect(SchoolPage(request)).rejects.toThrow('NEXT_NOT_FOUND');
+});
+
+test('homepage search failure is not returned as an empty successful page', async () => {
+  jest.mocked(fetchSchools).mockRejectedValueOnce(new APIFetchError('unavailable', 503));
+  await expect(HomePage({ searchParams: Promise.resolve({ search: 'school' }) })).rejects.toThrow('unavailable');
+});
+
+test('a place is absent only on 404; other failures propagate', async () => {
+  global.fetch = jest.fn().mockResolvedValue({ ok: false, status: 404 });
+  await expect(fetchPlace('town', 'example')).resolves.toBeNull();
+  jest.mocked(global.fetch).mockResolvedValue({ ok: false, status: 503 } as Response);
+  await expect(fetchPlace('town', 'example')).rejects.toMatchObject({ status: 503 });
+  await expect(fetchPlaces()).rejects.toMatchObject({ status: 503 });
+});
+
+test('the error boundary offers a retry without showing an empty search', () => {
+  const reset = jest.fn();
+  render(<ErrorPage error={new Error('offline')} reset={reset} />);
+  fireEvent.click(screen.getByRole('button', { name: 'Try again' }));
+  expect(reset).toHaveBeenCalledTimes(1);
+});

nextjs-app/__tests__/app/releaseIdentity.test.ts

@@ -0,0 +1,30 @@
+/** @jest-environment node */
+import { GET } from '@/app/(frontend)/release.json/route';
+import { readFile } from 'node:fs/promises';
+
+jest.mock('node:fs/promises', () => ({ readFile: jest.fn() }));
+const realFetch = global.fetch;
+const identity = { sha: 'a'.repeat(40), build_id: 'b'.repeat(32) };
+beforeEach(() => {
+  jest.mocked(readFile).mockResolvedValue(JSON.stringify(identity));
+  global.fetch = jest.fn(async () => Response.json(identity));
+});
+afterEach(() => { global.fetch = realFetch; jest.resetAllMocks(); });
+
+test('reports immutable file identity and backend identity without caching', async () => {
+  const response = await GET();
+  expect(response.status).toBe(200);
+  expect(response.headers.get('Cache-Control')).toBe('no-store');
+  expect(await response.json()).toEqual({ frontend: identity, backend: identity });
+  expect(fetch).toHaveBeenCalledWith(expect.stringMatching(/\/api\/release$/), expect.objectContaining({ cache: 'no-store', signal: expect.anything() }));
+});
+
+test('missing build metadata cannot pass the release gate', async () => {
+  jest.mocked(readFile).mockRejectedValueOnce(new Error('missing file'));
+  expect((await GET()).status).toBe(503);
+});
+
+test('backend failure cannot pass the release gate', async () => {
+  jest.mocked(fetch).mockResolvedValueOnce(new Response('', { status: 503 }));
+  expect((await GET()).status).toBe(503);
+});

nextjs-app/__tests__/components/HomeView.staleFetch.test.tsx

@@ -0,0 +1,81 @@
+import { act, fireEvent, render, screen } from '@testing-library/react';
+import { HomeView } from '@/components/HomeView';
+import { fetchSchools } from '@/lib/api';
+import { primaryFixture } from '../support/schoolFixtures';
+import type { SchoolsResponse, School } from '@/lib/types';
+
+let params = new URLSearchParams('postcode=SW1A+1AA');
+jest.mock('next/navigation', () => ({
+  useSearchParams: () => params,
+  usePathname: () => '/',
+  useRouter: () => ({ push: jest.fn(), replace: jest.fn() }),
+}));
+jest.mock('@/context/ComparisonContext', () => ({
+  useComparisonContext: () => ({ addSchool: jest.fn(), removeSchool: jest.fn(), selectedSchools: [] }),
+}));
+jest.mock('@/lib/api', () => ({
+  fetchSchools: jest.fn(),
+  fetchNationalAverages: jest.fn(async () => ({})),
+  fetchLAaverages: jest.fn(async () => ({ secondary: { attainment_8_by_la: {} } })),
+}));
+jest.mock('@/components/FilterBar', () => ({ FilterBar: () => null }));
+jest.mock('@/components/SchoolRow', () => ({ SchoolRow: ({ school }: {school: School}) => <div>{school.school_name}</div> }));
+jest.mock('@/components/SchoolMap', () => ({ SchoolMap: ({ schools }: {schools: School[]}) => <div data-testid="map">{schools.map(s => s.school_name).join(',')}</div> }));
+
+const filters = { local_authorities: [], school_types: [], years: [], phases: [], genders: [], admissions_policies: [] };
+function response(name: string): SchoolsResponse {
+  return { schools: [{ ...primaryFixture.schoolInfo, school_name: name }],
+    total: 2, page: 1, page_size: 1, total_pages: 2 };
+}
+function deferred() {
+  let resolve!: (value: SchoolsResponse) => void;
+  let reject!: (error: Error) => void;
+  const promise = new Promise<SchoolsResponse>((yes, no) => { resolve = yes; reject = no; });
+  return { promise, resolve, reject };
+}
+beforeEach(() => {
+  params = new URLSearchParams('postcode=SW1A+1AA');
+  jest.mocked(fetchSchools).mockReset();
+});
+
+test('load-more results from an old search are discarded, even after returning to it', async () => {
+  const pending = deferred();
+  jest.mocked(fetchSchools).mockReturnValueOnce(pending.promise);
+  const view = render(<HomeView initialSchools={response('Initial A')} filters={filters} />);
+  fireEvent.click(screen.getByRole('button', { name: 'Load more schools' }));
+  const signal = jest.mocked(fetchSchools).mock.calls[0][1]?.signal;
+  params = new URLSearchParams('postcode=SW2+1AA');
+  view.rerender(<HomeView initialSchools={response('Initial B')} filters={filters} />);
+  expect(signal?.aborted).toBe(true);
+  params = new URLSearchParams('postcode=SW1A+1AA');
+  view.rerender(<HomeView initialSchools={response('Fresh A')} filters={filters} />);
+  await act(async () => pending.resolve(response('Stale append')));
+  expect(screen.queryByText('Stale append')).not.toBeInTheDocument();
+  expect(screen.getByRole('button', { name: 'Load more schools' })).toBeEnabled();
+});
+
+test('an older map response cannot overwrite the current search', async () => {
+  const first = deferred(), second = deferred();
+  jest.mocked(fetchSchools).mockReturnValueOnce(first.promise).mockReturnValueOnce(second.promise);
+  const initial = response('Initial A');
+  const view = render(<HomeView initialSchools={initial} filters={filters} />);
+  fireEvent.click(screen.getByRole('button', { name: 'Map' }));
+  params = new URLSearchParams('postcode=SW2+1AA');
+  view.rerender(<HomeView initialSchools={response('Initial B')} filters={filters} />);
+  await act(async () => second.resolve(response('Current map')));
+  await act(async () => first.resolve(response('Stale map')));
+  expect(screen.getByTestId('map')).toHaveTextContent('Current map');
+  expect(screen.getByTestId('map')).not.toHaveTextContent('Stale map');
+});
+
+test('failed map requests can be retried by reopening the map', async () => {
+  const pending = deferred();
+  jest.mocked(fetchSchools).mockReturnValueOnce(pending.promise).mockResolvedValue(response('Retry result'));
+  render(<HomeView initialSchools={response('Initial')} filters={filters} />);
+  fireEvent.click(screen.getByRole('button', { name: 'Map' }));
+  await act(async () => pending.reject(new Error('offline')));
+  fireEvent.click(screen.getByRole('button', { name: 'List' }));
+  await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Map' })));
+  expect(fetchSchools).toHaveBeenCalledTimes(2);
+  expect(screen.getByTestId('map')).toHaveTextContent('Retry result');
+});

nextjs-app/app/(frontend)/error.tsx

@@ -0,0 +1,12 @@
+'use client';
+
+export default function ErrorPage({ reset }: { error: Error & { digest?: string }; reset: () => void }) {
+  return (
+    <main style={{ maxWidth: '48rem', margin: '4rem auto', padding: '1.5rem' }}>
+      <h1>We couldn’t load this page</h1>
+      <p>School information is temporarily unavailable. Please try again.</p>
+      <button type="button" onClick={reset}>Try again</button>
+      <p><a href="/">Return to school search</a></p>
+    </main>
+  );
+}

nextjs-app/app/(frontend)/page.tsx

@@ -85,73 +85,50 @@ export default async function HomePage({ searchParams }: HomePageProps) {
     params.has_sixth_form
   );
 
-  // Fetch data on server with error handling
-  try {
-    const [filtersData, dataInfo] = await Promise.all([fetchFilters(), fetchDataInfo().catch(() => null)]);
+  // Failures propagate to the retryable error boundary.
+  const [filtersData, dataInfo] = await Promise.all([fetchFilters(), fetchDataInfo().catch(() => null)]);
 
-    // Only fetch schools if there are search parameters
-    let schoolsData;
-    if (hasSearchParams) {
-      schoolsData = await fetchSchools({
-        search: params.search,
-        local_authority: params.local_authority,
-        school_type: params.school_type,
-        phase: params.phase,
-        postcode: params.postcode,
-        radius,
-        page,
-        page_size: 50,
-        gender: params.gender,
-        admissions_policy: params.admissions_policy,
-        has_sixth_form: params.has_sixth_form,
-      });
-    } else {
-      // Empty state by default
-      schoolsData = { schools: [], page: 1, page_size: 50, total: 0, total_pages: 0 };
-    }
-
-    const resolvedFilters = filtersData || { local_authorities: [], school_types: [], years: [], phases: [], genders: [], admissions_policies: [] };
-    // `unique_schools`, not `total_schools` — the latter is not a field this
-    // endpoint returns, and reading it silently yielded null on every request.
-    const total = dataInfo?.unique_schools ?? null;
-    const years = dataInfo?.years_available ?? [];
-    return (
-      <HomeView
-        autosuggest={autosuggest}
-        initialSchools={schoolsData}
-        filters={resolvedFilters}
-        totalSchools={total}
-        howItWorks={hasSearchParams ? null : <HowItWorksSection />}
-        editorial={hasSearchParams ? null : (
-          <EditorialSection
-            totalSchools={total}
-            localAuthorityCount={resolvedFilters.local_authorities.length}
-            earliestYearLabel={years.length ? formatAcademicYear(years[0]) : null}
-            latestYearLabel={years.length ? formatAcademicYear(years[years.length - 1]) : null}
-          />
-        )}
-      />
-    );
-  } catch (error) {
-    console.error('Error fetching data for home page:', error);
-
-    const emptyFilters = { local_authorities: [], school_types: [], years: [], phases: [], genders: [], admissions_policies: [] };
-    return (
-      <HomeView
-        autosuggest={autosuggest}
-        initialSchools={{ schools: [], page: 1, page_size: 50, total: 0, total_pages: 0 }}
-        filters={emptyFilters}
-        totalSchools={null}
-        howItWorks={hasSearchParams ? null : <HowItWorksSection />}
-        editorial={hasSearchParams ? null : (
-          <EditorialSection
-            totalSchools={null}
-            localAuthorityCount={0}
-            earliestYearLabel={null}
-            latestYearLabel={null}
-          />
-        )}
-      />
-    );
+  // Only fetch schools if there are search parameters
+  let schoolsData;
+  if (hasSearchParams) {
+    schoolsData = await fetchSchools({
+      search: params.search,
+      local_authority: params.local_authority,
+      school_type: params.school_type,
+      phase: params.phase,
+      postcode: params.postcode,
+      radius,
+      page,
+      page_size: 50,
+      gender: params.gender,
+      admissions_policy: params.admissions_policy,
+      has_sixth_form: params.has_sixth_form,
+    });
+  } else {
+    // Empty state by default
+    schoolsData = { schools: [], page: 1, page_size: 50, total: 0, total_pages: 0 };
   }
+
+  const resolvedFilters = filtersData || { local_authorities: [], school_types: [], years: [], phases: [], genders: [], admissions_policies: [] };
+  // `unique_schools`, not `total_schools` — the latter is not a field this
+  // endpoint returns, and reading it silently yielded null on every request.
+  const total = dataInfo?.unique_schools ?? null;
+  const years = dataInfo?.years_available ?? [];
+  return (
+    <HomeView
+      autosuggest={autosuggest}
+      initialSchools={schoolsData}
+      filters={resolvedFilters}
+      totalSchools={total}
+      howItWorks={hasSearchParams ? null : <HowItWorksSection />}
+      editorial={hasSearchParams ? null : (
+        <EditorialSection
+          totalSchools={total}
+          localAuthorityCount={resolvedFilters.local_authorities.length}
+          earliestYearLabel={years.length ? formatAcademicYear(years[0]) : null}
+          latestYearLabel={years.length ? formatAcademicYear(years[years.length - 1]) : null}
+        />
+      )}
+    />
+  );
 }

nextjs-app/app/(frontend)/release.json/route.ts

@@ -0,0 +1,17 @@
+import { readFile } from 'node:fs/promises';
+import path from 'node:path';
+
+export const dynamic = 'force-dynamic';
+export const runtime = 'nodejs';
+
+export async function GET() {
+  try {
+    const frontend = JSON.parse(await readFile(path.join(process.cwd(), 'build-info.json'), 'utf8'));
+    const base = process.env.FASTAPI_URL || 'http://localhost:8000/api';
+    const res = await fetch(`${base}/release`, { cache: 'no-store', signal: AbortSignal.timeout(5000) });
+    if (!res.ok) throw new Error('Backend identity unavailable');
+    return Response.json({ frontend, backend: await res.json() }, { headers: { 'Cache-Control': 'no-store', 'X-Robots-Tag': 'noindex' } });
+  } catch {
+    return Response.json({ detail: 'Release identity unavailable' }, { status: 503, headers: { 'Cache-Control': 'no-store', 'X-Robots-Tag': 'noindex' } });
+  }
+}

nextjs-app/app/(frontend)/school/[slug]/page.tsx

@@ -4,7 +4,7 @@
  * URL format: /school/138267-school-name-here
  */
 
-import { fetchSchoolDetails, fetchSchools, fetchNationalAverages } from '@/lib/api';
+import { APIFetchError, fetchSchoolDetails, fetchSchools, fetchNationalAverages } from '@/lib/api';
 import { notFound, redirect } from 'next/navigation';
 import { SchoolDetailShell } from '@/components/school/SchoolDetailShell';
 import { NearbyPlaces } from '@/components/school/NearbyPlaces';
@@ -146,8 +146,8 @@ export default async function SchoolPage({ params }: SchoolPageProps) {
       fetchNationalAverages().catch(() => null),
     ]);
   } catch (error) {
-    console.error(`Failed to fetch school ${urn}:`, error);
-    notFound();
+    if (error instanceof APIFetchError && error.status === 404) notFound();
+    throw error;
   }
 
   const { school_info, yearly_data, absence_data, ofsted, census, admissions, admissions_history, admission_distance, deprivation, finance, destinations } = data;

nextjs-app/components/HomeView.tsx

@@ -213,6 +213,16 @@ export function HomeView({ initialSchools, filters, totalSchools, howItWorks, ed
   const [isLoadingMap, setIsLoadingMap] = useState(false);
   const prevSearchParamsRef = useRef(searchParams.toString());
   const mapParamsRef = useRef<string>('');
+  const loadMoreController = useRef<AbortController | null>(null);
+  // Identity changes even for A → B → A, so an old A response stays stale.
+  const searchScope = useRef({ key: searchParams.toString() });
+  if (searchScope.current.key !== searchParams.toString()) {
+    searchScope.current = { key: searchParams.toString() };
+  }
+  useEffect(() => {
+    setIsLoadingMore(false);
+    return () => { loadMoreController.current?.abort(); };
+  }, [searchParams]);
   const [geoState, setGeoState] = useState<'idle' | 'requesting' | 'error'>('idle');
   const [geoError, setGeoError] = useState<string | null>(null);
   /*
@@ -274,17 +284,27 @@ export function HomeView({ initialSchools, filters, totalSchools, howItWorks, ed
     if (resultsView !== 'map' || !isLocationSearch) return;
     const paramsKey = searchParams.toString();
     if (paramsKey === mapParamsRef.current) return;
-    mapParamsRef.current = paramsKey;
+    const controller = new AbortController();
+    const scope = searchScope.current;
+    const current = () => !controller.signal.aborted && searchScope.current === scope;
     setIsLoadingMap(true);
     const params: Record<string, any> = {};
     searchParams.forEach((value, key) => { params[key] = value; });
     params.page = 1;
     params.page_size = 500;
-    fetchSchools(params, { cache: 'no-store' })
-      .then(r => setMapSchools(r.schools))
-      .catch(() => setMapSchools(initialSchools.schools))
-      .finally(() => setIsLoadingMap(false));
-  }, [resultsView, searchParams]);
+    fetchSchools(params, { cache: 'no-store', signal: controller.signal })
+      .then(r => {
+        if (!current()) return;
+        mapParamsRef.current = paramsKey;
+        setMapSchools(r.schools);
+      })
+      .catch(() => {
+        if (current()) setMapSchools(initialSchools.schools);
+        // No cache marker on failure: opening the map again retries.
+      })
+      .finally(() => { if (current()) setIsLoadingMap(false); });
+    return () => controller.abort();
+  }, [resultsView, searchParams, initialSchools.schools]);
 
   // Fetch LA averages when secondary or mixed schools are visible
   useEffect(() => {
@@ -305,19 +325,25 @@ export function HomeView({ initialSchools, filters, totalSchools, howItWorks, ed
     if (isLoadingMore || !hasMore) return;
     track('results_load_more', { next_page: currentPage + 1 });
     setIsLoadingMore(true);
+    const scope = searchScope.current;
+    const controller = new AbortController();
+    loadMoreController.current?.abort();
+    loadMoreController.current = controller;
+    const current = () => !controller.signal.aborted && searchScope.current === scope;
     try {
       const params: Record<string, any> = {};
       searchParams.forEach((value, key) => { params[key] = value; });
       params.page = currentPage + 1;
       params.page_size = initialSchools.page_size;
-      const response = await fetchSchools(params, { cache: 'no-store' });
+      const response = await fetchSchools(params, { cache: 'no-store', signal: controller.signal });
+      if (!current()) return;
       setAllSchools(prev => [...prev, ...response.schools]);
       setCurrentPage(response.page);
       setHasMore(response.page < response.total_pages);
     } catch {
       // silently ignore
     } finally {
-      setIsLoadingMore(false);
+      if (current()) setIsLoadingMore(false);
     }
   };
 

nextjs-app/jest.config.cjs

@@ -9,6 +9,7 @@ const createJestConfig = nextJest({
 const customJestConfig = {
   setupFilesAfterEnv: ['<rootDir>/jest.setup.js'],
   testEnvironment: 'jest-environment-jsdom',
+  modulePathIgnorePatterns: ['<rootDir>/.next/'],
   moduleNameMapper: {
     '^@/(.*)$': '<rootDir>/$1',
   },

nextjs-app/lib/places.ts

@@ -1,3 +1,5 @@
+import { APIFetchError } from './api';
+
 /**
  * Client for the places API.
  *
@@ -69,7 +71,7 @@ const API = process.env.FASTAPI_URL || process.env.NEXT_PUBLIC_API_URL
 
 export async function fetchPlaces(): Promise<PlaceSummary[]> {
   const res = await fetch(`${API}/places`, { next: { revalidate: 604800 } });
-  if (!res.ok) return [];
+  if (!res.ok) throw new APIFetchError("Unable to load places", res.status);
   return (await res.json()).places ?? [];
 }
 
@@ -79,6 +81,7 @@ export async function fetchPlace(
   const q = phase ? `?phase=${encodeURIComponent(phase)}` : '';
   const res = await fetch(`${API}/places/${kind}/${slug}${q}`,
     { next: { revalidate: 604800 } });
-  if (!res.ok) return null;
+  if (res.status === 404) return null;
+  if (!res.ok) throw new APIFetchError("Unable to load place", res.status);
   return res.json();
 }

pipeline/Dockerfile

@@ -40,4 +40,9 @@ ENV AIRFLOW_HOME=/opt/airflow
 ENV AIRFLOW__CORE__DAGS_FOLDER=/opt/pipeline/dags
 ENV PYTHONPATH=/opt/pipeline
 
+ARG BUILD_SHA=development
+ARG BUILD_ID=development
+LABEL io.schoolcompare.build-id=$BUILD_ID
+LABEL io.schoolcompare.commit=$BUILD_SHA
+
 CMD ["airflow", "api-server"]

pipeline/scripts/sync_typesense.py

@@ -11,6 +11,10 @@ Usage:
 from __future__ import annotations
 
 import argparse
+import json
+import logging
+import re
+import uuid
 import os
 import sys
 import time
@@ -112,63 +116,78 @@ def build_document(row: dict) -> dict:
     return doc
 
 
+def publish_collection(client, rows: list[dict]) -> str:
+    """Validate a new collection before moving the alias; keep rollback data.
+
+    Caller holds the database advisory lock across reading and publication so
+    overlapping school-data DAGs cannot publish or prune each other's work.
+    Failed drafts are left for the next successful publication to prune: a lost
+    alias-update response must never cause deletion of a potentially live index.
+    """
+    if not rows or len({r["urn"] for r in rows}) != len(rows):
+        raise ValueError("Search source must contain nonempty, unique school URNs")
+    name = f"schools_{int(time.time())}_{uuid.uuid4().hex[:12]}"
+    try:
+        previous = client.aliases["schools"].retrieve()["collection_name"]
+    except typesense.exceptions.ObjectNotFound:
+        previous = None
+    client.collections.create({**COLLECTION_SCHEMA, "name": name})
+    for i in range(0, len(rows), 500):
+        batch = [build_document(r) for r in rows[i:i + 500]]
+        results = client.collections[name].documents.import_(batch, {"action": "upsert"})
+        if isinstance(results, str):
+            results = [json.loads(line) for line in results.splitlines() if line.strip()]
+        if len(results) != len(batch) or any(r.get("success") is not True for r in results):
+            raise ValueError("Search import failed; live alias unchanged")
+    if client.collections[name].retrieve()["num_documents"] != len(rows):
+        raise ValueError("Search document count mismatch; live alias unchanged")
+    client.aliases.upsert("schools", {"collection_name": name})
+
+    # Retention is best-effort and must not make successful publication fail.
+    try:
+        keep = {name, previous}
+        keep.update(a["collection_name"] for a in client.aliases.retrieve()["aliases"])
+        for collection in client.collections.retrieve():
+            old = collection["name"]
+            if old not in keep and re.fullmatch(r"schools_\d+(?:_[0-9a-f]+)?", old):
+                client.collections[old].delete()
+    except Exception:
+        logging.getLogger(__name__).exception("Search published, but old collection cleanup failed")
+    return name
+
+
 def sync(typesense_url: str, api_key: str):
+    from urllib.parse import urlparse
+    url = urlparse(typesense_url)
     client = typesense.Client({
-        "nodes": [{"host": typesense_url.split("//")[-1].split(":")[0],
-                    "port": typesense_url.split(":")[-1],
-                    "protocol": "http"}],
+        "nodes": [{"host": url.hostname, "port": str(url.port or 8108),
+                   "protocol": url.scheme}],
         "api_key": api_key,
         "connection_timeout_seconds": 10,
     })
-
-    # Create timestamped collection for zero-downtime swap
-    ts = int(time.time())
-    collection_name = f"schools_{ts}"
-
-    print(f"Creating collection: {collection_name}")
-    schema = {**COLLECTION_SCHEMA, "name": collection_name}
-    client.collections.create(schema)
-
-    # Fetch data from marts — join fact_performance if it exists
     conn = get_db_connection()
-    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
-        # Check whether the merged fact table exists
-        cur.execute("""
-            SELECT table_name FROM information_schema.tables
-            WHERE table_schema = 'marts' AND table_name = 'fact_performance'
-        """)
-        has_fact_performance = cur.fetchone() is not None
-
-        query = QUERY_BASE
-        if has_fact_performance:
-            query = query.replace(
-                "l.longitude as lng",
-                "l.longitude as lng,\n        p.rwm_expected_pct,\n        p.progress_8_score",
-            )
-            query += QUERY_PERFORMANCE_JOIN
-
-        cur.execute(query)
-        rows = cur.fetchall()
-    conn.close()
-
-    print(f"Indexing {len(rows)} schools...")
-
-    # Batch import
-    batch_size = 500
-    for i in range(0, len(rows), batch_size):
-        batch = [build_document(r) for r in rows[i : i + batch_size]]
-        client.collections[collection_name].documents.import_(batch, {"action": "upsert"})
-        print(f"  Indexed {min(i + batch_size, len(rows))}/{len(rows)}")
-
-    # Swap alias
-    print("Swapping alias 'schools' → new collection")
     try:
-        client.aliases.upsert("schools", {"collection_name": collection_name})
-    except Exception:
-        # If alias doesn't exist yet, create it
-        client.aliases.upsert("schools", {"collection_name": collection_name})
-
-    print("Done.")
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            # Session-scoped lock is released even on errors when conn closes.
+            cur.execute("SELECT pg_advisory_lock(731042019)")
+            cur.execute("""
+                SELECT table_name FROM information_schema.tables
+                WHERE table_schema = 'marts' AND table_name = 'fact_performance'
+            """)
+            has_fact_performance = cur.fetchone() is not None
+            query = QUERY_BASE
+            if has_fact_performance:
+                query = query.replace(
+                    "l.longitude as lng",
+                    "l.longitude as lng, p.rwm_expected_pct, p.progress_8_score",
+                )
+                query += QUERY_PERFORMANCE_JOIN
+            cur.execute(query)
+            rows = cur.fetchall()
+        name = publish_collection(client, rows)
+        print(f"Published {len(rows)} schools in {name}")
+    finally:
+        conn.close()
 
 
 def main():

pipeline/tests/test_sync_typesense.py

@@ -0,0 +1,92 @@
+import importlib.util
+from pathlib import Path
+from unittest.mock import MagicMock
+import pytest
+
+
+@pytest.fixture
+def sync_module(monkeypatch):
+    folder = Path(__file__).resolve().parents[1] / 'scripts'
+    monkeypatch.syspath_prepend(str(folder))
+    spec = importlib.util.spec_from_file_location('sync_typesense', folder / 'sync_typesense.py')
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture
+def client():
+    c = MagicMock()
+    c.aliases.__getitem__.return_value.retrieve.return_value = {'collection_name': 'schools_2'}
+    c.aliases.retrieve.return_value = {'aliases': [{'collection_name': 'schools_99'}]}
+    c.collections.__getitem__.return_value.documents.import_.return_value = [{'success': True}]
+    c.collections.__getitem__.return_value.retrieve.return_value = {'num_documents': 1}
+    c.collections.retrieve.return_value = [{'name': n} for n in ['schools_1', 'schools_2', 'schools_99', 'unrelated']]
+    return c
+
+
+@pytest.fixture
+def rows():
+    return [{'urn': 100001, 'school_name': 'Example', 'phase_code': 2,
+             'school_type_code': 1, 'local_authority': 'Testshire', 'postcode': 'TS1 1AA',
+             'total_pupils': 250}]
+
+
+@pytest.mark.parametrize('results', [[{'success': False}], [], [{'success': True}, {'success': True}]])
+def test_partial_import_never_moves_alias(sync_module, client, rows, results):
+    client.collections.__getitem__.return_value.documents.import_.return_value = results
+    with pytest.raises(ValueError):
+        sync_module.publish_collection(client, rows)
+    client.aliases.upsert.assert_not_called()
+    client.collections.__getitem__.return_value.delete.assert_not_called()
+
+
+def test_count_mismatch_does_not_publish(sync_module, client, rows):
+    client.collections.__getitem__.return_value.retrieve.return_value = {'num_documents': 0}
+    with pytest.raises(ValueError):
+        sync_module.publish_collection(client, rows)
+    client.aliases.upsert.assert_not_called()
+
+
+@pytest.mark.parametrize('empty', [True, False])
+def test_invalid_source_never_creates_collection(sync_module, client, rows, empty):
+    with pytest.raises(ValueError):
+        sync_module.publish_collection(client, [] if empty else rows + rows)
+    client.collections.create.assert_not_called()
+
+
+def test_success_retains_previous_and_other_live_aliases(sync_module, client, rows):
+    # Distinct mock per collection allows checking exactly which one was deleted.
+    collections = {}
+    def get(name):
+        if name not in collections:
+            c = MagicMock()
+            c.documents.import_.return_value = '{"success":true}\n'
+            c.retrieve.return_value = {'num_documents': 1}
+            collections[name] = c
+        return collections[name]
+    client.collections.__getitem__.side_effect = get
+    name = sync_module.publish_collection(client, rows)
+    client.aliases.upsert.assert_called_once_with('schools', {'collection_name': name})
+    assert set(collections) == {name, 'schools_1'}
+    collections['schools_1'].delete.assert_called_once()
+
+
+def test_uncertain_alias_update_does_not_delete_candidate(sync_module, client, rows):
+    client.aliases.upsert.side_effect = RuntimeError('response lost')
+    with pytest.raises(RuntimeError):
+        sync_module.publish_collection(client, rows)
+    client.collections.__getitem__.return_value.delete.assert_not_called()
+
+
+def test_sync_closes_database_when_publication_fails(sync_module, monkeypatch):
+    conn = MagicMock()
+    monkeypatch.setattr(sync_module, 'get_db_connection', lambda: conn)
+    monkeypatch.setattr(sync_module.typesense, 'Client', lambda _: MagicMock())
+    def fail(*args): raise ValueError('import rejected')
+    monkeypatch.setattr(sync_module, 'publish_collection', fail)
+    with pytest.raises(ValueError):
+        sync_module.sync('http://localhost:8108', 'dummy')
+    conn.close.assert_called_once()
+    statements = [call.args[0] for call in conn.cursor.return_value.__enter__.return_value.execute.call_args_list]
+    assert statements[0] == 'SELECT pg_advisory_lock(731042019)'

scripts/ci/release.py

@@ -0,0 +1,136 @@
+"""Release identity checks and promotion of the exact digests that passed E2E.
+
+Uses only the standard library and Docker Buildx. No registry mutation happens
+until every image in the set has been resolved and its build labels validated.
+"""
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import time
+from urllib.request import Request, urlopen
+
+COMPONENTS = ('BACKEND', 'FRONTEND', 'PIPELINE')
+
+
+def docker(*args):
+    return subprocess.check_output(['docker', 'buildx', 'imagetools', *args], text=True).strip()
+
+
+def check_sha(sha):
+    if not re.fullmatch(r'[0-9a-f]{40}', sha):
+        raise ValueError('Expected a full commit SHA')
+    return sha
+
+
+def check_digest(digest):
+    if not re.fullmatch(r'sha256:[0-9a-f]{64}', digest):
+        raise ValueError('Expected an immutable image digest')
+    return digest
+
+
+def image_identity(ref):
+    image = json.loads(docker('inspect', ref, '--format', '{{json .Image}}'))
+    configs = [image] if 'config' in image else list(image.values())
+    identities = set()
+    for config in configs:
+        labels = config['config']['Labels']
+        identities.add((labels['io.schoolcompare.commit'], labels['io.schoolcompare.build-id']))
+    if len(identities) != 1:
+        raise ValueError('Image platforms disagree about their release identity')
+    return next(iter(identities))
+
+
+def resolve_images(sha, verified=False, expected_build_id=None):
+    check_sha(sha)
+    refs = []
+    build_ids = set()
+    for component in COMPONENTS:
+        image = f"{os.environ['REGISTRY']}/{os.environ[component + '_IMAGE_NAME']}"
+        if verified:
+            manifest = json.loads(docker('inspect', f'{image}:verified-{sha}', '--format', '{{json .Manifest}}'))
+            digest = check_digest(manifest['digest'])
+        else:
+            digest = check_digest(os.environ[component + '_DIGEST'])
+        ref = f'{image}@{digest}'
+        actual_sha, build_id = image_identity(ref)
+        if actual_sha != sha or not re.fullmatch(r'[0-9a-f]{32}', build_id):
+            raise ValueError(f'Unrecognised release identity for {component}')
+        if expected_build_id is not None and build_id != expected_build_id:
+            raise ValueError(f'Build identity mismatch for {component}')
+        build_ids.add(build_id)
+        refs.append((image, ref))
+    if len(build_ids) != 1:
+        raise ValueError('Refusing a mixed image set')
+    return {'sha': sha, 'build_id': build_ids.pop(), 'images': refs}
+
+
+def verify(sha, build_id):
+    release = resolve_images(sha, expected_build_id=build_id)
+    for image, ref in release['images']:
+        docker('create', '-t', f'{image}:verified-{sha}', ref)
+    return release
+
+
+def promote(sha):
+    release = resolve_images(sha, verified=True)
+    # Resolve all targets first; never discover a missing candidate halfway through.
+    for image, _ in release['images']:
+        try:
+            docker('create', '-t', f'{image}:prod-previous', f'{image}:prod')
+        except subprocess.CalledProcessError:
+            print(f'No rollback pointer saved for {image}', flush=True)
+    for image, ref in release['images']:
+        docker('create', '-t', f'{image}:prod', ref)
+    return release
+
+
+def matches(payload, sha, build_id):
+    return all(payload.get(component) == {'sha': sha, 'build_id': build_id}
+               for component in ('frontend', 'backend'))
+
+
+def wait(base_url, sha, build_id, timeout):
+    check_sha(sha)
+    if not re.fullmatch(r'[0-9a-f]{32}', build_id):
+        raise ValueError('Missing expected build identity')
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            req = Request(f'{base_url.rstrip("/")}/release.json?check={time.time_ns()}',
+                          headers={'Cache-Control': 'no-cache'})
+            with urlopen(req, timeout=min(10, max(.1, deadline - time.monotonic()))) as response:
+                payload = json.load(response)
+            if matches(payload, sha, build_id):
+                print(f'Verified deployed release {sha} / {build_id}')
+                return
+        except (OSError, ValueError):
+            pass
+        time.sleep(min(5, max(0, deadline - time.monotonic())))
+    raise RuntimeError('Deployment did not report the expected frontend/backend release')
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('action', choices=['wait', 'verify', 'promote'])
+    parser.add_argument('--timeout', type=float, default=300)
+    parser.add_argument('--release', type=Path)
+    parser.add_argument('--output', type=Path)
+    args = parser.parse_args()
+    identity = json.loads(args.release.read_text()) if args.release else {
+        'sha': os.environ.get('EXPECTED_SHA', ''),
+        'build_id': os.environ.get('EXPECTED_BUILD_ID', ''),
+    }
+    if args.action == 'wait':
+        wait(os.environ['BASE_URL'], identity['sha'], identity['build_id'], args.timeout)
+        return
+    result = (verify(identity['sha'], identity['build_id']) if args.action == 'verify'
+              else promote(identity['sha']))
+    if args.output:
+        args.output.write_text(json.dumps(result))
+
+
+if __name__ == '__main__':
+    main()

scripts/ci/tests/test_release.py

@@ -0,0 +1,65 @@
+import json
+from unittest.mock import Mock
+import pytest
+from scripts.ci import release
+
+SHA = 'a' * 40
+BUILD = 'b' * 32
+DIGESTS = ['sha256:' + c * 64 for c in '123']
+
+
+@pytest.fixture
+def docker(monkeypatch):
+    monkeypatch.setenv('REGISTRY', 'registry.example')
+    refs = {}
+    for component, digest in zip(release.COMPONENTS, DIGESTS):
+        monkeypatch.setenv(component + '_IMAGE_NAME', component.lower())
+        monkeypatch.setenv(component + '_DIGEST', digest)
+        refs[f'registry.example/{component.lower()}'] = digest
+    def run(*args):
+        if args[0] == 'create': return ''
+        if args[-1] == '{{json .Manifest}}':
+            return json.dumps({'digest': refs[args[1].split(':')[0]]})
+        return json.dumps({'config': {'Labels': {'io.schoolcompare.commit': SHA,
+                                                'io.schoolcompare.build-id': BUILD}}})
+    mock = Mock(side_effect=run)
+    monkeypatch.setattr(release, 'docker', mock)
+    return mock
+
+
+def test_wrong_deployed_build_is_rejected_even_at_same_commit():
+    assert not release.matches({'frontend': {'sha': SHA, 'build_id': BUILD},
+                                'backend': {'sha': SHA, 'build_id': 'c' * 32}}, SHA, BUILD)
+    assert release.matches({c: {'sha': SHA, 'build_id': BUILD} for c in ('frontend', 'backend')}, SHA, BUILD)
+
+
+def test_verification_tags_the_captured_digests(docker):
+    release.verify(SHA, BUILD)
+    creates = [c.args for c in docker.call_args_list if c.args[0] == 'create']
+    assert len(creates) == 3
+    for call, digest in zip(creates, DIGESTS):
+        assert call[-1].endswith('@' + digest)
+        assert call[2].endswith(':verified-' + SHA)
+
+
+def test_promotion_resolves_all_verified_images_before_mutation(docker):
+    result = release.promote(SHA)
+    assert result['build_id'] == BUILD
+    calls = [c.args for c in docker.call_args_list]
+    first_write = next(i for i, c in enumerate(calls) if c[0] == 'create')
+    assert first_write == 6  # each of three candidates needs manifest + config
+    assert all(c[-1].endswith('@' + d) for c, d in zip(calls[-3:], DIGESTS))
+
+
+def test_mixed_builds_fail_before_any_tag_is_changed(docker, monkeypatch):
+    identities = iter([(SHA, BUILD), (SHA, 'c' * 32), (SHA, BUILD)])
+    monkeypatch.setattr(release, 'image_identity', lambda _: next(identities))
+    with pytest.raises(ValueError, match='mixed'):
+        release.promote(SHA)
+    assert not any(c.args[0] == 'create' for c in docker.call_args_list)
+
+
+def test_missing_candidate_fails_before_any_tag_is_changed(docker):
+    docker.side_effect = RuntimeError('missing verified tag')
+    with pytest.raises(RuntimeError): release.promote(SHA)
+    assert not any(c.args[0] == 'create' for c in docker.call_args_list)

scripts/ci/tests/test_workflows.py

@@ -0,0 +1,34 @@
+"""Check the dependency graph that ties tested digests to deployable images."""
+from pathlib import Path
+import yaml
+
+ROOT = Path(__file__).resolve().parents[3]
+
+
+def test_staging_verifies_identity_before_and_after_journeys():
+    workflow = yaml.safe_load((ROOT / '.gitea/workflows/deploy.yml').read_text())
+    assert workflow['concurrency'] == {'group': 'staging-release', 'cancel-in-progress': False}
+    jobs = workflow['jobs']
+    for component in ('backend', 'frontend', 'pipeline'):
+        job = jobs['build-' + component]
+        assert 'prepare' in job['needs']
+        assert job['outputs']['digest'] == '${{ steps.build.outputs.digest }}'
+        build = next(step for step in job['steps'] if step.get('id') == 'build')
+        assert 'BUILD_ID=${{ needs.prepare.outputs.build_id }}' in build['with']['build-args']
+    steps = jobs['e2e-staging']['steps']
+    runs = [step.get('run', '') for step in steps]
+    test = runs.index('npx playwright test')
+    assert 'release.py wait' in runs[test - 1]
+    assert 'release.py wait' in runs[test + 1]
+    assert 'release.py verify' in runs[-1]
+    for component in ('backend', 'frontend', 'pipeline'):
+        assert 'build-' + component in jobs['e2e-staging']['needs']
+        assert component.upper() + '_DIGEST' in steps[-1]['env']
+
+
+def test_promotion_uses_verified_digest_resolver_and_build_identity_poll():
+    workflow = yaml.safe_load((ROOT / '.gitea/workflows/promote.yml').read_text())
+    steps = workflow['jobs']['promote-prod']['steps']
+    runs = [step.get('run', '') for step in steps]
+    assert 'python3 scripts/ci/release.py promote --output release.json' in runs
+    assert runs[-1] == 'python3 scripts/ci/release.py wait --release release.json'