Jyotisha/scripts/rectification/event_probes.py

"""Public-safe biographical probes from candidate dasha / varga differences.

Discriminators are feature-signature clusters over the full birth window.
known_event_quality may distinguish when signature groups disagree on the
event's theme varga type, the probe is anchored to confirmed evidence, and
the case cap is respected. Unanchored quality stays clarification-only.
"""

from __future__ import annotations

import re
from datetime import date, datetime, timedelta
from math import log2
from typing import Any, Sequence

from scripts.active_rectification_event_engine import (
    DOMAIN_CONFIG,
    _active_narayana,
    _active_vimshottari,
    _score_event,
)
import dasha_analyzer
import narayana_dasha
from scripts.rectification.candidate_contrast import (
    PROBE_PHASE_CANDIDATE_DISCRIMINATOR,
    PROBE_PHASE_EVENT_CLARIFICATION,
    PROBE_PHASE_EVIDENCE_COLLECTION,
    SIGNATURE_LAYERS,
    assert_distinguish_contract,
    candidate_ids_from_outcomes,
    candidate_set_version,
    candidate_split_hash,
    cluster_contexts_by_signature,
    discriminator_gate_open,
    distinguish_contract_errors,
    event_year,
    expand_times_through_clusters,
    missing_collection_domains,
    opportunity_from_probe,
)
from scripts.rectification.case_holdout import holdout_domain_years, holdout_event_ids
from scripts.rectification.probe_question_contract import (
    QUESTION_CONTRACT_VERSION,
    completed_style_options,
)
from scripts.rectification.refinement_packet import match_level

