Jyotisha/scripts/active_rectification_events.py

# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
# ─── How to run ───
# .venv/bin/python -m pytest -q tests/test_active_rectification_events.py
"""Deterministically adjudicate dated events against birth-time candidates."""

from __future__ import annotations

from collections.abc import Sequence
from typing import Any, Final, Literal, NotRequired, TypedDict, assert_never
from uuid import NAMESPACE_URL, uuid5

try:
    from scripts.rectification_policy import (
        MAX_CONFIRMATION_WIDTH_MINUTES,
        MIN_CONFIRMATION_DOMAINS,
        MIN_CONFIRMATION_EVENTS,
        MIN_CONFIRMATION_MARGIN_PERCENT,
    )
except ModuleNotFoundError:  # pragma: no cover - direct script execution
    from rectification_policy import (
        MAX_CONFIRMATION_WIDTH_MINUTES,
        MIN_CONFIRMATION_DOMAINS,
        MIN_CONFIRMATION_EVENTS,
        MIN_CONFIRMATION_MARGIN_PERCENT,
    )

EventPrecision = Literal["year", "month", "day"]
EventDomain = Literal[
    "education",
    "relocation",
    "relationship",
    "career",
    "finance",
    "health_pressure",
    "family",
    "appearance",
    "occupation",
]
Confidence = Literal["low", "medium", "high"]

ALGORITHM_VERSION: Final = "birth-time-event-scoring-v2"
PRECISION_WEIGHTS: Final[dict[EventPrecision, float]] = {
    "day": 1.0,
    "month": 0.8,
    "year": 0.5,
}


class CandidateEvidence(TypedDict):
    event_id: str
    domain: str
    candidate_time: str
    rule_ids: list[str]
    points: float


class LifeEvent(TypedDict):
    id: str
    domain: EventDomain
    event_kind: NotRequired[str]
    date: str
    precision: EventPrecision
    summary: NotRequired[str]


class RectificationEventRequest(TypedDict):
    birth_date: str
    start_time: str
    end_time: str
    lat: float
    lon: float
    tz: float
    ayanamsa: NotRequired[str]
    node_mode: NotRequired[str]
    minute_step: NotRequired[int]
    events: list[LifeEvent]


class CandidateScoreRow(TypedDict):
    time: str
    score: float
    evidence: list[CandidateEvidence]
    missing_layers: list[str]
    # Per scored event, the points of each evidence layer that was actually computed (comparability policy).
    layer_points: NotRequired[dict[str, dict[str, float]]]


class WinningSegment(TypedDict):
    start_time: str
    end_time: str
    representative_time: str
    width_minutes: int


class CandidateResult(TypedDict):
    result_id: str
    confidence: Confidence
    can_apply: bool
    winning_segment: WinningSegment | None
    event_count: int
    domain_count: int
    top_score: float
    second_score: float
    margin_percent: float
    reasons: list[str]
    evidence: list[CandidateEvidence]
    algorithm_version: str
    canonical_input_hash: str
    calculation_contract: dict[str, Any]
    stability_diagnostics: dict[str, Any]
    missing_layers: list[str]
    leader_status: NotRequired[str]
    provisional_leader: NotRequired[WinningSegment | None]
    comparable_leader: NotRequired[WinningSegment | None]
    coverage_comparison: NotRequired[dict[str, Any]]
    candidate_ranking_summary: NotRequired[list[dict[str, Any]]]
    candidate_summary: NotRequired[dict[str, Any]]


