diff --git a/.github/workflows/pull_request.yaml b/.github/workflows/pull_request.yaml
index 9443fa258..40ed5754b 100644
--- a/.github/workflows/pull_request.yaml
+++ b/.github/workflows/pull_request.yaml
@@ -72,6 +72,10 @@ jobs:
           fetch-depth: 0 # Fetch all history for all tags and branches
       - name: Install uv
         uses: astral-sh/setup-uv@v8.1.0
+        # Restore uv's package cache but never save one from a job that
+        # holds the data: its location can be moved (UV_CACHE_DIR).
+        with:
+          save-cache: false
       - name: Set up Python
         uses: actions/setup-python@v6
         with:
@@ -86,21 +90,314 @@ jobs:
         run: uv run --frozen make data
         env:
           TESTING: "1"
-      - name: Save calibration log (constituencies)
-        uses: actions/upload-artifact@v7
-        with:
-          name: constituency_calibration_log.csv
-          path: constituency_calibration_log.csv
-      
-      - name: Save calibration log (local authorities)
-        uses: actions/upload-artifact@v7
-        with:
-          name: la_calibration_log.csv
-          path: la_calibration_log.csv
-
       - name: Run tests
         run: uv run --frozen make test
         env:
           # Relaxes built-dataset outcome tolerances to match the reduced
           # fidelity of the TESTING build above (see test_cgt_band_donors).
           TESTING: "1"