# Public discriminator list. Four to six domains typically compete; 8 lets about
# three domains keep two years plus a couple of activation fallbacks without
# flooding the ask layer, which still ranks globally by information_gain.
MAX_PROBES = 8
# Collection still asks at most three missing-domain age-band questions.
MAX_COLLECTION_PROBES = 3
# N: keep the top scored probes per domain (boundary years, plus at most one
# activation if the domain is still under this cap).
MAX_PROBES_PER_DOMAIN = 3
# K: per-domain evaluation budget of unique (year, month) windows.
# N <= K. Endpoints and evenly spaced months are tried first (at most K).
# If those K miss every discriminator, scan the remainder until the first
# hit so a lone month is not dropped; do not keep scanning to fill N.
# Shrink K if runtime more than doubles; do not drop multi-pair union.
# MAX_PROBES is the published cap after a global information_gain sort.
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.
# Broad family existence (yearless / 3-year window / age band) is near-certain
# in adult life. Specified-year long-distance move is uncommon. Quality answers
# are closer to even because they condition on a known event.
_DEFAULT_EXISTENCE_PRIORS = {"yes": 0.35, "weak_yes": 0.15, "no": 0.40, "unsure": 0.10}
_DEFAULT_QUALITY_PRIORS = {"yes": 0.30, "weak_yes": 0.20, "no": 0.40, "unsure": 0.10}
ANSWER_PRIORS: dict[tuple[str, str], dict[str, float]] = {
    ("family", "existence"): {"yes": 0.85, "weak_yes": 0.05, "no": 0.05, "unsure": 0.05},
    ("relocation", "existence"): {"yes": 0.20, "weak_yes": 0.10, "no": 0.60, "unsure": 0.10},
    ("education", "existence"): {"yes": 0.45, "weak_yes": 0.15, "no": 0.30, "unsure": 0.10},
    ("relationship", "existence"): {"yes": 0.40, "weak_yes": 0.15, "no": 0.35, "unsure": 0.10},
    ("career", "existence"): {"yes": 0.40, "weak_yes": 0.15, "no": 0.35, "unsure": 0.10},
    ("education", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
    ("relationship", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
    ("career", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
    ("relocation", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
    ("family", "event_quality"): dict(_DEFAULT_QUALITY_PRIORS),
}
DOMAIN_QUALITY_LAYER = {
    "education": "d24",
    "relationship": "d9",
    "career": "d10",
    "relocation": "d4",
    "family": "d12",
}
QUALITY_DISTINGUISH_OPTIONS: dict[str, tuple[dict[str, str], ...]] = {
    "education": (
        {"label": "发挥明显失常", "answer_class": "yes"},
        {"label": "只是将就调剂", "answer_class": "weak_yes"},
        {"label": "基本如愿录取", "answer_class": "no"},
        {"label": "当时说不清楚", "answer_class": "unsure"},
    ),
    "relationship": (
        {"label": "明显受挫变难", "answer_class": "yes"},
        {"label": "只是将就相处", "answer_class": "weak_yes"},
        {"label": "整体比较顺利", "answer_class": "no"},
        {"label": "当时说不清楚", "answer_class": "unsure"},
    ),
    "career": (
        {"label": "明显受挫受压", "answer_class": "yes"},
        {"label": "只是将就应付", "answer_class": "weak_yes"},
        {"label": "整体比较顺利", "answer_class": "no"},
        {"label": "当时说不清楚", "answer_class": "unsure"},
    ),
    "relocation": (
        {"label": "搬迁特别折腾", "answer_class": "yes"},
        {"label": "只是将就安顿", "answer_class": "weak_yes"},
        {"label": "整体比较顺利", "answer_class": "no"},
        {"label": "当时说不清楚", "answer_class": "unsure"},
    ),
    "family": (
        {"label": "家里特别操心", "answer_class": "yes"},
        {"label": "只是普通操心", "answer_class": "weak_yes"},
        {"label": "整体比较顺利", "answer_class": "no"},
        {"label": "当时说不清楚", "answer_class": "unsure"},
    ),
}
LEVEL_RANK = {"none": 0, "weak": 1, "medium": 2, "strong": 3}
LEVEL_P = {"none": 0.15, "weak": 0.35, "medium": 0.62, "strong": 0.82}
SCORING_LAYERS = ("d1", "d9", "d10", "d4", "d5", "d24", "d7", "d12", "d2", "d11", "d30")
LAYER_DOMAIN = {
    "d9": "relationship",
    "d10": "career",
    "d4": "relocation",
    "d5": "education",
    "d24": "education",
    "d7": "family",
    "d12": "family",
    "d2": "finance",
    "d11": "finance",
    "d30": "health_pressure",
}
VOLUNTEER_ONLY = frozenset({"finance", "health_pressure"})
LAYER_VARGA = {
    "d9": "D9",
    "d10": "D10",
    "d4": "D4",
    "d5": "D5",
    "d24": "D24",
    "d7": "D7",
    "d12": "D12",
    "d2": "D2",
    "d11": "D11",
    "d30": "D30",
}
DOMAIN_CATALOG: dict[str, dict[str, Any]] = {
    "education": {
        "event_family": "升学、转学或换学习环境",
        "quality_family": "学业或考试发挥失常、压力特别大",
        "kind": "education_milestone",
        "age_lo": 16,
        "age_hi": 18,
        "varga": "D5 / D24",
    },
    "relocation": {
        "event_family": "搬家或长期住到外地",
        "quality_family": "搬家或住处特别折腾",
        "kind": "home_change",
        "age_lo": 18,
        "age_hi": 24,
        "varga": "D4",
    },
    "relationship": {
        "event_family": "开始一段认真关系、分手或结婚",
        "quality_family": "感情里压力特别大或相处明显变难",
        "kind": "relationship_change",
        "age_lo": 21,
        "age_hi": 26,
        "varga": "D9",
    },
    "career": {
        "event_family": "入职、换工作或职责加重",
        "quality_family": "职责加重或工作压力特别大",
        "kind": "career_change",
        "age_lo": 22,
        "age_hi": 30,
        "varga": "D10",
    },
    "family": {
        "event_family": "家人结婚、添丁或住院",
        "quality_family": "家人结婚、添丁、住院或家里特别操心",
        "kind": "family_event",
        "age_lo": 18,
        "age_hi": 30,
        "varga": "D12 / D7 / D3",
    },
    "finance": {
        "event_family": "收入明显变化、大笔支出或欠债",
        "quality_family": "收入或财务压力特别大",
        "kind": "finance_change",
        "age_lo": 22,
        "age_hi": 32,
        "varga": "D2 / D11",
    },
    "health_pressure": {
        "event_family": "生病、受伤或压力特别大",
        "quality_family": "生病、受伤或压力特别大",
        "kind": "self_health_event",
        "age_lo": 16,
        "age_hi": 40,
        "varga": "D30",
    },
}

# 入学年与高考通常同一学年或前后一年；感情/事业/搬家的邻近年也常是同一段经历。
EXISTENCE_NEARBY_YEARS = {
    "education": 1,
    "relationship": 1,
    "career": 1,
    "relocation": 1,
}
_SEMANTIC_YEAR = re.compile(r"^(?P<domain>[a-z_]+)\.(?P<year>(?:19|20)\d{2})(?:\.|$)")


def asked_years_for_domain(asked_probe_keys: Sequence[str] | None, domain: str) -> set[int]:
    years: set[int] = set()
    prefix = f"{domain}."
    for raw in asked_probe_keys or []:
        key = str(raw or "").strip()
        if not key.startswith(prefix):
            continue
        match = _SEMANTIC_YEAR.match(key)
        if match and match.group("domain") == domain:
            years.add(int(match.group("year")))
    return years


def _clock(value: str) -> int:
    return int(value[:2]) * 60 + int(value[3:5])


def _context_time(context: dict[str, Any]) -> str | None:
    feature = context.get("feature") if isinstance(context.get("feature"), dict) else {}
    raw = feature.get("time")
    if isinstance(raw, str) and len(raw) >= 5:
        return raw[:5]
    at = context.get("candidate_at")
    if isinstance(at, datetime):
        return at.strftime("%H:%M")
    return None


def _birth_year(value: object) -> int | None:
    text = str(value or "").strip()
    if len(text) < 4 or not text[:4].isdigit():
        return None
    year = int(text[:4])
    return year if 1900 <= year <= 2100 else None


def _event_year(event: dict[str, Any]) -> int | None:
    return event_year(event)


def _period_label(year: int, month: int | None = None) -> str:
    if isinstance(month, int) and 1 <= month <= 12:
        return f"{year} 年 {month} 月前后"
    return f"{year} 年前后"


def _year_label(year: int) -> str:
    return _period_label(year)


# Family events (parents marrying, illness) can occur in childhood.
_CHILDHOOD_OK_DOMAINS = frozenset({"family"})


def _domain_year_floor(birth_year: int, domain: str, global_lo: int) -> int:
    if domain in _CHILDHOOD_OK_DOMAINS:
        return global_lo
    catalog = DOMAIN_CATALOG.get(domain)
    if not catalog:
        return global_lo
    return max(global_lo, birth_year + int(catalog["age_lo"]))


def _age_band_year(birth_year: int, domain: str, today: date) -> int | None:
    catalog = DOMAIN_CATALOG.get(domain)
    if not catalog:
        return None
    age = (int(catalog["age_lo"]) + int(catalog["age_hi"])) // 2
    year = birth_year + age
    latest = min(today.year, birth_year + 80)
    earliest = birth_year + 5
    if year < earliest or year > latest:
        return None
    return year


def _layer_value(context: dict[str, Any], layer: str) -> int | None:
    feature = context.get("feature") if isinstance(context.get("feature"), dict) else {}
    if layer == "d1":
        raw = feature.get("ascendant_sign_index")
        if isinstance(raw, int):
            return raw
        index = context.get("ascendant_index")
        return index if isinstance(index, int) else None
    name = LAYER_VARGA.get(layer)
    if not name:
        return None
    vargas = feature.get("varga_ascendants") if isinstance(feature.get("varga_ascendants"), dict) else {}
    raw = vargas.get(name)
    if isinstance(raw, int):
        return raw
    charts = context.get("varga_charts") if isinstance(context.get("varga_charts"), dict) else {}
    chart = charts.get(name) if isinstance(charts.get(name), dict) else {}
    ascendant = chart.get("Ascendant") if isinstance(chart.get("Ascendant"), dict) else {}
    index = ascendant.get("sign_idx")
    return index if isinstance(index, int) else None


def _differing_layers(contexts: Sequence[dict[str, Any]]) -> set[str]:
    values: dict[str, set[int]] = {layer: set() for layer in SCORING_LAYERS}
    for context in contexts:
        for layer in SCORING_LAYERS:
            value = _layer_value(context, layer)
            if isinstance(value, int):
                values[layer].add(value)
    return {layer for layer, bucket in values.items() if len(bucket) > 1}


def _probe_domains(
    remaining_layers: set[str],
    events: Sequence[dict[str, Any]],
    *,
    d1_differs: bool = False,
) -> list[str]:
    volunteered = {
        str(event.get("domain"))
        for event in events
        if isinstance(event, dict) and event.get("domain")
    }
    ordered: list[str] = []
    for layer in SCORING_LAYERS:
        domain = LAYER_DOMAIN.get(layer)
        if not domain or domain in ordered:
            continue
        if layer not in remaining_layers:
            continue
        if domain in VOLUNTEER_ONLY and domain not in volunteered:
            continue
        ordered.append(domain)
    if not ordered and d1_differs:
        ordered.append("education")
    return ordered


def _static_contexts(built: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for context in built.get("static_contexts") or []:
        if isinstance(context, dict) and _context_time(context):
            rows.append(context)
    rows.sort(key=lambda item: _clock(str(_context_time(item))))
    return rows


def _remaining_contexts(built: dict[str, Any], candidate_times: Sequence[str]) -> list[dict[str, Any]]:
    by_time = {_context_time(item): item for item in _static_contexts(built)}
    remaining: list[dict[str, Any]] = []
    seen: set[str] = set()
    for raw in candidate_times:
        time = str(raw or "")[:5]
        context = by_time.get(time)
        if context is None or time in seen:
            continue
        seen.add(time)
        remaining.append(context)
    return remaining


def _scoreable(context: dict[str, Any]) -> bool:
    chart = context.get("chart")
    planets = context.get("planet_longitudes")
    vargas = context.get("varga_charts")
    return (
        isinstance(chart, dict)
        and isinstance(planets, dict)
        and isinstance(vargas, dict)
        and isinstance(planets.get("Moon"), (int, float))
        and isinstance(context.get("ascendant_index"), int)
    )


def _candidate_at(context: dict[str, Any], birth_date: str) -> datetime | None:
    at = context.get("candidate_at")
    if isinstance(at, datetime):
        return at
    time = _context_time(context)
    if not time:
        return None
    try:
        return datetime.strptime(f"{birth_date} {time}", "%Y-%m-%d %H:%M")
    except ValueError:
        return None


def _has_domain_activation(rule_ids: Sequence[str]) -> bool:
    return any(
        ("_domain_" in str(item) or str(item).endswith("_domain_house") or str(item).endswith("_domain_lord") or str(item).endswith("_domain_varga"))
        and not str(item).startswith("no_")
        for item in rule_ids
    )


def _discriminates(left: Sequence[str], right: Sequence[str]) -> bool:
    left_rank = LEVEL_RANK.get(match_level(left), 0)
    right_rank = LEVEL_RANK.get(match_level(right), 0)
    if abs(left_rank - right_rank) >= 2:
        return True
    return _has_domain_activation(left) != _has_domain_activation(right)


def _tracks_present(rule_ids: Sequence[str]) -> tuple[bool, bool]:
    text = [str(item) for item in rule_ids]
    return (
        any(item.startswith("vim_") for item in text),
        any(item.startswith("narayana_") for item in text),
    )


def _vim_start_dates(
    birth_date: str,
    moon_longitude: float,
    lo: int,
    hi: int,
    *,
    include_pratyantar: bool = False,
) -> list[date]:
    nakshatra, progress, _ = dasha_analyzer.lon_to_nakshatra(float(moon_longitude))
    timeline, _, _, _ = dasha_analyzer.build_dasha_timeline(birth_date, nakshatra, progress)
    starts: list[date] = []
    for major in timeline:
        start = major.get("start")
        if isinstance(start, datetime) and lo <= start.year <= hi:
            starts.append(start.date())
        for minor in dasha_analyzer.build_antardasha(major):
            minor_start = minor.get("start")
            if isinstance(minor_start, datetime) and lo <= minor_start.year <= hi:
                starts.append(minor_start.date())
            if not include_pratyantar:
                continue
            for prat in dasha_analyzer.build_antardasha(minor):
                prat_start = prat.get("start")
                if isinstance(prat_start, datetime) and lo <= prat_start.year <= hi:
                    starts.append(prat_start.date())
    return starts


def _vim_start_years(birth_date: str, moon_longitude: float, lo: int, hi: int) -> list[int]:
    return [item.year for item in _vim_start_dates(birth_date, moon_longitude, lo, hi)]


def _narayana_start_dates(
    ascendant_index: int,
    planet_longitudes: dict[str, float],
    birth_date: str,
    lo: int,
    hi: int,
) -> list[date] | None:
    periods = narayana_dasha.calc_narayana_mahadasha(ascendant_index, planet_longitudes)
    if not periods:
        return None
    birth = datetime.strptime(birth_date, "%Y-%m-%d")
    starts: list[date] = []
    for major in periods:
        start_age = major.get("start_age")
        if not isinstance(start_age, (int, float)):
            return None
        at = (birth + timedelta(days=float(start_age) * 365.2425)).date()
        if lo <= at.year <= hi:
            starts.append(at)
        antars = narayana_dasha.calc_narayana_antardasha(periods, int(major["sign_idx"]))
        for minor in antars:
            minor_age = minor.get("start_age")
            if not isinstance(minor_age, (int, float)):
                continue
            minor_at = (birth + timedelta(days=float(minor_age) * 365.2425)).date()
            if lo <= minor_at.year <= hi:
                starts.append(minor_at)
    return starts


def _narayana_start_years(
    ascendant_index: int,
    planet_longitudes: dict[str, float],
    birth_date: str,
    lo: int,
    hi: int,
) -> list[int] | None:
    starts = _narayana_start_dates(ascendant_index, planet_longitudes, birth_date, lo, hi)
    return None if starts is None else [item.year for item in starts]


def _as_start_date(value: date | datetime | int) -> date | None:
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    if isinstance(value, int) and 1900 <= value <= 2100:
        return date(value, 7, 1)
    return None


def _boundary_windows(left: Sequence[date | datetime | int], right: Sequence[date | datetime | int]) -> list[date]:
    windows: list[date] = []
    seen: set[tuple[int, int]] = set()
    for raw_left, raw_right in zip(left, right):
        one = _as_start_date(raw_left)
        two = _as_start_date(raw_right)
        if one is None or two is None:
            continue
        if one.year == two.year and abs((one - two).days) < MIN_BOUNDARY_DAYS:
            continue
        for item in (one, two):
            key = (item.year, item.month)
            if key in seen:
                continue
            seen.add(key)
            windows.append(item)
    return windows


def _boundary_years(left: list[int], right: list[int]) -> set[int]:
    return {item.year for item in _boundary_windows(left, right)}


def _representative_pairs(
    reps: Sequence[dict[str, Any]],
) -> list[tuple[dict[str, Any], dict[str, Any]]]:
    ordered = sorted(
        [item for item in reps if _scoreable(item) and _context_time(item)],
        key=lambda item: _clock(str(_context_time(item))),
    )
    if len(ordered) < 2:
        return []
    pairs: list[tuple[dict[str, Any], dict[str, Any]]] = []
    seen: set[tuple[str, str]] = set()

    def add(left: dict[str, Any], right: dict[str, Any]) -> None:
        left_time = str(_context_time(left) or "")
        right_time = str(_context_time(right) or "")
        if not left_time or not right_time or left_time == right_time:
            return
        key = (left_time, right_time) if left_time < right_time else (right_time, left_time)
        if key in seen:
            return
        seen.add(key)
        pairs.append((left, right))

    for index in range(len(ordered) - 1):
        add(ordered[index], ordered[index + 1])
    if len(ordered) > 2:
        add(ordered[0], ordered[-1])
    return pairs


def _vim_cache_key(birth_date: str, moon: float, lo: int, hi: int) -> tuple[Any, ...]:
    return (birth_date, round(float(moon), 6), lo, hi)


def _narayana_cache_key(
    ascendant_index: int,
    planet_longitudes: dict[str, Any],
    birth_date: str,
    lo: int,
    hi: int,
) -> tuple[Any, ...]:
    planet_key = tuple(
        sorted(
            (str(name), round(float(lon), 6))
            for name, lon in planet_longitudes.items()
            if isinstance(lon, (int, float))
        )
    )
    return (int(ascendant_index), planet_key, birth_date, lo, hi)


def _union_boundary_dates(
    reps: Sequence[dict[str, Any]],
    *,
    birth_date: str,
    lo: int,
    hi: int,
) -> list[date]:
    vim_cache: dict[tuple[Any, ...], list[date]] = {}
    narayana_cache: dict[tuple[Any, ...], list[date] | None] = {}
    dates_by_key: dict[tuple[int, int], date] = {}
    for left, right in _representative_pairs(reps):
        left_moon = float(left["planet_longitudes"]["Moon"])
        right_moon = float(right["planet_longitudes"]["Moon"])
        left_vim_key = _vim_cache_key(birth_date, left_moon, lo, hi)
        right_vim_key = _vim_cache_key(birth_date, right_moon, lo, hi)
        if left_vim_key not in vim_cache:
            vim_cache[left_vim_key] = _vim_start_dates(birth_date, left_moon, lo, hi)
        if right_vim_key not in vim_cache:
            vim_cache[right_vim_key] = _vim_start_dates(birth_date, right_moon, lo, hi)
        windows = list(_boundary_windows(vim_cache[left_vim_key], vim_cache[right_vim_key]))
        left_nara_key = _narayana_cache_key(
            int(left["ascendant_index"]),
            left["planet_longitudes"],
            birth_date,
            lo,
            hi,
        )
        right_nara_key = _narayana_cache_key(
            int(right["ascendant_index"]),
            right["planet_longitudes"],
            birth_date,
            lo,
            hi,
        )
        if left_nara_key not in narayana_cache:
            narayana_cache[left_nara_key] = _narayana_start_dates(
                int(left["ascendant_index"]),
                left["planet_longitudes"],
                birth_date,
                lo,
                hi,
            )
        if right_nara_key not in narayana_cache:
            narayana_cache[right_nara_key] = _narayana_start_dates(
                int(right["ascendant_index"]),
                right["planet_longitudes"],
                birth_date,
                lo,
                hi,
            )
        left_narayana = narayana_cache[left_nara_key]
        right_narayana = narayana_cache[right_nara_key]
        if left_narayana is not None and right_narayana is not None:
            windows.extend(_boundary_windows(left_narayana, right_narayana))
        for item in windows:
            dates_by_key.setdefault((item.year, item.month), item)
    return sorted(dates_by_key.values(), key=lambda item: (item.year, item.month))


def _bounded_candidate_dates(dates: Sequence[date], limit: int) -> list[date]:
    items = list(dates)
    if limit <= 0 or len(items) <= limit:
        return items
    if limit == 1:
        return items[:1]
    picked: list[date] = []
    seen: set[tuple[int, int]] = set()
    last_index = len(items) - 1
    for step in range(limit):
        index = (step * last_index + (limit - 1) // 2) // (limit - 1)
        item = items[index]
        key = (item.year, item.month)
        if key in seen:
            continue
        seen.add(key)
        picked.append(item)
    if len(picked) < limit:
        for item in items:
            key = (item.year, item.month)
            if key in seen:
                continue
            seen.add(key)
            picked.append(item)
            if len(picked) >= limit:
                break
    return picked


def _evaluation_order(dates: Sequence[date], limit: int) -> list[date]:
    sampled = _bounded_candidate_dates(dates, limit)
    sampled_keys = {(item.year, item.month) for item in sampled}
    remainder = [item for item in dates if (item.year, item.month) not in sampled_keys]
    return sampled + remainder


def _probe_sort_key(row: dict[str, Any]) -> tuple[float, str]:
    return (-float(row.get("information_gain") or 0), str(row.get("semantic_key") or ""))


def _best_probe_per_year(rows: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
    by_year: dict[int, dict[str, Any]] = {}
    for row in sorted(rows, key=_probe_sort_key):
        year = int(row["year"])
        if year not in by_year:
            by_year[year] = row
    return list(by_year.values())


def _try_activation_probe(
    *,
    reps: Sequence[dict[str, Any]],
    birth_date: str,
    birth_year: int,
    domain: str,
    now: date,
    blocked_years: set[int],
    holdout_keys: set[str],
    clusters: Sequence[dict[str, Any]],
    set_version: str,
) -> dict[str, Any] | None:
    midpoint = _age_band_year(birth_year, domain, now)
    if midpoint is None or midpoint in blocked_years or f"{domain}:{midpoint}" in holdout_keys:
        return None
    found = _evaluate_contexts(
        reps,
        birth_date=birth_date,
        domain=domain,
        year=midpoint,
        source="dasha_activation",
        clusters=clusters,
        set_version=set_version,
    )
    if found is None or distinguish_contract_errors(found):
        return None
    if not isinstance(found.get("year"), int) or int(found["year"]) <= 0:
        return None
    return found


def _score_year(
    context: dict[str, Any],
    *,
    birth_date: str,
    domain: str,
    year: int,
    month: int | None = None,
) -> dict[str, Any] | None:
    catalog = DOMAIN_CATALOG[domain]
    prefixes, _ = DOMAIN_CONFIG[domain]
    varga_charts = context.get("varga_charts") or {}
    domain_vargas = [varga_charts.get(prefix) for prefix in prefixes]
    if any(item is None for item in domain_vargas):
        return None
    candidate_at = _candidate_at(context, birth_date)
    moon = (context.get("planet_longitudes") or {}).get("Moon")
    if candidate_at is None or not isinstance(moon, (int, float)):
        return None
    month_value = month if isinstance(month, int) and 1 <= month <= 12 else None
    event_at = datetime(year, month_value, 15) if month_value else datetime(year, 7, 1)
    event = {
        "id": f"probe-{domain}-{year}" + (f"-{month_value:02d}" if month_value else ""),
        "domain": domain,
        "event_kind": catalog["kind"],
        "date": f"{year}-{month_value:02d}-15" if month_value else f"{year}-07-01",
        "precision": "month" if month_value else "year",
        "summary": catalog["event_family"],
    }
    try:
        vimshottari = _active_vimshottari(birth_date, float(moon), event_at)
        narayana = _active_narayana(
            int(context["ascendant_index"]),
            context["planet_longitudes"],
            candidate_at,
            event_at,
        )
    except (KeyError, TypeError, ValueError):
        return None
    if narayana[0] is None or narayana[1] is None:
        return None
    return _score_event(
        candidate_time=str(_context_time(context)),
        event=event,
        natal_chart=context["chart"],
        varga_charts=[item for item in domain_vargas if item is not None],
        vimshottari=vimshottari,
        narayana=narayana,
        arudha_padas=context.get("arudha_padas") or {},
    )


def _agent_brief(
    *,
    year_label: str,
    domain: str,
    family: str,
    quality: bool = False,
    exam: bool = False,
    nearby_note: str = "",
) -> str:
    nearby = nearby_note.strip()
    if exam:
        return (
            f"时间范围锁定 {year_label}；领域锁定 {domain}。"
            "语义目标是那次考试的实际体验。结合最近对话，只选一个容易回答的口语入口，"
            "问是否明显失常或压力很大；不要堆叠例子，不得改时间范围。"
            + (f"{nearby}" if nearby else "")
        )
    if quality:
        return (
            f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
            "结合最近对话，只选一个容易回答的口语入口来核对体验；"
            "不要逐字复述语义目标，不要堆叠例子，不得改时间范围。"
            + (f"{nearby}" if nearby else "")
        )
    lead = (
        f"{nearby}时间范围锁定 {year_label}；请问用户那段时间身上发生了什么变化；"
        if nearby
        else f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
        "结合最近对话，只选一个容易回答的口语入口，写一句自然的是/否题；"
    )
    if nearby:
        return (
            lead
            + f"领域锁定 {domain}；语义目标是 {family}。"
            "选项由服务端给出；不要发明年份，不得改时间范围。"
        )
    return (
        lead
        + "不要逐字复述语义目标，不要把所有例子堆进一句，不得改时间范围。"
    )


def _binary_entropy(probability: float) -> float:
    if probability <= 0.0 or probability >= 1.0:
        return 0.0
    return -(probability * log2(probability) + (1.0 - probability) * log2(1.0 - probability))


def _pair_entropy(left: float, right: float) -> float:
    total = left + right
    if total <= 0:
        return 0.0
    return _binary_entropy(left / total)


def _group_entropy(sizes: Sequence[int]) -> float:
    total = sum(int(item) for item in sizes)
    if total <= 0:
        return 0.0
    return round(-sum((item / total) * log2(item / total) for item in sizes if item > 0), 4)


def _information_gain(left_level: str, right_level: str) -> float:
    left_p = LEVEL_P.get(left_level, 0.5)
    right_p = LEVEL_P.get(right_level, 0.5)
    yes_p = 0.5 * left_p + 0.5 * right_p
    after = yes_p * _pair_entropy(left_p, right_p) + (1.0 - yes_p) * _pair_entropy(1.0 - left_p, 1.0 - right_p)
    return round(max(0.0, 1.0 - after), 4)


def _broad_existence_window(probe: dict[str, Any]) -> bool:
    if str(probe.get("source") or "") == "age_band":
        return True
    year = probe.get("year")
    if not isinstance(year, int) or year <= 0:
        return True
    span = probe.get("window_span_years")
    return isinstance(span, int) and span >= 3


def _answer_priors_for(probe: dict[str, Any]) -> dict[str, float]:
    kind = str(probe.get("choice_kind") or "existence")
    if kind not in {"existence", "event_quality"}:
        kind = "existence"
    domain = str(probe.get("domain") or "")
    if domain == "family" and kind == "existence" and not _broad_existence_window(probe):
        return dict(_DEFAULT_EXISTENCE_PRIORS)
    priors = ANSWER_PRIORS.get((domain, kind))
    if priors:
        return dict(priors)
    return dict(_DEFAULT_QUALITY_PRIORS if kind == "event_quality" else _DEFAULT_EXISTENCE_PRIORS)


def _expected_information_gain(raw_split_gain: float, priors: dict[str, float]) -> float:
    weight = (
        float(priors.get("yes") or 0)
        + float(priors.get("no") or 0)
        + 0.5 * float(priors.get("weak_yes") or 0)
    )
    return round(max(0.0, raw_split_gain * weight), 4)


def _apply_prior_ranking(probe: dict[str, Any]) -> dict[str, Any]:
    raw = float(probe.get("information_gain") or 0)
    priors = _answer_priors_for(probe)
    probe["raw_split_gain"] = raw
    probe["answer_priors"] = priors
    probe["information_gain"] = _expected_information_gain(raw, priors)
    return probe


def _dominant_existence_prior(probe: dict[str, Any], priors: dict[str, float]) -> bool:
    if str(probe.get("choice_kind") or "existence") != "existence":
        return False
    if str(probe.get("source") or "") == "known_event_quality":
        return False
    return max(priors.values()) > DOMINANT_ANSWER_PRIOR


def _event_month(event: dict[str, Any]) -> int | None:
    raw = str(event.get("date") or event.get("date_start") or "")
    if len(raw) >= 7 and raw[4] == "-":
        try:
            month = int(raw[5:7])
        except ValueError:
            return None
        if 1 <= month <= 12:
            return month
    return None


def _display_date_label(event: dict[str, Any]) -> str:
    year = _event_year(event)
    month = _event_month(event)
    if year is None:
        return "那次"
    if month:
        return f"{year} 年 {month} 月"
    return f"{year} 年"


def _event_month_index(year: int, month: int | None) -> int | None:
    if month is None or not 1 <= month <= 12:
        return None
    return year * 12 + month


def nearby_ledger_note(
    events: Sequence[dict[str, Any]],
    *,
    domain: str,
    year: int,
    month: int | None,
) -> str:
    if year <= 0:
        return ""
    probe_index = _event_month_index(year, month)
    best: dict[str, Any] | None = None
    best_delta = 99
    for event in events:
        if not isinstance(event, dict):
            continue
        other_domain = str(event.get("domain") or "")
        if other_domain == domain or other_domain not in DOMAIN_CATALOG:
            continue
        other_year = _event_year(event)
        if other_year is None:
            continue
        other_month = _event_month(event)
        other_index = _event_month_index(other_year, other_month)
        if probe_index is not None and other_index is not None:
            delta = abs(probe_index - other_index)
            if delta > 2:
                continue
        elif other_year != year:
            continue
        else:
            delta = 2 if probe_index is not None or other_index is not None else 0
        if delta < best_delta:
            best_delta = delta
            best = event
    if best is None:
        return ""
    family = str(DOMAIN_CATALOG[str(best.get("domain") or "")]["event_family"])
    return f"账本里 { _display_date_label(best) } 有{family}；题干先提那件事再问。"


def _annotate_nearby_ledger(
    probes: Sequence[dict[str, Any]],
    events: Sequence[dict[str, Any]],
) -> None:
    for probe in probes:
        if not isinstance(probe, dict):
            continue
        if str(probe.get("choice_kind") or "existence") != "existence":
            continue
        if str(probe.get("source") or "") not in {"dasha_boundary", "dasha_activation"}:
            continue
        year = probe.get("year")
        if not isinstance(year, int) or year <= 0:
            continue
        note = nearby_ledger_note(
            events,
            domain=str(probe.get("domain") or ""),
            year=year,
            month=int(probe["month"]) if isinstance(probe.get("month"), int) else None,
        )
        if not note:
            continue
        meaning = str(probe.get("user_meaning") or "")
        if note not in meaning:
            probe["user_meaning"] = f"{note}{meaning}"


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":
        return (
            f"{label}那次上大学，更接近如愿、将就调剂、发挥失常还是说不清。"
            "只问那次经历的实际体验，不得改时间范围。"
        )
    family = str(DOMAIN_CATALOG[domain]["event_family"])
    return (
        f"{label}那次{family}，当时更接近顺利、将就、明显受挫还是说不清。"
        "只问那次经历的实际体验，不得改时间范围。"
    )


def _quality_distinguish_probes(
    events: Sequence[dict[str, Any]],
    clusters: Sequence[dict[str, Any]],
    *,
    set_version: str,
    holdout_ids: set[str],
    holdout_keys: set[str],
) -> list[dict[str, Any]]:
    if len(clusters) < 2:
        return []
    rows: list[dict[str, Any]] = []
    for event in events:
        if not isinstance(event, dict):
            continue
        event_id = str(event.get("id") or "")
        domain = str(event.get("domain") or "")
        year = _event_year(event)
        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):
            continue
        groups: dict[int, list[str]] = {}
        for cluster in clusters:
            representative = cluster.get("representative") if isinstance(cluster, dict) else None
            if not isinstance(representative, dict):
                continue
            sign = _layer_value(representative, layer)
            if not isinstance(sign, int):
                continue
            bucket = groups.setdefault(sign, [])
            for time in cluster.get("times") or []:
                clock = str(time)[:5]
                if len(clock) == 5 and clock not in bucket:
                    bucket.append(clock)
        if len(groups) < 2:
            continue
        signs = sorted(groups)
        yes_times = sorted(groups[signs[-1]], key=_clock)
        no_times = sorted(groups[signs[0]], key=_clock)
        overlap = set(yes_times) & set(no_times)
        yes_times = [time for time in yes_times if time not in overlap]
        no_times = [time for time in no_times if time not in overlap]
        if len(yes_times) < 1 or len(no_times) < 1:
            continue
        month = _event_month(event)
        outcomes = [
            {"answer_class": "yes", "supports": yes_times, "conflicts": no_times},
            {"answer_class": "weak_yes", "supports": yes_times, "conflicts": no_times},
            {"answer_class": "no", "supports": no_times, "conflicts": yes_times},
            {"answer_class": "unsure", "supports": [], "conflicts": []},
        ]
        split = candidate_split_hash(
            candidate_set_version_value=set_version,
            domain=domain,
            year=year,
            month=month,
            groups=[yes_times, no_times],
        )
        gain = round(_group_entropy([len(yes_times), len(no_times)]), 4)
        if gain <= 0:
            continue
        label = _display_date_label(event)
        probe = _public_probe(
            year=year,
            month=month,
            domain=domain,
            source="known_event_quality",
            tracks=("vimshottari", "narayana"),
            tracks_agree=True,
            user_meaning=_quality_user_meaning(event, domain),
            event_family=str(DOMAIN_CATALOG[domain]["quality_family"]),
            information_gain=gain,
            semantic_key=f"{domain}.{year}.known_event_quality",
            candidate_split_hash=split,
            candidate_set_version=set_version,
            expected_outcomes=outcomes,
            candidate_ids=candidate_ids_from_outcomes(outcomes),
            left_time=yes_times[0],
            right_time=no_times[0],
            target_evidence_id=event_id,
            display_date_label=label,
            role="distinguish",
            phase=PROBE_PHASE_CANDIDATE_DISCRIMINATOR,
            style_options=list(QUALITY_DISTINGUISH_OPTIONS.get(domain) or ()),
        )
        if distinguish_contract_errors(probe):
            continue
        rows.append(_apply_prior_ranking(probe))
    return _select_quality_distinguish_rows(rows)


def _year_activated(rule_ids: Sequence[str]) -> bool:
    return _has_domain_activation(rule_ids) or LEVEL_RANK.get(match_level(rule_ids), 0) >= 2


def _public_probe(
    *,
    year: int,
    domain: str,
    source: str,
    tracks: Sequence[str],
    tracks_agree: bool,
    user_meaning: str,
    event_family: str,
    month: int | None = None,
    **extra: Any,
) -> dict[str, Any]:
    if source == "known_event_quality":
        phase = PROBE_PHASE_EVENT_CLARIFICATION
        role = "clarify"
    elif source == "age_band":
        phase = PROBE_PHASE_EVIDENCE_COLLECTION
        role = "collect"
    else:
        phase = PROBE_PHASE_CANDIDATE_DISCRIMINATOR
        role = "distinguish"
    month_value = month if isinstance(month, int) and 1 <= month <= 12 else None
    payload = {
        "year": year,
        "year_label": _period_label(year, month_value),
        "domain": domain,
        "event_family": event_family,
        "source": source,
        "tracks": list(tracks),
        "tracks_agree": tracks_agree,
        "unique_minute_claim": False,
        "user_meaning": user_meaning,
        "role": role,
        "phase": phase,
        "semantic_key": f"{domain}.{year}.{month_value:02d}" if month_value else f"{domain}.{year}",
        "information_gain": 0.0,
        "candidate_split_hash": f"{domain}:{year}" + (f"-{month_value:02d}" if month_value else ""),
        "expected_outcomes": [],
        "candidate_ids": [],
        "choice_kind": "event_quality" if source == "known_event_quality" else "existence",
    }
    if month_value:
        payload["month"] = month_value
    payload.update(extra)
    if payload["role"] == "distinguish":
        payload["candidate_ids"] = candidate_ids_from_outcomes(payload.get("expected_outcomes") or [])
    style_options = completed_style_options(payload.get("choice_kind"), payload.get("style_options"))
    if style_options:
        payload["style_options"] = style_options
    payload["question_contract_version"] = QUESTION_CONTRACT_VERSION
    return payload


QUALITY_HINTS: dict[str, tuple[str, ...]] = {
    "education": ("失利", "失常", "压力", "复读", "考砸", "发挥不好", "发挥异常"),
}
# Only education has a quality dimension distinct from existence.
# Career/family "quality_family" copies event_family and must not jump the
# adoption gate as known_event_quality.


def _quality_encoded(event: dict[str, Any], domain: str) -> bool:
    summary = str(event.get("summary") or "")
    return any(hint in summary for hint in QUALITY_HINTS.get(domain, ()))


def _year_quality_encoded(events: Sequence[dict[str, Any]], domain: str, year: int) -> bool:
    return any(
        isinstance(event, dict)
        and str(event.get("domain") or "") == domain
        and _event_year(event) == year
        and _quality_encoded(event, domain)
        for event in events
    )


def _event_years(events: Sequence[dict[str, Any]], domain: str) -> set[int]:
    years: set[int] = set()
    for event in events:
        if not isinstance(event, dict) or str(event.get("domain") or "") != domain:
            continue
        year = _event_year(event)
        if year is not None:
            years.add(year)
    return years


def _existence_blocked_years(domain: str, known_years: set[int]) -> set[int]:
    blocked = set(known_years)
    span = int(EXISTENCE_NEARBY_YEARS.get(domain, 0))
    if span <= 0:
        return blocked
    for year in known_years:
        blocked.update(range(year - span, year + span + 1))
    return blocked


def _quality_probes(
    events: Sequence[dict[str, Any]],
    domains: Sequence[str],
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    seen_domains: set[str] = set()
    for event in events:
        if not isinstance(event, dict):
            continue
        domain = str(event.get("domain") or "")
        year = _event_year(event)
        if domain not in domains or year is None or domain in seen_domains:
            continue
        if domain not in DOMAIN_CATALOG or domain not in QUALITY_HINTS:
            continue
        if _year_quality_encoded(events, domain, year):
            continue
        seen_domains.add(domain)
        summary = str(event.get("summary") or "")
        family = str(DOMAIN_CATALOG[domain]["quality_family"])
        exam = domain == "education" and ("高考" in summary or "考试" in summary)
        rows.append(_public_probe(
            year=year,
            domain=domain,
            source="known_event_quality",
            tracks=("vimshottari", "narayana"),
            tracks_agree=True,
            user_meaning=_agent_brief(
                year_label=_year_label(year),
                domain=domain,
                family=family,
                quality=True,
                exam=exam,
            ),
            event_family=family,
        ))
    return rows


def _evaluate_contexts(
    contexts: Sequence[dict[str, Any]],
    *,
    birth_date: str,
    domain: str,
    year: int,
    source: str,
    month: int | None = None,
    clusters: Sequence[dict[str, Any]] | None = None,
    set_version: str | None = None,
) -> dict[str, Any] | None:
    scored_rows: list[tuple[str, list[str]]] = []
    month_value = month if isinstance(month, int) and 1 <= month <= 12 else None
    for context in contexts:
        time = _context_time(context)
        if not time:
            continue
        scored = _score_year(
            context,
            birth_date=birth_date,
            domain=domain,
            year=year,
            month=month_value,
        )
        if scored is None:
            continue
        scored_rows.append((time, list(scored.get("rule_ids") or [])))
    if len(scored_rows) < 2:
        return None
    yes: list[tuple[str, list[str]]] = []
    no: list[tuple[str, list[str]]] = []
    for time, rules in scored_rows:
        if _year_activated(rules):
            yes.append((time, rules))
        else:
            no.append((time, rules))
    if not yes or not no:
        ranks = [
            (time, LEVEL_RANK.get(match_level(rules), 0), rules)
            for time, rules in scored_rows
        ]
        highest = max(item[1] for item in ranks)
        lowest = min(item[1] for item in ranks)
        if highest - lowest < 2:
            return None
        yes = [(time, rules) for time, rank, rules in ranks if rank == highest]
        no = [(time, rules) for time, rank, rules in ranks if rank < highest]
        if not yes or not no:
            return None
    if not any(_discriminates(left, right) for _, left in yes for _, right in no):
        return None
    yes_times = sorted((time for time, _ in yes), key=_clock)
    no_times = sorted((time for time, _ in no), key=_clock)
    if clusters:
        yes_times = expand_times_through_clusters(yes_times, clusters)
        no_times = expand_times_through_clusters(no_times, clusters)
        overlap = set(yes_times) & set(no_times)
        if overlap:
            yes_times = [time for time in yes_times if time not in overlap]
            no_times = [time for time in no_times if time not in overlap]
        if len(yes_times) < 1 or len(no_times) < 1:
            return None
    yes_level = max((match_level(rules) for _, rules in yes), key=lambda item: LEVEL_RANK[item])
    no_level = min((match_level(rules) for _, rules in no), key=lambda item: LEVEL_RANK[item])
    stronger = next(rules for _, rules in yes if match_level(rules) == yes_level)
    vim_hit, narayana_hit = _tracks_present(stronger)
    gain = round(
        _group_entropy([len(yes_times), len(no_times)]) + _information_gain(yes_level, no_level),
        4,
    )
    if gain <= 0:
        return None
    version = set_version or candidate_set_version([yes_times, no_times])
    outcomes = [
        {"answer_class": "yes", "supports": yes_times, "conflicts": no_times},
        {"answer_class": "weak_yes", "supports": yes_times, "conflicts": no_times},
        {"answer_class": "no", "supports": no_times, "conflicts": yes_times},
        {"answer_class": "unsure", "supports": [], "conflicts": []},
    ]
    split = candidate_split_hash(
        candidate_set_version_value=version,
        domain=domain,
        year=year,
        month=month_value,
        groups=[yes_times, no_times],
    )
    period = _period_label(year, month_value)
    probe = _public_probe(
        year=year,
        month=month_value,
        domain=domain,
        source=source,
        tracks=("vimshottari", "narayana"),
        tracks_agree=vim_hit and narayana_hit,
        user_meaning=_agent_brief(
            year_label=period,
            domain=domain,
            family=str(DOMAIN_CATALOG[domain]["event_family"]),
        ),
        event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
        information_gain=gain,
        semantic_key=f"{domain}.{year}.{month_value:02d}.{source}" if month_value else f"{domain}.{year}.{source}",
        candidate_split_hash=split,
        candidate_set_version=version,
        expected_outcomes=outcomes,
        candidate_ids=candidate_ids_from_outcomes(outcomes),
        left_time=yes_times[0],
        right_time=no_times[0],
        source_features=[{
            "technique": source,
            "layers": list(SIGNATURE_LAYERS),
            "calculationResultId": None,
        }],
    )
    if not probe or distinguish_contract_errors(probe):
        return None
    return probe


def _evaluate_year(
    left: dict[str, Any],
    right: dict[str, Any],
    *,
    birth_date: str,
    domain: str,
    year: int,
    source: str,
) -> dict[str, Any] | None:
    return _evaluate_contexts(
        [left, right],
        birth_date=birth_date,
        domain=domain,
        year=year,
        source=source,
    )


def event_clarification_probes(
    request: dict[str, Any],
    *,
    domains: Sequence[str] | None = None,
) -> list[dict[str, Any]]:
    events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
    known = [
        str(event.get("domain"))
        for event in events
        if str(event.get("domain") or "") in DOMAIN_CATALOG
    ]
    return _quality_probes(events, domains or known)


def evidence_collection_probes(
    request: dict[str, Any],
    *,
    today: date | None = None,
) -> list[dict[str, Any]]:
    birth_year = _birth_year(request.get("birth_date"))
    if birth_year is None:
        return []
    now = today or date.today()
    events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
    missing = missing_collection_domains(events, list(DOMAIN_CATALOG), VOLUNTEER_ONLY)
    rows: list[dict[str, Any]] = []
    for domain in missing:
        year = _age_band_year(birth_year, domain, now)
        if year is None:
            continue
        rows.append(_public_probe(
            year=year,
            domain=domain,
            source="age_band",
            tracks=("vimshottari", "narayana"),
            tracks_agree=False,
            user_meaning=_agent_brief(
                year_label=_year_label(year),
                domain=domain,
                family=str(DOMAIN_CATALOG[domain]["event_family"]),
            ),
            event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
        ))
        if len(rows) >= MAX_COLLECTION_PROBES:
            break
    return rows


def candidate_contrast_opportunities(
    request: dict[str, Any],
    built: dict[str, Any],
    *,
    scan: dict[str, Any],
    candidate_times: Sequence[str],
    representative_time: str | None,
    today: date | None = None,
) -> list[dict[str, Any]]:
    probes = discriminating_event_probes(
        request,
        built,
        scan=scan,
        candidate_times=candidate_times,
        representative_time=representative_time,
        today=today,
    )
    return [opportunity_from_probe(probe) for probe in probes]


def _discriminating_event_probe_lists(
    request: dict[str, Any],
    built: dict[str, Any],
    *,
    scan: dict[str, Any],
    candidate_times: Sequence[str],
    representative_time: str | None,
    precision_current: str | None = None,
    today: date | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    del precision_current, representative_time
    birth_date = str(request.get("birth_date") or "").strip()
    birth_year = _birth_year(birth_date)
    empty: tuple[list[dict[str, Any]], list[dict[str, Any]]] = ([], [])
    if birth_year is None:
        return empty
    try:
        datetime.strptime(birth_date, "%Y-%m-%d")
    except ValueError:
        return empty
    events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
    asked_probe_keys = [
        str(item).strip()
        for item in (request.get("asked_probe_keys") or [])
        if isinstance(item, str) and str(item).strip()
    ]
    if not discriminator_gate_open(events):
        return empty
    holdout_keys = holdout_domain_years(events)
    now = today or date.today()
    full = _static_contexts(built)
    if len(full) < 2:
        return empty
    clusters = cluster_contexts_by_signature(full)
    if len(clusters) < 2:
        remaining = _remaining_contexts(built, candidate_times) or full
        clusters = cluster_contexts_by_signature(remaining)
    if len(clusters) < 2:
        return empty
    reps = [cluster["representative"] for cluster in clusters if _scoreable(cluster["representative"])]
    if len(reps) < 2:
        reps = [item for item in full if _scoreable(item)]
    if len(reps) < 2:
        return empty
    set_version = candidate_set_version([cluster["times"] for cluster in clusters])
    remaining_layers = _differing_layers(full)
    if not remaining_layers:
        remaining_layers = {
            layer for layer in SCORING_LAYERS
            if scan.get(f"{layer}_candidates_differ")
        }
    domains = _probe_domains(
        remaining_layers,
        events,
        d1_differs="d1" in remaining_layers or bool(scan.get("d1_candidates_differ")),
    )
    lo, hi = birth_year + 5, min(now.year, birth_year + 80)
    boundary_dates = _union_boundary_dates(reps, birth_date=birth_date, lo=lo, hi=hi) if domains else []
    probes: list[dict[str, Any]] = []
    for domain in domains:
        if domain not in DOMAIN_CATALOG:
            continue
        known_years = _event_years(events, domain) | asked_years_for_domain(asked_probe_keys, domain)
        blocked_years = _existence_blocked_years(domain, known_years)
        domain_lo = _domain_year_floor(birth_year, domain, lo)
        eligible = [
            item
            for item in boundary_dates
            if domain_lo <= item.year <= hi
            and item.year not in blocked_years
            and f"{domain}:{item.year}" not in holdout_keys
        ]
        found: list[dict[str, Any]] = []
        evaluated = 0
        sample_size = min(MAX_BOUNDARY_CANDIDATES_PER_DOMAIN, len(eligible))
        for at in _evaluation_order(eligible, MAX_BOUNDARY_CANDIDATES_PER_DOMAIN):
            if evaluated >= sample_size and found:
                break
            row = _evaluate_contexts(
                reps,
                birth_date=birth_date,
                domain=domain,
                year=at.year,
                month=at.month,
                source="dasha_boundary",
                clusters=clusters,
                set_version=set_version,
            )
            evaluated += 1
            if row is None or distinguish_contract_errors(row):
                continue
            if not isinstance(row.get("year"), int) or int(row["year"]) <= 0:
                continue
            found.append(row)
            if evaluated > sample_size:
                break
        kept = _best_probe_per_year(found)[:MAX_PROBES_PER_DOMAIN]
        if len(kept) < MAX_PROBES_PER_DOMAIN:
            activation = _try_activation_probe(
                reps=reps,
                birth_date=birth_date,
                birth_year=birth_year,
                domain=domain,
                now=now,
                blocked_years=blocked_years,
                holdout_keys=set(holdout_keys),
                clusters=clusters,
                set_version=set_version,
            )
            if activation is not None:
                activation_key = (
                    str(activation["domain"]),
                    int(activation["year"]),
                    int(activation.get("month") or 0),
                    str(activation["source"]),
                )
                existing = {
                    (str(row["domain"]), int(row["year"]), int(row.get("month") or 0), str(row["source"]))
                    for row in kept
                }
                if activation_key not in existing:
                    kept.append(activation)
        probes.extend(kept)
    probes.extend(_quality_distinguish_probes(
        events,
        clusters,
        set_version=set_version,
        holdout_ids=set(holdout_event_ids(events)),
        holdout_keys=set(holdout_keys),
    ))
    _annotate_nearby_ledger(probes, events)
    probes.sort(key=_probe_sort_key)
    public, dropped = _partition_ranked_probes(probes)
    assert_distinguish_contract(public)
    return public, dropped


def _partition_ranked_probes(
    probes: Sequence[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    public: list[dict[str, Any]] = []
    dropped: list[dict[str, Any]] = []
    seen: set[tuple[str, int, int, str]] = set()
    for row in probes:
        anchored_quality = (
            row.get("source") == "known_event_quality"
            and row.get("role") == "distinguish"
            and row.get("target_evidence_id")
        )
        if row.get("source") == "known_event_quality" and not anchored_quality:
            continue
        if row.get("phase") != PROBE_PHASE_CANDIDATE_DISCRIMINATOR:
            continue
        if distinguish_contract_errors(row):
            continue
        ranked = row if "raw_split_gain" in row else _apply_prior_ranking(dict(row))
        priors = ranked.get("answer_priors") or _answer_priors_for(ranked)
        if _dominant_existence_prior(ranked, priors):
            dropped.append({
                "semantic_key": ranked.get("semantic_key"),
                "reason": "dominant_answer_prior",
                "domain": ranked.get("domain"),
                "year": ranked.get("year"),
                "source": ranked.get("source"),
                "answer_priors": priors,
            })
            continue
        if not isinstance(row.get("year"), int) or int(row["year"]) <= 0:
            continue
        key = (str(ranked["domain"]), int(ranked["year"]), int(ranked.get("month") or 0), str(ranked["source"]))
        encoded = str(ranked)
        if key in seen or "points" in encoded:
            continue
        seen.add(key)
        public.append(ranked)
        if len(public) >= MAX_PROBES:
            break
    public.sort(key=_probe_sort_key)
    return public, dropped


def discriminating_event_probe_set(
    request: dict[str, Any],
    built: dict[str, Any],
    *,
    scan: dict[str, Any],
    candidate_times: Sequence[str],
    representative_time: str | None,
    precision_current: str | None = None,
    today: date | None = None,
) -> dict[str, list[dict[str, Any]]]:
    probes, dropped = _discriminating_event_probe_lists(
        request,
        built,
        scan=scan,
        candidate_times=candidate_times,
        representative_time=representative_time,
        precision_current=precision_current,
        today=today,
    )
    return {"probes": probes, "dropped": dropped}


def discriminating_event_probes(
    request: dict[str, Any],
    built: dict[str, Any],
    *,
    scan: dict[str, Any],
    candidate_times: Sequence[str],
    representative_time: str | None,
    precision_current: str | None = None,
    today: date | None = None,
) -> list[dict[str, Any]]:
    probes, _dropped = _discriminating_event_probe_lists(
        request,
        built,
        scan=scan,
        candidate_times=candidate_times,
        representative_time=representative_time,
        precision_current=precision_current,
        today=today,
    )
    return probes


def prospective_windows_for_time(
    request: dict[str, Any],
    built: dict[str, Any],
    time: str,
    *,
    today: date | None = None,
) -> list[dict[str, Any]]:
    """1–2 event windows in the next 12 months for one candidate minute."""
    birth_date = str(request.get("birth_date") or "").strip()
    clock = str(time or "")[:5]
    if not birth_date or len(clock) < 5:
        return []
    now = today or date.today()
    horizon = now + timedelta(days=366)
    by_time = {_context_time(item): item for item in _static_contexts(built)}
    context = by_time.get(clock)
    if not isinstance(context, dict):
        return []
    moon = (context.get("planet_longitudes") or {}).get("Moon")
    if not isinstance(moon, (int, float)):
        return []
    try:
        starts = [
            item
            for item in _vim_start_dates(
                birth_date,
                float(moon),
                now.year,
                horizon.year,
                include_pratyantar=True,
            )
            if now <= item <= horizon
        ]
    except (KeyError, TypeError, ValueError):
        return []
    rows: list[dict[str, Any]] = []
    seen: set[str] = set()
    for start in sorted(set(starts)):
        key = f"{start.year:04d}-{start.month:02d}"
        if key in seen:
            continue
        seen.add(key)
        rows.append({
            "domain": "career",
            "from": key,
            "to": key,
        })
        if len(rows) >= 2:
            break
    return rows


def prospective_windows_by_time(
    request: dict[str, Any],
    built: dict[str, Any],
    times: Sequence[str],
    *,
    today: date | None = None,
) -> dict[str, list[dict[str, Any]]]:
    out: dict[str, list[dict[str, Any]]] = {}
    for raw in times:
        clock = str(raw or "")[:5]
        if len(clock) < 5 or clock in out:
            continue
        out[clock] = prospective_windows_for_time(request, built, clock, today=today)
    return out


def prospective_event_windows(
    request: dict[str, Any],
    built: dict[str, Any],
    *,
    candidate_times: Sequence[str],
    today: date | None = None,
) -> list[dict[str, Any]]:
    birth_date = str(request.get("birth_date") or "").strip()
    if not birth_date:
        return []
    now = today or date.today()
    lo, hi = now.year + 1, now.year + 3
    full = _static_contexts(built)
    if len(full) < 2:
        return []
    clusters = cluster_contexts_by_signature(full)
    if len(clusters) < 2:
        remaining = _remaining_contexts(built, candidate_times) or full
        clusters = cluster_contexts_by_signature(remaining)
    if len(clusters) < 2:
        return []
    labels = ("A", "B", "C")
    rows: list[dict[str, Any]] = []
    for index, cluster in enumerate(clusters[:3]):
        context = cluster.get("representative") if isinstance(cluster, dict) else None
        if not isinstance(context, dict):
            continue
        moon = (context.get("planet_longitudes") or {}).get("Moon")
        if not isinstance(moon, (int, float)):
            continue
        starts = _vim_start_dates(birth_date, float(moon), lo, hi, include_pratyantar=True)
        future = [item for item in starts if item.year >= lo]
        if not future:
            continue
        start = min(future)
        label = labels[index]
        window = f"{start.year} 年 {start.month} 月附近"
        rows.append({
            "candidate_label": label,
            "domain": "career",
            "window_label": window,
            "user_meaning": (
                f"候选 {label} 预测下一次事业变动更可能在 {window}。"
                "这是预测窗口，不是承诺；下次发生时回来补一条，可进一步分辨。"
            ),
            "used_for_scoring": False,
        })
    return rows