def build_candidate_result_summary(result: dict[str, Any]) -> dict[str, Any]:
    """Project a candidate result into stable, non-confirmatory next steps."""
    supported: dict[str, dict[str, Any]] = {}
    unconfirmed: dict[str, dict[str, Any]] = {}
    contradictory: dict[str, dict[str, Any]] = {}

    for item in result.get("evidence", []):
        if not isinstance(item, dict):
            continue
        domain = str(item.get("domain") or "unknown")
        points = float(item.get("points") or 0)
        bucket = supported if points > 0 else contradictory if points < 0 else unconfirmed
        row = bucket.setdefault(domain, {"domain": domain, "event_count": 0, "total_points": 0.0, "rule_ids": set()})
        row["event_count"] += 1
        row["total_points"] += points
        row["rule_ids"].update(str(rule_id) for rule_id in item.get("rule_ids", []))

    def rows(bucket: dict[str, dict[str, Any]], *, include_points: bool) -> list[dict[str, Any]]:
        output = []
        for row in bucket.values():
            item: dict[str, Any] = {
                "domain": row["domain"],
                "event_count": row["event_count"],
                "rule_ids": sorted(row["rule_ids"]),
            }
            if include_points:
                item["total_points"] = round(row["total_points"], 3)
            output.append(item)
        return sorted(output, key=lambda item: (-item.get("total_points", 0), item["domain"]))

    segment = result.get("winning_segment")
    next_steps: list[str] = []
    if int(result.get("event_count") or 0) < 5:
        next_steps.append("collect_at_least_five_events")
    if float(result.get("margin_percent") or 0) <= 0 or "tied_leader" in (result.get("reasons") or []):
        next_steps.append("resolve_candidate_tie_or_narrow_window")
    if isinstance(segment, dict) and int(segment.get("width_minutes") or 0) > 5:
        next_steps.append("narrow_window_before_minute_claim")
    next_steps.append("do_not_apply_as_birth_time_truth")
    return {
        "claim_status": "candidate_range_not_birth_time_truth",
        "candidate_range": segment,
        "supporting_evidence": rows(supported, include_points=True),
        "unconfirmed_evidence": rows(unconfirmed, include_points=False),
        "contradictory_evidence": rows(contradictory, include_points=True),
        "reasons": list(result.get("reasons") or []),
        "next_step_codes": next_steps,
    }


def independent_events(events: Sequence[LifeEvent]) -> tuple[list[LifeEvent], list[str]]:
    """Keep the first event of each (domain, date, precision); return the dropped ids.

    The score of an event depends only on those three fields, so a repeated entry is the same test
    re-run, not independent support (占星师 2026-11-08 验收清单 §6).
    """
    seen: set[tuple[str, str, str]] = set()
    kept: list[LifeEvent] = []
    duplicates: list[str] = []
    for event in events:
        key = (str(event["domain"]), str(event["date"]).strip(), str(event["precision"]))
        if key in seen:
            duplicates.append(str(event["id"]))
            continue
        seen.add(key)
        kept.append(event)
    return kept, duplicates


def precision_weight(precision: EventPrecision) -> float:
    """Return the fixed evidence weight for a declared date precision."""
    match precision:
        case "day" | "month" | "year":
            return PRECISION_WEIGHTS[precision]
        case unreachable:
            assert_never(unreachable)


def _minute_value(value: str) -> int:
    hour, minute = value.split(":", maxsplit=1)
    return int(hour) * 60 + int(minute)


def _is_next_minute(previous: str, current: str) -> bool:
    return (_minute_value(current) - _minute_value(previous)) % (24 * 60) == 1


def _top_segments(rows: Sequence[CandidateScoreRow], top_score: float) -> list[list[CandidateScoreRow]]:
    segments: list[list[CandidateScoreRow]] = []
    for row in rows:
        if row["score"] != top_score:
            continue
        if segments and _is_next_minute(segments[-1][-1]["time"], row["time"]):
            segments[-1].append(row)
        else:
            segments.append([row])
    return segments


