Add find-weird-recommendations skill + its two detectors

A focused sibling to audit-ara-portfolio: that skill audits baselines/plans/SAP;
this one audits the *recommendations themselves* — why a measure was or wasn't
offered. Motivated by the portfolio-814 review (Khalim's HHRSH-on-community-
heating, missing-HHRSH, missing-secondary-heating-removal, and a neighbour split).

Adds:
- .claude/skills/find-weird-recommendations/SKILL.md — scan -> neighbour scan ->
  live re-model deep-dive -> root-cause -> codify, with a seeded known-bug
  catalogue and the query-safety rules inherited from audit-ara-portfolio.
- scripts/audit/anomalies.py: new `plan-stops-short-of-goal` HIGH check — the
  default plan ends below the goal band on an unlimited-budget scenario (the
  deterministic worklist for "why didn't this get recommended X"). Adds
  scenario_budget to the bundle/query so budget-capped scenarios are excluded.
- scripts/audit/neighbour_divergence.py: groups a portfolio by (postcode,
  property_type, built_form) and flags effective-SAP outliers vs the cohort
  median. Never touches the 26m-row recommendation table, so it is safe
  portfolio-wide.
- Tests for both (12 passing).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.claude/skills/find-weird-recommendations/SKILL.md

@@ -0,0 +1,242 @@
+---
+name: find-weird-recommendations
+description: Hunt for dubious *recommendations* in a modelled portfolio — measures that were wrongly withheld (SAP could have moved to the goal band but didn't), measures that contradict the dwelling (e.g. HHRSH on community heating), and neighbours of the same type/postcode that were modelled differently. Deterministic scan → neighbour scan → deep-dive with the live re-model → root-cause → codify. Use when the tech team flags "weird" or "dubious" recommendations on a portfolio and wants to understand why the engine did it. Asks for portfolio id and scenario id.
+---
+
+# Find weird recommendations
+
+Sibling to `audit-ara-portfolio`. That skill audits **baselines, plans and SAP
+scores**; this one audits the **recommendations themselves** — *why a measure was
+or wasn't offered*. It exists because the tech team keeps finding individual
+dodgy recommendations (portfolio 814: Khalim's HHRSH-on-community-heating,
+missing-HHRSH, missing-secondary-heating-removal, and a neighbour split) and we
+want a repeatable way to turn "this one looks wrong" into a root cause and a
+durable check that makes the engine better.
+
+The engine's job on an unlimited-budget scenario is to move each dwelling to the
+goal band (usually C) with a sensible, physically-valid set of measures.
+"Weird" = the engine did something a competent surveyor wouldn't: **left SAP on
+the table**, **recommended a measure the dwelling can't take**, or **treated two
+identical neighbours differently**. Each has a deterministic tell; this skill
+runs the tells, then uses the live re-model to explain each hit.
+
+## Input
+
+Ask for **portfolio_id** and **scenario_id** if the user didn't give them. The
+scenario matters here: its `goal_value` is the band we expect the plan to reach,
+and its `budget` decides whether a shortfall is a bug or just "the money ran
+out". Without a scenario_id the deterministic scan audits each Property's
+*default* plan (the FE-shown one).
+
+## Query safety (READ FIRST — inherited from `audit-ara-portfolio`)
+
+The `recommendation` table is **~26m rows with NO index on `plan_id`**. Any query
+reaching it via `plan_id` (a `JOIN ... ON r.plan_id = pl.id`, or a correlated
+`EXISTS`) forces a full seq-scan and can take the shared DB down — this is what
+blocked the portfolio-796 audit. The full rules are in the `audit-ara-portfolio`
+skill; the ones that bite *here*:
+
+- **Confirm any ad-hoc `recommendation` SQL with the user, and show the
+  `EXPLAIN` plan first** (no `ANALYZE`). If it contains `Seq Scan on
+  recommendation`, do NOT run it.
+- **Reach `recommendation` only via the indexed `property_id`, scoped to a
+  handful of flagged property ids** — never portfolio-wide, never via `plan_id`.
+- The two scripts below are safe by construction: `anomalies.py`'s recommendation
+  rollup is opt-in + EXPLAIN-gated, and `neighbour_divergence.py` never touches
+  `recommendation` at all. The measure-set comparison that Khalim's neighbour
+  case needs is done in the **deep-dive** (Phase 4) with the live re-model, not a
+  portfolio-wide `recommendation` query.
+
+## Phase 1 — Deterministic recommendation scan
+
+```
+python -m scripts.audit.anomalies --portfolio <portfolio_id> --scenario <scenario_id>
+```
+
+Writes `modelling_audit.md` / `.csv`. The check that carries this skill:
+
+- **`plan-stops-short-of-goal`** (HIGH) — the default plan ends *below* the goal
+  band on an **unlimited-budget** scenario. This is the deterministic worklist
+  for the "why didn't this get recommended X" class (Khalim's pid 742121). With
+  no budget cap, the only reasons to fall short are (a) a measure was wrongly
+  withheld — a bug — or (b) the dwelling genuinely can't reach the band. Phase 4
+  decides which. Budget-capped scenarios are excluded (falling short is expected).
+
+Also read `plan-below-baseline-band`, `already-meets-goal-with-works`, and — if
+you pass `--with-recommendations` (EXPLAIN-gated; confirm with the user first) —
+`excessive-solar-sap` / `low-solar-bill-savings`. These bound the "the plan did
+something odd" surface.
+
+## Phase 2 — Neighbour-divergence scan
+
+```
+python -m scripts.audit.neighbour_divergence --portfolio <portfolio_id>
+```
+
+Writes `neighbour_divergence.md` / `.csv`. Groups the portfolio into cohorts of
+(postcode, property_type, built_form) and flags any member whose **effective
+baseline SAP** sits `--min-gap` (default 12) points from its cohort median.
+Near-identical neighbours should model alike; a SAP split is what drives a
+recommendation split (Khalim's neighbour case). This is the SAP half of the
+heuristic — cheap and portfolio-wide because it never touches `recommendation`.
+The **measure-set** half is Phase 4: re-model both neighbours and diff their
+plans.
+
+Floor area is reported, not keyed on — discount a genuinely bigger/smaller
+neighbour before trusting a hit.
+
+## Phase 3 — Triage the hits
+
+For each group, HIGH first: note the count, read a few rows, and give a one-line
+**root-cause hypothesis** + a verdict (real bug / expected / threshold to tune).
+Cross-check the two scans against each other — a `plan-stops-short-of-goal`
+property that is *also* a divergent neighbour is a high-value case (the two
+signals agree).
+
+## Phase 4 — Deep-dive with the live re-model (the core of this skill)
+
+Reproduce end-to-end, NO DB writes (never pass `--persist`):
+
+```
+python -m scripts.run_modelling_e2e <property_id> [<neighbour_id> ...] --scenario-id <scenario_id>
+```
+
+This re-fetches the live EPC + solar and runs the full DDD Modelling stage. It
+writes three files — read all three:
+
+- `modelling_e2e.md` — the chosen plan, measure by measure, with post-SAP.
+- `modelling_e2e_candidates.csv` — **every candidate Option the generators
+  produced, including the ones the Optimiser didn't pick.** This is how you tell
+  a *withheld* measure from a *rejected* one:
+  - Measure **absent from candidates** → a **generator eligibility gate**
+    excluded it. This is the bug surface for "why no HHRSH / no
+    secondary_heating_removal". Open the generator and read its `_*_eligible`
+    predicate against the live EPC.
+  - Measure **present in candidates but not chosen** → the Optimiser judged it
+    not worth it (cost/SAP/goal already met). Usually correct; confirm the goal
+    was reached without it.
+- `modelling_e2e.csv` — the row-level plan.
+
+For a neighbour pair, run **both** in one invocation and diff their candidate
+sets and chosen plans — that is the measure-set comparison Khalim asked for,
+done safely (no `recommendation` table).
+
+### Known recommendation-bug patterns (seed catalogue — grew from portfolio 814)
+
+Match each deep-dive against these before hypothesising a novel cause:
+
+- **HHRSH offered on a community-heated dwelling** (pids 742174, 742175).
+  **Confirmed bug.** The DDD generator's `_hhr_storage_eligible`
+  ([`domain/modelling/generators/heating_recommendation.py`](../../../domain/modelling/generators/heating_recommendation.py))
+  returns eligible when the dwelling is `off_gas` — but a community-heated
+  dwelling is off-gas, and the legacy engine's
+  `is_high_heat_retention_valid` ([`recommendations/HeatingRecommender.py`](../../../recommendations/HeatingRecommender.py))
+  gated on `main_fuel["is_community"]` and refused it. The community-heating
+  guard was **lost in the legacy→DDD translation**. Confirm by checking the EPC's
+  main fuel / heat-network flag, then fix by restoring the gate in
+  `_hhr_storage_eligible` (community heating ⇒ not HHR-eligible). A community
+  network is a shared asset a single dwelling can't rip out for storage heaters.
+- **Measure withheld → plan stops short of goal** (pid 742121, "why no HHRSH").
+  The dwelling can't reach the goal band, and the expected bundle is *absent from
+  candidates*. Read the generator's eligibility predicate against the live EPC —
+  an over-tight gate (a fuel/description/category condition that shouldn't
+  exclude this dwelling) is the usual cause.
+- **Missing `secondary_heating_removal`** (pid 742264, "gov site shows secondary
+  heating"). Check whether the live EPC carries a secondary-heating system and
+  whether
+  [`domain/modelling/generators/secondary_heating_recommendation.py`](../../../domain/modelling/generators/secondary_heating_recommendation.py)
+  detects it. A gap between the gov register's secondary-heating field and what
+  our EpcPropertyData mapper carries is a common upstream cause — verify the
+  mapped value, not just the register.
+- **Neighbour split** — two same-cohort dwellings, one gets a bundle the other
+  doesn't. Almost always traces to a **baseline-input divergence** (a different
+  mapped fuel / heating code / floor area / an override on one and not the
+  other), which then flips a generator's eligibility gate. Diff the two EPCs'
+  mapped inputs first; the recommendation split is downstream of that.
+
+## Phase 5 — Root-cause and cross-reference open work
+
+Isolate **where** it goes wrong: a generator eligibility gate, the EPC→
+EpcPropertyData mapping, an override, or the Optimiser. Then check nothing is
+already in flight:
+
+```
+gh pr list --repo Hestia-Homes/Model --state open
+gh pr list --repo Hestia-Homes/Model --state all --search "<keyword>"
+```
+
+Map each confirmed cause to an existing PR/ADR or flag it as new work. State,
+per case, the smallest correct fix (usually: tighten/loosen one generator
+predicate, or fix one mapper field) and which real property ids reproduce it.
+
+## Phase 6 — Self-improve (the compounding loop)
+
+When the review confirms a **novel, systematic** recommendation problem, codify
+it so every future run catches it — same gates as `audit-ara-portfolio` Phase 6:
+
+1. **Systematic** — reproduced on **≥ 5** properties and root-caused, not a
+   one-off (a single weird property is a ticket, not a check).
+2. **Not already covered** — no existing check fires on it; no open/merged PR or
+   ADR already addresses the cause.
+3. **Pressure-tested** — for any threshold/heuristic, run `/grill-me` on the
+   proposed check: false-positive rate on this portfolio? threshold defensible
+   against the real distribution? overlaps an existing check?
+
+**What to change, smallest first:**
+- **A per-property tell** → a decorated `(PropertyAudit) -> Optional[str]` check
+  in `scripts/audit/anomalies.py` (docstring records the motivating pids + the
+  one-line cause, like `plan-stops-short-of-goal` does). Extend the
+  `PropertyAudit` bundle + `_QUERY` if it needs a field — keep every query bounded
+  by the portfolio and off the `recommendation` table's `plan_id`.
+- **A cross-property tell** (a neighbour/cohort pattern) → a detector in
+  `scripts/audit/neighbour_divergence.py`. The obvious next one: promote the
+  measure-set diff into a scripted check that reaches `recommendation` via
+  property-id-scoped, EXPLAIN-gated queries per cohort (kept out of v1 to stay
+  portfolio-wide-safe — do it once the SAP tell's false-positive rate is known).
+- **This catalogue** → add any newly-confirmed bug pattern to Phase 4's list with
+  its pids and root cause, so the next reviewer starts ahead.
+
+Commit each codified check on its own referencing the motivating run, then re-run
+Phases 1–2 to confirm it fires on the motivating cases and nothing surprising
+else. The check registry — with provenance — is the durable output of every hunt.
+
+## Notes
+
+- Read-only on the DB. `run_modelling_e2e` is a dry run — never `--persist`.
+- **Two engines exist.** Portfolios are modelled by the **DDD** engine
+  (`domain/modelling/generators/…`, what `run_modelling_e2e` runs); the legacy
+  `recommendations/…` engine is the historical reference. When a gate looks
+  wrong, compare the DDD predicate against its legacy counterpart — several
+  portfolio-814 bugs are gates the DDD translation **dropped** (community heating
+  is the confirmed one). The legacy code is the spec of intent, not dead code.
+- **Stored plan can be STALE vs live.** The persisted default plan is an earlier
+  run's output; `run_modelling_e2e` re-models against current logic + live EPC.
+  A stored-vs-live gap is a big driver of Phase 1 hits and is fixed by
+  re-modelling, not by debugging the calculator — confirm a sample with the
+  re-model before blaming a generator (see `audit-ara-portfolio` Notes).
+- A `plan-stops-short-of-goal` hit is **not automatically a bug** — the user's
+  own framing: "surely there's a reason, but if a sensible reason we should
+  understand why". The deliverable for a physically-unreachable dwelling is the
+  *explanation* (which measures were tried, why the band is out of reach), not a
+  code change.
+- **On flat/maisonette-heavy portfolios, `plan-stops-short-of-goal` fires in
+  bulk — mostly by design, not bugs.** Characterise the cohort by property_type
+  and post-band before deep-diving (cheap: `property` + `epc_property` + `plan`,
+  no `recommendation` table). Portfolio 814 (94% flats/maisonettes, goal band B)
+  landed 311/338 short of B, but **every** D-or-worse lander and every £0-works
+  shortfall was a flat/maisonette — all 19 houses reached C+. The structural
+  cause: the heating-decarbonisation levers are gated away from flats —
+  `_ashp_eligible` offers ASHP **only to houses/bungalows** (`_is_house_or_bungalow`),
+  the gas-boiler upgrade needs an existing **wet** boiler (point-of-use gas water
+  heaters like WHC 908 don't qualify — see pid 742121, a mains-gas maisonette at
+  E that correctly gets only lighting), and HHRSH needs electric/off-gas. So a
+  mains-gas or community-heated flat with insulated walls + DG has almost no lever
+  and lands at D/E legitimately. Two takeaways: (1) triage the *houses* short of
+  goal first — those are the real bugs; (2) the flat cohort is a **product**
+  question (should ASHP/communal-heat measures be offered to flats?), not a
+  per-property modelling bug — surface it as such.
+- **A measure can be SELECTED while *lowering* SAP** if it's a forced companion
+  (pid 742175: `mechanical_ventilation` at **−4.1 SAP**, bundled after
+  `cavity_wall_insulation`). Net plan SAP still rose, but the companion drags it
+  and worsens the bill — worth a look when a plan's post-SAP barely moves or dips
+  (overlaps `plan-score-below-baseline`). Not yet root-caused; note the pids.

scripts/audit/anomalies.py

@@ -74,6 +74,7 @@ class PropertyAudit:
     portfolio_id: int
     scenario_id: Optional[int]
     scenario_goal_band: Optional[str]
+    scenario_budget: Optional[float]  # None == unlimited budget
     lodged_sap: Optional[float]
     lodged_band: Optional[str]
     effective_sap: Optional[float]
@@ -148,6 +149,32 @@ def _already_meets_goal_with_works(a: PropertyAudit) -> Optional[str]:
     return f"already {a.effective_band} >= goal {a.scenario_goal_band} but cost_of_works £{a.cost_of_works:.0f}"
 
 
+@check("plan-stops-short-of-goal", Severity.HIGH)
+def _plan_stops_short_of_goal(a: PropertyAudit) -> Optional[str]:
+    """The default plan ends BELOW the scenario's goal band on a scenario that is
+    NOT budget-capped — with an unlimited budget a plan should reach the goal
+    unless it is physically impossible for that dwelling, so a shortfall is either
+    a measure wrongly withheld (a bug) or a "sensible reason" the engine should be
+    able to explain.
+
+    This is the deterministic worklist for Khalim's "why didn't this get
+    recommended <measure>" class (portfolio 814, pid 742121 — a dwelling left
+    short of C with no HHRSH bundle). It only flags candidates; the deep-dive
+    (``run_modelling_e2e`` + the candidates CSV) decides withheld-measure vs
+    physically-unreachable. Budget-capped scenarios are excluded because there a
+    shortfall is expected — the money simply ran out."""
+    goal, post = _band_rank(a.scenario_goal_band), _band_rank(a.post_band)
+    if goal is None or post is None or post <= goal:
+        return None
+    if a.scenario_budget is not None:
+        # Budget-capped: falling short is expected once the money runs out.
+        return None
+    return (
+        f"post {a.post_band} ({a.post_sap}) short of goal {a.scenario_goal_band} "
+        f"on an unlimited-budget scenario (cost_of_works £{a.cost_of_works or 0:.0f})"
+    )
+
+
 @check("post-band-score-mismatch", Severity.MEDIUM)
 def _post_band_score_mismatch(a: PropertyAudit) -> Optional[str]:
     """The persisted post band disagrees with the band the post SAP implies — a
@@ -276,7 +303,7 @@ def _negative_bill_savings(a: PropertyAudit) -> Optional[str]:
 _QUERY = text(
     """
     SELECT p.id, p.uprn, p.portfolio_id,
-           pl.scenario_id, s.goal_value AS goal_band,
+           pl.scenario_id, s.goal_value AS goal_band, s.budget AS scenario_budget,
            pbp.lodged_sap_score, pbp.lodged_epc_band,
            pbp.effective_sap_score, pbp.effective_epc_band, pbp.rebaseline_reason,
            pl.post_sap_points, pl.post_epc_rating, pl.cost_of_works,
@@ -394,6 +421,7 @@ def _load(
                     portfolio_id=m["portfolio_id"],
                     scenario_id=m["scenario_id"],
                     scenario_goal_band=m["goal_band"],
+                    scenario_budget=m["scenario_budget"],
                     lodged_sap=m["lodged_sap_score"],
                     lodged_band=m["lodged_epc_band"],
                     effective_sap=m["effective_sap_score"],

scripts/audit/neighbour_divergence.py

@@ -0,0 +1,226 @@
+"""Flag neighbouring dwellings that were modelled *differently despite looking
+the same* — the SAP half of the "neighbours should agree" heuristic.
+
+Motivation (portfolio 814): Khalim found a dwelling that gets an HHRSH bundle
+while its next-door neighbour of the same type does not. Neighbours in one
+postcode, of the same property type and built form, are usually near-identical
+stock; a big split in their *effective baseline SAP* is a strong tell that one of
+the pair was mis-mapped, mis-overridden, or mis-rebaselined — and a SAP split is
+what then drives a recommendation split. This script surfaces those pairs
+cheaply so the deep-dive (``run_modelling_e2e`` on both neighbours) can compare
+their actual measure sets and root-cause the divergence.
+
+It reaches only ``property`` + ``epc_property`` + ``property_baseline_performance``,
+all via the indexed ``property_id`` — it NEVER touches the 26m-row
+``recommendation`` table, so it is safe to run portfolio-wide (unlike the
+measure-set comparison, which the skill does per flagged cohort in the deep-dive).
+
+Run:
+    python -m scripts.audit.neighbour_divergence --portfolio 814
+    python -m scripts.audit.neighbour_divergence --portfolio 814 --min-gap 15
+
+Writes ``neighbour_divergence.md`` + ``neighbour_divergence.csv`` and prints a
+summary. Read-only: never writes to the DB.
+
+A cohort is the set of a portfolio's properties sharing (postcode, property_type,
+built_form). Within a cohort of >= 2, any member whose effective SAP is >=
+``--min-gap`` points from the cohort median is flagged. Floor area is reported,
+not keyed on, so the reviewer can discount a genuinely bigger/smaller neighbour;
+promoting an area guard into the cohort key is a clean future change once the
+false-positive rate is measured on a real portfolio (skill Phase 6).
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+from dataclasses import dataclass
+from statistics import median
+from typing import Optional
+
+from sqlalchemy import text
+
+from scripts.e2e_common import build_engine, load_env
+
+# Hard ceiling so a bad plan aborts instead of saturating the shared DB, mirroring
+# scripts/audit/anomalies.py. Every table here is reached via the indexed
+# property_id and scoped to one portfolio, so this is a backstop, not a crutch.
+_STATEMENT_TIMEOUT_MS = 120_000
+
+# Default SAP gap (points from the cohort median) at which a neighbour is flagged.
+# 12 is a starting threshold, not a tuned one — a full band is ~10-11 SAP points,
+# so >= 12 means the neighbour sits a clear band apart from otherwise-identical
+# stock. Re-justify against a real portfolio's distribution before trusting it
+# (skill Phase 6); expose it as --min-gap so a run can sweep the threshold.
+_DEFAULT_MIN_GAP = 12.0
+
+
+@dataclass(frozen=True)
+class Neighbour:
+    """One modelled property placed in its postcode/type/form cohort."""
+
+    property_id: int
+    uprn: Optional[int]
+    postcode: Optional[str]
+    property_type: Optional[str]
+    built_form: Optional[str]
+    floor_area_m2: Optional[float]
+    effective_sap: Optional[float]
+    effective_band: Optional[str]
+
+
+@dataclass(frozen=True)
+class Divergence:
+    property_id: int
+    uprn: Optional[int]
+    cohort_key: str
+    cohort_size: int
+    detail: str
+
+
+# DISTINCT ON picks the latest ingested epc_property row per property (its
+# property_id is NOT unique — ingestion keeps history), ordered by id DESC.
+_QUERY = text(
+    """
+    SELECT p.id, p.uprn, p.postcode,
+           ep.property_type, ep.built_form, ep.total_floor_area_m2,
+           pbp.effective_sap_score, pbp.effective_epc_band
+    FROM property p
+    JOIN property_baseline_performance pbp ON pbp.property_id = p.id
+    LEFT JOIN LATERAL (
+        SELECT property_type, built_form, total_floor_area_m2
+        FROM epc_property e
+        WHERE e.property_id = p.id
+        ORDER BY e.id DESC
+        LIMIT 1
+    ) ep ON TRUE
+    WHERE p.portfolio_id = :portfolio_id
+    ORDER BY p.id
+    """
+)
+
+
+def _load(portfolio_id: int) -> list[Neighbour]:
+    engine = build_engine()
+    out: list[Neighbour] = []
+    with engine.connect() as conn:
+        conn.execute(text(f"SET statement_timeout = {_STATEMENT_TIMEOUT_MS}"))
+        for row in conn.execute(_QUERY, {"portfolio_id": portfolio_id}):
+            m = row._mapping  # noqa: SLF001 — SQLAlchemy row mapping, mirrors anomalies.py
+            out.append(
+                Neighbour(
+                    property_id=m["id"],
+                    uprn=m["uprn"],
+                    postcode=m["postcode"],
+                    property_type=m["property_type"],
+                    built_form=m["built_form"],
+                    floor_area_m2=m["total_floor_area_m2"],
+                    effective_sap=m["effective_sap_score"],
+                    effective_band=m["effective_epc_band"],
+                )
+            )
+    return out
+
+
+def _cohort_key(n: Neighbour) -> Optional[str]:
+    """A stable cohort label, or None when the neighbour can't be placed (no
+    postcode or no property type — it has no comparable peers to diverge from)."""
+    if not n.postcode or not n.property_type:
+        return None
+    form = n.built_form or "?"
+    return f"{n.postcode.strip().upper()} · {n.property_type} · {form}"
+
+
+def find_divergences(
+    neighbours: list[Neighbour], min_gap: float
+) -> list[Divergence]:
+    cohorts: dict[str, list[Neighbour]] = {}
+    for n in neighbours:
+        key = _cohort_key(n)
+        if key is None or n.effective_sap is None:
+            continue
+        cohorts.setdefault(key, []).append(n)
+
+    found: list[Divergence] = []
+    for key, members in cohorts.items():
+        if len(members) < 2:
+            continue
+        saps = [m.effective_sap for m in members if m.effective_sap is not None]
+        cohort_median = median(saps)
+        for m in members:
+            if m.effective_sap is None:
+                continue
+            gap = m.effective_sap - cohort_median
+            if abs(gap) < min_gap:
+                continue
+            area = f"{m.floor_area_m2:.0f}m²" if m.floor_area_m2 is not None else "?m²"
+            found.append(
+                Divergence(
+                    property_id=m.property_id,
+                    uprn=m.uprn,
+                    cohort_key=key,
+                    cohort_size=len(members),
+                    detail=(
+                        f"effective SAP {m.effective_sap:.0f} ({m.effective_band}) "
+                        f"vs cohort median {cohort_median:.0f} (Δ{gap:+.0f}, "
+                        f"{area}, {len(members)} neighbours)"
+                    ),
+                )
+            )
+    found.sort(key=lambda d: (d.cohort_key, d.property_id))
+    return found
+
+
+def _write_reports(divergences: list[Divergence], scanned: int) -> None:
+    with open("neighbour_divergence.csv", "w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow(["property_id", "uprn", "cohort", "cohort_size", "detail"])
+        for d in divergences:
+            w.writerow([d.property_id, d.uprn, d.cohort_key, d.cohort_size, d.detail])
+
+    by_cohort: dict[str, list[Divergence]] = {}
+    for d in divergences:
+        by_cohort.setdefault(d.cohort_key, []).append(d)
+    lines = [
+        "# Neighbour SAP-divergence audit",
+        "",
+        f"Scanned **{scanned}** properties · flagged **{len(divergences)}** "
+        f"divergent neighbours across **{len(by_cohort)}** cohorts.",
+        "",
+    ]
+    for key in sorted(by_cohort):
+        rows = by_cohort[key]
+        lines.append(f"## {key} — {len(rows)} divergent")
+        lines.append("")
+        for d in rows:
+            lines.append(f"- property **{d.property_id}** (uprn {d.uprn}): {d.detail}")
+        lines.append("")
+    with open("neighbour_divergence.md", "w") as f:
+        f.write("\n".join(lines))
+
+
+def main() -> None:
+    load_env()
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--portfolio", type=int, required=True, help="portfolio_id")
+    parser.add_argument(
+        "--min-gap",
+        type=float,
+        default=_DEFAULT_MIN_GAP,
+        help=f"SAP points from cohort median to flag (default {_DEFAULT_MIN_GAP})",
+    )
+    args = parser.parse_args()
+
+    neighbours = _load(args.portfolio)
+    divergences = find_divergences(neighbours, args.min_gap)
+    _write_reports(divergences, len(neighbours))
+
+    print(
+        f"scanned {len(neighbours)} properties · {len(divergences)} divergent "
+        f"neighbours (>= {args.min_gap:.0f} SAP from cohort median)"
+    )
+    print("wrote neighbour_divergence.md / neighbour_divergence.csv")
+
+
+if __name__ == "__main__":
+    main()

tests/scripts/test_audit_anomalies.py

@@ -6,29 +6,36 @@ from scripts.audit.anomalies import (
     PropertyAudit,
     _effective_lodged_divergence,
     _implausible_lodged_score,
+    _plan_stops_short_of_goal,
 )
 
 
 def _make_audit(
     *,
-    lodged_sap: Optional[float],
-    effective_sap: Optional[float],
+    lodged_sap: Optional[float] = None,
+    effective_sap: Optional[float] = None,
     rebaseline_reason: str = "both",
+    scenario_goal_band: Optional[str] = None,
+    scenario_budget: Optional[float] = None,
+    post_band: Optional[str] = None,
+    post_sap: Optional[float] = None,
+    cost_of_works: Optional[float] = None,
 ) -> PropertyAudit:
     return PropertyAudit(
         property_id=1,
         uprn=None,
         portfolio_id=796,
         scenario_id=None,
-        scenario_goal_band=None,
+        scenario_goal_band=scenario_goal_band,
+        scenario_budget=scenario_budget,
         lodged_sap=lodged_sap,
         lodged_band=None,
         effective_sap=effective_sap,
         effective_band=None,
         rebaseline_reason=rebaseline_reason,
-        post_sap=None,
-        post_band=None,
-        cost_of_works=None,
+        post_sap=post_sap,
+        post_band=post_band,
+        cost_of_works=cost_of_works,
         energy_bill_savings=None,
         energy_consumption_savings=None,
         solar_sap_points=None,
@@ -92,3 +99,53 @@ class TestImplausibleLodgedScore:
 
         # Assert
         assert result is None
+
+
+class TestPlanStopsShortOfGoal:
+    def test_fires_when_short_of_goal_on_unlimited_budget(self) -> None:
+        # Arrange — goal C, plan lands at D, budget unlimited (None): a shortfall
+        # the engine should be able to explain (Khalim's 742121 class).
+        audit = _make_audit(
+            scenario_goal_band="C",
+            scenario_budget=None,
+            post_band="D",
+            post_sap=63.0,
+            cost_of_works=8000.0,
+        )
+
+        # Act
+        result = _plan_stops_short_of_goal(audit)
+
+        # Assert
+        assert result is not None
+        assert "D" in result
+        assert "C" in result
+
+    def test_silent_when_plan_meets_goal(self) -> None:
+        # Arrange — goal C, plan reaches C: nothing to explain.
+        audit = _make_audit(
+            scenario_goal_band="C", scenario_budget=None, post_band="C", post_sap=70.0
+        )
+
+        # Act
+        result = _plan_stops_short_of_goal(audit)
+
+        # Assert
+        assert result is None
+
+    def test_silent_when_budget_capped(self) -> None:
+        # Arrange — goal C, plan short at D, but the scenario is budget-capped:
+        # falling short is expected once the money runs out, not an anomaly.
+        audit = _make_audit(
+            scenario_goal_band="C",
+            scenario_budget=5000.0,
+            post_band="D",
+            post_sap=63.0,
+            cost_of_works=5000.0,
+        )
+
+        # Act
+        result = _plan_stops_short_of_goal(audit)
+
+        # Assert
+        assert result is None

tests/scripts/test_neighbour_divergence.py

@@ -0,0 +1,76 @@
+from typing import Optional
+
+from scripts.audit.neighbour_divergence import Neighbour, find_divergences
+
+
+def _n(
+    property_id: int,
+    *,
+    effective_sap: Optional[float],
+    postcode: str = "AB1 2CD",
+    property_type: str = "Flat",
+    built_form: str = "Mid-Terrace",
+    floor_area_m2: float = 60.0,
+) -> Neighbour:
+    return Neighbour(
+        property_id=property_id,
+        uprn=None,
+        postcode=postcode,
+        property_type=property_type,
+        built_form=built_form,
+        floor_area_m2=floor_area_m2,
+        effective_sap=effective_sap,
+        effective_band=None,
+    )
+
+
+class TestFindDivergences:
+    def test_flags_the_outlier_neighbour(self) -> None:
+        # Arrange — three identical-cohort flats, one sits a clear band below.
+        cohort = [
+            _n(1, effective_sap=70.0),
+            _n(2, effective_sap=72.0),
+            _n(3, effective_sap=50.0),  # Δ-20 vs median 70
+        ]
+
+        # Act
+        result = find_divergences(cohort, min_gap=12.0)
+
+        # Assert — only the outlier is flagged.
+        assert [d.property_id for d in result] == [3]
+
+    def test_silent_when_cohort_agrees(self) -> None:
+        # Arrange — neighbours within a few SAP points of each other.
+        cohort = [_n(1, effective_sap=70.0), _n(2, effective_sap=68.0)]
+
+        # Act
+        result = find_divergences(cohort, min_gap=12.0)
+
+        # Assert
+        assert result == []
+
+    def test_singletons_never_flagged(self) -> None:
+        # Arrange — one flat here, one house there: neither has a peer to diverge from.
+        lonely = [
+            _n(1, effective_sap=30.0, property_type="Flat"),
+            _n(2, effective_sap=90.0, property_type="House"),
+        ]
+
+        # Act
+        result = find_divergences(lonely, min_gap=12.0)
+
+        # Assert
+        assert result == []
+
+    def test_different_postcode_is_a_different_cohort(self) -> None:
+        # Arrange — same type/form but different postcodes: not neighbours.
+        split = [
+            _n(1, effective_sap=40.0, postcode="AB1 2CD"),
+            _n(2, effective_sap=80.0, postcode="ZZ9 9ZZ"),
+        ]
+
+        # Act
+        result = find_divergences(split, min_gap=12.0)
+
+        # Assert
+        assert result == []