Jyotisha/tests/test_rectification_comparability_bug1801.py

"""BUG-1801: a rectification leader is only comparable when every candidate computed the same evidence.

占星师 2026-11-08 验收清单 §6 / §9 / §10: ``adjudicate_candidate_rows`` built ``winning_segment`` from raw
totals even when candidates had different evidence coverage, so a candidate could lead by "computing less,
losing fewer points". The raw-total leader is now ``provisional_leader``; ``comparable_leader`` re-ranks over
the evidence units every candidate has. Synthetic, anonymous rows only.
"""
from __future__ import annotations

from scripts.active_rectification_event_engine import adjudicate_event_candidate_rows
from scripts.active_rectification_events import (
    adjudicate_candidate_rows,
    assess_leader_comparability,
    independent_events,
)
from scripts.chart_choice import astrologer, astrologer_engine

DOMAINS = ["career", "relationship", "finance", "career", "relationship"]


def _row(time: str, points: dict[str, float], missing: list[str] | None = None) -> dict:
    return {
        "time": time,
        "score": round(sum(points.values()), 4),
        "evidence": [
            {"event_id": event_id, "domain": DOMAINS[int(event_id[1:]) - 1], "candidate_time": time,
             "rule_ids": ["synthetic"], "points": value}
            for event_id, value in points.items()
        ],
        "missing_layers": missing or [],
    }


def _adjudicate(rows: list[dict]) -> dict:
    return adjudicate_candidate_rows(
        rows, event_count=5, domain_count=3, request_fingerprint="synthetic-bug1801",
        leave_one_event_out={"status": "pass", "runs": []},
    )


def test_her_reproduction_12_00_is_only_a_provisional_leader() -> None:
    """Her case: 12:00 / 12:01, totals 10 / 8, the second lacks the Narayana AD for one event; 5 events, 3 domains."""
    full = _row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2})
    short = _row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_MD_AD"])

    result = _adjudicate([full, short])

    assert result["top_score"] == 10 and result["second_score"] == 8
    assert result["winning_segment"]["representative_time"] == "12:00"
    assert result["provisional_leader"] == result["winning_segment"]
    assert result["leader_status"] == "provisional_unequal_coverage"
    assert result["confidence"] == "low"
    assert result["can_apply"] is False
    assert "missing_mandatory_layers" in result["reasons"]
    assert "unequal_evidence_coverage" in result["reasons"]
    # Over the four events both candidates have, they tie: the comparable leader is the 2-minute segment,
    # not the single minute 12:00.
    assert result["comparable_leader"]["width_minutes"] == 2
    assert result["coverage_comparison"]["leader_changed"] is True
    comparison = result["coverage_comparison"]
    assert comparison["comparison"] == "common_coverage"
    assert comparison["common_unit_count"] == 4
    assert {row["time"]: row["score"] for row in comparison["common_scores"]} == {"12:00": 8, "12:01": 8}


def test_unequal_coverage_flips_the_leader_once_only_common_layers_are_compared() -> None:
    """12:00 never computed the event that penalises it, so it wins raw totals by computing less."""
    computed_less = _row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_MD_AD"])
    computed_all = _row("12:01", {"e1": 3, "e2": 3, "e3": 2, "e4": 2, "e5": -3})

    result = _adjudicate([computed_less, computed_all])

    assert result["provisional_leader"]["representative_time"] == "12:00"
    assert result["comparable_leader"]["representative_time"] == "12:01"
    assert result["coverage_comparison"]["leader_changed"] is True
    assert result["can_apply"] is False
    # The missing event is dropped for both candidates: it is neither 0 for 12:00 nor kept as -3 for 12:01.
    assert {row["time"]: row["score"] for row in result["coverage_comparison"]["common_scores"]} == {"12:00": 8, "12:01": 10}


def test_equal_coverage_keeps_the_existing_semantics() -> None:
    rows = [_row("12:00", {"e1": 3, "e2": 2, "e3": 2, "e4": 2, "e5": 2}),
            _row("12:01", {"e1": 1, "e2": 1, "e3": 1, "e4": 1, "e5": 1})]

    result = _adjudicate(rows)

    assert result["leader_status"] == "comparable"
    assert result["comparable_leader"] == result["winning_segment"] == result["provisional_leader"]
    assert "unequal_evidence_coverage" not in result["reasons"]
    assert result["coverage_comparison"]["comparison"] == "raw_totals"


def test_a_folded_auxiliary_layer_cannot_be_reduced_and_gives_no_comparable_leader() -> None:
    """Old rows fold Ashtakavarga into each event's points; one candidate lacking it is not reducible."""
    rows = [_row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2}, ["Ashtakavarga"]),
            _row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 1.8})]

    comparison = assess_leader_comparability(rows, None)

    assert comparison["status"] == "provisional_unequal_coverage"
    assert comparison["comparison"] == "not_reducible"
    assert comparison["unreducible_layers"] == ["Ashtakavarga"]
    assert comparison["comparable_leader"] is None


