diff --git a/migration/prune_empty_customers.py b/migration/prune_empty_customers.py
new file mode 100644
index 0000000..ec633d9
--- /dev/null
+++ b/migration/prune_empty_customers.py
@@ -0,0 +1,119 @@
+"""
+Migration step 5: drop customers that carry no business records at all.
+
+A customer is "empty" when it owns zero properties, zero policies and zero
+transactions — the legacy DATGRAL row exists, but nothing in either business
+line ever attached to it. These are dead ID slots and never-activated
+prospects from the Access era, and they pad the staff customer list with
+rows that can't be acted on.
+
+This runs LAST in the customer graph, not inside transform_customers.py,
+because emptiness is only knowable after properties, policies and
+transactions have loaded. Deciding it here also means the rule stays in one
+place instead of being re-derived from the staged Parquet by duplicating each
+downstream transform's source-matching logic.
+
+Safe by construction: a customer with zero rows in all three tables has
+nothing pointing at it, so the delete cannot orphan anything. Only its own
+`customer_legacy_refs` rows go with it.
+
+Nothing is silently destroyed — every deleted customer is written to
+`output/pruned_customers.csv` (with its legacy provenance) before the delete,
+and the whole step is reproducible from the Access sources by re-running the
+pipeline without it.
+
+Run:  ./.venv/bin/python prune_empty_customers.py --env dev [--dry-run]
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+from pathlib import Path
+
+from dbenv import connect
+
+AUDIT = Path(__file__).parent / "output" / "pruned_customers.csv"
+
+# The emptiness test. Kept as one string so the audit dump and the delete can
+# never disagree about what "empty" means.
+EMPTY_WHERE = """
+    NOT EXISTS (SELECT 1 FROM properties   x WHERE x.customerId = c.id)
+AND NOT EXISTS (SELECT 1 FROM policies     x WHERE x.customerId = c.id)
+AND NOT EXISTS (SELECT 1 FROM transactions x WHERE x.customerId = c.id)
+"""
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--env", default="dev", help="target environment (reads deploy/.env.<env>)")
+    ap.add_argument("--dry-run", action="store_true",
+                    help="report and write the audit CSV, but delete nothing")
+    args = ap.parse_args()
+
+    conn = connect(args.env)
+    cur = conn.cursor()
+    print(f"[prune] target env: {args.env}")
+
+    cur.execute("SELECT COUNT(*) FROM customers")
+    before = cur.fetchone()[0]
+
+    # Snapshot what is about to go, provenance included, for the audit trail.
+    cur.execute(f"""
+        SELECT c.id, c.name, c.nameSource, c.nameMissing, c.city, c.state,
+               c.email, c.phone, c.mobile, c.status,
+               GROUP_CONCAT(CONCAT(r.sourceSystem, ':', r.sourceTable, ':', r.legacyId)
+                            ORDER BY r.sourceSystem SEPARATOR ' | ')
+        FROM customers c
+        LEFT JOIN customer_legacy_refs r ON r.customerId = c.id
+        WHERE {EMPTY_WHERE}
+        GROUP BY c.id
+        ORDER BY c.nameMissing, c.name
+    """)
+    rows = cur.fetchall()
+
+    AUDIT.parent.mkdir(parents=True, exist_ok=True)
+    with AUDIT.open("w", newline="", encoding="utf-8") as fh:
+        w = csv.writer(fh)
+        w.writerow(["id", "name", "nameSource", "nameMissing", "city", "state",
+                    "email", "phone", "mobile", "status", "legacyRefs"])
+        w.writerows(rows)
+
+    named = sum(1 for r in rows if not r[3])
+    nameless = len(rows) - named
+
+    if args.dry_run:
+        print(f"  dry run — {len(rows)} would be pruned, nothing deleted")
+    else:
+        cur.execute(f"DELETE r FROM customer_legacy_refs r JOIN customers c ON c.id = r.customerId "
+                    f"WHERE {EMPTY_WHERE}")
+        refs_deleted = cur.rowcount
+        cur.execute(f"DELETE c FROM customers c WHERE {EMPTY_WHERE}")
+        deleted = cur.rowcount
+        conn.commit()
+        print(f"  deleted {deleted} customers, {refs_deleted} legacy refs")
+
+    cur.execute("SELECT COUNT(*) FROM customers")
+    after = cur.fetchone()[0]
+
+    print("=== Empty-customer prune complete ===")
+    print(f"  customers before : {before}")
+    print(f"  empty (no property, policy or transaction) : {len(rows)}")
+    print(f"     named          : {named}")
+    print(f"     without a name : {nameless}")
+    print(f"  customers after  : {after}")
+    print(f"  audit trail      : {AUDIT}")
+
+    # Whatever survived must still have every downstream FK intact.
+    for table in ("properties", "policies", "transactions"):
+        cur.execute(f"SELECT COUNT(*) FROM {table} t "
+                    f"LEFT JOIN customers c ON c.id = t.customerId WHERE c.id IS NULL")
+        orphans = cur.fetchone()[0]
+        assert orphans == 0, f"{table}: {orphans} orphaned rows after prune"
+    print("  validation: OK (0 orphans)")
+    conn.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/migration/run_all.py b/migration/run_all.py
index 21868a3..00c4987 100644
--- a/migration/run_all.py
+++ b/migration/run_all.py
@@ -43,6 +43,7 @@ STEPS = [
     "transform_properties.py",    # properties + services + trust accounts
     "transform_policies.py",      # policies + installments/vehicles/drivers/benef/claims/adjusters
     "transform_transactions.py",  # shared ledger + type_transactions + exchange_rates
+    "prune_empty_customers.py",   # drop customers with no property/policy/transaction
     "transform_bank.py",          # SCOTHIA bank register (no customer FK; independent)
     "blob_extract.py",            # document pointers; must follow properties + policies
 ]