+      # After the tests, whatever they did. The raw calibration logs stay on
+      # the runner: a target supported by a few households, divided by a
+      # published count, could give one household's value back. Only the
+      # summary the fit comparison needs is uploaded (utils/fit_report.py).
+      - name: Summarise the calibration fit
+        if: ${{ !cancelled() }}
+        run: uv run --frozen python -m policyengine_uk_data.utils.fit_report summarise constituency_calibration_log.csv --out fit_summary.json
+      - name: Save the fit summary
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@v7
+        with:
+          name: fit_summary.json
+          path: fit_summary.json
+      # After the tests, and whatever they did: the measurement downloads the
+      # ONS projection, and an ONS outage must not skip the test suite.
+      - name: Measure the income distribution and single ages
+        if: ${{ !cancelled() }}
+        # A failed measurement fails the comparison (missing report), not
+        # the test job, which stays about the tests.
+        continue-on-error: true
+        run: uv run --frozen python -m policyengine_uk_data.utils.distribution_validation measure policyengine_uk_data/storage/enhanced_frs_2024_25.h5 --out distribution_report.json
+      - name: Save the distribution report
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@v7
+        with:
+          name: distribution_report.json
+          path: distribution_report.json
+      - name: Check out the aggregate-impact tool
+        if: ${{ !cancelled() }}
+        # A missing result fails the comparison, not the test job.
+        continue-on-error: true
+        uses: actions/checkout@v6
+        with:
+          repository: PolicyEngine/policyengine-uk
+          ref: 6406b2ceae24f174eca41bfa1dbb1026305d2a46
+          path: pe-uk-tool
+          sparse-checkout: |
+            policyengine_uk/utils/aggregate_impact
+            policyengine_uk/data/fiscal_partition.yaml
+      - name: Measure national aggregates with the model fixed
+        if: ${{ !cancelled() }}
+        # A missing result fails the comparison, not the test job.
+        continue-on-error: true
+        run: |
+          tool="$RUNNER_TEMP/tool"
+          mkdir -p "$tool"
+          cp -R pe-uk-tool/policyengine_uk/utils/aggregate_impact "$tool/pe_uk_aggregate_impact"
+          model="policyengine-uk==$(uv run --frozen python -c 'import importlib.metadata as m; print(m.version("policyengine-uk"))')"
+          PYTHONPATH="$tool" uv run --frozen python -m pe_uk_aggregate_impact measure \
+            --label head --dataset-file policyengine_uk_data/storage/enhanced_frs_2024_25.h5 --dataset-label "this pull request's build" \
+            --fallback-partition pe-uk-tool/policyengine_uk/data/fiscal_partition.yaml \
+            --no-guard --git-sha "$model" --out "$RUNNER_TEMP/aggregate" || true
+          cp "$RUNNER_TEMP/aggregate/result.json" aggregate_result.json \
+            || echo "::warning::The aggregate measurement wrote no result; see the log above."
+      - name: Save the aggregate result (aggregates only)
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@v7
+        with:
+          name: aggregate_result.json
+          path: aggregate_result.json
+  # The policyengine-uk aggregate-impact tool (policyengine_uk/utils/
+  # aggregate_impact) at a pinned commit measures each build with the locked
+  # model, and the comparison job compares them with the model fixed
+  # (`compare --axis data`). Only result.json leaves a job; its private/
+  # folder (household detail) stays on the runner.
+  # The no-worsening check (W3). The base is built the same way as the head:
+  # TESTING, one clone and the head's lockfile, so both builds run one model
+  # and environment and the comparison measures the data change. As in the
+  # test job, only aggregate reports leave it: the fit summary, the
+  # distribution report and the aggregate result. No dataset or calibration
+  # log is uploaded, and nothing is cached: setup-uv restores uv's package
+  # cache and saves none.
+  base-build:
+    name: Build the base for comparison
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    env:
+      HUGGING_FACE_TOKEN: ${{ secrets.HUGGING_FACE_TOKEN }}
+      PYTHONUNBUFFERED: "1"
+      PE_UK_DATA_OA_CLONES: "1"
+    steps:
+      - name: Check out the head, for its lockfile
+        uses: actions/checkout@v6
+        with:
+          path: head
+      - name: Check out the base
+        uses: actions/checkout@v6
+        with:
+          ref: ${{ github.event.pull_request.base.sha }}
+          path: base
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.1.0
+        # Restore uv's package cache but never save one from a job that
+        # holds the data: its location can be moved (UV_CACHE_DIR).
+        with:
+          save-cache: false
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: 3.13
+      - name: Sync the head's locked environment for the base
+        run: |
+          cp head/uv.lock base/uv.lock
+          cd base && uv sync --frozen --all-extras
+      - name: Download data inputs
+        working-directory: base
+        run: uv run --frozen make download
+      - name: Build the base datasets
+        working-directory: base
+        run: uv run --frozen make data
+        env:
+          TESTING: "1"
+      - name: Summarise the base's calibration fit
+        run: |
+          cp head/policyengine_uk_data/utils/fit_report.py base/policyengine_uk_data/utils/fit_report.py
+          uv run --frozen --project base python -m policyengine_uk_data.utils.fit_report summarise base/constituency_calibration_log.csv --out base_fit_summary.json
+      - name: Save the base fit summary
+        uses: actions/upload-artifact@v7
+        with:
+          name: base_fit_summary.json
+          path: base_fit_summary.json
+      - name: Measure the base's income distribution and single ages
+        working-directory: base
+        # The head's measurement code, on the base's build.
+        run: |
+          mkdir -p policyengine_uk_data/targets/reference_data
+          cp ../head/policyengine_uk_data/utils/distribution_validation.py policyengine_uk_data/utils/
+          cp ../head/policyengine_uk_data/targets/reference_data/hbai_fye_2025.yaml policyengine_uk_data/targets/reference_data/
+          uv run --frozen python -m policyengine_uk_data.utils.distribution_validation measure policyengine_uk_data/storage/enhanced_frs_2024_25.h5 --out ../base_distribution_report.json
+      - name: Save the base distribution report
+        uses: actions/upload-artifact@v7
+        with:
+          name: base_distribution_report.json
+          path: base_distribution_report.json
+      - name: Check out the aggregate-impact tool
+        if: ${{ !cancelled() }}
+        uses: actions/checkout@v6
+        with:
+          repository: PolicyEngine/policyengine-uk
+          ref: 6406b2ceae24f174eca41bfa1dbb1026305d2a46
+          path: pe-uk-tool
+          sparse-checkout: |
+            policyengine_uk/utils/aggregate_impact
+            policyengine_uk/data/fiscal_partition.yaml
+      - name: Measure national aggregates with the model fixed
+        if: ${{ !cancelled() }}
+        run: |
+          tool="$RUNNER_TEMP/tool"
+          mkdir -p "$tool"
+          cp -R pe-uk-tool/policyengine_uk/utils/aggregate_impact "$tool/pe_uk_aggregate_impact"
+          model="policyengine-uk==$(uv run --frozen --project base python -c 'import importlib.metadata as m; print(m.version("policyengine-uk"))')"
+          PYTHONPATH="$tool" uv run --frozen --project base python -m pe_uk_aggregate_impact measure \
+            --label base --dataset-file base/policyengine_uk_data/storage/enhanced_frs_2024_25.h5 --dataset-label "the base's build" \
+            --fallback-partition pe-uk-tool/policyengine_uk/data/fiscal_partition.yaml \
+            --no-guard --git-sha "$model" --out "$RUNNER_TEMP/aggregate" || true
+          cp "$RUNNER_TEMP/aggregate/result.json" base_aggregate_result.json \
+            || echo "::warning::The aggregate measurement wrote no result; see the log above."
+      - name: Save the aggregate result (aggregates only)
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@v7
+        with:
+          name: base_aggregate_result.json
+          path: base_aggregate_result.json
+  fit-comparison:
+    name: Calibration fit against the base
+    needs: [test, base-build]
+    # Report a failed side too: an earlier passing comment must not stand.
+    if: ${{ !cancelled() }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    permissions:
+      contents: read
+      issues: write
+      pull-requests: write
+    steps:
+      - uses: actions/checkout@v6
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.1.0
+        # Restore uv's package cache but never save one from a job that
+        # holds the data: its location can be moved (UV_CACHE_DIR).
+        with:
+          save-cache: false
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: 3.13
+      - name: Sync locked environment
+        run: uv sync --frozen
+      - name: Download the fit summaries
+        uses: actions/download-artifact@v8
+        continue-on-error: true
+        with:
+          pattern: "*fit_summary.json"
+          path: logs
+          merge-multiple: true
+      - name: Download the distribution reports
+        uses: actions/download-artifact@v8
+        continue-on-error: true
+        with:
+          pattern: "*distribution_report.json"
+          path: reports
+          merge-multiple: true
+      - name: Download the aggregate results
+        uses: actions/download-artifact@v8
+        continue-on-error: true
+        with:
+          pattern: "*aggregate_result.json"
+          path: aggregates
+          merge-multiple: true
+      - name: Check out the aggregate-impact tool
+        uses: actions/checkout@v6
+        with:
+          repository: PolicyEngine/policyengine-uk
+          ref: 6406b2ceae24f174eca41bfa1dbb1026305d2a46
+          path: pe-uk-tool
+          sparse-checkout: |
+            policyengine_uk/utils/aggregate_impact
+      - name: Compare the final fits
+        id: compare
+        env:
+          RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+        run: |
+          marker='<!-- policyengine-uk-data-fit-comparison -->'
+          if [ ! -f logs/fit_summary.json ] || [ ! -f logs/base_fit_summary.json ]; then
+            {
+              echo "$marker"
+              echo '## Calibration fit against the base build: FAIL'
+              echo
+              echo "One of the builds produced no fit summary, so the fit could not be compared. See the [workflow run]($RUN_URL)."
+            } > fit.md
+            echo "status=1" >> "$GITHUB_OUTPUT"
+          else
+            set +e
+            uv run --frozen python -m policyengine_uk_data.utils.fit_report compare \
+              logs/base_fit_summary.json logs/fit_summary.json --markdown body.md
+            echo "status=$?" >> "$GITHUB_OUTPUT"
+            set -e
+            { echo "$marker"; cat body.md 2>/dev/null || echo '## Calibration fit against the base build: FAIL'; echo; echo "[Workflow run]($RUN_URL)"; } > fit.md
+          fi
+          if [ -f reports/base_distribution_report.json ] && [ -f reports/distribution_report.json ]; then
+            set +e
+            uv run --frozen python -m policyengine_uk_data.utils.distribution_validation compare \
+              reports/base_distribution_report.json reports/distribution_report.json --markdown distribution.md
+            distribution_status=$?
+            set -e
+            { echo; cat distribution.md 2>/dev/null || echo '## Income distribution and single ages: FAIL (the comparison did not finish)'; } >> fit.md
+            if [ "$distribution_status" != 0 ]; then echo "status=1" >> "$GITHUB_OUTPUT"; fi
+          else
+            { echo; echo '## Income distribution and single ages: FAIL'; echo; echo "The report is missing for one build, so the distributions could not be compared."; } >> fit.md
+            echo "status=1" >> "$GITHUB_OUTPUT"
+          fi
+          # National aggregates with the model fixed.
+          tool="$RUNNER_TEMP/tool"
+          mkdir -p "$tool" aggregates/base aggregates/head
+          cp -R pe-uk-tool/policyengine_uk/utils/aggregate_impact "$tool/pe_uk_aggregate_impact"
+          if [ -f aggregates/base_aggregate_result.json ] && [ -f aggregates/aggregate_result.json ]; then
+            mv aggregates/base_aggregate_result.json aggregates/base/result.json
+            mv aggregates/aggregate_result.json aggregates/head/result.json
+            set +e
+            PYTHONPATH="$tool" uv run --frozen python -m pe_uk_aggregate_impact compare --axis data \
+              --base aggregates/base --head aggregates/head --markdown aggregates.md --run-url "$RUN_URL"
+            aggregate_status=$?
+            set -e
+            if [ "$aggregate_status" != 0 ]; then echo "status=1" >> "$GITHUB_OUTPUT"; fi
+            # The tool's comment marker belongs to policyengine-uk's own comment.
+            if [ -s aggregates.md ]; then
+              { echo; sed '/^<!-- policyengine-uk-aggregate-impact -->$/d' aggregates.md; } >> fit.md
+            else
+              { echo; echo '## National aggregates: FAIL (the comparison did not finish)'; } >> fit.md
+              echo "status=1" >> "$GITHUB_OUTPUT"
+            fi
+          else
+            { echo; echo '## National aggregates: FAIL'; echo; echo "The result is missing for one build (see the build jobs' logs), so the aggregates could not be compared."; } >> fit.md
+            echo "status=1" >> "$GITHUB_OUTPUT"
+          fi
+          cat fit.md >> "$GITHUB_STEP_SUMMARY"
+      - name: Post or update the pull request comment
+        if: ${{ !cancelled() && github.event.pull_request.head.repo.full_name == github.repository }}
+        env:
+          GH_TOKEN: ${{ github.token }}
+          PR: ${{ github.event.pull_request.number }}
+        run: |
+          marker='<!-- policyengine-uk-data-fit-comparison -->'
+          if [ ! -s fit.md ]; then
+            # The job failed before the comparison could write a comment.
+            {
+              echo "$marker"
+              echo '## Calibration fit against the base build: FAIL'
+              echo
+              echo "The comparison did not run. See the [workflow run]($GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID)."
+            } > fit.md
+          fi
+          existing=$(gh api --paginate "repos/$GITHUB_REPOSITORY/issues/$PR/comments" \
+            --jq ".[] | select(.user.login == \"github-actions[bot]\") | select(.body | startswith(\"$marker\")) | .id" \
+            | tail -n 1)
+          if [ -n "$existing" ]; then
+            gh api -X PATCH "repos/$GITHUB_REPOSITORY/issues/comments/$existing" -F body=@fit.md > /dev/null
+          else
+            gh api -X POST "repos/$GITHUB_REPOSITORY/issues/$PR/comments" -F body=@fit.md > /dev/null
+          fi
+      - name: Fail on a fit regression
+        if: steps.compare.outputs.status != '0'
+        run: |
+          echo "::error::The calibration fit, distribution or national aggregate comparison failed; see the job summary."
+          exit 1
diff --git a/changelog.d/aggregate-data-comparison.added.md b/changelog.d/aggregate-data-comparison.added.md
new file mode 100644
index 000000000..e15b5d1f2
--- /dev/null
+++ b/changelog.d/aggregate-data-comparison.added.md
@@ -0,0 +1 @@
+Compare each pull request build's national aggregates (receipts, spending, caseloads, poverty, deciles and the standard reform costings) with its base build's, measured with the locked model by policyengine-uk's aggregate-impact tool, and report them in the pull request comment.
diff --git a/changelog.d/distribution-checks.added.md b/changelog.d/distribution-checks.added.md
new file mode 100644
index 000000000..5077bf911
--- /dev/null
+++ b/changelog.d/distribution-checks.added.md
@@ -0,0 +1 @@
+Measure each pull request build's HBAI income distribution (medians, means, Gini and relative low income by group against DWP's HBAI FYE 2025 estimates and intervals) and single-age population (against the ONS 2024-based projection), and report them against the base build in the pull request comment. They flag but do not fail yet.
diff --git a/changelog.d/fit-comparison.added.md b/changelog.d/fit-comparison.added.md
new file mode 100644
index 000000000..89d7d993e
--- /dev/null
+++ b/changelog.d/fit-comparison.added.md
@@ -0,0 +1 @@
+Build each pull request's base the same way as its head, with the head's lockfile, and fail when the share of national or local targets within 10% falls by more than 2 percentage points over the targets both builds have (`targets/fit_policy.yaml`). National targets in a worse error band on the head are flagged in a pull request comment. The raw calibration logs stay on the runner; only each target's error band (within 10%, 10-25%, 25-50%, 50-100%, over 100%) is uploaded, with no estimate or error (`python -m policyengine_uk_data.utils.fit_report summarise` and `compare`).
diff --git a/policyengine_uk_data/targets/fit_policy.yaml b/policyengine_uk_data/targets/fit_policy.yaml
new file mode 100644
index 000000000..5b71730bc
--- /dev/null
+++ b/policyengine_uk_data/targets/fit_policy.yaml
@@ -0,0 +1,47 @@
+# When a pull request's build fails or flags against its base build's fit
+# (utils/fit_report.py). Both builds run the same way in pull request CI:
+# TESTING, 32 epochs, one output-area clone, the head's lockfile.
+#
+# The share allowance was adopted under cos d1344 (15 October 2026), from the final-epoch fits of
+# five builds of the same data code: docs, changelog and CI pull requests
+# while main's data code was unchanged from b34a4ea2 to 2b9c1c2c (runs
+# 38006434333, 37959581668, 38009046475, 37960852139, 37864653693).
+schema_version: 1
+
+share_within_10pct:
+  # Fail when the share of national, or of local, targets within 10% of
+  # their values falls by more than this. Across the 20 ordered pairs of the
+  # five builds the national share moved by at most 0.93pp and the local by
+  # at most 0.69pp, so 2pp is about twice the same-code spread.
+  # Alternatives: 1pp leaves no margin over that spread while #598 (builds
+  # of the same code fall into two weight sets) is unexplained. 5pp would
+  # pass the 2.4pp local drop of #551's build against the build of the
+  # 1.58.1 release branch (runs 38661397028 and 38228500493).
+  allowed_drop: 0.02
+
+national_target:
+  # Flag, for review, each national target whose error band is worse on the
+  # head than on the base. Bands of |estimate / target - 1|: within 10%,
+  # 10-25%, 25-50%, 50-100%, over 100% (fit_report.BANDS). Across the 20
+  # ordered pairs of the five same-code builds above, 2 to 12 of the 644
+  # national targets move to a worse band per pair (mean 7), so a list stays
+  # short. #551's build moves 31 and #579's 27 against the 1.58.1 release
+  # branch's build.
+  #
+  # This replaces flagging an error rise of more than 0.2 (adopted under
+  # d1344 on 15 October), which needed each target's error in the uploaded
+  # summary. Sol's re-review of #599 found no safe way to publish them: the
+  # dataset repeats households (output-area clones, SPI and CGT donors), so
+  # a count of rows can't show how many respondents support an estimate, and
+  # one household can dominate a target. A band is a bound, not a value.
+  # The old rule gave 0 to 6 flags per same-code pair. Bands still can't see
+  # a move inside one band (say 30% to 45%). The replacement is a W3
+  # threshold call under d1344's delegation (thresholds on evidence, with an
+  # independent review agreeing), pending that review. Per-target moves
+  # don't fail until #598 is fixed.
+  #
+  # Run ids of the five builds and their saved log folders: 38006434333
+  # (docs-ebdbbce5), 37959581668 (docs-2a071400), 38009046475
+  # (changelog-6af935fb), 37960852139 (changelog-bb41c3ca), 37864653693
+  # (ci-push-gate-03b3ecd9); the 1.58.1 release branch is 38228500493.
+  flag: worse_band
diff --git a/policyengine_uk_data/targets/reference_data/hbai_fye_2025.yaml b/policyengine_uk_data/targets/reference_data/hbai_fye_2025.yaml
new file mode 100644
index 000000000..c209ab767
--- /dev/null
+++ b/policyengine_uk_data/targets/reference_data/hbai_fye_2025.yaml
@@ -0,0 +1,27 @@
+# Households below average income, FYE 2025 (DWP, 26 March 2026), from the
+# published data tables:
+# https://assets.publishing.service.gov.uk/media/69c2902a7e02b81c0d1c7762/hbai-data_tables-2025.zip
+# uncertainty-hbai-2024-25-tables.ods, tables 8a (medians, means, Gini) and
+# 8b (low income), read on 15 October 2026. Estimates and 95% confidence
+# intervals for the United Kingdom, FRS 2024/25, which policyengine-uk models
+# as period 2024. Incomes are equivalised, in £ a week at 2024/25 prices;
+# rates and Gini coefficients are per cent.
+schema_version: 1
+source: DWP Households below average income FYE 2025
+period: 2024
+measures:
+  median_bhc: {estimate: 719.481062, low: 703.567322, high: 735.575892, table: 8a}
+  median_ahc: {estimate: 623.144059, low: 605.209749, high: 637.071794, table: 8a}
+  mean_bhc: {estimate: 854.979494, low: 836.367495, high: 870.351587, table: 8a}
+  mean_ahc: {estimate: 749.82155, low: 732.329379, high: 764.496483, table: 8a}
+  gini_bhc: {estimate: 32.736632, low: 31.889881, high: 33.978308, table: 8a}
+  gini_ahc: {estimate: 36.780956, low: 35.897706, high: 38.293054, table: 8a}
+  # Percentage of individuals below 60% of the relative median.
+  relative_low_income_bhc_all: {estimate: 15.911572, low: 14.446526, high: 17.637657, table: 8b}
+  relative_low_income_bhc_children: {estimate: 20.519181, low: 17.356404, high: 23.955585, table: 8b}
+  relative_low_income_bhc_working_age: {estimate: 14.281508, low: 12.908347, high: 16.018814, table: 8b}
+  relative_low_income_bhc_pensioners: {estimate: 15.895071, low: 13.690455, high: 17.991754, table: 8b}
+  relative_low_income_ahc_all: {estimate: 19.640844, low: 17.9939, high: 21.271606, table: 8b}
+  relative_low_income_ahc_children: {estimate: 27.41076, low: 23.49369, high: 30.644085, table: 8b}
+  relative_low_income_ahc_working_age: {estimate: 18.579734, low: 17.012601, high: 20.343162, table: 8b}
+  relative_low_income_ahc_pensioners: {estimate: 13.866371, low: 11.618244, high: 15.898207, table: 8b}
diff --git a/policyengine_uk_data/tests/test_distribution_validation.py b/policyengine_uk_data/tests/test_distribution_validation.py
new file mode 100644
index 000000000..7738583d3
--- /dev/null
+++ b/policyengine_uk_data/tests/test_distribution_validation.py
@@ -0,0 +1,245 @@
+"""Single-age and HBAI distribution checks (utils/distribution_validation.py)."""
+
+import numpy as np
+import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from policyengine_uk_data.utils.distribution_validation import (
+    SCHEMA,
+    TOP_AGE_BAND,
+    compare,
+    hbai_measures,
+    load_hbai_reference,
+    render,
+    single_ages,
+    summarise_single_ages,
+    weighted_gini,
+    weighted_median,
+)
+
+samples = st.lists(
+    st.tuples(st.floats(-1e4, 1e5, allow_nan=False), st.integers(1, 5)),
+    min_size=1,
+    max_size=40,
+)
+
+
+def _expand(pairs):
+    return np.repeat([v for v, _ in pairs], [w for _, w in pairs]).astype(float)
+
+
+@settings(max_examples=300, deadline=None)
+@given(pairs=samples)
+def test_weighted_median_equals_the_median_of_the_expanded_sample(pairs):
+    """Differential: integer weights are repeated records."""
+    values = np.array([v for v, _ in pairs])
+    weights = np.array([w for _, w in pairs], float)
+    expanded = np.sort(_expand(pairs))
+    # The lower median of the expanded sample.
+    lower = expanded[(len(expanded) - 1) // 2]
+    upper = expanded[len(expanded) // 2]
+    assert lower <= weighted_median(values, weights) <= upper
+    # Scaling every weight leaves it unchanged.
+    assert weighted_median(values, weights * 7.5) == weighted_median(values, weights)
+
+
+@settings(max_examples=300, deadline=None)
+@given(
+    pairs=st.lists(
+        st.tuples(st.floats(0.01, 1e5), st.integers(1, 5)), min_size=2, max_size=40
+    )
+)
+def test_weighted_gini_equals_the_mean_difference_gini_of_the_expanded_sample(pairs):
+    expanded = _expand(pairs)
+    n = len(expanded)
+    mean_difference = np.abs(expanded[:, None] - expanded[None, :]).sum() / (n * n)
+    reference = mean_difference / (2 * expanded.mean())
+    values = np.array([v for v, _ in pairs])
+    weights = np.array([w for _, w in pairs], float)
+    assert weighted_gini(values, weights) == pytest.approx(reference, abs=1e-9)
+
+
+def test_gini_of_equal_incomes_is_zero():
+    assert weighted_gini([5.0, 5.0, 5.0], [1, 2, 3]) == pytest.approx(0)
+
+
+def test_hbai_measures_on_a_known_population():
+    income = np.array([100.0, 200.0, 300.0, 400.0, 1000.0])
+    weights = np.ones(5)
+    child = np.array([1, 0, 0, 0, 0], bool)
+    pensioner = np.array([0, 0, 0, 0, 1], bool)
+    m = hbai_measures(income, income - 50, weights, child, pensioner)
+    assert m["median_bhc"] == 300.0
+    assert m["mean_bhc"] == 400.0
+    # Below 60% of 300: 100 only, a child.
+    assert m["relative_low_income_bhc_all"] == pytest.approx(20.0)
+    assert m["relative_low_income_bhc_children"] == pytest.approx(100.0)
+    assert m["relative_low_income_bhc_working_age"] == pytest.approx(0.0)
+    assert m["relative_low_income_bhc_pensioners"] == pytest.approx(0.0)
+
+
+def test_the_hbai_reference_is_the_published_table():
+    """Spot values from DWP HBAI FYE 2025, tables 8a and 8b."""
+    ref = load_hbai_reference()
+    assert ref["period"] == 2024
+    m = ref["measures"]
+    assert m["median_bhc"]["estimate"] == pytest.approx(719.481062)
+    assert m["median_ahc"]["estimate"] == pytest.approx(623.144059)
+    assert m["relative_low_income_ahc_children"]["estimate"] == pytest.approx(27.41076)
+    assert len(m) == 14
+
+
+def test_single_ages_band_the_oldest_and_split_by_sex():
+    age = np.array([0, 0, 79, 80, 85, 90])
+    male = np.array([1, 0, 1, 1, 0, 1], bool)
+    weights = np.array([2.0, 3.0, 1.0, 1.0, 4.0, 5.0])
+    reference = {
+        (s, a): 1.0 for s in ("male", "female") for a in range(TOP_AGE_BAND + 1)
+    }
+    rows = single_ages(age, male, weights, reference)
+    assert rows["male"]["0"]["value"] == 2.0 and rows["female"]["0"]["value"] == 3.0
+    assert rows["male"]["80"]["value"] == 6.0 and rows["female"]["80"]["value"] == 4.0
+    summary = summarise_single_ages(rows)
+    assert summary["error_80_plus"] == {"male": 5.0, "female": 3.0}
+
+
+def _report(**hbai_values):
+    ref = load_hbai_reference()["measures"]
+    hbai = {}
+    for name, r in ref.items():
+        value = hbai_values.get(name, r["estimate"])
+        hbai[name] = {
+            "value": value,
+            "reference": r["estimate"],
+            "low": r["low"],
+            "high": r["high"],
+            "inside": r["low"] <= value <= r["high"],
+        }
+    single = {
+        "mean_abs_error_0_79": 0.2,
+        "max_abs_error_0_79": 0.9,
+        "max_abs_error_15_19": 0.6,
+        "error_80_plus": {"male": 0.0, "female": 0.0},
+    }
+    return {
+        "schema": SCHEMA,
+        "periods": {"hbai": 2024, "single_age": 2025},
+        "hbai": hbai,
+        "summary": {"single_age": single},
+    }
+
+
+def test_compare_flags_a_measure_that_leaves_its_interval_but_does_not_fail():
+    result = compare(_report(), _report(median_bhc=760.0))
+    assert result["status"] == "pass"
+    assert result["flags"] == [
+        "HBAI median_bhc leaves its 95% interval: 719.48 -> 760.00 (HBAI 719.48, 703.57 to 735.58)."
+    ]
+    text = render(result)
+    assert "760.00 (outside)" in text and "do not fail yet" in text
+
+
+def test_compare_with_itself_passes_with_no_flags():
+    result = compare(_report(), _report())
+    assert result["status"] == "pass" and result["flags"] == []
+    assert all(r["further_from_reference"] == 0 for r in result["rows"])
+
+
+@pytest.mark.parametrize(
+    "change, message",
+    [
+        (lambda r: r.update(schema="other"), "malformed"),
+        (lambda r: r["periods"].update(hbai=2025), "different periods"),
+        (lambda r: r["hbai"]["median_bhc"].update(value=float("nan")), "not finite"),
+    ],
+)
+def test_compare_fails_on_a_broken_or_mismatched_report(change, message):
+    head = _report()
+    change(head)
+    result = compare(_report(), head)
+    assert result["status"] == "fail"
+    assert any(message in f for f in result["failures"])
+
+
+def test_negative_incomes_before_housing_costs_are_reset_to_zero_as_hbai_does():
+    """Sol review of #600: HBAI resets negative BHC to zero and derives AHC from
+    the adjusted figure, so AHC can still be negative."""
+    from policyengine_uk_data.utils.distribution_validation import (
+        WEEKS_PER_YEAR,
+        hbai_incomes,
+    )
+
+    weekly_bhc = np.array([-100.0, 100.0, 200.0, 300.0, 400.0])
+    annual_bhc = weekly_bhc * WEEKS_PER_YEAR
+    annual_ahc = (weekly_bhc - 50) * WEEKS_PER_YEAR
+    bhc, ahc = hbai_incomes(annual_bhc, annual_ahc, np.ones(5), np.ones(5))
+    np.testing.assert_allclose(bhc, [0, 100, 200, 300, 400])
+    np.testing.assert_allclose(ahc, [-50, 50, 150, 250, 350])
+    m = hbai_measures(bhc, ahc, np.ones(5), np.zeros(5, bool), np.zeros(5, bool))
+    assert m["mean_bhc"] == pytest.approx(200) and m["mean_ahc"] == pytest.approx(150)
+    assert m["gini_bhc"] == pytest.approx(40.0)
+    assert m["gini_ahc"] == pytest.approx(53.333333, rel=1e-6)
+    # No children or pensioners: those groups have no rate, not a NaN.
+    assert m["relative_low_income_bhc_children"] is None
+    assert m["relative_low_income_ahc_pensioners"] is None
+    # Median 200, line 120: incomes 0 and 100 are below it.
+    assert m["relative_low_income_bhc_working_age"] == pytest.approx(40.0)
+
+
+def test_an_incomplete_or_non_finite_report_fails():
+    """Sol review of #600: every reference measure and age summary must be there."""
+    head = _report()
+    del head["hbai"]["median_bhc"]
+    result = compare(_report(), head)
+    assert result["status"] == "fail"
+    assert result["failures"] == ["The head report has no median_bhc."]
+    head = _report()
+    head["summary"]["single_age"]["mean_abs_error_0_79"] = float("nan")
+    result = compare(_report(), head)
+    assert result["status"] == "fail" and "not finite" in result["failures"][0]
+    head = _report()
+    name = next(iter(head["hbai"]))
+    head["hbai"][name]["value"] = None  # an empty group
+    result = compare(_report(), head)
+    assert result["status"] == "fail"
+    assert f"HBAI {name} is missing or not finite on one side." in result["failures"]
+
+
+def test_the_workflow_fails_when_a_distribution_report_is_missing():
+    from pathlib import Path
+
+    text = (
+        Path(__file__).parents[2] / ".github" / "workflows" / "pull_request.yaml"
+    ).read_text()
+    block = text[text.index("The report is missing for one build") :]
+    assert 'echo "status=1" >> "$GITHUB_OUTPUT"' in block.split("\n          fi", 1)[0]
+    assert "## Income distribution and single ages: FAIL" in text
+
+
+def test_the_measurement_runs_after_the_tests_and_cannot_skip_them():
+    """Delta reviews of #600: an ONS outage must not skip the tests or the
+    fit summary, nor turn the test job red."""
+    from pathlib import Path
+
+    import yaml
+
+    workflow = Path(__file__).parents[2] / ".github" / "workflows" / "pull_request.yaml"
+    steps = yaml.safe_load(workflow.read_text())["jobs"]["test"]["steps"]
+    names = [step.get("name") for step in steps]
+    measure = names.index("Measure the income distribution and single ages")
+    assert names.index("Run tests") < measure
+    assert names.index("Save the fit summary") < measure
+    assert steps[measure]["if"] == "${{ !cancelled() }}"
+    assert steps[measure]["continue-on-error"] is True
+    # The fit summary runs after the tests too, whatever they did.
+    summarise = names.index("Summarise the calibration fit")
+    assert names.index("Run tests") < summarise
+    assert steps[summarise]["if"] == "${{ !cancelled() }}"
+    assert steps[names.index("Save the fit summary")]["if"] == "${{ !cancelled() }}"
+    # On the base, the distribution measurement runs after the fit summary.
+    base = yaml.safe_load(workflow.read_text())["jobs"]["base-build"]["steps"]
+    base_names = [step.get("name") for step in base]
+    assert base_names.index("Save the base fit summary") < base_names.index(
+        "Measure the base's income distribution and single ages"
+    )
diff --git a/policyengine_uk_data/tests/test_fit_report.py b/policyengine_uk_data/tests/test_fit_report.py
new file mode 100644
index 000000000..dc14487fe
--- /dev/null
+++ b/policyengine_uk_data/tests/test_fit_report.py
@@ -0,0 +1,324 @@
+"""The final calibration fit summary and the no-worsening comparison (utils/fit_report.py)."""
+
+import json
+
+import numpy as np
+import pandas as pd
+import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from policyengine_uk_data.utils.fit_report import (
+    BANDS,
+    FitPolicy,
+    band,
+    compare,
+    load_policy,
+    main,
+    summarise,
+)
+
+POLICY = FitPolicy(share_drop_allowance=0.02)
+
+
+def _log(national: dict, local=(), epochs=(0, 30)):
+    """A calibration log: national {metric: (estimate, target)}, local [(est, target)]."""
+    rows = []
+    for epoch in epochs:
+        for metric, (estimate, target) in national.items():
+            rows.append(("UK", 0, metric, estimate, target, epoch))
+        for i, (estimate, target) in enumerate(local):
+            rows.append((f"area {i}", f"E{i}", f"local/{i}", estimate, target, epoch))
+    return pd.DataFrame(
+        rows, columns=["name", "index", "metric", "estimate", "target", "epoch"]
+    )
+
+
+LOCAL = [(100, 100)] * 10
+
+
+def test_summarise_reads_the_last_epoch_and_drops_zero_targets():
+    log = _log(
+        {"a": (105, 100), "b": (150, 100), "zero": (5, 0)}, local=[(9, 10), (20, 10)]
+    )
+    log.loc[log.epoch == 0, "estimate"] = 1  # earlier epochs are ignored
+    s = summarise(log)
+    assert s["epoch"] == 30
+    # Bands of |estimate / target - 1|: 0.05 is within 10%, 0.5 is 25-50%.
+    assert s["national"] == {"a": 0, "b": 2}
+    assert "zero" not in s["national"]
+    assert s["local"] == {"E0|local/0": 0, "E1|local/1": 3}
+
+
+@settings(max_examples=100, deadline=None)
+@given(
+    estimates=st.lists(st.floats(0.01, 1e12, allow_nan=False), min_size=1, max_size=20),
+    target=st.floats(1.0, 1e12),
+)
+def test_the_summary_publishes_no_estimate_or_error(estimates, target):
+    """Sol re-reviews of #599: rows don't count respondents and one household
+    can dominate a target, so nothing per target but within 10% is published."""
+    log = _log(
+        {f"t{i}": (e, target) for i, e in enumerate(estimates)},
+        local=[(e, target) for e in estimates],
+    )
+    s = summarise(log)
+    for group in ("national", "local"):
+        assert all(type(v) is int and 0 <= v < len(BANDS) for v in s[group].values())
+    text = json.dumps(s)
+    for e in estimates:
+        assert repr(e) not in text and repr(abs(e / target - 1)) not in text
+
+
+def test_a_malformed_log_is_refused():
+    with pytest.raises(ValueError, match="no rows"):
+        summarise(_log({}, epochs=()))
+    with pytest.raises(ValueError, match="non-finite"):
+        summarise(_log({"a": (np.inf, 100)}))
+    with pytest.raises(ValueError, match="repeats"):
+        summarise(pd.concat([_log({"a": (1, 1)}), _log({"a": (2, 1)})]))
+
+
+def _summary(national_errors, local=LOCAL, epochs=(0, 30)):
+    return summarise(
+        _log(
+            {f"t{i}": (100 * (1 + e), 100) for i, e in enumerate(national_errors)},
+            local,
+            epochs,
+        )
+    )
+
+
+def test_a_drop_beyond_the_allowance_fails_and_within_it_passes():
+    base = _summary([0.0] * 100)
+    head = _summary([0.5, 0.5, 0.5] + [0.0] * 97)
+    result = compare(base, head, POLICY)
+    assert result["status"] == "fail"
+    assert (
+        "national targets within 10% falls from 100.0% to 97.0%"
+        in result["failures"][0]
+    )
+    assert [f["target"] for f in result["flags"]] == ["t0", "t1", "t2"]
+    assert compare(base, _summary([0.5] + [0.0] * 99), POLICY)["status"] == "pass"
+
+
+def test_a_target_in_a_worse_band_flags_and_a_better_one_is_a_note():
+    base = _summary([0.0, 0.5] + [0.0] * 98)
+    head = _summary([0.5, 0.0] + [0.0] * 98)
+    result = compare(base, head, POLICY)
+    assert result["status"] == "pass"
+    assert result["flags"] == [{"target": "t0", "base": "within 10%", "head": "25-50%"}]
+    assert "1 national targets are in a better band on the head." in result["notes"]
+    # Sol re-review of #599: a target worsening outside 10% flags too.
+    assert compare(_summary([0.15]), _summary([3.0]), POLICY)["flags"] == [
+        {"target": "t0", "base": "10-25%", "head": "over 100%"}
+    ]
+    # A move inside one band is not a flag.
+    assert compare(_summary([0.01]), _summary([0.09]), POLICY)["flags"] == []
+    assert compare(_summary([0.3]), _summary([0.45]), POLICY)["flags"] == []
+
+
+@pytest.mark.parametrize(
+    "error, expected",
+    [(0.0, 0), (0.1, 0), (0.1000001, 1), (0.25, 1), (0.5, 2), (1.0, 3), (1.01, 4)],
+)
+def test_band_edges_are_inclusive(error, expected):
+    assert band(error) == expected
+
+
+def test_an_estimate_exactly_on_a_band_edge_is_inside_it():
+    """Delta review of #599: 110 / 100 - 1 is 0.10000000000000009."""
+    s = summarise(_log({"edge": (110, 100), "quarter": (125, 100)}, local=LOCAL))
+    assert s["national"] == {"edge": 0, "quarter": 1}
+
+
+def test_a_repeated_local_target_is_refused():
+    log = _log({"a": (1, 1)}, local=[(1, 1)])
+    with pytest.raises(ValueError, match="repeats a local target"):
+        summarise(pd.concat([log, log[log.name != "UK"]]))
+
+
+def test_added_targets_do_not_count_against_the_existing_ones():
+    """Sol review of #599, P2: 100 unchanged targets plus 3 poor new ones pass."""
+    base = _summary([0.0] * 100)
+    head = summarise(
+        _log(
+            {
+                **{f"t{i}": (100, 100) for i in range(100)},
+                **{f"new{i}": (200, 100) for i in range(3)},
+            },
+            LOCAL,
+        )
+    )
+    result = compare(base, head, POLICY)
+    assert result["status"] == "pass"
+    assert result["summary"]["national"]["compared"] == 100
+    assert "3 national targets are new: new0, new1, new2" in result["notes"]
+
+
+@pytest.mark.parametrize(
+    "head_log",
+    [
+        _log({"a": (1, 1)}, local=()),  # only national rows
+        _log({}, local=LOCAL),  # only local rows
+        _log({"a": (1, 0)}, local=[(1, 0)]),  # only zero targets
+    ],
+)
+def test_an_empty_target_group_fails(head_log):
+    """Sol review of #599, P2: empty groups used to give NaN shares and pass."""
+    result = compare(_summary([0.0] * 10), summarise(head_log), POLICY)
+    assert result["status"] == "fail"
+    assert "has no" in result["failures"][0]
+
+
+def test_an_incomplete_last_epoch_fails():
+    log = _log({f"t{i}": (100, 100) for i in range(10)}, LOCAL)
+    log = log[~((log.epoch == 30) & (log.metric == "t9"))]
+    result = compare(_summary([0.0] * 10), summarise(log), POLICY)
+    assert result["status"] == "fail" and "incomplete" in result["failures"][0]
+
+
+def test_builds_of_different_epochs_cannot_be_compared():
+    base = _summary([0.0], epochs=(0, 30))
+    assert (
+        compare(base, _summary([0.0], epochs=(0, 510, 511)), POLICY)["status"] == "fail"
+    )
+    # 32 epochs log last at 30, or at 31 once the final epoch is saved (#594).
+    result = compare(base, _summary([0.0], epochs=(0, 30, 31)), POLICY)
+    assert result["status"] == "pass"
+    assert "saves the final epoch" in result["notes"][0]
+
+
+errors = st.lists(st.floats(0, 3), min_size=1, max_size=40)
+
+
+@settings(max_examples=200, deadline=None)
+@given(national=errors, local=errors)
+def test_a_build_compared_with_itself_passes_with_nothing_flagged(national, local):
+    s = _summary(national, local=[(100 * (1 + e), 100) for e in local])
+    result = compare(s, s, POLICY)
+    assert result["status"] == "pass" and result["flags"] == []
+    assert all(row["change"] == 0 for row in result["summary"].values())
+
+
+@settings(max_examples=200, deadline=None)
+@given(base=errors, data=st.data())
+def test_improving_every_target_never_fails_and_changes_are_antisymmetric(base, data):
+    better = [data.draw(st.floats(0, e)) for e in base]
+    forward = compare(_summary(base), _summary(better), POLICY)
+    assert forward["status"] == "pass" and forward["flags"] == []
+    backward = compare(_summary(better), _summary(base), POLICY)
+    for name in ("national", "local"):
+        assert forward["summary"][name]["change"] == pytest.approx(
+            -backward["summary"][name]["change"]
+        )
+
+
+def test_the_shipped_policy_loads():
+    policy = load_policy()
+    assert policy.share_drop_allowance == 0.02
+
+
+def test_the_commands_summarise_and_exit_one_on_a_failure(tmp_path):
+    base_log, head_log = tmp_path / "base.csv", tmp_path / "head.csv"
+    targets = {f"t{i}": (100, 100) for i in range(50)}
+    _log(targets, LOCAL).to_csv(base_log, index=False)
+    _log(dict(targets, **{f"t{i}": (200, 100) for i in range(5)}), LOCAL).to_csv(
+        head_log, index=False
+    )
+    for log in (base_log, head_log):
+        assert (
+            main(["summarise", str(log), "--out", str(log.with_suffix(".json"))]) == 0
+        )
+    out = tmp_path / "fit.md"
+    code = main(
+        [
+            "compare",
+            str(tmp_path / "base.json"),
+            str(tmp_path / "head.json"),
+            "--markdown",
+            str(out),
+        ]
+    )
+    assert code == 1
+    assert "## Calibration fit against the base build: fail" in out.read_text()
+    assert (
+        main(["compare", str(tmp_path / "base.json"), str(tmp_path / "base.json")]) == 0
+    )
+
+
+def test_the_pull_request_workflow_uploads_only_summaries():
+    """The datasets are licensed microdata, and raw calibration logs stay on the runner.
+
+    Only aggregate reports leave a job: the fit summaries, the distribution reports
+    and the aggregate results.
+    """
+    from pathlib import Path
+
+    import yaml
+
+    workflow = Path(__file__).parents[2] / ".github" / "workflows" / "pull_request.yaml"
+    jobs = yaml.safe_load(workflow.read_text())["jobs"]
+    uploads = sorted(
+        step["with"]["path"]
+        for job in jobs.values()
+        for step in job.get("steps", [])
+        if step.get("uses", "").startswith("actions/upload-artifact")
+    )
+    assert uploads == [
+        "aggregate_result.json",
+        "base_aggregate_result.json",
+        "base_distribution_report.json",
+        "base_fit_summary.json",
+        "distribution_report.json",
+        "fit_summary.json",
+    ]
+    # The aggregate tool writes household detail under private/; only its
+    # result.json is copied out for upload.
+    text = workflow.read_text()
+    assert 'cp "$RUNNER_TEMP/aggregate/result.json"' in text
+    for job in jobs.values():
+        for step in job.get("steps", []):
+            assert not step.get("uses", "").startswith("actions/cache"), step
+    assert jobs["fit-comparison"]["needs"] == ["test", "base-build"]
+    post = next(
+        s
+        for s in jobs["fit-comparison"]["steps"]
+        if s.get("name", "").startswith("Post")
+    )
+    assert "if [ ! -s fit.md ]" in post["run"]
+
+
+def test_the_aggregate_comparison_is_wired_consistently():
+    """Delta review of #602: one tool commit on every side, the aggregate
+    steps independent of the other measurements, and a crash fails."""
+    from pathlib import Path
+
+    import yaml
+
+    workflow = Path(__file__).parents[2] / ".github" / "workflows" / "pull_request.yaml"
+    text = workflow.read_text()
+    jobs = yaml.safe_load(text)["jobs"]
+    refs = {
+        step["with"]["ref"]
+        for job in jobs.values()
+        for step in job.get("steps", [])
+        if step.get("name") == "Check out the aggregate-impact tool"
+    }
+    assert len(refs) == 1 and len(next(iter(refs))) == 40, refs
+    for job_id in ("test", "base-build"):
+        for step in jobs[job_id]["steps"]:
+            if "aggregate" in str(step.get("name", "")).lower():
+                assert step.get("if") == "${{ !cancelled() }}", (job_id, step["name"])
+    for step in jobs["test"]["steps"]:
+        if step.get("name") in (
+            "Check out the aggregate-impact tool",
+            "Measure national aggregates with the model fixed",
+        ):
+            assert step.get("continue-on-error") is True, step["name"]
+    compare = next(
+        s for s in jobs["fit-comparison"]["steps"] if s.get("id") == "compare"
+    )["run"]
+    crash = compare[compare.index("aggregate_status=$?") :]
+    assert crash.index('echo "status=1"') < crash.index("aggregates.md")
+    assert "## National aggregates: FAIL (the comparison did not finish)" in crash
diff --git a/policyengine_uk_data/utils/distribution_validation.py b/policyengine_uk_data/utils/distribution_validation.py
new file mode 100644
index 000000000..67ff43b13
--- /dev/null
+++ b/policyengine_uk_data/utils/distribution_validation.py
@@ -0,0 +1,424 @@
+"""Single-age population and HBAI income distribution of a built dataset.
+
+Calibration fits grouped targets. These checks measure what the groups can
+hide, on the built dataset:
+
+- **HBAI.** Person-weighted medians, means and Gini coefficients of
+  equivalised household income, before and after housing costs. Also the
+  share of individuals below 60% of the median, for everyone, HBAI
+  dependent children, working-age adults and those at State Pension age.
+  Each is compared with DWP's HBAI FYE 2025 estimate and 95% confidence
+  interval (``targets/reference_data/hbai_fye_2025.yaml``), at period 2024.
+- **Single ages.** Weighted people by sex and single year of age, at the
+  calibration year, against the ONS 2024-based projection the calibration
+  targets come from. Ages from 80 up are one band, because the dataset's
+  oldest records are in the 80-84 band (#593).
+
+``compare`` sets a pull request's build against its base's. While the
+same-code spread of these measures is unmeasured (#598), it reports moves
+and flags a measure that leaves the HBAI confidence interval, but fails
+only on a broken or mismatched report.
+
+    python -m policyengine_uk_data.utils.distribution_validation measure DATASET.h5 --out report.json
+    python -m policyengine_uk_data.utils.distribution_validation compare BASE.json HEAD.json [--markdown out.md]
+
+Reports hold aggregates only: no household's values.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import sys
+from pathlib import Path
+from typing import Mapping, Optional
+
+import numpy as np
+import yaml
+
+SCHEMA = "policyengine-uk-data-distribution/1"
+HBAI_REFERENCE = (
+    Path(__file__).parents[1] / "targets" / "reference_data" / "hbai_fye_2025.yaml"
+)
+# The FRS loader annualises weekly amounts with 365.25 / 7 (datasets/frs.py).
+WEEKS_PER_YEAR = 365.25 / 7
+TOP_AGE_BAND = 80
+SEXES = {"male": ("MALE", "Males"), "female": ("FEMALE", "Females")}
+
+
+def weighted_median(values, weights) -> float:
+    values, weights = np.asarray(values, float), np.asarray(weights, float)
+    order = np.argsort(values, kind="mergesort")
+    cumulative = np.cumsum(weights[order])
+    return float(values[order][np.searchsorted(cumulative, cumulative[-1] / 2)])
+
+
+def weighted_gini(values, weights) -> float:
+    """Gini coefficient of a weighted distribution (0 to 1)."""
+    values, weights = np.asarray(values, float), np.asarray(weights, float)
+    order = np.argsort(values, kind="mergesort")
+    x, w = values[order], weights[order]
+    cw = np.cumsum(w)
+    cxw = np.cumsum(x * w)
+    total_w, total_xw = cw[-1], cxw[-1]
+    # Area under the Lorenz curve by the trapezium rule over weighted points.
+    lorenz = cxw / total_xw
+    previous = np.concatenate([[0.0], lorenz[:-1]])
+    area = np.sum(w / total_w * (lorenz + previous) / 2)
+    return float(1 - 2 * area)
+
+
+def hbai_incomes(bhc, ahc, equivalisation_bhc, equivalisation_ahc):
+    """Equivalised weekly incomes as HBAI defines them.
+
+    HBAI resets negative incomes before housing costs to zero, and derives
+    the income after housing costs from the adjusted figure, so it can still
+    be negative (DWP, HBAI background information and methodology, FYE 2025,
+    "Negative incomes"). ``bhc`` and ``ahc`` are annual household incomes.
+    """
+    bhc, ahc = np.asarray(bhc, float), np.asarray(ahc, float)
+    housing_costs = bhc - ahc
+    adjusted = np.maximum(bhc, 0.0)
+    return (
+        adjusted / np.asarray(equivalisation_bhc, float) / WEEKS_PER_YEAR,
+        (adjusted - housing_costs)
+        / np.asarray(equivalisation_ahc, float)
+        / WEEKS_PER_YEAR,
+    )
+
+
+def hbai_measures(income_bhc, income_ahc, weights, child, pensioner) -> dict:
+    """The HBAI reference measures from person-level arrays.
+
+    ``income_*`` are each person's household equivalised income in £ a week;
+    ``weights`` are person weights; ``child`` and ``pensioner`` are the HBAI
+    groups (working-age adults are everyone else).
+    """
+    child, pensioner = np.asarray(child, bool), np.asarray(pensioner, bool)
+    groups = {
+        "all": np.ones_like(child),
+        "children": child,
+        "working_age": ~child & ~pensioner,
+        "pensioners": pensioner,
+    }
+    out = {}
+    for suffix, income in (("bhc", income_bhc), ("ahc", income_ahc)):
+        income = np.asarray(income, float)
+        median = weighted_median(income, weights)
+        out[f"median_{suffix}"] = median
+        out[f"mean_{suffix}"] = float(np.average(income, weights=weights))
+        out[f"gini_{suffix}"] = 100 * weighted_gini(income, weights)
+        low = income < 0.6 * median
+        for group, mask in groups.items():
+            w = weights * mask
+            total = np.sum(w)
+            # An empty group has no rate; compare fails on it.
+            out[f"relative_low_income_{suffix}_{group}"] = (
+                float(100 * np.sum(w * low) / total) if total > 0 else None
+            )
+    return out
+
+
+def load_hbai_reference(path=None) -> dict:
+    with open(path or HBAI_REFERENCE) as f:
+        raw = yaml.safe_load(f)
+    if raw.get("schema_version") != 1:
+        raise ValueError("HBAI reference: schema_version must be 1")
+    for name, m in raw["measures"].items():
+        if not m["low"] <= m["estimate"] <= m["high"]:
+            raise ValueError(f"HBAI reference {name}: estimate outside its interval")
+    return raw
+
+
+def single_ages(age, is_male, weights, reference: Mapping) -> dict:
+    """People by sex and single age, ages from ``TOP_AGE_BAND`` up as one band.
+
+    ``reference`` maps (sex, age) to the ONS count, with ages from
+    ``TOP_AGE_BAND`` up summed by the caller.
+    """
+    age = np.minimum(np.floor(np.asarray(age, float)), TOP_AGE_BAND).astype(int)
+    out = {}
+    for sex, mask in (("male", is_male), ("female", ~np.asarray(is_male, bool))):
+        counts = np.bincount(
+            age[mask], weights=np.asarray(weights)[mask], minlength=TOP_AGE_BAND + 1
+        )
+        out[sex] = {
+            str(a): {"value": float(counts[a]), "reference": float(reference[(sex, a)])}
+            for a in range(TOP_AGE_BAND + 1)
+        }
+    return out
+
+
+def ons_single_ages(year: int) -> dict:
+    """The ONS 2024-based projection by sex and age, with 80+ as one band."""
+    import pandas as pd
+
+    from policyengine_uk_data.targets.sources.ons_demographics import (
+        _download_uk_projection,
+        _year_column,
+    )
+
+    df = _download_uk_projection()
+    column = _year_column(df, year)
+    if column is None:
+        raise ValueError(f"the ONS projection has no {year}")
+    ages = pd.to_numeric(
+        df["Age"].astype(str).str.extract(r"(\d+)")[0], errors="coerce"
+    )
+    out = {}
+    for sex, (_, ons_sex) in SEXES.items():
+        rows = df["Sex"] == ons_sex
+        for a in range(TOP_AGE_BAND):
+            out[(sex, a)] = float(df.loc[rows & (ages == a), column].sum())
+        out[(sex, TOP_AGE_BAND)] = float(
+            df.loc[rows & (ages >= TOP_AGE_BAND), column].sum()
+        )
+    return out
+
+
+def summarise_single_ages(rows: Mapping) -> dict:
+    errors = {
+        (sex, int(a)): cell["value"] / cell["reference"] - 1
+        for sex, by_age in rows.items()
+        for a, cell in by_age.items()
+        if cell["reference"] > 0
+    }
+    under = [abs(e) for (_, a), e in errors.items() if a < TOP_AGE_BAND]
+    teens = [abs(e) for (_, a), e in errors.items() if 15 <= a <= 19]
+    return {
+        "mean_abs_error_0_79": float(np.mean(under)),
+        "max_abs_error_0_79": float(np.max(under)),
+        "max_abs_error_15_19": float(np.max(teens)),
+        "error_80_plus": {
+            sex: float(
+                rows[sex][str(TOP_AGE_BAND)]["value"]
+                / rows[sex][str(TOP_AGE_BAND)]["reference"]
+                - 1
+            )
+            for sex in rows
+        },
+    }
+
+
+def measure(
+    dataset_path, hbai_period: Optional[int] = None, age_year: int = 2025
+) -> dict:
+    """The distribution report of a built dataset (aggregates only)."""
+    from policyengine_uk import Microsimulation
+
+    reference = load_hbai_reference()
+    period = hbai_period or int(reference["period"])
+    sim = Microsimulation(dataset=str(dataset_path))
+
+    def person(variable, year):
+        return np.asarray(sim.calculate(variable, year, map_to="person").values, float)
+
+    weights = person("household_weight", period)
+    income_bhc, income_ahc = hbai_incomes(
+        person("hbai_household_net_income", period),
+        person("hbai_household_net_income_ahc", period),
+        person("household_equivalisation_bhc", period),
+        person("household_equivalisation_ahc", period),
+    )
+    measures = hbai_measures(
+        income_bhc,
+        income_ahc,
+        weights,
+        person("is_hbai_dependent_child", period) > 0,
+        person("is_SP_age", period) > 0,
+    )
+    hbai = {}
+    for name, ref in reference["measures"].items():
+        value = measures[name]
+        hbai[name] = {
+            "value": value,
+            "reference": ref["estimate"],
+            "low": ref["low"],
+            "high": ref["high"],
+            "inside": value is not None and bool(ref["low"] <= value <= ref["high"]),
+        }
+    gender = np.asarray(sim.calculate("gender", age_year).values).astype(str)
+    ages = single_ages(
+        person("age", age_year),
+        gender == "MALE",
+        person("household_weight", age_year),
+        ons_single_ages(age_year),
+    )
+    data = Path(dataset_path).read_bytes()
+    return {
+        "schema": SCHEMA,
+        "dataset": Path(dataset_path).name,
+        "sha256": hashlib.sha256(data).hexdigest(),
+        "periods": {"hbai": period, "single_age": age_year},
+        "hbai": hbai,
+        "single_age": ages,
+        "summary": {
+            "hbai_outside_interval": sorted(
+                n for n, m in hbai.items() if not m["inside"]
+            ),
+            "single_age": summarise_single_ages(ages),
+        },
+    }
+
+
+def _finite(x) -> bool:
+    return isinstance(x, (int, float)) and math.isfinite(x)
+
+
+def compare(base: Mapping, head: Mapping) -> dict:
+    """A head build's distribution against its base's.
+
+    Fails only on a broken or mismatched report; flags a measure that
+    leaves its HBAI interval, or moves away from its reference.
+    """
+    failures, flags = [], []
+    for side, report in (("base", base), ("head", head)):
+        if report.get("schema") != SCHEMA:
+            failures.append(f"The {side} distribution report is missing or malformed.")
+    if failures:
+        return {
+            "status": "fail",
+            "failures": failures,
+            "flags": [],
+            "rows": [],
+            "single_age": {},
+        }
+    required = set(load_hbai_reference()["measures"])
+    for side, report in (("base", base), ("head", head)):
+        missing = sorted(required - set(report.get("hbai") or {}))
+        if missing:
+            failures.append(f"The {side} report has no {', '.join(missing)}.")
+        summary = (report.get("summary") or {}).get("single_age") or {}
+        for key in ("mean_abs_error_0_79", "max_abs_error_0_79", "max_abs_error_15_19"):
+            if not _finite(summary.get(key)):
+                failures.append(
+                    f"The {side} report's single-age {key} is missing or not finite."
+                )
+    if failures:
+        return {
+            "status": "fail",
+            "failures": failures,
+            "flags": [],
+            "rows": [],
+            "single_age": {},
+        }
+    if base["periods"] != head["periods"]:
+        failures.append(
+            f"The reports measured different periods ({base['periods']} and {head['periods']})."
+        )
+    rows = []
+    for name, h in head["hbai"].items():
+        b = base["hbai"].get(name)
+        if b is None or not (_finite(h["value"]) and _finite(b["value"])):
+            failures.append(f"HBAI {name} is missing or not finite on one side.")
+            continue
+        worse = abs(h["value"] - h["reference"]) - abs(b["value"] - b["reference"])
+        row = {
+            "measure": name,
+            "base": b["value"],
+            "head": h["value"],
+            "reference": h["reference"],
+            "low": h["low"],
+            "high": h["high"],
+            "base_inside": b["inside"],
+            "head_inside": h["inside"],
+            "further_from_reference": worse,
+        }
+        rows.append(row)
+        if b["inside"] and not h["inside"]:
+            flags.append(
+                f"HBAI {name} leaves its 95% interval: {b['value']:.2f} -> {h['value']:.2f} "
+                f"(HBAI {h['reference']:.2f}, {h['low']:.2f} to {h['high']:.2f})."
+            )
+    sb, sh = base["summary"]["single_age"], head["summary"]["single_age"]
+    single = {
+        k: {"base": sb[k], "head": sh[k]}
+        for k in ("mean_abs_error_0_79", "max_abs_error_0_79", "max_abs_error_15_19")
+    }
+    return {
+        "status": "fail" if failures else "pass",
+        "failures": failures,
+        "flags": flags,
+        "rows": rows,
+        "single_age": single,
+    }
+
+
+def render(result: Mapping) -> str:
+    lines = [
+        f"## Income distribution and single ages against the base build: {result['status']}",
+        "",
+    ]
+    if result["failures"]:
+        lines += ["### Failures", ""] + [f"- {f}" for f in result["failures"]] + [""]
+    if result["rows"]:
+        lines += [
+            "HBAI FYE 2025 (DWP), period 2024: equivalised income in £ a week, rates and Gini in per cent. "
+            "A value outside HBAI's 95% interval is marked (outside).",
+            "",
+            "| Measure | Base | Head | HBAI | 95% interval |",
+            "|---|---:|---:|---:|---|",
+        ]
+        for r in result["rows"]:
+            mark = lambda inside: "" if inside else " (outside)"  # noqa: E731
+            lines.append(
+                f"| {r['measure']} | {r['base']:.2f}{mark(r['base_inside'])} | {r['head']:.2f}{mark(r['head_inside'])} "
+                f"| {r['reference']:.2f} | {r['low']:.2f} to {r['high']:.2f} |"
+            )
+        lines.append("")
+    if result["single_age"]:
+        lines += [
+            "| Single ages against the ONS projection | Base | Head |",
+            "|---|---:|---:|",
+        ]
+        labels = {
+            "mean_abs_error_0_79": "Mean error, ages 0 to 79",
+            "max_abs_error_0_79": "Largest error, ages 0 to 79",
+            "max_abs_error_15_19": "Largest error, ages 15 to 19",
+        }
+        for key, label in labels.items():
+            cell = result["single_age"][key]
+            lines.append(
+                f"| {label} | {100 * cell['base']:.1f}% | {100 * cell['head']:.1f}% |"
+            )
+        lines.append("")
+    lines += ["### Flags", ""] + ([f"- {f}" for f in result["flags"]] or ["None."])
+    lines += [
+        "",
+        "These measures report and flag but do not fail yet: their same-code spread is unmeasured (#598).",
+    ]
+    return "\n".join(lines) + "\n"
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(
+        prog="python -m policyengine_uk_data.utils.distribution_validation"
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+    m = sub.add_parser("measure", help="measure a built dataset")
+    m.add_argument("dataset")
+    m.add_argument("--out", required=True)
+    c = sub.add_parser("compare", help="compare a head build's report with its base's")
+    c.add_argument("base")
+    c.add_argument("head")
+    c.add_argument("--markdown")
+    args = parser.parse_args(argv)
+    if args.command == "measure":
+        report = measure(args.dataset)
+        Path(args.out).write_text(json.dumps(report, indent=1))
+        print(json.dumps(report["summary"], indent=1))
+        return 0
+    base = json.loads(Path(args.base).read_text())
+    head = json.loads(Path(args.head).read_text())
+    result = compare(base, head)
+    text = render(result)
+    if args.markdown:
+        Path(args.markdown).write_text(text)
+    print(text)
+    return 1 if result["status"] == "fail" else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/policyengine_uk_data/utils/fit_report.py b/policyengine_uk_data/utils/fit_report.py
new file mode 100644
index 000000000..b4db12d7d
--- /dev/null
+++ b/policyengine_uk_data/utils/fit_report.py
@@ -0,0 +1,293 @@
+"""Final calibration fit of a build, summarised safely, and the comparison of two.
+
+A build writes ``constituency_calibration_log.csv``: each target's estimate
+and value every tenth epoch and at the last one, nationally (rows named
+``UK``) and for each constituency. The log stays on the runner that built
+it. ``summarise`` reduces its last epoch to what the comparison needs, and
+only the summary is uploaded: for every national and local target, the band
+its error |estimate / target - 1| falls in (``BANDS``: within 10%, 10-25%,
+25-50%, 50-100%, over 100%), with zero targets left out. No estimate or
+error leaves the runner. An
+estimate built from a few households, or dominated by one, divided by a
+published count, could give that household's value back, and the dataset
+repeats households (output-area clones, SPI and CGT donors), so a count of
+rows can't show how many respondents support a figure (Sol re-review of
+#599). A band, at least 10% of a public target wide, is a bound, not a
+value.
+
+``compare`` checks that a pull request does not worsen the fit of the build
+it would replace, built the same way (same environment, epochs and clones),
+using the policy in ``targets/fit_policy.yaml``:
+
+- **Fail** when the share of national, or of local, targets within 10%
+  falls by more than the allowance, over the targets both builds have.
+  Added and removed targets are reported, not counted.
+- **Flag**, for review, each national target whose band is worse on the
+  head than on the base. Per-target moves do not fail while builds of the
+  same code move 2 to 12 national targets to a worse band (#598).
+
+    python -m policyengine_uk_data.utils.fit_report summarise LOG.csv --out summary.json
+    python -m policyengine_uk_data.utils.fit_report compare BASE.json HEAD.json [--markdown out.md]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+import pandas as pd
+import yaml
+
+POLICY_PATH = Path(__file__).parents[1] / "targets" / "fit_policy.yaml"
+SCHEMA = "policyengine-uk-data-fit-summary/3"
+# Upper edges of the error bands; band 0 is within 10%.
+BAND_EDGES = (0.10, 0.25, 0.50, 1.00)
+BANDS = ("within 10%", "10-25%", "25-50%", "50-100%", "over 100%")
+_COLUMNS = ["name", "index", "metric", "estimate", "target", "epoch"]
+
+
+@dataclass(frozen=True)
+class FitPolicy:
+    share_drop_allowance: float
+
+
+def load_policy(path=None) -> FitPolicy:
+    with open(path or POLICY_PATH) as f:
+        raw = yaml.safe_load(f)
+    if raw.get("schema_version") != 1:
+        raise ValueError("fit policy: schema_version must be 1")
+    if raw.get("national_target", {}).get("flag") != "worse_band":
+        raise ValueError("fit policy: national_target.flag must be worse_band")
+    return FitPolicy(
+        share_drop_allowance=float(raw["share_within_10pct"]["allowed_drop"]),
+    )
+
+
+def band(error: float) -> int:
+    """The index in ``BANDS`` of an error: 0 when within 10% (inclusive).
+
+    The error is rounded to 9 decimal places first, so an estimate exactly
+    10% off (110 / 100 - 1 = 0.10000000000000009 in floating point) is
+    within 10%.
+    """
+    return int(np.searchsorted(BAND_EDGES, round(float(error), 9), side="left"))
+
+
+def summarise(log) -> dict:
+    """The last epoch of a calibration log, reduced to what may leave the runner."""
+    if isinstance(log, pd.DataFrame):
+        df = log
+    else:
+        df = pd.read_csv(log, usecols=_COLUMNS)
+    if df.empty:
+        raise ValueError("the calibration log has no rows")
+    epochs = sorted(df.epoch.unique())
+    last = int(epochs[-1])
+    rows = df[df.epoch == last]
+    previous = int((df.epoch == epochs[-2]).sum()) if len(epochs) > 1 else None
+    national = rows[rows.name == "UK"]
+    if national.metric.duplicated().any():
+        raise ValueError(
+            "the calibration log repeats a national target at its last epoch"
+        )
+    national = national[national.target != 0].set_index("metric")
+    local = rows[(rows.name != "UK") & (rows.target != 0)]
+    if local.duplicated(["index", "metric"]).any():
+        raise ValueError("the calibration log repeats a local target at its last epoch")
+    n_errors = (national.estimate / national.target - 1).abs()
+    l_errors = (local.estimate / local.target - 1).abs()
+    if not np.isfinite(n_errors).all() or not np.isfinite(l_errors).all():
+        raise ValueError("the calibration log has non-finite estimates")
+    return {
+        "schema": SCHEMA,
+        "epoch": last,
+        "rows_last_epoch": int(len(rows)),
+        "rows_previous_epoch": previous,
+        "national": {metric: band(error) for metric, error in n_errors.items()},
+        "local": {
+            f"{i}|{m}": band(e)
+            for i, m, e in zip(local["index"], local.metric, l_errors)
+        },
+    }
+
+
+def _share(flags: list) -> float:
+    return float(np.mean(flags)) if flags else float("nan")
+
+
+def compare(base: dict, head: dict, policy: FitPolicy) -> dict:
+    """Failures, flags and the summary for a head build against its base."""
+    failures, notes = [], []
+    for side, s in (("base", base), ("head", head)):
+        if s.get("schema") != SCHEMA:
+            failures.append(f"The {side} fit summary is missing or malformed.")
+            continue
+        if not s["national"] or not s["local"]:
+            missing = "national" if not s["national"] else "local"
+            failures.append(f"The {side} build's last epoch has no {missing} targets.")
+        previous = s["rows_previous_epoch"]
+        if previous and s["rows_last_epoch"] < previous:
+            failures.append(
+                f"The {side} build's last logged epoch ({s['epoch']}) has fewer rows "
+                f"({s['rows_last_epoch']}) than the one before ({previous}): the log "
+                "is incomplete."
+            )
+    if failures:
+        return {
+            "status": "fail",
+            "failures": failures,
+            "notes": [],
+            "flags": [],
+            "summary": {},
+            "targets_compared": 0,
+        }
+    # A log's last epoch is a multiple of ten or, since #594, the final one;
+    # the same epoch count gives last epochs less than ten apart.
+    if abs(base["epoch"] - head["epoch"]) >= 10:
+        failures.append(
+            f"The builds' logs end at epochs {base['epoch']} and {head['epoch']}, so "
+            "they ran different epoch counts and their fits cannot be compared."
+        )
+    elif base["epoch"] != head["epoch"]:
+        notes.append(
+            f"The base log ends at epoch {base['epoch']} and the head's at "
+            f"{head['epoch']}: one of them saves the final epoch (#594)."
+        )
+    summary = {}
+    for name in ("national", "local"):
+        b, h = base[name], head[name]
+        common = sorted(set(b) & set(h))
+        added, removed = sorted(set(h) - set(b)), sorted(set(b) - set(h))
+        before = _share([b[k] == 0 for k in common])
+        after = _share([h[k] == 0 for k in common])
+        summary[name] = {
+            "base": before,
+            "head": after,
+            "change": after - before,
+            "compared": len(common),
+        }
+        if added:
+            notes.append(
+                f"{len(added)} {name} targets are new: {', '.join(added[:10])}"
+            )
+        if removed:
+            notes.append(
+                f"{len(removed)} {name} targets are gone: {', '.join(removed[:10])}"
+            )
+        if not common:
+            failures.append(f"The builds have no {name} targets in common.")
+        elif after < before - policy.share_drop_allowance:
+            failures.append(
+                f"The share of {name} targets within 10% falls from {before:.1%} to "
+                f"{after:.1%}, over the {len(common)} targets both builds have, by "
+                f"more than the {policy.share_drop_allowance:.1%} allowed."
+            )
+    common = sorted(set(base["national"]) & set(head["national"]))
+    b_band, h_band = base["national"], head["national"]
+    flags = [
+        {"target": t, "base": BANDS[b_band[t]], "head": BANDS[h_band[t]]}
+        for t in common
+        if h_band[t] > b_band[t]
+    ]
+    # Largest moves first.
+    flags.sort(
+        key=lambda f: (BANDS.index(f["base"]) - BANDS.index(f["head"]), f["target"])
+    )
+    better = sum(1 for t in common if h_band[t] < b_band[t])
+    if better:
+        notes.append(f"{better} national targets are in a better band on the head.")
+    return {
+        "status": "fail" if failures else "pass",
+        "failures": failures,
+        "notes": notes,
+        "flags": flags,
+        "summary": summary,
+        "targets_compared": summary["national"]["compared"],
+    }
+
+
+def render(result: dict, policy: FitPolicy) -> str:
+    lines = [f"## Calibration fit against the base build: {result['status']}", ""]
+    if result["summary"]:
+        lines += [
+            "| Targets within 10% (those both builds have) | Base | Head | Change |",
+            "|---|---:|---:|---:|",
+        ]
+        for name, row in result["summary"].items():
+            lines.append(
+                f"| {name.capitalize()} ({row['compared']:,}) | {row['base']:.1%} "
+                f"| {row['head']:.1%} | {100 * row['change']:+.1f}pp |"
+            )
+        lines += ["", f"A drop of more than {policy.share_drop_allowance:.1%} fails."]
+    if result["failures"]:
+        lines += ["", "### Failures", ""] + [f"- {f}" for f in result["failures"]]
+    if result["notes"]:
+        lines += ["", "### Notes", ""] + [f"- {n}" for n in result["notes"]]
+    flags = result["flags"]
+    lines += [
+        "",
+        f"### National targets in a worse error band on the head ({len(flags)})",
+        "",
+    ]
+    if flags:
+        lines += ["| Target | Base | Head |", "|---|---|---|"]
+        lines += [
+            f"| `{f['target']}` | {f['base']} | {f['head']} |" for f in flags[:30]
+        ]
+        if len(flags) > 30:
+            lines.append(f"\n{len(flags) - 30} more in the job log.")
+    else:
+        lines.append("None.")
+    lines += [
+        "",
+        "Flags are for review: builds of the same code move 2 to 12 national "
+        "targets to a worse band (#598). Bands are of |estimate / target - 1|; "
+        "no target's estimate or error is published.",
+    ]
+    return "\n".join(lines) + "\n"
+
+
+def main(argv: Optional[list] = None) -> int:
+    parser = argparse.ArgumentParser(
+        prog="python -m policyengine_uk_data.utils.fit_report"
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+    s = sub.add_parser("summarise", help="reduce a calibration log to a safe summary")
+    s.add_argument("log")
+    s.add_argument("--out", required=True)
+    c = sub.add_parser(
+        "compare", help="compare a head build's fit summary with its base's"
+    )
+    c.add_argument("base")
+    c.add_argument("head")
+    c.add_argument("--markdown", help="write the comparison here as markdown")
+    c.add_argument("--policy", help="fit policy YAML (default targets/fit_policy.yaml)")
+    args = parser.parse_args(argv)
+    if args.command == "summarise":
+        summary = summarise(args.log)
+        Path(args.out).write_text(json.dumps(summary))
+        print(
+            f"epoch {summary['epoch']}: {len(summary['national'])} national, "
+            f"{len(summary['local'])} local targets"
+        )
+        return 0
+    policy = load_policy(args.policy)
+    base = json.loads(Path(args.base).read_text())
+    head = json.loads(Path(args.head).read_text())
+    result = compare(base, head, policy)
+    text = render(result, policy)
+    if args.markdown:
+        Path(args.markdown).write_text(text)
+    print(text)
+    for flag in result["flags"]:
+        print(f"flag: {flag['target']} {flag['base']} -> {flag['head']}")
+    return 1 if result["status"] == "fail" else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())