diff --git a/scripts/expired_prediction_pairs_harness.py b/scripts/expired_prediction_pairs_harness.py
index bb619b469..ec46c318f 100644
--- a/scripts/expired_prediction_pairs_harness.py
+++ b/scripts/expired_prediction_pairs_harness.py
@@ -203,7 +203,161 @@ def _predict_arm(
     return predictor.predict(target, comparables)
 
 
-def run(postcodes: list[str]) -> str:  # pragma: no cover - live IO composition
+def _part0(epc: EpcPropertyData) -> Optional[object]:
+    parts = epc.sap_building_parts
+    return parts[0] if parts else None
+
+
+def _actual_age_band(epc: EpcPropertyData) -> Optional[object]:
+    part = _part0(epc)
+    return getattr(part, "construction_age_band", None)
+
+
+def _actual_wall(epc: EpcPropertyData) -> Optional[object]:
+    part = _part0(epc)
+    return getattr(part, "wall_construction", None)
+
+
+def _actual_fuel(epc: EpcPropertyData) -> Optional[object]:
+    details = epc.sap_heating.main_heating_details
+    return details[0].main_fuel_type if details else None
+
+
+def _tfa_within_band(actual_tfa: float, hist_tfa: Optional[float]) -> Optional[bool]:
+    if hist_tfa is None:
+        return None
+    return abs(actual_tfa - hist_tfa) <= 0.05 * hist_tfa
+
+
+@dataclass(frozen=True)
+class LadderStep:
+    """One conditioning filter's fate in the sequential relax ladder: how many
+    of the incoming cohort matched, and whether it engaged (matches >= k)."""
+
+    matches: int
+    cohort_before: int
+    engaged: bool
+
+
+def simulate_conditioning_ladder(
+    base: list[ComparableProperty],
+    *,
+    age_band: Optional[str],
+    main_fuel: Optional[int],
+    total_floor_area_m2: Optional[float],
+    minimum_cohort: int = 5,
+) -> dict[str, Optional[LadderStep]]:
+    """Replay select_comparables' age->fuel->TFA conditioning sequence over the
+    plain arm's selected cohort, recording per filter whether it ENGAGED
+    (>= minimum_cohort matches survive) or RELAXED. None = attribute unresolved,
+    filter never active."""
+    steps: dict[str, Optional[LadderStep]] = {}
+    cohort = list(base)
+
+    def apply(name: str, active: bool, matches: list[ComparableProperty]) -> None:
+        nonlocal cohort
+        if not active:
+            steps[name] = None
+            return
+        engaged = len(matches) >= minimum_cohort
+        steps[name] = LadderStep(len(matches), len(cohort), engaged)
+        if engaged:
+            cohort = matches
+
+    apply(
+        "construction_age_band",
+        age_band is not None,
+        [c for c in cohort if _actual_age_band(c.epc) == age_band],
+    )
+    apply(
+        "main_fuel",
+        main_fuel is not None,
+        [c for c in cohort if _actual_fuel(c.epc) == main_fuel],
+    )
+    apply(
+        "total_floor_area",
+        total_floor_area_m2 is not None,
+        [
+            c
+            for c in cohort
+            if total_floor_area_m2 is not None
+            and abs(c.epc.total_floor_area_m2 - total_floor_area_m2)
+            <= 0.05 * total_floor_area_m2
+        ],
+    )
+    return steps
+
+
+def format_diagnosis(rows: list[dict[str, object]]) -> str:
+    """Aggregate the per-pair telemetry into the two diagnosis tables: did each
+    conditioning filter ever ENGAGE, and does the historic value still AGREE
+    with the newly lodged one (the staleness measurement)."""
+    if not rows:
+        return ""
+    filters = ("construction_age_band", "main_fuel", "total_floor_area")
+    lines = [
+        "",
+        "## Diagnosis — filter engagement (conditioned arm)",
+        "",
+        "| filter | resolved | engaged | relaxed (too few matches) |",
+        "|---|---|---|---|",
+    ]
+    for name in filters:
+        resolved = engaged = 0
+        for row in rows:
+            step = row.get(f"ladder_{name}")
+            if step is None:
+                continue
+            resolved += 1
+            if isinstance(step, LadderStep) and step.engaged:
+                engaged += 1
+        lines.append(
+            f"| {name} | {resolved}/{len(rows)} | {engaged}/{resolved or 1} "
+            f"| {resolved - engaged}/{resolved or 1} |"
+        )
+    attrs = (
+        "property_type",
+        "built_form",
+        "wall_construction",
+        "construction_age_band",
+        "main_fuel",
+        "tfa_within_5pct",
+    )
+    lines += [
+        "",
+        "## Diagnosis — historic vs newly-lodged agreement (staleness)",
+        "",
+        "| attribute | historic resolved | agrees with new cert |",
+        "|---|---|---|",
+    ]
+    for name in attrs:
+        resolved = agrees = 0
+        for row in rows:
+            value = row.get(f"agrees_{name}")
+            if value is None:
+                continue
+            resolved += 1
+            if value:
+                agrees += 1
+        pct = f" ({agrees / resolved:.0%})" if resolved else ""
+        lines.append(f"| {name} | {resolved}/{len(rows)} | {agrees}/{resolved or 1}{pct} |")
+    sizes: list[int] = [
+        size
+        for row in rows
+        if isinstance((size := row.get("plain_cohort_size")), int)
+    ]
+    if sizes:
+        lines += [
+            "",
+            f"Mean plain-arm cohort size: {sum(sizes) / len(sizes):.1f} "
+            f"(min {min(sizes)}, max {max(sizes)}); relax threshold k=5.",
+        ]
+    return "\n".join(lines)
+
+
+def run(  # pragma: no cover - live IO composition
+    postcodes: list[str], telemetry_path: Optional[Path] = None
+) -> str:
     from domain.sap10_calculator.calculator import Sap10Calculator
     from infrastructure.epc_client.epc_client_service import EpcClientService
     from repositories.comparable_properties.epc_comparable_properties_repository import (
@@ -239,6 +393,7 @@ def run(postcodes: list[str]) -> str:  # pragma: no cover - live IO composition
 
     plain_scores: list[PairScore] = []
     conditioned_scores: list[PairScore] = []
+    telemetry: list[dict[str, object]] = []
     pairs = 0
     for index, raw_postcode in enumerate(postcodes):
         postcode = str(Postcode(raw_postcode))
@@ -277,6 +432,52 @@ def run(postcodes: list[str]) -> str:  # pragma: no cover - live IO composition
             )
             plain = _predict_arm(plain_target, loo_cohort, predictor)
             conditioned = _predict_arm(conditioned_target, loo_cohort, predictor)
+            if plain_target is not None:
+                base = list(select_comparables(plain_target, loo_cohort).members)
+                ladder = simulate_conditioning_ladder(
+                    base,
+                    age_band=conditioning.construction_age_band,
+                    main_fuel=conditioning.main_fuel,
+                    total_floor_area_m2=conditioning.total_floor_area_m2,
+                )
+                telemetry.append(
+                    {
+                        "postcode": postcode,
+                        "uprn": uprn,
+                        "plain_cohort_size": len(base),
+                        **{f"ladder_{k}": v for k, v in ladder.items()},
+                        "agrees_property_type": (
+                            None
+                            if conditioning.property_type is None
+                            else conditioning.property_type == actual.property_type
+                        ),
+                        "agrees_built_form": (
+                            None
+                            if conditioning.built_form is None
+                            else conditioning.built_form == actual.built_form
+                        ),
+                        "agrees_wall_construction": (
+                            None
+                            if conditioning.wall_construction is None
+                            else conditioning.wall_construction == _actual_wall(actual)
+                        ),
+                        "agrees_construction_age_band": (
+                            None
+                            if conditioning.construction_age_band is None
+                            else conditioning.construction_age_band
+                            == _actual_age_band(actual)
+                        ),
+                        "agrees_main_fuel": (
+                            None
+                            if conditioning.main_fuel is None
+                            else conditioning.main_fuel == _actual_fuel(actual)
+                        ),
+                        "agrees_tfa_within_5pct": _tfa_within_band(
+                            actual.total_floor_area_m2,
+                            conditioning.total_floor_area_m2,
+                        ),
+                    }
+                )
             if plain is not None:
                 plain_scores.append(
                     PairScore(compare_prediction(plain, actual), sap_residual(plain, actual))
@@ -289,7 +490,21 @@ def run(postcodes: list[str]) -> str:  # pragma: no cover - live IO composition
                     )
                 )
 
