fix(pipeline): keep the destinations marts out of the scheduled builds

fact_ks4_destinations and fact_ks5_destinations join dim_school, so the
daily build's stg_gias_establishments+ and the monthly Ofsted build's
dim_school+ both selected them. They also read stg_ees_ks4/ks5_destinations,
which only the manually triggered EES DAG builds. Where that DAG hasn't run
since the destinations models landed, dbt_build fails with "relation
staging.stg_ees_ks4_destinations does not exist", and sync_typesense and
invalidate_cache never run. Production's register data has been stuck at
about 25 Aug 2026.

Both builds now exclude the descendants of the two EES staging models, as
the daily build already does for the KS2/KS4 lineage models. The EES DAG
still rebuilds the marts when their data changes.

test_dag_selectors reads the model graph from the SQL (CI has no dbt) and
checks that every scheduled build only reads models it or the daily build
builds. It failed for the daily and monthly Ofsted builds before this change.

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

e2e/tests/journeys.spec.ts

@@ -410,6 +410,29 @@ test('search and the school page agree on how many pupils a secondary has', asyn
   expect(school.total_pupils).toBe(detail.school_info.total_pupils);
 });
 
+test('a secondary search row compares its Attainment 8 with the LA average', async ({ page }) => {
+  // The comparison vanished unnoticed: the averages were fetched with
+  // force-cache, so one stored failure hid it in that browser for good.
+  // Playwright disables the HTTP cache when it intercepts requests, so this
+  // guards the comparison itself; the unit test pins the cache mode.
+  const la = await (await page.request.get('/api/la-averages')).json();
+  const averages: Record<string, number> = la.secondary?.attainment_8_by_la ?? {};
+  const res = await page.request.get('/api/schools?search=school&phase=secondary&page_size=50');
+  expect(res.ok()).toBeTruthy();
+  const school = ((await res.json()).schools ?? []).find(
+    (s: { attainment_8_score?: number | null; local_authority?: string; school_type?: string }) =>
+      s.attainment_8_score != null && s.local_authority != null && averages[s.local_authority] != null
+      && !/special|pupil referral|alternative provision/i.test(s.school_type ?? ''));
+  test.skip(!school, 'no mainstream secondary with an LA average here');
+
+  await searchByName(page, school.school_name);
+  const link = page.locator(`a[href^="/school/${school.urn}-"]`).first();
+  await expect(link).toBeVisible({ timeout: 15_000 });
+  const stats = link.locator('xpath=ancestor::div[contains(@class, "__rowContent")][1]')
+    .locator('[class*="__line3"]');
+  await expect(stats.getByText(/vs LA avg/)).toBeVisible();
+});
+
 test('a phase outside primary/secondary filters to that phase, not to everything', async ({ page }) => {
   // The search page offers every GIAS phase, but the API only knew the grouped
   // ones and silently dropped the rest — so "Nursery" returned primaries.

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

@@ -1,6 +1,6 @@
 import { act, fireEvent, render, screen } from '@testing-library/react';
 import { HomeView } from '@/components/HomeView';
-import { fetchSchools } from '@/lib/api';
+import { fetchLAaverages, fetchSchools } from '@/lib/api';
 import { primaryFixture } from '../support/schoolFixtures';
 import type { SchoolsResponse, School } from '@/lib/types';
 
@@ -84,3 +84,23 @@ test('failed map requests can be retried by reopening the map', async () => {
   expect(fetchSchools).toHaveBeenCalledTimes(2);
   expect(screen.getByTestId('map')).toHaveTextContent('Retry result');
 });
+
+test('LA averages are not fetched with force-cache, so one failure is not replayed for good', async () => {
+  // force-cache serves any stored response, however old, without asking the
+  // server. A request that failed once (a staging deploy restart, the July
+  // proxy outage) was stored and replayed on every later visit, and the
+  // "vs LA avg" delta vanished from every secondary row in that browser.
+  // The default mode honours the API's Cache-Control and never reuses an
+  // error.
+  params = new URLSearchParams('search=high');
+  const secondary: SchoolsResponse = {
+    ...response('Alpha High'),
+    schools: [{ ...primaryFixture.schoolInfo, school_name: 'Alpha High', phase: 'Secondary', attainment_8_score: 50 }],
+  };
+  render(<HomeView initialSchools={secondary} filters={filters} />);
+  await act(async () => {});
+  expect(fetchLAaverages).toHaveBeenCalled();
+  for (const [options] of jest.mocked(fetchLAaverages).mock.calls) {
+    expect(options?.cache).not.toBe('force-cache');
+  }
+});

nextjs-app/components/HomeView.tsx

@@ -358,10 +358,14 @@ export function HomeView({ initialSchools, filters, totalSchools, howItWorks, ed
     return () => controller.abort();
   }, [resultsView, searchParams, initialSchools.schools]);
 
-  // Fetch LA averages when secondary or mixed schools are visible
+  // Fetch LA averages when secondary or mixed schools are visible. Default
+  // cache mode, never force-cache: force-cache replays any stored response
+  // without asking the server, so one failed request (a deploy restart) hid
+  // every "vs LA avg" delta in that browser for good. The API's Cache-Control
+  // already lets the browser reuse a good answer for five minutes.
   useEffect(() => {
     if (!isSecondaryView && !isMixedView) return;
-    fetchLAaverages({ cache: 'force-cache' })
+    fetchLAaverages()
       .then(data => setLaAverages(data.secondary.attainment_8_by_la))
       .catch(() => {});
   }, [isSecondaryView, isMixedView]);

pipeline/dags/school_data_pipeline.py

@@ -106,9 +106,12 @@ print(f'Validation passed: {{count}} GIAS rows')
         """,
     )
 
+    # Marts fed by annual EES staging models are rebuilt by the EES DAG, even
+    # when they join dim_school. Selecting them here fails in any database
+    # where that DAG hasn't run (pipeline/tests/test_dag_selectors.py).
     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+ gias_code_names+ --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+ stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
     )
 
     sync_typesense = BashOperator(
@@ -143,7 +146,7 @@ with DAG(
 
     dbt_build_ofsted = BashOperator(
         task_id="dbt_build",
-        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_ofsted_inspections+ int_ofsted_latest+ fact_ofsted_inspection+ dim_school+",
+        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_ofsted_inspections+ int_ofsted_latest+ fact_ofsted_inspection+ dim_school+ --exclude stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
     )
 
     sync_typesense_ofsted = BashOperator(

pipeline/tests/test_dag_selectors.py

@@ -0,0 +1,98 @@
+"""Every scheduled dbt build must only build models whose parents exist.
+
+The daily GIAS build selects `stg_gias_establishments+`, so any mart that joins
+dim_school joins the daily build too. When such a mart also reads a staging
+model that only the manually triggered EES DAG builds, the daily build fails in
+any database where that DAG has not run since. Sync and cache invalidation then
+never run either. The destinations marts did this from late August 2026.
+
+The graph is read from the model SQL, because CI has no dbt.
+"""
+import re
+from collections import defaultdict
+from pathlib import Path
+
+import pytest
+
+PIPELINE = Path(__file__).resolve().parents[1]
+MODELS = PIPELINE / 'transform' / 'models'
+DAG_FILE = PIPELINE / 'dags' / 'school_data_pipeline.py'
+
+REF = re.compile(r"ref\(\s*'([a-z0-9_]+)'\s*\)")
+DBT_BUILD = re.compile(r'dbt_build\w*\s*=\s*BashOperator\(.*?build --profiles-dir \. --target production ([^"]+)"', re.S)
+DAG_ID = re.compile(r'dag_id="([a-z0-9_]+)"')
+
+DAILY = 'school_data_daily'
+
+# dim_school reads int_ofsted_latest only when the relation exists
+# (adapter.get_relation), so a missing table is not a failure.
+OPTIONAL_PARENTS = {'int_ofsted_latest'}
+
+
+def model_parents():
+    """{model: models it refs}. Seeds are left out: they are loaded once and always exist."""
+    sql = {p.stem: p.read_text() for p in MODELS.rglob('*.sql')}
+    return {name: set(REF.findall(text)) & set(sql) for name, text in sql.items()}
+
+
+def downstream(node, children):
+    seen, stack = {node}, [node]
+    while stack:
+        for child in children[stack.pop()]:
+            if child not in seen:
+                seen.add(child)
+                stack.append(child)
+    return seen
+
+
+def expand(tokens, children):
+    out = set()
+    for token in tokens:
+        out |= downstream(token[:-1], children) if token.endswith('+') else {token}
+    return out
+
+
+def scheduled_builds():
+    """{dag_id: dbt selection arguments} for every dbt build in the DAG file."""
+    text = DAG_FILE.read_text()
+    starts = [(m.start(), m.group(1)) for m in DAG_ID.finditer(text)]
+    builds = {}
+    for i, (start, dag_id) in enumerate(starts):
+        end = starts[i + 1][0] if i + 1 < len(starts) else len(text)
+        found = DBT_BUILD.search(text, start, end)
+        if found:
+            builds[dag_id] = found.group(1)
+    return builds
+
+
+def selected_models(args, parents):
+    children = defaultdict(set)
+    for model, ps in parents.items():
+        for p in ps:
+            children[p].add(model)
+    select = re.search(r'--select (.+?)(?= --exclude|$)', args).group(1).split()
+    excluded = re.search(r'--exclude (.+)$', args)
+    exclude = excluded.group(1).split() if excluded else []
+    return (expand(select, children) - expand(exclude, children)) & set(parents)
+
+
+PARENTS = model_parents()
+BUILDS = scheduled_builds()
+DAILY_MODELS = selected_models(BUILDS[DAILY], PARENTS)
+
+
+def test_every_dag_with_a_dbt_build_is_parsed():
+    assert set(BUILDS) == {
+        'school_data_daily', 'school_data_monthly_ofsted', 'school_data_annual_ees',
+        'school_data_annual_idaci', 'school_data_annual_distance',
+    }
+
+
+@pytest.mark.parametrize('dag_id', sorted(BUILDS))
+def test_selected_models_only_read_models_that_exist(dag_id):
+    selected = selected_models(BUILDS[dag_id], PARENTS)
+    # The daily build is the base layer: other DAGs may rely on what it builds.
+    available = selected | OPTIONAL_PARENTS | (DAILY_MODELS if dag_id != DAILY else set())
+    missing = {model: sorted(PARENTS[model] - available) for model in sorted(selected)
+               if PARENTS[model] - available}
+    assert missing == {}, f'{dag_id} builds models whose parents it never builds: {missing}'