From 348dd493c7cba66b329a384410e4bb21de97cd4d Mon Sep 17 00:00:00 2001
From: Max Ghenis <mghenis@gmail.com>
Date: Thu, 1 Oct 2026 09:48:35 -0400
Subject: [PATCH 1/3] Seed the calibration weight dropout

calibrate_local_areas replaced 5% of the log-weights with their mean each
epoch, choosing them with torch.rand_like. That draws from torch's global
generator, which torch seeds differently in every process and nothing in
the build seeded, so two builds of the same commit with the same inputs gave
different household weights. The dropout now has its own torch.Generator,
seeded from a new seed argument (default 0), and leaves the global
generator alone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../deterministic-calibration.fixed.md        |   1 +
 .../tests/test_calibrate_determinism.py       | 116 ++++++++++++++++++
 policyengine_uk_data/utils/calibrate.py       |  14 ++-
 3 files changed, 130 insertions(+), 1 deletion(-)
 create mode 100644 changelog.d/deterministic-calibration.fixed.md
 create mode 100644 policyengine_uk_data/tests/test_calibrate_determinism.py

diff --git a/changelog.d/deterministic-calibration.fixed.md b/changelog.d/deterministic-calibration.fixed.md
new file mode 100644
index 00000000..877dd757
--- /dev/null
+++ b/changelog.d/deterministic-calibration.fixed.md
@@ -0,0 +1 @@
+- Seed the weight dropout in `calibrate_local_areas` with its own generator (`seed`, default 0). The dropout masks were drawn from torch's global generator, which torch seeds differently in every process, so two builds of the same commit with the same inputs gave different household weights and aggregates.
diff --git a/policyengine_uk_data/tests/test_calibrate_determinism.py b/policyengine_uk_data/tests/test_calibrate_determinism.py
new file mode 100644
index 00000000..0d8d5287
--- /dev/null
+++ b/policyengine_uk_data/tests/test_calibrate_determinism.py
@@ -0,0 +1,116 @@
+"""Calibration gives the same weights for the same inputs and seed.
+
+Invariants of ``calibrate_local_areas``:
+
+- the saved weights are a function of the inputs and ``seed`` alone: not of
+  the process, and not of whatever has consumed torch's global generator;
+- calibration does not consume the global generator;
+- ``seed`` is used: different seeds draw different dropout masks.
+
+Before the dropout masks had their own generator, two builds of one commit
+differed in every household weight, because torch seeds its global generator
+differently in each process.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+import numpy as np
+import pytest
+
+if (
+    importlib.util.find_spec("torch") is None
+    or importlib.util.find_spec("policyengine_uk") is None
+):
+    pytest.skip(
+        "torch/policyengine_uk not available in test environment",
+        allow_module_level=True,
+    )
+
+import h5py
+import torch
+
+from policyengine_uk_data.tests.test_calibrate_save import (
+    _StubDataset,
+    _make_toy_inputs,
+)
+from policyengine_uk_data.utils import calibrate as calibrate_module
+from policyengine_uk_data.utils.calibrate import calibrate_local_areas
+
+
+def _toy_weights(**kwargs) -> np.ndarray:
+    """Area-by-household weights saved by a short toy calibration.
+
+    120 weights and 21 epochs, so dropout (5% per weight per epoch) acts
+    many times and the last save (epoch 20) reflects it.
+    """
+    matrix_fn, national_matrix_fn = _make_toy_inputs(n_households=40, area_count=3)
+    dataset = _StubDataset(np.linspace(1.0, 5.0, 40))
+    with tempfile.TemporaryDirectory() as folder:
+        with patch.object(calibrate_module, "STORAGE_FOLDER", Path(folder)):
+            calibrate_local_areas(
+                dataset=dataset,
+                matrix_fn=matrix_fn,
+                national_matrix_fn=national_matrix_fn,
+                area_count=3,
+                weight_file="weights.h5",
+                dataset_key="2025",
+                epochs=21,
+                verbose=False,
+                **kwargs,
+            )
+        with h5py.File(Path(folder) / "weights.h5", "r") as f:
+            return f["2025"][:]
+
+
+def _toy_weights_digest() -> str:
+    return hashlib.sha256(_toy_weights().tobytes()).hexdigest()
+
+
+@pytest.mark.parametrize("seed", [0, 1, 2**31 - 1])
+@pytest.mark.parametrize("global_state, other_global_state", [(0, 1), (7, 12345)])
+def test_weights_depend_on_seed_not_on_the_global_generator(
+    seed, global_state, other_global_state
+):
+    torch.manual_seed(global_state)
+    first = _toy_weights(seed=seed)
+    torch.manual_seed(other_global_state)
+    torch.rand(7)
+    second = _toy_weights(seed=seed)
+    assert np.array_equal(first, second)
+
+
+def test_calibration_does_not_consume_the_global_generator():
+    torch.manual_seed(123)
+    expected = torch.rand(5)
+    torch.manual_seed(123)
+    _toy_weights()
+    assert torch.equal(torch.rand(5), expected)
+
+
+def test_different_seeds_give_different_weights():
+    assert not np.array_equal(_toy_weights(seed=0), _toy_weights(seed=1))
+
+
+def test_fresh_processes_give_identical_weights():
+    """The failure this guards against only shows across processes."""
+    command = [
+        sys.executable,
+        "-c",
+        "from policyengine_uk_data.tests.test_calibrate_determinism import "
+        "_toy_weights_digest; print(_toy_weights_digest())",
+    ]
+    digests = [
+        subprocess.run(command, capture_output=True, text=True, check=True)
+        .stdout.strip()
+        .splitlines()[-1]
+        for _ in range(2)
+    ]
+    assert digests[0] == digests[1] == _toy_weights_digest()
diff --git a/policyengine_uk_data/utils/calibrate.py b/policyengine_uk_data/utils/calibrate.py
index a31dcc1f..c47a7682 100644
--- a/policyengine_uk_data/utils/calibrate.py
+++ b/policyengine_uk_data/utils/calibrate.py
@@ -14,6 +14,7 @@
 
 
 DEFAULT_ZERO_WEIGHT_PRIOR_TOTAL_SHARE = 0.5
+DEFAULT_CALIBRATION_SEED = 0
 
 
 def default_weight_dataset_key() -> str:
@@ -221,6 +222,7 @@ def calibrate_local_areas(
     nested_progress=None,
     time_period: int | str | None = None,
     zero_weight_prior_total_share: float = DEFAULT_ZERO_WEIGHT_PRIOR_TOTAL_SHARE,
+    seed: int = DEFAULT_CALIBRATION_SEED,
 ):
     """
     Generic calibration function for local areas (constituencies, local authorities, etc.)
