diff --git a/scripts/expired_prediction_pairs_harness.py b/scripts/expired_prediction_pairs_harness.py
index ec46c318f..bcab7a4da 100644
--- a/scripts/expired_prediction_pairs_harness.py
+++ b/scripts/expired_prediction_pairs_harness.py
@@ -395,16 +395,40 @@ def run(  # pragma: no cover - live IO composition
     conditioned_scores: list[PairScore] = []
     telemetry: list[dict[str, object]] = []
     pairs = 0
+
+    def emit_telemetry(row: dict[str, object]) -> None:
+        # Append incrementally so a late crash loses nothing.
+        telemetry.append(row)
+        if telemetry_path is None:
+            return
+        import dataclasses as _dc
+        import json as _json
+
+        serialisable = {
+            k: (_dc.asdict(v) if isinstance(v, LadderStep) else v)
+            for k, v in row.items()
+        }
+        with telemetry_path.open("a") as handle:
+            handle.write(_json.dumps(serialisable) + "\n")
+
+    if telemetry_path is not None and telemetry_path.exists():
+        telemetry_path.unlink()
     for index, raw_postcode in enumerate(postcodes):
         postcode = str(Postcode(raw_postcode))
         historic = latest_pre_2012_by_uprn(
             historic_repo.get_for_postcode(Postcode(raw_postcode))
         )
         # Pair-check first: only a postcode with a SAP-10.2 relodgement of a
-        # pre-2012 UPRN pays for the cohort fetch.
+        # pre-2012 UPRN pays for the cohort fetch. A cert the mapper can't yet
+        # map (strict-raise) can't be a validation target either — skip it and
+        # keep sweeping; one bad cert must not kill a multi-hour run.
         paired: list[tuple[str, HistoricEpc, EpcPropertyData]] = []
         for uprn, record in historic.items():
-            actual = epc_client.get_by_uprn(int(uprn))
+            try:
+                actual = epc_client.get_by_uprn(int(uprn))
+            except Exception as error:
+                print(f"  {uprn}: unmappable cert skipped: {error}", file=sys.stderr)
+                continue
             if actual is not None and actual.sap_version == _VALIDATION_SAP_VERSION:
                 paired.append((uprn, record, actual))
         print(
@@ -415,7 +439,11 @@ def run(  # pragma: no cover - live IO composition
         )
         if not paired:
             continue
-        cohort = comparables_repo.candidates_for(postcode)
+        try:
+            cohort = comparables_repo.candidates_for(postcode)
+        except Exception as error:
+            print(f"  {postcode}: cohort fetch failed: {error}", file=sys.stderr)
+            continue
         for uprn, record, actual in paired:
             pairs += 1
             loo_cohort = [c for c in cohort if c.epc.uprn != int(uprn)]
@@ -440,11 +468,20 @@ def run(  # pragma: no cover - live IO composition
                     main_fuel=conditioning.main_fuel,
                     total_floor_area_m2=conditioning.total_floor_area_m2,
                 )
-                telemetry.append(
+                emit_telemetry(
                     {
                         "postcode": postcode,
                         "uprn": uprn,
                         "plain_cohort_size": len(base),
+                        # Raw values alongside the booleans, so band-width and
+                        # near-miss questions are answerable post hoc.
+                        "historic_age_band": conditioning.construction_age_band,
+                        "actual_age_band": _actual_age_band(actual),
+                        "historic_tfa": conditioning.total_floor_area_m2,
+                        "actual_tfa": actual.total_floor_area_m2,
+                        "historic_fuel": conditioning.main_fuel,
+                        "actual_fuel": _actual_fuel(actual),
+                        "historic_fuel_text": record.main_fuel,
                         **{f"ladder_{k}": v for k, v in ladder.items()},
                         "agrees_property_type": (
                             None
@@ -490,17 +527,6 @@ def run(  # pragma: no cover - live IO composition
                     )
                 )
 
-    if telemetry_path is not None:
-        import dataclasses as _dc
-        import json as _json
-
-        with telemetry_path.open("w") as handle:
-            for row in telemetry:
-                serialisable = {
-                    k: (_dc.asdict(v) if isinstance(v, LadderStep) else v)
-                    for k, v in row.items()
-                }
-                handle.write(_json.dumps(serialisable) + "\n")
     report = format_report(
         aggregate(plain_scores), aggregate(conditioned_scores), pairs
     )