-    return format_report(aggregate(plain_scores), aggregate(conditioned_scores), pairs)
+    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
+    )
+    return report + format_diagnosis(telemetry)
 
 
 def main() -> None:  # pragma: no cover - CLI entry
@@ -297,6 +512,9 @@ def main() -> None:  # pragma: no cover - CLI entry
     parser.add_argument("postcodes", nargs="*", help="postcodes to scan")
     parser.add_argument("--postcodes-file", type=Path, default=None)
     parser.add_argument("--out", type=Path, default=None, help="write the report here")
+    parser.add_argument(
+        "--telemetry", type=Path, default=None, help="write per-pair JSONL here"
+    )
     args = parser.parse_args()
     postcodes: list[str] = list(args.postcodes)
     if args.postcodes_file is not None:
@@ -307,7 +525,7 @@ def main() -> None:  # pragma: no cover - CLI entry
         ]
     if not postcodes:
         parser.error("no postcodes given")
-    report = run(postcodes)
+    report = run(postcodes, telemetry_path=args.telemetry)
     if args.out is not None:
         args.out.write_text(report + "\n")
     print(report)
diff --git a/tests/scripts/test_expired_prediction_pairs_harness.py b/tests/scripts/test_expired_prediction_pairs_harness.py
index 2781dc566..2bf4e5268 100644
--- a/tests/scripts/test_expired_prediction_pairs_harness.py
+++ b/tests/scripts/test_expired_prediction_pairs_harness.py
@@ -91,3 +91,58 @@ def test_report_prints_both_arms_side_by_side():
     assert "| floor_area_m2 | 4.0 | 4.0 |" in report
     assert "| mean abs | 8.0 | 2.0 |" in report
     assert "1 pairs" in report