@@ -239,6 +241,8 @@ def calibrate_local_areas(
         area_name: Name of the area type for logging
         zero_weight_prior_total_share: Share of prior household mass to reserve for
             rows whose incoming household_weight is zero.
+        seed: Seed for the dropout masks. The same inputs and seed give the same
+            weights in every process.
     """
     if dataset_key is None:
         dataset_key = default_weight_dataset_key()
@@ -381,11 +385,19 @@ def pct_close(w, t=0.1, local=True, national=True):
 
         return numerator / denominator
 
+    # Dropout masks come from their own seeded generator. torch seeds its
+    # global generator differently in every process, so masks drawn from it
+    # (the earlier `torch.rand_like`) gave every build different weights.
+    dropout_generator = torch.Generator().manual_seed(seed)
+
     def dropout_weights(weights, p):
         if p == 0:
             return weights
         # Replace p% of the weights with the mean value of the rest of them
-        mask = torch.rand_like(weights) < p
+        mask = (
+            torch.rand(weights.shape, generator=dropout_generator, dtype=weights.dtype)
+            < p
+        )
         mean = weights[~mask].mean()
         masked_weights = weights.clone()
         masked_weights[mask] = mean

From a7650c4dfa13e3dfdae31b93069fc30423554959 Mon Sep 17 00:00:00 2001
From: Max Ghenis <mghenis@gmail.com>
Date: Mon, 12 Oct 2026 05:14:08 -0400
Subject: [PATCH 2/3] Cover calibration backends with a shared seed and
 byte-identity tests

---
 .../deterministic-calibration.fixed.md        |  2 +-
 .../tests/test_calibrate_determinism.py       | 10 +--
 .../tests/test_calibrate_l0.py                | 66 ++++++++++++++++++-
 policyengine_uk_data/utils/calibrate.py       |  7 +-
 policyengine_uk_data/utils/calibrate_l0.py    | 51 ++++++++------
 5 files changed, 107 insertions(+), 29 deletions(-)

diff --git a/changelog.d/deterministic-calibration.fixed.md b/changelog.d/deterministic-calibration.fixed.md
index 877dd757..6e228f8c 100644
--- a/changelog.d/deterministic-calibration.fixed.md
+++ b/changelog.d/deterministic-calibration.fixed.md
@@ -1 +1 @@
-- Seed the weight dropout in `calibrate_local_areas` with its own generator (`seed`, default 0). The dropout masks were drawn from torch's global generator, which torch seeds differently in every process, so two builds of the same commit with the same inputs gave different household weights and aggregates.
+- Seed calibration randomness (`seed`, default 0): dense weight dropout uses its own generator, and L0 gate initialization and sampling use an isolated CPU RNG context. Previously, unseeded torch draws gave builds of the same commit different household weights and aggregates. Fixed inputs, targets, dependency versions, thread count and platform now give identical weight bytes without consuming the caller's RNG stream.
diff --git a/policyengine_uk_data/tests/test_calibrate_determinism.py b/policyengine_uk_data/tests/test_calibrate_determinism.py
index 0d8d5287..9f318f64 100644
--- a/policyengine_uk_data/tests/test_calibrate_determinism.py
+++ b/policyengine_uk_data/tests/test_calibrate_determinism.py
@@ -2,8 +2,8 @@
 
 Invariants of ``calibrate_local_areas``:
 
-- the saved weights are a function of the inputs and ``seed`` alone: not of
-  the process, and not of whatever has consumed torch's global generator;
+- with fixed inputs, targets, thread count and platform, saved weight bytes
+  depend on ``seed``: not on the process or torch's global generator;
 - calibration does not consume the global generator;
 - ``seed`` is used: different seeds draw different dropout masks.
 
@@ -84,7 +84,9 @@ def test_weights_depend_on_seed_not_on_the_global_generator(
     torch.manual_seed(other_global_state)
     torch.rand(7)
     second = _toy_weights(seed=seed)
-    assert np.array_equal(first, second)
+    assert first.shape == second.shape
+    assert first.dtype == second.dtype
+    assert first.tobytes() == second.tobytes()
 
 
 def test_calibration_does_not_consume_the_global_generator():
@@ -96,7 +98,7 @@ def test_calibration_does_not_consume_the_global_generator():
 
 
 def test_different_seeds_give_different_weights():
-    assert not np.array_equal(_toy_weights(seed=0), _toy_weights(seed=1))
+    assert _toy_weights(seed=0).tobytes() != _toy_weights(seed=1).tobytes()
 
 
 def test_fresh_processes_give_identical_weights():
diff --git a/policyengine_uk_data/tests/test_calibrate_l0.py b/policyengine_uk_data/tests/test_calibrate_l0.py
index 356a1e3e..0edd894c 100644
--- a/policyengine_uk_data/tests/test_calibrate_l0.py
+++ b/policyengine_uk_data/tests/test_calibrate_l0.py
@@ -7,12 +7,15 @@
 import sys
 import types
 
+import h5py
 import numpy as np
 import pandas as pd
 import pytest
+import torch
 from scipy import sparse as sp
 
 from policyengine_uk_data.datasets.frs_release import CURRENT_FRS_RELEASE
+from policyengine_uk_data.utils.calibrate import DEFAULT_CALIBRATION_SEED
 from policyengine_uk_data.utils.calibrate_l0 import (
     _build_sparse_calibration_matrix,
     calibrate_l0,
@@ -147,8 +150,6 @@ def test_l0_reduces_error(self):
         )
 
         # Check predictions
-        import torch
-
         with torch.no_grad():
             y_pred = model.forward(M, deterministic=True).numpy()
 
@@ -186,6 +187,67 @@ def test_sparsity_with_strong_l0(self):
         assert sparsity > 0.1, f"Sparsity {sparsity:.1%} too low with strong L0 penalty"
 
 
+def test_l0_seeded_weights_are_byte_identical_and_preserve_rng(monkeypatch, tmp_path):
+    """Exercise real gate initialization, sampling, optimization and persistence."""
+    monkeypatch.setattr(
+        "policyengine_uk_data.utils.calibrate_l0.STORAGE_FOLDER", tmp_path
+    )
+
+    class Dataset:
+        def __init__(self):
+            self.household = pd.DataFrame({"household_weight": np.arange(1.0, 9.0)})
+
+        def copy(self):
+            clone = Dataset()
+            clone.household = self.household.copy()
+            return clone
+
+    def matrix_fn(_dataset, time_period=None):
+        return (
+            pd.DataFrame(
+                {
+                    "households": np.ones(8),
+                    "income": np.array([1.0, 3.0, 5.0, 7.0, 2.0, 4.0, 6.0, 8.0]),
+                }
+            ),
+            pd.DataFrame({"households": [40.0], "income": [200.0]}),
+            np.ones((1, 8)),
+        )
+
+    def national_matrix_fn(_dataset, time_period=None):
+        return pd.DataFrame({"households": np.ones(8)}), pd.Series({"households": 40.0})
+
+    def run_calibration(name, **kwargs):
+        state_before = torch.random.get_rng_state().clone()
+        calibrated = calibrate_l0(
+            dataset=Dataset(),
+            matrix_fn=matrix_fn,
+            national_matrix_fn=national_matrix_fn,
+            area_count=1,
+            weight_file=name,
+            dataset_key="2025",
+            epochs=16,
+            **kwargs,
+        )
+        assert torch.equal(torch.random.get_rng_state(), state_before)
+        returned_weights = calibrated.household.household_weight.to_numpy()
+        with h5py.File(tmp_path / name, "r") as weights_file:
+            saved_weights = weights_file["2025"][:]
+        assert saved_weights.tobytes() == returned_weights.tobytes()
+        return saved_weights.tobytes()
+
+    # Different ambient CPU RNG states must not alter either calibration.
+    with torch.random.fork_rng(devices=[]):
+        torch.default_generator.manual_seed(1234)
+        first = run_calibration("first.h5")
+        torch.rand(64)
+        repeat = run_calibration("repeat.h5", seed=DEFAULT_CALIBRATION_SEED)
+        different = run_calibration("different.h5", seed=DEFAULT_CALIBRATION_SEED + 1)
+
+    assert first == repeat
+    assert first != different
+
+
 class TestL0CalibrationPeriods:
     @pytest.mark.parametrize(
         "dataset_key",
diff --git a/policyengine_uk_data/utils/calibrate.py b/policyengine_uk_data/utils/calibrate.py
index e0c3ea1c..db45bf5c 100644
--- a/policyengine_uk_data/utils/calibrate.py
+++ b/policyengine_uk_data/utils/calibrate.py
@@ -14,6 +14,8 @@
 
 
 DEFAULT_ZERO_WEIGHT_PRIOR_TOTAL_SHARE = 0.5
+# Shared default for calibration randomness. Each call starts a fresh RNG;
+# reproducibility also requires fixed inputs, targets, torch threads and platform.
 DEFAULT_CALIBRATION_SEED = 0
 
 
@@ -241,8 +243,9 @@ def calibrate_local_areas(
         area_name: Name of the area type for logging
         zero_weight_prior_total_share: Share of prior household mass to reserve for
             rows whose incoming household_weight is zero.
-        seed: Seed for the dropout masks. The same inputs and seed give the same
-            weights in every process.
+        seed: Seed for the dropout masks (default DEFAULT_CALIBRATION_SEED).
+            Fixed inputs, targets, torch thread count and platform give
+            byte-identical weights in every process, without using the global RNG.
     """
     if dataset_key is None:
         dataset_key = default_weight_dataset_key()
diff --git a/policyengine_uk_data/utils/calibrate_l0.py b/policyengine_uk_data/utils/calibrate_l0.py
index 6ff8a06f..c5fcfd25 100644
--- a/policyengine_uk_data/utils/calibrate_l0.py
+++ b/policyengine_uk_data/utils/calibrate_l0.py
@@ -20,6 +20,7 @@
 from scipy import sparse as sp
 
 from policyengine_uk_data.storage import STORAGE_FOLDER
+from policyengine_uk_data.utils.calibrate import DEFAULT_CALIBRATION_SEED
 
 logger = logging.getLogger(__name__)
 
@@ -163,6 +164,7 @@ def calibrate_l0(
     get_performance=None,
     nested_progress=None,
     time_period: int | str | None = None,
+    seed: int = DEFAULT_CALIBRATION_SEED,
 ):
     """Calibrate local area weights using L0-regularised optimisation.
 
@@ -190,6 +192,10 @@ def calibrate_l0(
         nested_progress: Progress tracker for nested display.
         time_period: Target period to pass to matrix functions. Defaults to
             dataset_key when the key is a year.
+        seed: Seed for gate initialization and training samples. Shares the
+            dense calibrator's default. Identical inputs and seed give identical
+            weight bytes on CPU with fixed dependency versions, thread count
+            and platform.
 
     Returns:
         dataset with calibrated household_weight.
@@ -239,26 +245,31 @@ def calibrate_l0(
         epochs,
     )
 
-    # Create and fit the L0 model
-    model = SparseCalibrationWeights(
-        n_features=n_records,
-        init_weights=init_weights,
-        init_keep_prob=init_keep_prob,
-        device="cpu",
-    )
-
-    model.fit(
-        M=M,
-        y=y,
-        lambda_l0=lambda_l0,
-        lambda_l2=lambda_l2,
-        lr=lr,
-        epochs=epochs,
-        loss_type="relative",
-        verbose=verbose,
-        verbose_freq=50,
-        target_groups=target_groups,
-    )
+    # l0-python draws initialization jitter and HardConcrete gates from the
+    # global CPU generator and does not accept a generator or seed. Isolate
+    # those draws so earlier random consumers cannot affect calibration and
+    # calibration does not change the caller's random stream.
+    with torch.random.fork_rng(devices=[]):
+        torch.default_generator.manual_seed(seed)
+        model = SparseCalibrationWeights(
+            n_features=n_records,
+            init_weights=init_weights,
+            init_keep_prob=init_keep_prob,
+            device="cpu",
+        )
+
+        model.fit(
+            M=M,
+            y=y,
+            lambda_l0=lambda_l0,
+            lambda_l2=lambda_l2,
+            lr=lr,
+            epochs=epochs,
+            loss_type="relative",
+            verbose=verbose,
+            verbose_freq=50,
+            target_groups=target_groups,
+        )
 
     # Extract final weights
     with torch.no_grad():

From 60a87ff7ca8e75abea0f09bb49d37fd92901fc39 Mon Sep 17 00:00:00 2001
From: Max Ghenis <mghenis@gmail.com>
Date: Mon, 12 Oct 2026 05:26:48 -0400
Subject: [PATCH 3/3] Prepare controlled four-build reproducibility plan for
 the UK hub

---
 jobs/adopt-ud507/BUILD_PLAN.md               |  76 ++++
 jobs/adopt-ud507/QUEUE_ROW.md                |   7 +
 jobs/adopt-ud507/cached_build.py             |  57 +++
 jobs/adopt-ud507/compare.py                  | 375 +++++++++++++++++++
 jobs/adopt-ud507/freeze_inputs.sh            |  16 +
 jobs/adopt-ud507/run_build.sh                | 116 ++++++
 jobs/adopt-ud507/seed_http_cache.py          |  56 +++
 jobs/adopt-ud507/setup_build_worktree.sh     |  52 +++
 jobs/adopt-ud507/totals.py                   |  66 ++++
 jobs/adopt-ud507/verify_historical_spread.py |  43 +++
 10 files changed, 864 insertions(+)
 create mode 100644 jobs/adopt-ud507/BUILD_PLAN.md
 create mode 100644 jobs/adopt-ud507/QUEUE_ROW.md
 create mode 100644 jobs/adopt-ud507/cached_build.py
 create mode 100644 jobs/adopt-ud507/compare.py
 create mode 100644 jobs/adopt-ud507/freeze_inputs.sh
 create mode 100644 jobs/adopt-ud507/run_build.sh
 create mode 100644 jobs/adopt-ud507/seed_http_cache.py
 create mode 100644 jobs/adopt-ud507/setup_build_worktree.sh
 create mode 100644 jobs/adopt-ud507/totals.py
 create mode 100644 jobs/adopt-ud507/verify_historical_spread.py

diff --git a/jobs/adopt-ud507/BUILD_PLAN.md b/jobs/adopt-ud507/BUILD_PLAN.md
new file mode 100644
index 00000000..04957869
--- /dev/null
+++ b/jobs/adopt-ud507/BUILD_PLAN.md
@@ -0,0 +1,76 @@
+# #507 reproducibility build plan
+
+Status: **PENDING HUB INSTALLATION; no builds or microsimulations started by this job.** This workspace cannot write `~/reviews/uk-hub`. The hub must install these artifacts in `~/reviews/uk-hub/jobs/adopt-ud507/` and add the proposed row from `QUEUE_ROW.md` to `~/reviews/uk-hub/rebuild-queue.md`. Run only after the hub replaces that row's Status with a grant containing **`GO mm/dd hh:mmZ`**, for example `**GO 10/12 12:00Z (UK hub).**` The driver requires exactly one `ud#507` queue row, checks its Status before waiting and again before launch, and rejects `NO GO`, `LAPSED`, `EXPIRED` or `REVOKED` even if another GO token is present.
+
+Run four fresh builds, one at a time: `branch1`, `branch2`, `main1`, `main2`. The branch uses the final reviewed head of `origin/deterministic-calibration`; resolve it once before setup and preserve it in `expected_refs.json`. Both main controls use **`2b9c1c2c64ced6e1d95fed1ed398a22627233669`**. The saved pre-resume merge was `d9d13a182a7fdaa8f38a27f0c19985639e2c5888`; use the final head including the new tests and L0 fix.
+
+All four use the same frozen HTTP responses and cached imputation models, Python 3.13, the committed `uv.lock` (policyengine-uk **2.122.2**, policyengine-core **3.32.13**), **512 epochs**, **one OA clone**, and **one CPU thread**. `TESTING` and `PYTHONPATH` are unset. No wrapper seeds the process-wide RNG: the branch owns calibration seed 0, while main retains its unseeded weight dropout.
+
+`setup_build_worktree.sh` derives a temporary copy of the requested harness at `~/reviews/uk-data-spi-reported-2026-10-01/setup_build_worktree.sh`. Its required SHA256 is `3207a17bdfc1e441cba58ba34c063e2413ba7fc2242dcacff25c338c3fcfb1de`. The only functional change removes that harness's automatic `cherry-pick -x 348dd49`: the branch already contains #507, and applying it to main would invalidate the controls. It also adjusts the success message. Setup aborts if the source harness changes, the destination exists, the requested ref changes, the branch lacks the original #507 commit, or main's dropout is seeded.
+
+Frozen models come from the harness's `prereqs/*.pkl`, except `income_spi_2022_23.pkl`, taken from the 7 October `uk-data-combined-1008` build to match #544's updated metadata. Frozen HTTP responses come from the harness's cache, plus its one missing post-batch HMRC salary-sacrifice CSV. `seed_http_cache.py` checks that CSV against SHA256 `04e0dab6354b01c96d04f9e379f237ea2083476bd9e920c882a236cdcf88101d` and uses the combined build's saved response metadata. No live target fetches are allowed. Any missing cache response, changed model, or changed cache manifest fails the build.
+
+The hub installs the committed scripts and freezes the refs with these commands. Preserve the existing job's progress and scripts first; keep its other evidence files.
+
+```zsh
+set -euo pipefail
+REPO=~/PolicyEngine/policyengine-uk-data
+J=~/reviews/uk-hub/jobs/adopt-ud507
+git -C "$REPO" fetch origin deterministic-calibration main
+BRANCH_REF=$(git -C "$REPO" rev-parse origin/deterministic-calibration)
+MAIN_REF=2b9c1c2c64ced6e1d95fed1ed398a22627233669
+mkdir -p "$J/previous-attempt-20261012"
+if [[ -f "$J/PROGRESS.md" ]]; then cp -c "$J/PROGRESS.md" "$J/previous-attempt-20261012/PROGRESS.md"; fi
+for FILE in BUILD_PLAN.md QUEUE_ROW.md setup_build_worktree.sh freeze_inputs.sh run_build.sh cached_build.py seed_http_cache.py compare.py totals.py verify_historical_spread.py; do
+  if [[ -f "$J/$FILE" ]]; then cp -c "$J/$FILE" "$J/previous-attempt-20261012/$FILE"; fi
+  git -C "$REPO" show "$BRANCH_REF:jobs/adopt-ud507/$FILE" > "$J/$FILE"
+done
+printf '\n- 2026-10-12: Installed #507 build artifacts at branch %s; code and tests ready, impact pending the four queued builds. Existing progress preserved.\n' "$BRANCH_REF" >> "$J/PROGRESS.md"
+python3 - "$J" "$BRANCH_REF" "$MAIN_REF" <<'PY'
+import json
+import sys
+from pathlib import Path
+job, branch, main = sys.argv[1:]
+Path(job, "expected_refs.json").write_text(json.dumps({"branch": branch, "main": main}, indent=2) + "\n")
+PY
+```
+
+After installing the queue row and receiving GO, use these exact setup and build commands. `freeze_inputs.sh` runs once and refuses to overwrite an existing frozen directory. The four worktrees must be new; cached models are copied from the single frozen snapshot into each. Setup runs the pinned harness's `uv sync --frozen --all-extras --python 3.13`.
+
+```zsh
+zsh "$J/freeze_inputs.sh" "$REPO"
+zsh "$J/setup_build_worktree.sh" ud507-branch1 "$BRANCH_REF" branch
+zsh "$J/setup_build_worktree.sh" ud507-branch2 "$BRANCH_REF" branch
+zsh "$J/setup_build_worktree.sh" ud507-main1 "$MAIN_REF" main
+zsh "$J/setup_build_worktree.sh" ud507-main2 "$MAIN_REF" main
+zsh "$J/run_build.sh" ud507-branch1 branch1
+zsh "$J/run_build.sh" ud507-branch2 branch2
+zsh "$J/run_build.sh" ud507-main1 main1
+zsh "$J/run_build.sh" ud507-main2 main2
+```
+
+Before each build, the driver requires statvfs free disk above 100 GB, compressor below 30 GB, load5 below 36 on three checks 60 seconds apart, no other data build, and `reservation.py allowed uk-hub` exit 0. The hub must retain its GO and report starts and finishes under its queue rules. A task-specific monitor checks free disk every 30 seconds while the build runs; below 30 GB, or if the disk check fails, it sends TERM only to that build's captured child PID and logs the event in `disk_floor.log`. Traps stop the monitor and its own build if the driver exits early. The driver gives up after six hours of waiting without starting a build. Expect approximately 1.2 GB of retained artifacts per build; gzip logs and delete storage build copies after each build. Raw survey prerequisites remain the same shared harness inputs.
+
+Then run four microsimulations sequentially in the same locked environment, followed by comparison. `totals.py` checks the thread caps, compressor and engine versions before constructing a microsimulation. Re-read the hub's reservation and running-process state before this finisher stage.
+
+```zsh
+export OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 MKL_NUM_THREADS=1
+export VECLIB_MAXIMUM_THREADS=1 NUMEXPR_NUM_THREADS=1
+export OMP_WAIT_POLICY=PASSIVE PYTHONHASHSEED=0
+unset TESTING PYTHONPATH
+W=~/PolicyEngine/_worktrees/ud507-branch1
+for LABEL in branch1 branch2 main1 main2; do
+  "$W/.venv/bin/python" "$J/totals.py" "$J/builds/$LABEL/enhanced_frs_2024_25.h5" "$J/impact/$LABEL.json"
+done
+"$W/.venv/bin/python" "$J/compare.py"
+```
+
+`compare.py` writes `RESULT.md` and `RESULT.json`. PASS requires exact recorded refs and committed calibration source, matching locks/settings/frozen manifests, successful builds, no target-cache misses or OBR receipts fallback, and complete nonempty weight arrays. Branch constituency and local-authority weight arrays must have identical shapes, dtypes and bytes, and branch dataset tables and decompressed long-geography content must match. Main's weight arrays must differ. All four pre-calibration FRS tables and eFRS person/benefit-unit tables must agree; other input noise fails the comparison. Aggregate JSONs must match their dataset hashes and engine versions. The report gives both builds' income tax and household net income totals and their absolute spread for 2025 and 2026. Impact remains pending these queued builds; do not present the historical measurements as the controlled experiment's result.
+
+The PR's rounded historical figures are verified by reading saved aggregate JSONs only:
+
+```zsh
+python3 "$J/verify_historical_spread.py"
+```
+
+Sources are `~/reviews/frs-cvpay-property-income-2026-09-30/impact/efrs_{main_matched,main_matched_2,main_a,main_b}.json`. All report policyengine-uk head `44240bd8d7d7e97f2980ccc8134e76e61541966f`. Across those four outputs, 2026 income tax ranges from £312.411684864bn (`main_b`) to £313.196281856bn (`main_matched`): **£0.784596992bn**. Household net income ranges from £1756.085878784bn (`main_a`) to £1759.036964864bn (`main_b`): **£2.951086080bn**. These support the quoted roughly £0.8bn and £3bn maxima; the maxima use different pairs. The verifier asserts the shared engine head and both spreads without running a build or simulation.
diff --git a/jobs/adopt-ud507/QUEUE_ROW.md b/jobs/adopt-ud507/QUEUE_ROW.md
new file mode 100644
index 00000000..546e1cf9
--- /dev/null
+++ b/jobs/adopt-ud507/QUEUE_ROW.md
@@ -0,0 +1,7 @@
+# Proposed rebuild queue row
+
+Status: **PENDING HUB INSTALLATION**. The external queue is read-only in this job. The hub should append the single row below to `~/reviews/uk-hub/rebuild-queue.md`, replace `NEXT` with its next unused row number, and resolve the final reviewed branch head into the installed job's `expected_refs.json` before giving GO. Replace Status with a grant containing `GO mm/dd hh:mmZ`; remove the pending `NO GO` text. No build or microsimulation has been started.
+
+| # | Item | Owner | Precondition | Status |
+|---|---|---|---|---|
+| NEXT | ud#507 calibration reproducibility: FOUR fresh builds, one at a time (`branch1`, `branch2`, `main1`, `main2`), then four sequential aggregate microsims. Branch = final reviewed `origin/deterministic-calibration`; both main controls = `2b9c1c2c64ced6e1d95fed1ed398a22627233669`. Pinned hub setup harness with automatic #507 cherry-pick omitted, shared frozen HTTP and pkl manifests, 512 epochs, one OA clone, one CPU thread, policyengine-uk 2.122.2/core 3.32.13. Compare weight bytes and 2025/2026 income tax/household net income spreads; approximately 1.2 GB retained per build. Runbook: `~/reviews/uk-hub/jobs/adopt-ud507/BUILD_PLAN.md`; preserve the PR as WIP draft until impact is measured. | UK hub (adopt-ud507; code job prepares artifacts only) | Hub installs artifacts/refs and writes GO in this row; one build at a time; statvfs free >100 GB; compressor <30 GB; load5 <36 three checks; reservation.py allowed uk-hub exit 0; same frozen inputs; main controls retain unseeded calibration. | **PENDING HUB INSTALLATION — NO GO.** Code/tests ready; four builds and impact measurements pending. |
diff --git a/jobs/adopt-ud507/cached_build.py b/jobs/adopt-ud507/cached_build.py
new file mode 100644
index 00000000..fdc77525
--- /dev/null
+++ b/jobs/adopt-ud507/cached_build.py
@@ -0,0 +1,57 @@
+"""Run the hub's dataset command with a frozen, read-only HTTP cache.
+
+Unlike the original harness, a missing response terminates the build instead
+of fetching a live target. No random generator is seeded by this wrapper.
+Usage, from a build worktree: python cached_build.py <cache> <http.log>
+"""
+
+import hashlib
+import json
+import runpy
+import sys
+from pathlib import Path
+
+import requests
+from requests.structures import CaseInsensitiveDict
+
+
+def main():
+    cache, http_log = Path(sys.argv[1]), Path(sys.argv[2])
+
+    def cached_get(url, **kwargs):
+        key = hashlib.sha256(
+            json.dumps(
+                [url, kwargs.get("params")], sort_keys=True, default=str
+            ).encode()
+        ).hexdigest()[:24]
+        meta_path, body_path = cache / f"{key}.json", cache / f"{key}.body"
+        if not meta_path.exists() or not body_path.exists():
+            with http_log.open("a") as log:
+                log.write(f"MISS frozen cache {url}\n")
+            # Target loaders catch ordinary exceptions and silently fall back.
+            # SystemExit stops the build before that can change its target set.
+            raise SystemExit(f"Frozen target cache has no response for {url}")
+        meta, body = json.loads(meta_path.read_text()), body_path.read_bytes()
+        response = requests.Response()
+        response.status_code, response._content, response.url = (
+            meta["status"],
+            body,
+            meta["url"],
+        )
+        response.headers = CaseInsensitiveDict(meta["headers"])
+        response.encoding, response.reason = meta["encoding"], meta["reason"]
+        with http_log.open("a") as log:
+            log.write(
+                f"cache {meta['status']} {len(body)}b "
+                f"{hashlib.sha256(body).hexdigest()} {url}\n"
+            )
+        return response
+
+    requests.get = cached_get
+    runpy.run_path(
+        "policyengine_uk_data/datasets/create_datasets.py", run_name="__main__"
+    )
+
+
+if __name__ == "__main__":
+    main()
diff --git a/jobs/adopt-ud507/compare.py b/jobs/adopt-ud507/compare.py
new file mode 100644
index 00000000..ecce29ed
--- /dev/null
+++ b/jobs/adopt-ud507/compare.py
@@ -0,0 +1,375 @@
+"""Compare #507's four builds. Aggregates only: counts, totals and hashes, never records.
+
+PASS needs all of:
+- provenance: every build exited 0; branch1/branch2 share the branch commit and the seeded calibrate.py, main1/main2
+  share the main commit and the unseeded one; clean trees; the same thread settings and cached models; every target
+  download served from the cache (no MISS) with the same bodies in all four; no OBR receipts failure (notice 7);
+- branch1 and branch2: byte-identical constituency and local authority weights, identical eFRS and FRS tables,
+  identical long-geography file, and identical income tax and household net income totals;
+- main1 and main2: different weights (main's dropout is unseeded).
+It reports main's spread in income tax and household net income, and how far the weights moved.
+
+Usage (with a build worktree's venv, which has h5py, pandas and tables):
+  ~/PolicyEngine/_worktrees/ud507-branch/.venv/bin/python compare.py
+Reads builds/<label>/ and impact/<label>.json for branch1, branch2, main1, main2. Writes RESULT.md and RESULT.json.
+Exit 0 = PASS, 1 = FAIL.
+"""
+
+import gzip
+import hashlib
+import json
+from collections import Counter
+from pathlib import Path
+
+import h5py
+import numpy as np
+import pandas as pd
+
+JOB = Path(__file__).resolve().parent
+PAIRS = {"branch": ("branch1", "branch2"), "main": ("main1", "main2")}
+WEIGHT_FILES = ("parliamentary_constituency_weights.h5", "local_authority_weights.h5")
+STORES = ("enhanced_frs_2024_25.h5", "frs_2024_25.h5")
+METRICS = ("income_tax", "household_net_income")
+YEARS = ("2025", "2026")
+failures, notes, report = [], [], {}
+
+
+def build(label):
+    return JOB / "builds" / label
+
+
+def text(label, name):
+    path = build(label) / name
+    if path.exists():
+        return path.read_text()
+    zipped = path.with_name(path.name + ".gz")
+    if zipped.exists():
+        with gzip.open(zipped, "rt") as source:
+            return source.read()
+    return ""
+
+
+def check(ok, message):
+    if not ok:
+        failures.append(message)
+    return ok
+
+
+def downloads(label):
+    lines = text(label, "http.log").splitlines()
+    misses = [line for line in lines if line.startswith("MISS")]
+    # "cache <status> <bytes>b <sha12> <url>": the body each URL was served.
+    return misses, Counter(tuple(line.split()[1:]) for line in lines)
+
+
+def weights_diff(a, b):
+    out = {}
+    for name in WEIGHT_FILES:
+        with (
+            h5py.File(build(a) / name, "r") as fa,
+            h5py.File(build(b) / name, "r") as fb,
+        ):
+            check(set(fa) == set(fb), f"{name}: keys differ between {a} and {b}")
+            check(
+                set(fa) == {"2025"} and set(fb) == {"2025"},
+                f"{name}: missing expected 2025 weight arrays",
+            )
+            for key in sorted(set(fa) & set(fb)):
+                x, y = fa[key][:], fb[key][:]
+                areas = 650 if name.startswith("parliamentary") else 360
+                check(
+                    x.ndim == 2 and x.shape[0] == areas and x.size > 0,
+                    f"{a}/{name}/{key}: empty or invalid weight shape",
+                )
+                check(
+                    y.ndim == 2 and y.shape[0] == areas and y.size > 0,
+                    f"{b}/{name}/{key}: empty or invalid weight shape",
+                )
+                check(
+                    np.isfinite(x).all() and np.isfinite(y).all(),
+                    f"{name}/{key}: weights include non-finite values",
+                )
+                same_layout = x.dtype == y.dtype and x.shape == y.shape
+                identical = same_layout and x.tobytes() == y.tobytes()
+                entry = {
+                    "shape": list(x.shape),
+                    "dtype": str(x.dtype),
+                    "byte_identical": identical,
+                }
+                if same_layout and not identical:
+                    entry["cells_differing"] = int((x != y).sum())
+                    entry["sum_abs_diff_over_total"] = float(
+                        np.abs(x.astype(float) - y).sum() / x.astype(float).sum()
+                    )
+                out[f"{name}/{key}"] = entry
+    return out
+
+
+def tables_diff(a, b):
+    out = {}
+    for name in STORES:
+        with (
+            pd.HDFStore(build(a) / name, "r") as sa,
+            pd.HDFStore(build(b) / name, "r") as sb,
+        ):
+            check(
+                set(sa.keys()) == set(sb.keys()),
+                f"{name}: tables differ between {a} and {b}",
+            )
+            expected_tables = {"/household", "/person", "/benunit", "/time_period"}
+            check(
+                set(sa.keys()) == expected_tables and set(sb.keys()) == expected_tables,
+                f"{name}: incomplete expected dataset tables",
+            )
+            for key in sorted(set(sa.keys()) & set(sb.keys())):
+                x, y = sa[key], sb[key]
+                if isinstance(x, pd.Series):
+                    x, y = x.to_frame(), y.to_frame()
+                differing = sorted(
+                    c
+                    for c in set(x.columns) | set(y.columns)
+                    if c not in x or c not in y or not x[c].equals(y[c])
+                )
+                out[f"{name}{key}"] = {
+                    "rows": [len(x), len(y)],
+                    "columns_differing": differing,
+                    "layout_identical": x.index.equals(y.index)
+                    and x.columns.equals(y.columns),
+                }
+    return out
+
+
+def household_weight_distance(a, b):
+    x, y = (pd.read_hdf(build(label) / STORES[0], "household") for label in (a, b))
+    if (
+        len(x) != len(y)
+        or not (x.household_id.to_numpy() == y.household_id.to_numpy()).all()
+    ):
+        return {"aligned": False}
+    w, v = x.household_weight.to_numpy(float), y.household_weight.to_numpy(float)
+    return {
+        "aligned": True,
+        "rows": len(w),
+        "rows_differing": int((w != v).sum()),
+        "totals": [float(w.sum()), float(v.sum())],
+        "sum_abs_diff_over_total": float(np.abs(w - v).sum() / w.sum()),
+    }
+
+
+def long_geography_digest(label):
+    with gzip.open(build(label) / "local_geography_weights.csv.gz", "rb") as f:
+        return hashlib.sha256(f.read()).hexdigest()
+
+
+def totals(label):
+    path = JOB / "impact" / f"{label}.json"
+    if not check(path.exists(), f"impact/{label}.json missing (run placebo/totals.py)"):
+        return None
+    data = json.loads(path.read_text())
+    check(
+        data["dataset_sha256"] == text(label, "dataset.sha256").strip(),
+        f"{label}: totals came from a different dataset",
+    )
+    check(
+        data["policyengine_uk_version"] == "2.122.2"
+        and data["policyengine_core_version"] == "3.32.13",
+        f"{label}: totals used a different engine environment",
+    )
+    return {y: {m: data[y][m]["total_bn"] for m in METRICS} for y in YEARS}
+
+
+labels = [label for pair in PAIRS.values() for label in pair]
+expected_refs = json.loads((JOB / "expected_refs.json").read_text())
+
+# Provenance.
+served = {}
+for label in labels:
+    check(
+        (build(label) / "finished").exists()
+        and text(label, "build.log").rstrip().endswith("exit 0"),
+        f"{label}: build did not finish with exit 0",
+    )
+    check(
+        text(label, "status").strip() == "",
+        f"{label}: tracked files modified in the build worktree",
+    )
+    kind = "branch" if label.startswith("branch") else "main"
+    check(
+        text(label, "commit").strip() == expected_refs[kind],
+        f"{label}: build is not of the exact requested {kind} commit",
+    )
+    check(
+        text(label, "calibrate.sha256")
+        == text(label, "calibrate_committed.sha256")
+        != "",
+        f"{label}: calibration source does not match its committed source",
+    )
+    check(
+        "Failed to download/parse OBR receipts" not in text(label, "build.log"),
+        f"{label}: OBR receipts fell back (notice 7)",
+    )
+    misses, served[label] = downloads(label)
+    check(not misses, f"{label}: {len(misses)} downloads missed the frozen cache")
+    check(
+        text(label, "models_after.sha256") == text(label, "models_before.sha256"),
+        f"{label}: an imputation model changed during the build",
+    )
+    check(
+        text(label, "targets_before.sha256") == text(label, "targets_after.sha256"),
+        f"{label}: the frozen target cache changed during the build",
+    )
+for kind, (a, b) in PAIRS.items():
+    check(
+        text(a, "commit") == text(b, "commit") != "",
+        f"{kind} builds are not of one commit",
+    )
+    source = text(a, "calibrate_source")
+    seeded = "dropout_generator = torch.Generator()" in source
+    check(
+        seeded == (kind == "branch") and source == text(b, "calibrate_source"),
+        f"{kind} builds: wrong calibrate.py",
+    )
+check(
+    text("branch1", "commit") != text("main1", "commit"),
+    "branch and main builds share a commit",
+)
+check(
+    len({text(label, "env") for label in labels}) == 1,
+    "thread settings differ between builds",
+)
+check("torch_threads 1" in text("branch1", "env"), "builds did not use one CPU thread")
+check(
+    len({text(label, "lock.sha256") for label in labels}) == 1,
+    "builds used different dependency locks",
+)
+check(
+    len({text(label, "models_before.sha256") for label in labels}) == 1,
+    "builds started from different cached models",
+)
+check(
+    all(
+        text(label, "models_before.sha256")
+        == (JOB / "frozen/models.sha256").read_text()
+        for label in labels
+    ),
+    "builds did not use the frozen model manifest",
+)
+check(
+    all(
+        text(label, "targets_before.sha256")
+        == (JOB / "frozen/targets.sha256").read_text()
+        for label in labels
+    ),
+    "builds did not use the frozen target manifest",
+)
+check(
+    len({frozenset(served[label].items()) for label in labels}) == 1,
+    "builds were served different target downloads",
+)
+
+# Outputs.
+for kind, (a, b) in PAIRS.items():
+    weights = weights_diff(a, b)
+    tables = tables_diff(a, b)
+    entry = {
+        "weights": weights,
+        "tables_differing": {
+            k: v["columns_differing"]
+            for k, v in tables.items()
+            if v["columns_differing"]
+        },
+        "household_weight": household_weight_distance(a, b),
+        "long_geography_identical": long_geography_digest(a)
+        == long_geography_digest(b),
+    }
+    ta, tb = totals(a), totals(b)
+    if ta and tb:
+        entry["totals_bn"] = {
+            y: {m: [ta[y][m], tb[y][m], tb[y][m] - ta[y][m]] for m in METRICS}
+            for y in YEARS
+        }
+        entry["spread_bn"] = {
+            y: {m: abs(tb[y][m] - ta[y][m]) for m in METRICS} for y in YEARS
+        }
+    report[kind] = entry
+    all_identical = all(w["byte_identical"] for w in weights.values())
+    check(
+        len(weights) == len(WEIGHT_FILES), f"{kind}: incomplete weight-array comparison"
+    )
+    check(
+        all(v["layout_identical"] for v in tables.values()),
+        f"{kind}: dataset row or column layouts differ",
+    )
+    if kind == "branch":
+        check(all_identical, "branch builds: weights differ")
+        check(not entry["tables_differing"], "branch builds: eFRS or FRS tables differ")
+        check(
+            entry["long_geography_identical"],
+            "branch builds: long-geography file differs",
+        )
+        if "totals_bn" in entry:
+            check(
+                all(entry["totals_bn"][y][m][2] == 0 for y in YEARS for m in METRICS),
+                "branch builds: totals differ",
+            )
+    else:
+        check(
+            not all_identical,
+            "main builds: weights identical, so main's dropout did not vary",
+        )
+        check(
+            not any(key.startswith(STORES[1]) for key in entry["tables_differing"]),
+            "main builds differ before calibration: an additional source of noise",
+        )
+
+# All four pre-calibration datasets must agree. Person and benefit-unit tables
+# must also agree after calibration; only household weights and the monetary
+# household proxies rescaled from those weights may change.
+for label in labels[1:]:
+    comparison = tables_diff("branch1", label)
+    protected = {
+        key: value
+        for key, value in comparison.items()
+        if key.startswith(STORES[1])
+        or key in (f"{STORES[0]}/person", f"{STORES[0]}/benunit")
+    }
+    check(
+        bool(protected)
+        and all(
+            not value["columns_differing"] and value["layout_identical"]
+            for value in protected.values()
+        ),
+        f"branch1 vs {label}: inputs or non-household calibrated tables differ",
+    )
+
+report["verdict"] = "PASS" if not failures else "FAIL"
+report["failures"], report["notes"] = failures, notes
+(JOB / "RESULT.json").write_text(json.dumps(report, indent=1))
+
+md = [f"# uk-data#507 build comparison: {report['verdict']}", ""]
+md += [f"- FAIL: {f}" for f in failures] + [f"- Note: {n}" for n in notes]
+md += [
+    "",
+    "| Pair | Weight arrays byte-identical | Household weights differing | Σ|Δw|/Σw | eFRS/FRS tables with differing columns |",
+    "|---|---|---|---|---|",
+]
+for kind in PAIRS:
+    e = report[kind]
+    hw = e["household_weight"]
+    md.append(
+        f"| {kind} | {sum(w['byte_identical'] for w in e['weights'].values())} of {len(e['weights'])} "
+        f"| {hw.get('rows_differing', 'n/a')} of {hw.get('rows', 'n/a')} | {hw.get('sum_abs_diff_over_total', float('nan')):.2e} "
+        f"| {', '.join(e['tables_differing']) or 'none'} |"
+    )
+md += [
+    "",
+    "| Pair | Year | Measure | Build 1 (£bn) | Build 2 (£bn) | Difference (£bn) |",
+    "|---|---|---|---|---|---|",
+]
+for kind in PAIRS:
+    for y, measures in report[kind].get("totals_bn", {}).items():
+        for m, (x, z, d) in measures.items():
+            md.append(f"| {kind} | {y} | {m} | {x:.3f} | {z:.3f} | {d:+.3f} |")
+(JOB / "RESULT.md").write_text("\n".join(md) + "\n")
+print("\n".join(md))
+raise SystemExit(0 if not failures else 1)
diff --git a/jobs/adopt-ud507/freeze_inputs.sh b/jobs/adopt-ud507/freeze_inputs.sh
new file mode 100644
index 00000000..99f93e9b
--- /dev/null
+++ b/jobs/adopt-ud507/freeze_inputs.sh
@@ -0,0 +1,16 @@
+#!/bin/zsh
+# Hub only, before its four builds: freeze_inputs.sh <branch checkout>
+set -eu
+J=${0:A:h}
+E=~/reviews/uk-data-spi-reported-2026-10-01
+C=~/PolicyEngine/_worktrees/uk-data-combined-1008/policyengine_uk_data/storage
+[ ! -e "$J/frozen" ] || { echo 'Frozen inputs already exist; preserve them.' >&2; exit 1; }
+mkdir -p "$J/frozen/models"
+for MODEL in "$E"/prereqs/*.pkl; do cp -c "$MODEL" "$J/frozen/models/${MODEL:t}"; done
+# #544 changed the income-model metadata. Use the model retrained on 7 Oct
+# with exactly current main's inputs, instead of silently retraining four times.
+cp -c -f "$C/income_spi_2022_23.pkl" "$J/frozen/models/income_spi_2022_23.pkl"
+(cd "$J/frozen/models" && shasum -a 256 *.pkl) > "$J/frozen/models.sha256"
+python3 "$J/seed_http_cache.py" "$1"
+(cd "$J/frozen/http_cache" && shasum -a 256 *) > "$J/frozen/targets.sha256"
+echo 'Frozen cached models and targets ready. No dataset build ran.'
diff --git a/jobs/adopt-ud507/run_build.sh b/jobs/adopt-ud507/run_build.sh
new file mode 100644
index 00000000..dab2e34d
--- /dev/null
+++ b/jobs/adopt-ud507/run_build.sh
@@ -0,0 +1,116 @@
+#!/bin/zsh
+# Usage: run_build.sh <worktree name> <label>
+# One production-settings build (512 epochs, PE_UK_DATA_OA_CLONES=1, TESTING unset) of the worktree's commit, with every
+# download served from this job's frozen cache. Waits for the hub's gates (statvfs free > 100 GB, compressor < 30 GB,
+# load5 < 36 on three checks 60 s apart, no other uk-data build, reservation.py allowed uk-hub exit 0); gives up after
+# GATE_WAIT_HOURS (default 6) with exit 3. From the harness's run_gated_build.sh (sha256 4a3b4b0a...): same build command
+# and cleanup, with the hub's gates, the thread caps, and outputs (data, weights, download log, model hashes) in builds/<label>/.
+set -euo pipefail
+J=${0:A:h}
+W=~/PolicyEngine/_worktrees/$1
+L=$2
+B=$J/builds/$L
+QUEUE=~/reviews/uk-hub/rebuild-queue.md
+build_pid=''
+monitor_pid=''
+cleanup_children() {
+  [ -z "$monitor_pid" ] || kill -TERM "$monitor_pid" 2>/dev/null || true
+  [ -z "$build_pid" ] || kill -TERM "$build_pid" 2>/dev/null || true
+}
+trap cleanup_children EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+require_go() {
+  python3 - "$QUEUE" <<'PY'
+import re
+import sys
+from pathlib import Path
+
+rows = [line for line in Path(sys.argv[1]).read_text().splitlines()
+        if line.startswith("|") and re.search(r"\bud#507\b", line)]
+if len(rows) != 1:
+    raise SystemExit("Expected exactly one ud#507 queue row")
+status = rows[0].split("|")[-2]
+if re.search(r"\b(?:NO GO|LAPSED|EXPIRED|REVOKED)\b", status, re.I):
+    raise SystemExit("ud#507 has no current hub GO")
+if not re.search(r"\bGO \d{2}/\d{2} \d{2}:\d{2}Z\b", status):
+    raise SystemExit("ud#507 requires GO mm/dd hh:mmZ in its queue Status")
+print(rows[0])
+PY
+}
+export OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 MKL_NUM_THREADS=1 VECLIB_MAXIMUM_THREADS=1 NUMEXPR_NUM_THREADS=1 OMP_WAIT_POLICY=PASSIVE PYTHONHASHSEED=0
+unset TESTING PYTHONPATH
+[ -e $B/finished ] && [ "$(gzip -cd $B/build.log.gz | tail -n 1)" = "exit 0" ] && { echo "$L already built"; exit 0; }
+mkdir -p $B
+require_go > $B/queue_go.txt
+rm -f $B/finished $B/http.log
+gate() {
+  local free comp load5 others res
+  require_go >/dev/null || return 1
+  free=$(python3 -c 'import os;s=os.statvfs("/System/Volumes/Data");print(int(s.f_bavail*s.f_frsize/1e9))')
+  comp=$(vm_stat | awk 'NR==1 {match($0, /[0-9]+ bytes/); page=substr($0,RSTART,RLENGTH)+0} /occupied by compressor/{gsub(/\./,"",$5); print int($5*page/1e9)}')
+  load5=$(sysctl -n vm.loadavg | awk '{print int($3)}')
+  others=$(ps -axo args= | grep -cE '[c]reate_datasets|[c]ached_build\.py|[b]uild_harness\.py' || true)
+  if python3 ~/reviews/fleet-ops/tools/reservation.py allowed uk-hub >/dev/null 2>&1; then res=0; else res=$?; fi
+  echo "$(date -u +%FT%TZ) free=${free}GB compressor=${comp}GB load5=$load5 other_builds=$others reservation=$res" >> $B/gates.log
+  (( free > 100 && comp < 30 && load5 < 36 && others == 0 && res == 0 ))
+}
+deadline=$(( $(date +%s) + ${GATE_WAIT_HOURS:-6} * 3600 ))
+until gate && sleep 60 && gate && sleep 60 && gate; do
+  (( $(date +%s) > deadline )) && { echo "$(date -u +%FT%TZ) gates never held; no build ran" >> $B/gates.log; exit 3; }
+  sleep 300
+done
+cd $W
+S=policyengine_uk_data/storage
+rm -f $S/enhanced_frs_2024_25.h5 $S/frs_2024_25.h5 $S/*_tiny.h5(N) $S/parliamentary_constituency_weights.h5 $S/local_authority_weights.h5 $S/local_geography_weights.csv.gz
+git rev-parse HEAD > $B/commit
+git status --porcelain --untracked-files=no > $B/status
+[ ! -s $B/status ] || { echo 'Tracked build sources changed.' >&2; exit 1; }
+grep -nE "dropout_generator = torch.Generator\(\)|mask = torch.rand_like\(weights\) < p" policyengine_uk_data/utils/calibrate.py > $B/calibrate_source
+shasum -a 256 policyengine_uk_data/utils/calibrate.py | awk '{print $1}' > $B/calibrate.sha256
+git show HEAD:policyengine_uk_data/utils/calibrate.py | shasum -a 256 | awk '{print $1}' > $B/calibrate_committed.sha256
+cmp $B/calibrate.sha256 $B/calibrate_committed.sha256
+(cd $S && shasum -a 256 *.pkl) > $B/models_before.sha256
+cmp $J/frozen/models.sha256 $B/models_before.sha256 || exit 1
+(cd $J/frozen/http_cache && shasum -a 256 *) > $B/targets_before.sha256
+cmp $J/frozen/targets.sha256 $B/targets_before.sha256 || exit 1
+shasum -a 256 uv.lock | awk '{print $1}' > $B/lock.sha256
+{ env | grep -E '^(OMP|OPENBLAS|MKL|VECLIB|NUMEXPR)_|^PYTHONHASHSEED=' | sort; .venv/bin/python -c 'import importlib.metadata as m, torch; print("torch_threads", torch.get_num_threads()); print("torch", m.version("torch")); print("policyengine-uk", m.version("policyengine-uk")); print("policyengine-core", m.version("policyengine-core")); print("epochs", 512); print("oa_clones", 1); assert torch.get_num_threads() == 1; assert m.version("policyengine-uk") == "2.122.2"; assert m.version("policyengine-core") == "3.32.13"'; } > $B/env
+date -u +%FT%TZ > $B/started
+require_go > $B/queue_go.txt
+rc=0
+PE_UK_DATA_OA_CLONES=1 .venv/bin/python $J/cached_build.py $J/frozen/http_cache $B/http.log > $B/build.log 2>&1 &
+build_pid=$!
+(
+  while kill -0 "$build_pid" 2>/dev/null; do
+    free_bytes=$(python3 -c 'import os;s=os.statvfs("/System/Volumes/Data");print(s.f_bavail*s.f_frsize)') || free_bytes=0
+    if (( free_bytes < 30000000000 )); then
+      echo "$(date -u +%FT%TZ) disk floor reached (${free_bytes} bytes); terminating this build PID $build_pid" >> $B/disk_floor.log
+      kill -TERM "$build_pid" 2>/dev/null || true
+      exit 0
+    fi
+    sleep 30
+  done
+) &
+monitor_pid=$!
+wait "$build_pid" || rc=$?
+build_pid=''
+kill -TERM "$monitor_pid" 2>/dev/null || true
+wait "$monitor_pid" 2>/dev/null || true
+monitor_pid=''
+(cd $S && shasum -a 256 *.pkl) > $B/models_after.sha256
+(cd $J/frozen/http_cache && shasum -a 256 *) > $B/targets_after.sha256
+cmp $J/frozen/models.sha256 $B/models_after.sha256 || rc=1
+cmp $J/frozen/targets.sha256 $B/targets_after.sha256 || rc=1
+if [ $rc -eq 0 ]; then
+  for f in enhanced_frs_2024_25.h5 frs_2024_25.h5 parliamentary_constituency_weights.h5 local_authority_weights.h5 local_geography_weights.csv.gz; do cp -c $S/$f $B/$f; done
+  shasum -a 256 $B/enhanced_frs_2024_25.h5 | awk '{print $1}' > $B/dataset.sha256
+fi
+for f in constituency_calibration_log.csv la_calibration_log.csv; do [ -e $f ] && gzip -c $f > $B/$f.gz; done
+rm -f constituency_calibration_log.csv la_calibration_log.csv
+# Retain only the APFS-cloned aggregate artifacts in this job's builds folder.
+rm -f $S/enhanced_frs_2024_25.h5 $S/frs_2024_25.h5 $S/*_tiny.h5(N) $S/parliamentary_constituency_weights.h5 $S/local_authority_weights.h5 $S/local_geography_weights.csv.gz
+echo "exit $rc" >> $B/build.log
+gzip -f $B/build.log
+date -u +%FT%TZ > $B/finished
+exit $rc
diff --git a/jobs/adopt-ud507/seed_http_cache.py b/jobs/adopt-ud507/seed_http_cache.py
new file mode 100644
index 00000000..8009753f
--- /dev/null
+++ b/jobs/adopt-ud507/seed_http_cache.py
@@ -0,0 +1,56 @@
+"""Make this job's frozen download cache: the hub harness's cache plus the one URL current main adds.
+
+The harness cache (~/reviews/uk-data-spi-reported-2026-10-01/http_cache, 12 responses) predates the batch, which
+moved the HMRC salary-sacrifice source to a new URL. The combined-1008 build of 57ce68a fetched that URL live once
+on 7 Oct (12,742 bytes, sha256 04e0dab6...), and the response is byte-identical to the tracked
+policyengine_uk_data/storage/hmrc_pension_relief_tables_6_1_6_2_2024_25.csv. This adds it under cached_build.py's
+key, so all four builds are served every target download from the cache and none fetches live.
+
+Usage: python3 seed_http_cache.py <a uk-data checkout at 2b9c1c2c or later>
+"""
+
+import hashlib
+import json
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+JOB = Path(__file__).resolve().parent
+HARNESS_CACHE = Path.home() / "reviews/uk-data-spi-reported-2026-10-01/http_cache"
+COMBINED_META = (
+    Path.home()
+    / "PolicyEngine/_worktrees/uk-data-combined-1008/_out/hmrc_salary_sacrifice_download.json"
+)
+HMRC_CSV = "policyengine_uk_data/storage/hmrc_pension_relief_tables_6_1_6_2_2024_25.csv"
+HMRC_SHA256 = "04e0dab6354b01c96d04f9e379f237ea2083476bd9e920c882a236cdcf88101d"
+
+
+def key(url):
+    # As cached_build.py: requests.get(url, ...) with no params.
+    return hashlib.sha256(
+        json.dumps([url, None], sort_keys=True, default=str).encode()
+    ).hexdigest()[:24]
+
+
+cache = JOB / "frozen" / "http_cache"
+if cache.exists():
+    sys.exit(f"{cache} exists; preserve the frozen cache")
+subprocess.run(["cp", "-c", "-R", str(HARNESS_CACHE), str(cache)], check=True)
+
+body = (Path(sys.argv[1]) / HMRC_CSV).read_bytes()
+assert hashlib.sha256(body).hexdigest() == HMRC_SHA256, "tracked HMRC CSV changed"
+meta = json.loads(COMBINED_META.read_text())
+assert meta["status"] == 200 and len(body) == int(meta["headers"]["Content-Length"])
+k = key(meta["requested"])
+(cache / f"{k}.body").write_bytes(body)
+shutil.copyfile(COMBINED_META, cache / f"{k}.json")
+
+entries = sorted(cache.glob("*.json"))
+for m in entries:
+    requested = json.loads(m.read_text())["requested"]
+    assert m.stem == key(requested), f"{m.name} is not keyed by its URL"
+    assert (cache / f"{m.stem}.body").exists(), f"{m.stem}.body missing"
+print(
+    f"{cache}: {len(entries)} responses, all keyed by URL; added {k} for {meta['requested']}"
+)
diff --git a/jobs/adopt-ud507/setup_build_worktree.sh b/jobs/adopt-ud507/setup_build_worktree.sh
new file mode 100644
index 00000000..3e901804
--- /dev/null
+++ b/jobs/adopt-ud507/setup_build_worktree.sh
@@ -0,0 +1,52 @@
+#!/bin/zsh
+# Hub only, after GO: setup_build_worktree.sh <name> <exact-ref> <branch|main>
+# Reuse the pinned hub harness, omitting its seed cherry-pick: the PR already
+# has that commit, while controls must retain main's unseeded dropout.
+set -eu
+J=${0:A:h}
+E=~/reviews/uk-data-spi-reported-2026-10-01
+W=~/PolicyEngine/_worktrees/$1
+REF=$2
+KIND=$3
+HARNESS=$E/setup_build_worktree.sh
+EXPECT=3207a17bdfc1e441cba58ba34c063e2413ba7fc2242dcacff25c338c3fcfb1de
+[ "$(shasum -a 256 "$HARNESS" | awk '{print $1}')" = "$EXPECT" ] || {
+  echo 'The source harness changed; review its delta before building.' >&2
+  exit 1
+}
+[ -e "$J/frozen/models.sha256" ] || { echo 'Freeze inputs first.' >&2; exit 1; }
+[ ! -e "$W" ] || { echo "Use a fresh worktree: $W already exists." >&2; exit 1; }
+DERIVED=$(mktemp)
+trap 'rm -f "$DERIVED"' EXIT
+python3 - "$HARNESS" "$DERIVED" <<'PY'
+import sys
+from pathlib import Path
+
+source = Path(sys.argv[1]).read_text()
+seed_line = 'git -c user.name=build -c user.email=build@localhost cherry-pick -x 348dd49 >/dev/null\n'
+assert source.count(seed_line) == 1
+source = source.replace(seed_line, '# #507 controls build exact refs; skip seed cherry-pick.\n')
+source = source.replace(' (base $(git rev-parse --short HEAD~1))', ' (exact requested ref)')
+Path(sys.argv[2]).write_text(source)
+PY
+zsh "$DERIVED" "$1" "$REF"
+cd "$W"
+[ "$(git rev-parse HEAD)" = "$(git rev-parse "$REF^{commit}")" ]
+case "$KIND" in
+  branch)
+    git merge-base --is-ancestor 348dd493c7cba66b329a384410e4bb21de97cd4d HEAD
+    rg -q 'dropout_generator = torch.Generator\(\).*manual_seed\(seed\)' policyengine_uk_data/utils/calibrate.py
+    ;;
+  main)
+    [ "$(git rev-parse HEAD)" = "$(git rev-parse 2b9c1c2c64ce)" ]
+    rg -q 'mask = torch.rand_like\(weights\) < p' policyengine_uk_data/utils/calibrate.py
+    ;;
+  *) echo 'kind must be branch or main' >&2; exit 1 ;;
+esac
+S=policyengine_uk_data/storage
+for MODEL in "$J"/frozen/models/*.pkl; do cp -c -f "$MODEL" "$S/${MODEL:t}"; done
+(cd "$S" && shasum -a 256 *.pkl) > "$W/models_setup.sha256"
+cmp "$J/frozen/models.sha256" "$W/models_setup.sha256"
+.venv/bin/python -c 'import importlib.metadata as m; assert m.version("policyengine-uk") == "2.122.2"; assert m.version("policyengine-core") == "3.32.13"'
+git diff --exit-code
+echo "$1 ready at $(git rev-parse --short HEAD) ($KIND; pristine exact ref)"
diff --git a/jobs/adopt-ud507/totals.py b/jobs/adopt-ud507/totals.py
new file mode 100644
index 00000000..ccb33a56
--- /dev/null
+++ b/jobs/adopt-ud507/totals.py
@@ -0,0 +1,66 @@
+"""Hub finisher: income tax and household net income, aggregate output only.
+
+Run only after the hub approves the queued builds; one microsimulation at a
+time, in the same locked environment used for all four builds.
+Usage: python totals.py <enhanced_frs.h5> <output.json>
+"""
+
+import hashlib
+import importlib.metadata
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+import numpy as np
+from policyengine_uk import Microsimulation
+
+
+def main():
+    dataset, output = Path(sys.argv[1]), Path(sys.argv[2])
+    for variable in (
+        "OMP_NUM_THREADS",
+        "OPENBLAS_NUM_THREADS",
+        "MKL_NUM_THREADS",
+        "VECLIB_MAXIMUM_THREADS",
+        "NUMEXPR_NUM_THREADS",
+    ):
+        if os.environ.get(variable) != "1":
+            raise SystemExit(f"Export {variable}=1 before measuring totals")
+    memory = subprocess.check_output(["vm_stat"], text=True)
+    page_size = int(re.search(r"page size of (\d+) bytes", memory).group(1))
+    pages = int(re.search(r"Pages occupied by compressor:\s+(\d+)\.", memory).group(1))
+    if pages * page_size >= 30e9:
+        raise SystemExit("Compressor is at least 30 GB; wait before microsimulation")
+    with dataset.open("rb") as source:
+        digest = hashlib.file_digest(source, "sha256").hexdigest()
+    result = {
+        "dataset_sha256": digest,
+        "policyengine_uk_version": importlib.metadata.version("policyengine-uk"),
+        "policyengine_core_version": importlib.metadata.version("policyengine-core"),
+    }
+    if result["policyengine_uk_version"] != "2.122.2":
+        raise SystemExit("Use the frozen policyengine-uk 2.122.2 build environment")
+    if result["policyengine_core_version"] != "3.32.13":
+        raise SystemExit("Use the frozen policyengine-core 3.32.13 build environment")
+    sim = Microsimulation(dataset=str(dataset))
+    for year in (2025, 2026):
+        result[str(year)] = {}
+        for variable, weight in (
+            ("income_tax", "person_weight"),
+            ("household_net_income", "household_weight"),
+        ):
+            values = np.asarray(sim.calculate(variable, year).values, dtype=np.float64)
+            weights = np.asarray(sim.calculate(weight, year).values, dtype=np.float64)
+            result[str(year)][variable] = {
+                "total_bn": float(np.sum(values * weights, dtype=np.float64) / 1e9)
+            }
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(result, indent=2) + "\n")
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/jobs/adopt-ud507/verify_historical_spread.py b/jobs/adopt-ud507/verify_historical_spread.py
new file mode 100644
index 00000000..0627b7d3
--- /dev/null
+++ b/jobs/adopt-ud507/verify_historical_spread.py
@@ -0,0 +1,43 @@
+"""Verify the PR's rounded £0.8bn/£3bn claim from saved aggregates only."""
+
+import json
+from pathlib import Path
+
+
+def main():
+    evidence = Path.home() / "reviews/frs-cvpay-property-income-2026-09-30/impact"
+    labels = ("main_matched", "main_matched_2", "main_a", "main_b")
+    data = {
+        label: json.loads((evidence / f"efrs_{label}.json").read_text())
+        for label in labels
+    }
+    heads = {entry["policyengine_uk_head"] for entry in data.values()}
+    assert heads == {"44240bd8d7d7e97f2980ccc8134e76e61541966f"}
+    report = {
+        "sources": str(evidence),
+        "policyengine_uk_head": heads.pop(),
+        "years": {},
+    }
+    for year in ("2025", "2026"):
+        report["years"][year] = {}
+        for metric in ("income_tax_bn", "household_net_income_bn"):
+            values = {label: entry[year][metric] for label, entry in data.items()}
+            low, high = min(values, key=values.get), max(values, key=values.get)
+            report["years"][year][metric] = {
+                "values_bn": values,
+                "minimum_build": low,
+                "maximum_build": high,
+                "range_bn": values[high] - values[low],
+            }
+    assert (
+        abs(report["years"]["2026"]["income_tax_bn"]["range_bn"] - 0.784596992) < 1e-10
+    )
+    assert (
+        abs(report["years"]["2026"]["household_net_income_bn"]["range_bn"] - 2.95108608)
+        < 1e-10
+    )
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == "__main__":
+    main()