def test_per_layer_points_drop_an_auxiliary_layer_for_every_candidate() -> None:
    with_sav = _row("12:00", {"e1": 2.2, "e2": 2.2})
    with_sav["layer_points"] = {"e1": {"dasha_varga_core": 2, "Ashtakavarga": 0.2},
                                "e2": {"dasha_varga_core": 2, "Ashtakavarga": 0.2}}
    without_sav = _row("12:01", {"e1": 2.1, "e2": 2.0}, ["Ashtakavarga"])
    without_sav["layer_points"] = {"e1": {"dasha_varga_core": 2.1}, "e2": {"dasha_varga_core": 2.0}}

    comparison = assess_leader_comparability([with_sav, without_sav], None)

    assert comparison["comparison"] == "common_coverage"
    assert comparison["comparable_leader"]["representative_time"] == "12:01"
    assert {row["time"]: row["score"] for row in comparison["common_scores"]} == {"12:00": 4.0, "12:01": 4.1}


def test_a_repeated_fact_is_one_test_not_independent_support() -> None:
    events = [
        {"id": "a", "domain": "career", "date": "2019-07", "precision": "month"},
        {"id": "b", "domain": "career", "date": "2019-07", "precision": "month", "summary": "same entry"},
        {"id": "c", "domain": "relationship", "date": "2019-07", "precision": "month"},
    ]
    kept, duplicates = independent_events(events)
    assert [event["id"] for event in kept] == ["a", "c"] and duplicates == ["b"]

    request = {"birth_date": "2000-01-01", "start_time": "12:00", "end_time": "12:01",
               "lat": 0.0, "lon": 0.0, "tz": 0.0, "events": events}
    rows = [_row("12:00", {"e1": 2, "e3": 1}), _row("12:01", {"e1": 1, "e3": 1})]
    for row in rows:
        for item, event_id in zip(row["evidence"], ("a", "c"), strict=True):
            item["event_id"] = event_id
    result = adjudicate_event_candidate_rows(request, rows)

    assert result["event_count"] == 2
    assert result["duplicate_event_ids"] == ["b"]


def test_live_stage_dedupes_repeated_entries() -> None:
    events = [{"id": "x", "category": "career", "date": "2013-07"},
              {"id": "y", "category": "career", "date": "2013-07"},
              {"id": "z", "category": "marriage", "date": "2013-07"}]
    kept, duplicates = astrologer._independent_events(events)
    assert [event["id"] for event in kept] == ["x", "z"] and duplicates == ["y"]


def test_live_python_engine_records_a_missing_narayana_instead_of_scoring_it_zero(monkeypatch) -> None:
    """原值：live 阶段 Python 引擎 Narayana 定不出时该级按 0 分计、missing_layers 不记。
    新值：该事件不计分、记 Narayana_MD_AD，覆盖不同时领先只是暂列。原因：占星师 2026-11-08 验收清单 §6。"""
    body = {"lat": 39.9, "lon": 116.4, "tz": 8, "reported_at": "1990-05-03T08:05",
            "start_at": "1990-05-03T08:00", "end_at": "1990-05-03T08:10",
            "events": [{"category": "education", "date": "2008"}, {"category": "career", "date": "2013-07"},
                       {"category": "marriage", "date": "2016-10-01"}, {"category": "father", "date": "2019"},
                       {"category": "travel", "date": "2014"}]}
    real = astrologer._active_narayana

    def blocked_at_0805(asc, planets, birth_at, event_at, *tables):
        if birth_at.strftime("%H:%M") == "08:05" and event_at.year == 2013:
            return None, None
        return real(asc, planets, birth_at, event_at, *tables)

    monkeypatch.setattr(astrologer, "_active_narayana", blocked_at_0805)
    result = astrologer.evaluate(body)["result"]

    engine = result["python_engine"]
    assert "Narayana_MD_AD" in engine["missing_layers"]
    assert engine["leader_status"] in {"provisional_unequal_coverage", "no_single_leader"}
    assert "unequal_evidence_coverage" in engine["reasons"]
    assert result["leader_coverage"]["status"] == "provisional_unequal_coverage"
    assert result["leader_coverage"]["missing_layers"] == ["narayana_ad", "narayana_md"]


def test_live_python_engine_verdict_carries_the_two_leaders() -> None:
    events = [{"id": f"e{i}", "category": category, "date": "2010"} for i, category in
              enumerate(["career", "marriage", "finance_pos", "career", "marriage"], 1)]
    rows = [{**_row("08:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_MD_AD"]), "kp_cusps": {"1": {}}},
            {**_row("08:01", {"e1": 3, "e2": 3, "e3": 2, "e4": 2, "e5": -3}), "kp_cusps": {"1": {}}}]

    verdict = astrologer_engine.verdict(events, rows)

    assert verdict["top_time"] == "08:00"
    assert verdict["leader_status"] == "provisional_unequal_coverage"
    assert verdict["comparable_top_time"] == "08:01"
    assert verdict["can_apply"] is False
    assert "各候选分钟算出的层不一样，总分不能直接比" in verdict["reasons_cn"]