+
+
+def test_ladder_simulation_engages_only_with_enough_matches():
+    # Arrange — a 6-strong base cohort: 5 band-C (engages at k=5), then within
+    # the band-C survivors only 2 on fuel 26 (relaxes), and 5 within ±5% of
+    # 100 m² (engages on the un-shrunk cohort).
+    from scripts.expired_prediction_pairs_harness import simulate_conditioning_ladder
+    from tests.domain.epc_prediction.test_comparable_properties import _comparable
+
+    base = [
+        _comparable(
+            property_type="0",
+            certificate_number=f"C{i}",
+            construction_age_band="C",
+            main_fuel=26 if i < 2 else 29,
+            total_floor_area_m2=100.0 + i,
+        )
+        for i in range(5)
+    ] + [
+        _comparable(
+            property_type="0",
+            certificate_number="G0",
+            construction_age_band="G",
+            main_fuel=26,
+            total_floor_area_m2=100.0,
+        )
+    ]
+
+    # Act
+    steps = simulate_conditioning_ladder(
+        base, age_band="C", main_fuel=26, total_floor_area_m2=100.0
+    )
+
+    # Assert — age engaged (5 matches), fuel relaxed (2 < 5 within band-C
+    # survivors), TFA engaged (all 5 survivors within the band).
+    age = steps["construction_age_band"]
+    fuel = steps["main_fuel"]
+    tfa = steps["total_floor_area"]
+    assert age is not None and age.engaged and age.matches == 5
+    assert fuel is not None and not fuel.engaged and fuel.matches == 2
+    assert tfa is not None and tfa.engaged and tfa.matches == 5
+
+
+def test_ladder_simulation_skips_unresolved_attributes():
+    from scripts.expired_prediction_pairs_harness import simulate_conditioning_ladder
+
+    steps = simulate_conditioning_ladder(
+        [], age_band=None, main_fuel=None, total_floor_area_m2=None
+    )
+
+    assert steps == {
+        "construction_age_band": None,
+        "main_fuel": None,
+        "total_floor_area": None,
+    }