def _winning_segment(rows: Sequence[CandidateScoreRow]) -> WinningSegment:
    width = len(rows)
    representative = rows[(width - 1) // 2]["time"]
    return {
        "start_time": rows[0]["time"],
        "end_time": rows[-1]["time"],
        "representative_time": representative,
        "width_minutes": width,
    }


def _clock_distance(left: str, right: str) -> int:
    distance = abs(_minute_value(left) - _minute_value(right))
    return min(distance, 24 * 60 - distance)


def _time_at_offset(value: str, offset: int) -> str:
    total = (_minute_value(value) + offset) % (24 * 60)
    return f"{total // 60:02d}:{total % 60:02d}"


def build_stability_diagnostics(
    rows: Sequence[CandidateScoreRow],
    *,
    winning_segment: WinningSegment | None,
) -> dict[str, Any]:
    """Describe exact-minute neighbor separation without claiming calibrated accuracy."""
    representative = winning_segment["representative_time"] if winning_segment else None
    by_time = {row["time"]: row for row in rows}
    representative_row = by_time.get(representative) if representative else None
    neighborhoods: list[dict[str, Any]] = []
    for radius in (1, 2, 5):
        required_times = {
            _time_at_offset(representative, -radius),
            _time_at_offset(representative, radius),
        } if representative else set()
        neighbors = [
            row for row in rows
            if representative is not None
            and 0 < _clock_distance(row["time"], representative) <= radius
        ]
        if representative_row is None or not required_times.issubset(by_time):
            neighborhoods.append({
                "radius_minutes": radius,
                "status": "blocked",
                "lead_points": None,
                "compared_candidate_count": len(neighbors),
                "reason": "candidate_range_does_not_cover_both_sides_of_neighborhood",
            })
            continue
        best_neighbor = max(row["score"] for row in neighbors)
        lead = round(representative_row["score"] - best_neighbor, 4)
        neighborhoods.append({
            "radius_minutes": radius,
            "status": "pass" if winning_segment["width_minutes"] == 1 and lead > 0 else "fail",
            "lead_points": lead,
            "compared_candidate_count": len(neighbors),
            "reason": (
                "unique_minute_leads_neighbor_candidates"
                if winning_segment["width_minutes"] == 1 and lead > 0
                else "minute_not_uniquely_separated_from_neighbors"
            ),
        })
    return {
        "scope": "candidate_neighbor_stability",
        "representative_time": representative,
        "neighborhoods": neighborhoods,
        "all_required_passed": all(item["status"] == "pass" for item in neighborhoods),
        "boundary": "Neighbor separation is a diagnostic only until thresholds are frozen before public holdout replay.",
    }


COMPARABILITY_POLICY: Final = "common_evidence_coverage_v1"
# Layers whose absence drops the whole event for that candidate (the event is not scored at all).
EVENT_GATING_LAYERS: Final[frozenset[str]] = frozenset({
    "Narayana", "Narayana_MD_AD", "Vimshottari_MD_AD_PD",
    "D2", "D3", "D4", "D6", "D7", "D8", "D9", "D10", "D11", "D12", "D24", "D30", "D60",
})


# 占星师原文层名 → 产品层名（只在「该候选整件事缺失」时适用，见 _explained_by_alias）。
ASTROLOGER_EVENT_LAYER_ALIASES: Final[dict[str, str]] = {
    "Narayana_AD": "Narayana_MD_AD",
}


def _layers_by_event(unit: dict[tuple[str, str], float]) -> dict[str, set[str]]:
    grouped: dict[str, set[str]] = {}
    for (event_id, layer) in unit:
        grouped.setdefault(event_id, set()).add(layer)
    return grouped


def _explained_by_alias(layer: str, rows: Sequence[CandidateScoreRow], units: Sequence[dict[tuple[str, str], float]]) -> bool:
    """占星师原文层名能不能当作产品层名来解释这一处差异（BUG-1827）。

    只有同时满足三条才算：该候选至少有一件事整件缺失；它自己覆盖单元里每一件事的层集合
    与共同覆盖一致（没有「事件在、但少了某层」）；它的缺失层里没有别的门控层。
    """
    if layer not in ASTROLOGER_EVENT_LAYER_ALIASES:
        return False
    holders = [index for index, row in enumerate(rows) if layer in row["missing_layers"]]
    if not holders:
        return False
    common = set.intersection(*[set(unit) for unit in units]) if units else set()
    common_by_event = _layers_by_event(common)
    all_events = {event for unit in units for (event, _) in unit}
    for index in holders:
        own_by_event = _layers_by_event(units[index])
        # ① 至少有一件事整件缺失（别的候选有、它没有）。
        if set(own_by_event) == all_events:
            return False
        # ② 它有的每一件事，层集合都必须与共同覆盖一致。
        if any(common_by_event.get(event) != layers for event, layers in own_by_event.items()):
            return False
        # ③ 缺失层里不能再有别的门控层。
        if any(item in EVENT_GATING_LAYERS and item != layer for item in rows[index]["missing_layers"]):
            return False
    return True


def _coverage_units(row: CandidateScoreRow) -> dict[tuple[str, str], float]:
    """(event, layer) -> points for every evidence unit this candidate actually computed.

    Rows that carry ``layer_points`` expose each layer; older rows expose one opaque unit per scored event.
    """
    explicit = row.get("layer_points")
    if isinstance(explicit, dict):
        return {
            (str(event_id), str(layer)): float(points)
            for event_id, layers in explicit.items()
            for layer, points in layers.items()
        }
    return {(str(item["event_id"]), "event_total"): float(item["points"]) for item in row["evidence"]}


def _excluded_dasha_contributions(rows: Sequence[CandidateScoreRow]) -> list[dict[str, str]]:
    """拆不开的 MD/AD 合分：只要有一个候选的该项 decomposable 为假，就对所有候选一起排除。"""
    excluded: dict[str, dict[str, str]] = {}
    for row in rows:
        for contribution in row.get("contributions") or []:
            if not isinstance(contribution, dict):
                continue
            if contribution.get("layer") != "dasha_varga_core":
                continue
            if contribution.get("decomposable") is not False:
                continue
            event_id = str(contribution.get("event_id") or "")
            if event_id:
                excluded[event_id] = {
                    "event_id": event_id,
                    "layer": "dasha_varga_core",
                    "reason": "md_ad_not_decomposable",
                }
    return [excluded[event_id] for event_id in sorted(excluded)]


def _explained_by_exclusion(layer: str, rows: Sequence[CandidateScoreRow],
                            excluded: Sequence[dict[str, str]]) -> bool:
    """差异层 L 是否已由「一致排除」解释（FIX-3）。

    每个带 L 的候选都必须有贡献明细、其 ``missing_reason == f"{L}_unresolved"`` 的项
    全部落在 ``excluded_contributions`` 里，并且该候选没有别的缺失层。
    没有 contributions 的旧形状行不适用。
    """
    covered = {(item["event_id"], item["layer"]) for item in excluded}
    holders = [row for row in rows if layer in row["missing_layers"]]
    if not holders:
        return False
    for row in holders:
        contributions = row.get("contributions")
        if not isinstance(contributions, list) or not contributions:
            return False
        unresolved = [
            item for item in contributions
            if isinstance(item, dict) and item.get("missing_reason") == f"{layer}_unresolved"
        ]
        if not unresolved:
            return False
        for item in unresolved:
            if (str(item.get("event_id") or ""), "dasha_varga_core") not in covered:
                return False
        if [other for other in row["missing_layers"] if other != layer]:
            return False
    return True


def missing_mandatory_layers(rows: Sequence[CandidateScoreRow]) -> list[dict[str, Any]]:
    """逐条列出「哪个候选、缺哪一层、影响哪几件事」（占星师第七轮 P1 第 3 条）。

    没有缺失时返回 ``[]``（任何情况下都不是 ``null``）；旧形状行没有贡献明细，
    ``event_ids`` 写 ``null`` 并标 ``event_ids_source``，不假装知道影响了哪几件事。
    """
    entries: list[dict[str, Any]] = []
    for row in rows:
        contributions = row.get("contributions")
        by_layer: dict[str, list[str]] = {}
        has_contributions = isinstance(contributions, list) and bool(contributions)
        for contribution in contributions or []:
            if not isinstance(contribution, dict):
                continue
            reason = contribution.get("missing_reason")
            layer = contribution.get("layer")
            if not isinstance(reason, str) or not isinstance(layer, str):
                continue
            if reason.startswith("event_not_scored:"):
                layer = reason.split(":", 1)[1]
            elif reason.endswith("_unresolved"):
                # dasha_varga_core 里「只缺小运」：缺失的层名在原因里（Narayana_AD_unresolved）。
                layer = reason[: -len("_unresolved")]
            else:
                continue
            event_id = str(contribution.get("event_id") or "")
            if event_id:
                by_layer.setdefault(layer, []).append(event_id)
        for layer in row["missing_layers"]:
            entry: dict[str, Any] = {"candidate_time": row["time"], "layer": layer}
            if layer in by_layer:
                entry["event_ids"] = sorted(set(by_layer[layer]))
            elif has_contributions:
                entry["event_ids"] = []
            else:
                entry["event_ids"] = None
                entry["event_ids_source"] = "legacy_row_without_contributions"
            entries.append(entry)
    entries.sort(key=lambda item: (item["candidate_time"], item["layer"]))
    return entries


def assess_leader_comparability(
    rows: Sequence[CandidateScoreRow],
    provisional_leader: WinningSegment | None,
) -> dict[str, Any]:
    """Explicit comparability policy (占星师 2026-11-08 验收清单 §6 / §9 / §10).

    Raw totals are only comparable when every candidate computed the same evidence units. When coverage
    differs, the raw-total leader stays ``provisional_leader`` (and keeps ``winning_segment``), and the
    candidates are re-ranked over the units that *every* candidate has; that leader is ``comparable_leader``.
    A unit one candidate lacks is dropped for all candidates, so it never counts as 0, negative or an old
    algorithm score, and a candidate cannot lead by computing less. When the differing layer is folded into
    another unit (no per-layer points), the comparison is ``not_reducible`` and there is no comparable leader.
    """
    units = [_coverage_units(row) for row in rows]
    row_layers = [set(row["missing_layers"]) for row in rows]
    unit_keys = [set(item) for item in units]
    equal = all(keys == unit_keys[0] for keys in unit_keys) and all(layers == row_layers[0] for layers in row_layers)
    common = set.intersection(*unit_keys) if unit_keys else set()
    base: dict[str, Any] = {
        "policy": COMPARABILITY_POLICY,
        "candidate_count": len(rows),
        "common_unit_count": len(common),
        "candidate_unit_counts": {
            "min": min((len(keys) for keys in unit_keys), default=0),
            "max": max((len(keys) for keys in unit_keys), default=0),
        },
    }
    separable = {layer for keys in unit_keys for _, layer in keys} - {"event_total"}
    differing = set.union(*row_layers) - set.intersection(*row_layers)
    # 占星师第七轮 P1 第 1 条：拆不开的 MD/AD 合分对所有候选一致排除，缺的不当 0、全的也不计。
    excluded_contributions = _excluded_dasha_contributions(rows)
    if excluded_contributions:
        dropped = {(item["event_id"], item["layer"]) for item in excluded_contributions}
        units = [{key: value for key, value in unit.items() if key not in dropped} for unit in units]
        unit_keys = [set(unit) for unit in units]
        common = set.intersection(*unit_keys) if unit_keys else set()
        base["common_unit_count"] = len(common)
        base["candidate_unit_counts"] = {
            "min": min((len(keys) for keys in unit_keys), default=0),
            "max": max((len(keys) for keys in unit_keys), default=0),
        }
        equal = all(keys == unit_keys[0] for keys in unit_keys) and all(layers == row_layers[0] for layers in row_layers)
    explained_aliases = {
        layer: ASTROLOGER_EVENT_LAYER_ALIASES[layer]
        for layer in differing
        if layer not in EVENT_GATING_LAYERS and layer not in separable
        and _explained_by_alias(layer, rows, units)
    }
    explained_by_exclusion = {
        layer for layer in differing
        if layer not in EVENT_GATING_LAYERS and layer not in separable
        and _explained_by_exclusion(layer, rows, excluded_contributions)
    }
    unexplained = sorted(
        layer for layer in differing
        if layer not in EVENT_GATING_LAYERS and layer not in separable
        and layer not in explained_aliases and layer not in explained_by_exclusion
    )
    if equal:
        return {
            **base,
            "status": "comparable",
            "comparison": "raw_totals",
            "candidates_with_extra_coverage": 0,
            **({"excluded_contributions": excluded_contributions} if excluded_contributions else {}),
            **({"layer_aliases_applied": explained_aliases} if explained_aliases else {}),
            "comparable_leader": provisional_leader,
            "leader_changed": False,
        }
    extra = sum(1 for keys in unit_keys if keys != common)
    if unexplained:
        return {
            **base,
            "status": "provisional_unequal_coverage",
            "comparison": "not_reducible",
            "unreducible_layers": unexplained,
            **({"excluded_contributions": excluded_contributions} if excluded_contributions else {}),
            "candidates_with_extra_coverage": extra,
            "comparable_leader": None,
            "leader_changed": None,
        }
    common_rows: list[CandidateScoreRow] = [
        {
            "time": row["time"],
            "score": round(sum(points for key, points in unit.items() if key in common), 4),
            "evidence": [],
            "missing_layers": [],
        }
        for row, unit in zip(rows, units, strict=True)
    ]
    common_top = max(row["score"] for row in common_rows)
    common_segments = _top_segments(common_rows, common_top)
    comparable = _winning_segment(common_segments[0]) if len(common_segments) == 1 else None
    return {
        **base,
        "status": "provisional_unequal_coverage",
        "comparison": "common_coverage",
        **({"layer_aliases_applied": explained_aliases} if explained_aliases else {}),
        **({"excluded_contributions": excluded_contributions} if excluded_contributions else {}),
        "candidates_with_extra_coverage": extra,
        "common_top_score": common_top,
        "common_scores": [{"time": row["time"], "score": row["score"]} for row in common_rows],
        "comparable_leader": comparable,
        "comparable_tied": len(common_segments) != 1,
        "leader_changed": (
            comparable is None
            or provisional_leader is None
            or (comparable["start_time"], comparable["end_time"])
            != (provisional_leader["start_time"], provisional_leader["end_time"])
        ),
    }


def adjudicate_candidate_rows(
    rows: Sequence[CandidateScoreRow],
    *,
    event_count: int,
    domain_count: int,
    request_fingerprint: str,
    canonical_input_hash: str = "",
    calculation_contract: dict[str, Any] | None = None,
    leave_one_event_out: dict[str, Any] | None = None,
) -> CandidateResult:
    """Rank precomputed candidate rows and apply conservative confidence gates."""
    if not rows:
        return {
            "result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{request_fingerprint}")),
            "confidence": "low",
            "can_apply": False,
            "winning_segment": None,
            "event_count": event_count,
            "domain_count": domain_count,
            "top_score": 0.0,
            "second_score": 0.0,
            "margin_percent": 0.0,
            "reasons": ["no_candidate_rows"],
            "evidence": [],
            "algorithm_version": ALGORITHM_VERSION,
            "canonical_input_hash": canonical_input_hash,
            "calculation_contract": calculation_contract or {},
            "stability_diagnostics": {
                "neighbor_stability": build_stability_diagnostics([], winning_segment=None),
                "leave_one_event_out": leave_one_event_out or {"status": "not_evaluated", "runs": []},
            },
            "missing_layers": [],
            "missing_mandatory_layers": [],
            "top_candidate_count": 0,
            "top_segment_count": 0,
            "unique_time": False,
            "leader_status": "no_leader",
            "provisional_leader": None,
            "comparable_leader": None,
            "coverage_comparison": {"policy": COMPARABILITY_POLICY, "status": "no_candidate_rows"},
        }

    ranked_scores = sorted({row["score"] for row in rows}, reverse=True)
    top_score = ranked_scores[0]
    second_score = ranked_scores[1] if len(ranked_scores) > 1 else top_score
    margin = round((top_score - second_score) / max(abs(top_score), 1.0) * 100, 2)
    segments = _top_segments(rows, top_score)
    missing_layers = sorted({layer for row in rows for layer in row["missing_layers"]})
    missing_entries = missing_mandatory_layers(rows)
    reasons: list[str] = []

    if len(segments) != 1:
        reasons.append("tied_leader")
    if event_count < 3:
        reasons.append("insufficient_events")
    if domain_count < 2:
        reasons.append("insufficient_domains")
    # 占星师第七轮 P1 第 3 条：这个 reason 当且仅当列表非空。
    if missing_entries:
        reasons.append("missing_mandatory_layers")

    segment = _winning_segment(segments[0]) if len(segments) == 1 else None
    comparability = assess_leader_comparability(rows, segment)
    if comparability["status"] != "comparable":
        reasons.append("unequal_evidence_coverage")
    if segment and segment["width_minutes"] > 15:
        reasons.append("winning_interval_too_wide")
    if margin < 10:
        reasons.append("lead_margin_below_medium_threshold")

    if reasons:
        confidence: Confidence = "low"
    elif (
        event_count >= MIN_CONFIRMATION_EVENTS
        and domain_count >= MIN_CONFIRMATION_DOMAINS
        and segment is not None
        and segment["width_minutes"] <= MAX_CONFIRMATION_WIDTH_MINUTES
        and margin >= MIN_CONFIRMATION_MARGIN_PERCENT
    ):
        confidence = "high"
    else:
        confidence = "medium"

    top_rows = segments[0] if len(segments) == 1 else []
    representative_row = top_rows[(len(top_rows) - 1) // 2] if top_rows else None
    evidence = list(representative_row["evidence"]) if representative_row else []
    neighbor_stability = build_stability_diagnostics(rows, winning_segment=segment)
    if not neighbor_stability["all_required_passed"]:
        reasons.append("neighbor_stability_not_passed")
    if (leave_one_event_out or {}).get("status") != "pass":
        reasons.append("leave_one_event_out_not_passed")
    can_apply = (
        confidence == "high"
        and segment is not None
        and neighbor_stability["all_required_passed"]
        and (leave_one_event_out or {}).get("status") == "pass"
        and not missing_layers
        and comparability["status"] == "comparable"
        and not any(reason in reasons for reason in (
            "unequal_evidence_coverage",
            "tied_leader",
            "insufficient_events",
            "insufficient_domains",
            "winning_interval_too_wide",
            "lead_margin_below_medium_threshold",
        ))
    )
    ranking_summary: list[dict[str, Any]] = []
    for score in ranked_scores[:3]:
        score_rows = sorted(
            (row for row in rows if row["score"] == score),
            key=lambda row: _minute_value(row["time"]),
        )
        if not score_rows:
            continue
        representative = score_rows[(len(score_rows) - 1) // 2]
        ranking_summary.append({
            "rank": len(ranking_summary) + 1,
            "time": representative["time"],
            "score": score,
            "tied_minute_count": len(score_rows),
        })
    # 占星师 2026-11-09 第七轮 P1 第 2 条：三个计数分开，代表分钟不能冒充唯一解。
    top_candidate_count = sum(1 for row in rows if row["score"] == top_score)
    top_segment_count = len(segments)
    return {
        "result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{request_fingerprint}")),
        "confidence": confidence,
        "can_apply": can_apply,
        "top_candidate_count": top_candidate_count,
        "top_segment_count": top_segment_count,
        "unique_time": top_candidate_count == 1,
        "winning_segment": segment,
        "event_count": event_count,
        "domain_count": domain_count,
        "top_score": top_score,
        "second_score": second_score,
        "margin_percent": margin,
        "reasons": reasons,
        "evidence": evidence,
        "algorithm_version": ALGORITHM_VERSION,
        "canonical_input_hash": canonical_input_hash,
        "calculation_contract": calculation_contract or {},
        "stability_diagnostics": {
            "neighbor_stability": neighbor_stability,
            "leave_one_event_out": leave_one_event_out or {"status": "not_evaluated", "runs": []},
        },
        "missing_layers": missing_layers,
        "missing_mandatory_layers": missing_entries,
        # winning_segment stays the raw-total leader; it is only "comparable" when coverage is equal.
        "leader_status": (
            "no_single_leader" if segment is None
            else "comparable" if comparability["status"] == "comparable"
            else "provisional_unequal_coverage"
        ),
        "provisional_leader": segment,
        "comparable_leader": comparability["comparable_leader"],
        "coverage_comparison": {key: value for key, value in comparability.items() if key != "comparable_leader"},
        "candidate_ranking_summary": ranking_summary,
    }


def score_life_events(request: RectificationEventRequest) -> CandidateResult:
    """Compute actual candidate rows, then apply the versioned confidence gates."""
    from scripts.active_rectification_event_engine import compute_event_candidate_result

    result = compute_event_candidate_result(request)
    result["candidate_summary"] = build_candidate_result_summary(result)
    return result