fix(rectification): bind quality cards to the followup probe

Graduation no longer gets the college-experience question, and identical D24 splits only ask once.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/rectification/event_probes.py

@@ -62,6 +62,11 @@ MAX_PROBES_PER_DOMAIN = 3
 MAX_BOUNDARY_CANDIDATES_PER_DOMAIN = 8
 MIN_BOUNDARY_DAYS = 45
 MAX_QUALITY_DISTINGUISH_PROBES = 2
+QUALITY_DISTINGUISH_EVENT_KINDS = frozenset({
+    "education_start",
+    "education_change",
+    "education_interruption",
+})
 ANSWER_PRIOR_TABLE_VERSION = "rectification-answer-priors-v1"
 DOMINANT_ANSWER_PRIOR = 0.8
 # Conservative population rates, not fitted from product users.
@@ -884,6 +889,43 @@ def _display_date_label(event: dict[str, Any]) -> str:
     return f"{year} 年"
 
 
+def _event_kind_name(event: dict[str, Any]) -> str:
+    return str(event.get("event_kind") or event.get("kind") or "")
+
+
+def _quality_event_allowed(event: dict[str, Any], domain: str) -> bool:
+    if domain != "education":
+        return False
+    return _event_kind_name(event) in QUALITY_DISTINGUISH_EVENT_KINDS
+
+
+def _quality_split_sets(probe: dict[str, Any]) -> tuple[frozenset[str], frozenset[str]]:
+    outcomes = probe.get("expected_outcomes") or []
+    yes = next((row for row in outcomes if isinstance(row, dict) and row.get("answer_class") == "yes"), {})
+    no = next((row for row in outcomes if isinstance(row, dict) and row.get("answer_class") == "no"), {})
+    yes_times = yes.get("supports") if isinstance(yes, dict) else ()
+    no_times = no.get("supports") if isinstance(no, dict) else ()
+    return frozenset(str(item) for item in (yes_times or ())), frozenset(str(item) for item in (no_times or ()))
+
+
+def _quality_probe_rank(probe: dict[str, Any]) -> tuple[float, int, int]:
+    year = int(probe["year"]) if isinstance(probe.get("year"), int) else 9999
+    month = int(probe["month"]) if isinstance(probe.get("month"), int) else 12
+    return (-float(probe.get("information_gain") or 0), year, month)
+
+
+def _select_quality_distinguish_rows(candidates: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
+    best: dict[tuple[str, frozenset[str], frozenset[str]], dict[str, Any]] = {}
+    for probe in candidates:
+        key = (str(probe.get("domain") or ""), *_quality_split_sets(probe))
+        current = best.get(key)
+        if current is None or _quality_probe_rank(probe) < _quality_probe_rank(current):
+            best[key] = probe
+    selected = list(best.values())
+    selected.sort(key=_quality_probe_rank)
+    return selected[:MAX_QUALITY_DISTINGUISH_PROBES]
+
+
 def _quality_user_meaning(event: dict[str, Any], domain: str) -> str:
     label = _display_date_label(event)
     if domain == "education":
@@ -918,6 +960,8 @@ def _quality_distinguish_probes(
         layer = DOMAIN_QUALITY_LAYER.get(domain)
         if not event_id or year is None or layer is None or domain not in DOMAIN_CATALOG:
             continue
+        if not _quality_event_allowed(event, domain):
+            continue
         if event_id in holdout_ids or f"{domain}:{year}" in holdout_keys:
             continue
         if _quality_encoded(event, domain):
@@ -989,9 +1033,7 @@ def _quality_distinguish_probes(
         if distinguish_contract_errors(probe):
             continue
         rows.append(_apply_prior_ranking(probe))
-        if len(rows) >= MAX_QUALITY_DISTINGUISH_PROBES:
-            break
-    return rows
+    return _select_quality_distinguish_rows(rows)
 
 
 def _year_activated(rule_ids: Sequence[str]) -> bool: