"""Memoization for candidate-minute invariants in the rectification engine.

Historical golden tests/golden/rectification_engine_memoization_v1.json was
produced from origin/staging @ a8d29d1b before any memoization landed.
On 2026-09-20 only two identity leaves were refreshed from the real engine:
representative_candidate_id 85ca5490-07ab-54b7-acf0-59385f705015 ->
e8c11277-022c-519c-95e9-6a0a04e8bf23, and snapshot algorithm_version
rectification-v5-matrix-scoring-7 -> rectification-v5-matrix-scoring-8.
The authorized cross-midnight version bump changes UUID identity, not this
same-day fixture's scores. Historical numeric values and feature hashes remain.
That file is now frozen (sha256 pinned below) and no longer the comparison
target: scoring-8/9 scores belong to the old functional benefic / malefic
grouping.

Current golden tests/golden/rectification_engine_memoization_v2.json
(2026-10-02, BUG-1181) was written by ``write_golden`` from the real engine at
57782aea (functional profile bphs_ch34_general_with_sign_exceptions_v2) with
ALGORITHM_VERSION rectification-v5-matrix-scoring-10. Scores move for the same
input (candidate 12:00: 8.6274 -> 8.4977), which is why the algorithm identity
was bumped; it already carries the dated-v1 receipt metadata. Never rewrite it
in place: a future scoring change adds a v3 file and freezes this one.

Current golden tests/golden/rectification_engine_memoization_v3.json
(2026-10-02, BUG-1193, upstream sync 5) was written by ``write_golden`` after
R1 raised the range floor to 4 dated events / 3 domains. Candidate scores are
identical to v2 (the ported combustion / Chara Karaka / Yogini / Shodhana
changes do not feed the event matrix; 77 v5 cases A/B byte-identical), so the
algorithm stays scoring-10. Only the decision receipt moved: policy
rectification-candidate-policy-v3 -> v4, event_quality / domain_diversity
minimum 3/2 -> 4/3, and the representative candidate UUID (its namespace
carries the policy version). v2 is frozen by sha256 below.

Do not compare that payload with a whole-structure ``==``. Cross-machine
libm / pyswisseph rounding already drifted ``margin_percent`` by 1.1e-3
(TASK-rectification-engine-memoization-fix-20260915). Equivalence of the
four cached layers is proven in-process instead.
"""

from __future__ import annotations

import hashlib
import inspect
import json
import math
from datetime import date, datetime
from pathlib import Path
from typing import Any
from unittest.mock import patch
from uuid import NAMESPACE_URL, uuid5

import pytest

from scripts.active_rectification_event_engine import (
    _candidate_datetimes,
    _controlled_transit_rules,
    _shadbala_verified_components_auxiliary,
    build_candidate_static_context,
    compute_event_candidate_rows,
)
from scripts.rectification.api_service import score_candidates
from scripts.rectification.candidate_contrast import opportunity_from_probe
from scripts.rectification.contracts import normalize_rectification_request
from scripts.rectification.refinement_packet import build_refinement_packet
from scripts.rectification.scoring_service import sample_event_dates, scoreable_request
import scripts.active_rectification_event_engine as event_engine
import scripts.rectification.event_probes as event_probes
import scripts.rectification.scoring_service as scoring_service
import shadbala

ROOT = Path(__file__).resolve().parents[1]
GOLDEN_PATH = ROOT / "tests" / "golden" / "rectification_engine_memoization_v3.json"
PREVIOUS_GOLDEN_PATH = ROOT / "tests" / "golden" / "rectification_engine_memoization_v2.json"
PREVIOUS_GOLDEN_SHA256 = "abb15efae4d86387d930cd463b82a22fe95ccd1779e85ca6cc4cdf07a063cb73"
PREVIOUS_SOURCE_COMMIT = "57782aea8ae28f5dd165a08d221ec27dbef391f8"
HISTORICAL_GOLDEN_PATH = ROOT / "tests" / "golden" / "rectification_engine_memoization_v1.json"
HISTORICAL_GOLDEN_SHA256 = "6266e448d7dad204b264cbe8f9bcf4c36e02286794764a94edb9e2ce4d0e3015"
CURRENT_ALGORITHM_VERSION = "rectification-v5-matrix-scoring-10"
FROZEN_TODAY = date(2026, 9, 16)
TIMING_KEYS = frozenset({"column_compare_ms"})
HISTORICAL_SOURCE_COMMIT = "a8d29d1b6cc37ff865ddec6c8bccdf9aa889ee53"
CURRENT_POLICY_VERSION = "rectification-candidate-policy-v4"
# Engine code of the v3 golden: upstream-sync5 tip before R1; the policy-v4
# receipt (R1, BUG-1193) is the commit that adds the v3 file on top.
SOURCE_COMMIT = "7fb165ecf66ba1726f2fc74ffa2353fadc65ba39"
CACHE_LAYER_KEYS = (
    "ashtakavarga_result",
    "shadbala_result",
    "vimshottari_timeline",
    "narayana_periods",
)
# TASK-rectification-engine-memoization-fix-20260915 §2 measured max
# cross-machine drift of 1.1e-3 on decision_receipt.margin_percent
# (5.2995 vs golden 5.3006). Candidate scores, already rounded to 4
# decimals, drifted by 1 ulp (1.0e-4). Tolerance is the wider of
# abs=2e-3 and rel=5e-4 so 1.1e-3 still passes while a 1e-2 mutation
# (the reverse test below) fails.
GOLDEN_FLOAT_ABS = 2e-3
GOLDEN_FLOAT_REL = 5e-4


def public_score_request() -> dict[str, Any]:
    """Fictional events on the public 1990-01-01 Beijing smoke chart."""
    return {
        "birth_date": "1990-01-01",
        "start_time": "12:00",
        "end_time": "12:02",
        "lat": 39.9,
        "lon": 116.4,
        "tz": 8,
        "events": [
            {
                "id": "00000000-0000-4000-8000-000000000001",
                "domain": "education",
                "event_kind": "education_start",
                "date_start": "2008-01-01",
                "date_end": "2008-12-31",
                "precision": "year",
                "summary": "入学",
            },
            {
                "id": "00000000-0000-4000-8000-000000000002",
                "domain": "career",
                "event_kind": "career_entry",
                "date_start": "2012-06-01",
                "date_end": "2012-06-30",
                "precision": "month",
                "summary": "入职",
            },
        ],
    }


def _strip_timing(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            key: _strip_timing(item)
            for key, item in value.items()
            if key not in TIMING_KEYS
        }
    if isinstance(value, list):
        return [_strip_timing(item) for item in value]
    return value


def _normalized_request() -> dict[str, Any]:
    return normalize_rectification_request(public_score_request(), today=FROZEN_TODAY)


class _FrozenDate(date):
    """`event_probes` reads `date.today()` for its year windows; the golden
    must not depend on the wall clock (BUG-733 family)."""

    @classmethod
    def today(cls) -> date:
        return FROZEN_TODAY


def _golden_payload() -> dict[str, Any]:
    with patch.object(event_probes, "date", _FrozenDate):
        scored = score_candidates(_normalized_request())
    return {
        "source_commit": SOURCE_COMMIT,
        "candidate_scores": scored["candidate_scores"],
        "decision_receipt": _strip_timing(scored["decision_receipt"]),
        "candidate_feature_snapshot": scored["candidate_feature_snapshot"],
    }


def write_golden(path: Path = GOLDEN_PATH) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(_golden_payload(), ensure_ascii=True, indent=2) + "\n", encoding="utf-8")
    return path


def _count_calls(monkeypatch, owner: Any, name: str, *, from_engine: bool = False) -> list[int]:
    original = getattr(owner, name)
    counter = [0]

    def wrapped(*args: Any, **kwargs: Any) -> Any:
        if from_engine:
            caller = inspect.stack()[1].filename.replace("\\", "/")
            if not caller.endswith("/active_rectification_event_engine.py"):
                return original(*args, **kwargs)
        counter[0] += 1
        return original(*args, **kwargs)

    monkeypatch.setattr(owner, name, wrapped)
    return counter


def _is_json_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _assert_tiered_equal(actual: Any, expected: Any, *, path: str) -> None:
    if isinstance(expected, dict):
        assert isinstance(actual, dict), path
        assert set(actual) == set(expected), f"{path} keys {set(actual)!r} != {set(expected)!r}"
        for key in expected:
            _assert_tiered_equal(actual[key], expected[key], path=f"{path}.{key}")
        return
    if isinstance(expected, list):
        assert isinstance(actual, list), path
        assert len(actual) == len(expected), f"{path} length {len(actual)} != {len(expected)}"
        for index, (left, right) in enumerate(zip(actual, expected)):
            _assert_tiered_equal(left, right, path=f"{path}[{index}]")
        return
    if isinstance(expected, int) and isinstance(actual, int) and not isinstance(expected, bool) and not isinstance(actual, bool):
        assert actual == expected, f"{path}: {actual!r} != {expected!r}"
        return
    if _is_json_number(expected) or _is_json_number(actual):
        left = float(actual)
        right = float(expected)
        tolerance = max(GOLDEN_FLOAT_ABS, GOLDEN_FLOAT_REL * abs(right))
        assert math.isclose(left, right, rel_tol=0.0, abs_tol=tolerance), (
            f"{path}: {left!r} vs {right!r} exceeds {tolerance}"
        )
        return
    assert actual == expected, f"{path}: {actual!r} != {expected!r}"


def _assert_memoization_payloads(actual: dict[str, Any], expected: dict[str, Any]) -> None:
    _assert_tiered_equal(actual["candidate_scores"], expected["candidate_scores"], path="candidate_scores")
    _assert_tiered_equal(actual["decision_receipt"], expected["decision_receipt"], path="decision_receipt")


def _event_engine_request() -> dict[str, Any]:
    """Dated events for compute_event_candidate_rows (not the v5 score payload)."""
    return {
        "birth_date": "1990-01-01",
        "start_time": "12:00",
        "end_time": "12:02",
        "lat": 39.9,
        "lon": 116.4,
        "tz": 8,
        "events": [
            {
                "id": "00000000-0000-4000-8000-000000000001",
                "domain": "education",
                "event_kind": "education_start",
                "date": "2008-07-01",
                "precision": "day",
                "summary": "入学",
            },
            {
                "id": "00000000-0000-4000-8000-000000000002",
                "domain": "career",
                "event_kind": "career_entry",
                "date": "2012-06-15",
                "precision": "day",
                "summary": "入职",
            },
        ],
    }


def _contexts_with_layer_cache_cleared(contexts: list[dict[str, Any]]) -> list[dict[str, Any]]:
    cleared: list[dict[str, Any]] = []
    for context in contexts:
        clone = dict(context)
        for key in CACHE_LAYER_KEYS:
            clone[key] = None
        cleared.append(clone)
    return cleared


def _assert_dated_payload_against_golden(actual: dict[str, Any], expected: dict[str, Any]) -> None:
    """Compare with the scoring-10 / policy-v4 golden; identity is spelled out, not trusted."""
    metadata = {
        "candidate_window_contract": "dated-v1",
        "candidate_intervals": [{"start_at": "1990-01-01T12:00", "end_at": "1990-01-01T12:02"}],
        "candidate_timezone_offset": 8,
        "candidate_timezone_id": "",
    }
    receipt = actual["decision_receipt"]
    golden_receipt = expected["decision_receipt"]
    assert set(receipt) == set(golden_receipt), f"receipt keys {set(receipt) ^ set(golden_receipt)!r}"
    for key, value in metadata.items():
        assert receipt[key] == value, key
        assert golden_receipt[key] == value, key
    assert expected["candidate_feature_snapshot"]["algorithm_version"] == CURRENT_ALGORITHM_VERSION
    assert actual["candidate_feature_snapshot"]["algorithm_version"] == CURRENT_ALGORITHM_VERSION
    # Independently spell out the result/candidate UUID namespace formula. Do not
    # call the production identity helper or accept any arbitrary UUID string.
    identity_request = {
        key: value for key, value in _normalized_request().items()
        if key not in {"asked_probe_keys", "dropped_asked_probe_keys", "declined_domains", "column_times", "refresh_probes"}
    }
    fingerprint = hashlib.sha256(json.dumps(
        identity_request, ensure_ascii=True, sort_keys=True, separators=(",", ":"),
    ).encode()).hexdigest()
    result_id = uuid5(NAMESPACE_URL, f"{CURRENT_ALGORITHM_VERSION}:{fingerprint}")
    representative_time = golden_receipt["representative_time"]
    candidate_id = uuid5(NAMESPACE_URL, f"{CURRENT_POLICY_VERSION}:{result_id}:{representative_time}")
    assert golden_receipt["representative_candidate_id"] == str(candidate_id)
    assert receipt["representative_candidate_id"] == str(candidate_id)
    # BUG-733/985: fingerprints hash unrounded floats; do not turn the
    # snapshot into a cross-process exact-hash contract.
    _assert_memoization_payloads(actual, expected)


def test_historical_scoring_8_golden_is_frozen_and_no_longer_current() -> None:
    """BUG-1181: the old golden stays byte-identical; current scores moved, hence scoring-10."""
    raw = HISTORICAL_GOLDEN_PATH.read_bytes()
    assert hashlib.sha256(raw).hexdigest() == HISTORICAL_GOLDEN_SHA256
    historical = json.loads(raw)
    assert historical["source_commit"] == HISTORICAL_SOURCE_COMMIT
    assert historical["candidate_feature_snapshot"]["algorithm_version"] == "rectification-v5-matrix-scoring-8"
    current = json.loads(GOLDEN_PATH.read_text(encoding="utf-8"))
    assert current["source_commit"] == SOURCE_COMMIT
    with pytest.raises(AssertionError):
        _assert_memoization_payloads(current, historical)


def test_previous_policy_v3_golden_is_frozen_with_identical_scores() -> None:
    """BUG-1193: v2 stays byte-identical; scores equal v3 (no scoring-11), only the receipt policy moved."""
    raw = PREVIOUS_GOLDEN_PATH.read_bytes()
    assert hashlib.sha256(raw).hexdigest() == PREVIOUS_GOLDEN_SHA256
    previous = json.loads(raw)
    current = json.loads(GOLDEN_PATH.read_text(encoding="utf-8"))
    assert previous["source_commit"] == PREVIOUS_SOURCE_COMMIT
    assert previous["candidate_feature_snapshot"]["algorithm_version"] == CURRENT_ALGORITHM_VERSION
    assert current["candidate_feature_snapshot"]["algorithm_version"] == CURRENT_ALGORITHM_VERSION
    _assert_tiered_equal(current["candidate_scores"], previous["candidate_scores"], path="candidate_scores")
    assert previous["decision_receipt"]["policy_version"] == "rectification-candidate-policy-v3"
    assert current["decision_receipt"]["policy_version"] == CURRENT_POLICY_VERSION
    assert previous["decision_receipt"]["gates"]["event_quality"]["minimum"] == 3
    assert current["decision_receipt"]["gates"]["event_quality"]["minimum"] == 4
    assert previous["decision_receipt"]["gates"]["domain_diversity"]["minimum"] == 2
    assert current["decision_receipt"]["gates"]["domain_diversity"]["minimum"] == 3
    with pytest.raises(AssertionError):
        _assert_memoization_payloads(current, previous)


def test_score_candidates_matches_baseline_golden() -> None:
    expected = json.loads(GOLDEN_PATH.read_text(encoding="utf-8"))
    actual = json.loads(json.dumps(_golden_payload(), ensure_ascii=True))
    _assert_dated_payload_against_golden(actual, expected)


@pytest.mark.parametrize("field,value", [
    ("candidate_window_contract", "dated-v0"),
    ("candidate_intervals", [{"start_at": "1990-01-02T12:00", "end_at": "1990-01-02T12:02"}]),
    ("candidate_timezone_offset", 9),
    ("candidate_timezone_id", "UTC"),
    ("representative_candidate_id", "00000000-0000-4000-8000-000000000001"),
    ("unapproved_metadata", True),
    ("confirmation_allowed", True),
])
def test_dated_golden_projection_rejects_metadata_identity_and_gate_mutations(field, value) -> None:
    expected = json.loads(GOLDEN_PATH.read_text(encoding="utf-8"))
    actual = json.loads(json.dumps(_golden_payload(), ensure_ascii=True))
    _assert_dated_payload_against_golden(actual, expected)
    assert actual["decision_receipt"].get(field) != value
    actual["decision_receipt"][field] = value
    with pytest.raises(AssertionError):
        _assert_dated_payload_against_golden(actual, expected)


def test_golden_float_shift_of_1e_minus_2_fails() -> None:
    expected = json.loads(GOLDEN_PATH.read_text(encoding="utf-8"))
    mutated = json.loads(json.dumps(expected))
    mutated["candidate_scores"][0]["score"] += 1e-2
    with pytest.raises(AssertionError):
        _assert_memoization_payloads(expected, mutated)


def test_golden_discrete_field_mismatch_fails() -> None:
    expected = json.loads(GOLDEN_PATH.read_text(encoding="utf-8"))
    mutated = json.loads(json.dumps(expected))
    mutated["candidate_scores"][0]["time"] = "99:99"
    with pytest.raises(AssertionError):
        _assert_memoization_payloads(expected, mutated)


def test_shadbala_verified_fields_match_with_and_without_birth_minute() -> None:
    request = {
        "birth_date": "1990-01-01",
        "start_time": "12:17",
        "end_time": "12:17",
        "lat": 39.9,
        "lon": 116.4,
        "tz": 8,
        "events": [],
    }
    candidate_at = datetime(1990, 1, 1, 12, 17)
    context = build_candidate_static_context(request, candidate_at)
    chart = context["chart"]
    planets = chart.get("planets", {})
    sun = float(planets["Sun"]["lon"])
    moon = float(planets["Moon"]["lon"])
    sign = str(chart["ascendant"].get("sign"))
    birth_hour = candidate_at.hour + candidate_at.minute / 60
    with_minute = shadbala.calc_shadbala(
        planets,
        sign,
        birth_hour,
        sun,
        moon,
        birth_minute=float(candidate_at.minute),
    )
    without_minute = shadbala.calc_shadbala(planets, sign, birth_hour, sun, moon)
    fields_with = {
        planet: (
            float((row.get("sthana_bala") or {}).get("total", 0)),
            float(row.get("drik_bala", 0)),
            float(row.get("naisargika_bala", 0)),
        )
        for planet, row in (with_minute.get("planets") or {}).items()
    }
    fields_without = {
        planet: (
            float((row.get("sthana_bala") or {}).get("total", 0)),
            float(row.get("drik_bala", 0)),
            float(row.get("naisargika_bala", 0)),
        )
        for planet, row in (without_minute.get("planets") or {}).items()
    }
    assert fields_with == fields_without
    reused_rules, reused_points = _shadbala_verified_components_auxiliary(
        chart, birth_hour, ("Sun", "Moon", "Mars"), shadbala_result=with_minute,
    )
    fresh_rules, fresh_points = _shadbala_verified_components_auxiliary(
        chart, birth_hour, ("Sun", "Moon", "Mars"),
    )
    assert reused_rules == fresh_rules
    assert reused_points == fresh_points


def test_shadbala_and_ashtakavarga_run_once_per_candidate(monkeypatch) -> None:
    shadbala_calls = _count_calls(monkeypatch, event_engine.shadbala, "calc_shadbala")
    ashtakavarga_calls = _count_calls(monkeypatch, event_engine.ashtakavarga, "calc_ashtakavarga")
    request = _normalized_request()
    scored = score_candidates(request)
    candidate_count = len(scored["candidate_scores"])
    assert candidate_count >= 2
    assert len(request["events"]) >= 2
    year_samples = sample_event_dates(request["events"][0])
    assert request["events"][0]["precision"] == "year"
    assert len(year_samples) >= 2
    assert shadbala_calls[0] == candidate_count
    assert ashtakavarga_calls[0] == candidate_count


def test_dasha_timelines_run_once_per_candidate(monkeypatch) -> None:
    # Year-only events skip dasha-transition proximity, which also calls these
    # two functions from event_probes. The scoring path itself must stay 1×/minute.
    body = public_score_request()
    body["events"][1]["precision"] = "year"
    body["events"][1]["date_start"] = "2012-01-01"
    body["events"][1]["date_end"] = "2012-12-31"
    request = normalize_rectification_request(body, today=FROZEN_TODAY)
    vim_calls = _count_calls(
        monkeypatch, event_engine.dasha_analyzer, "build_dasha_timeline", from_engine=True,
    )
    narayana_calls = _count_calls(
        monkeypatch, event_engine.narayana_dasha, "calc_narayana_mahadasha", from_engine=True,
    )
    scored = score_candidates(request)
    candidate_count = len(scored["candidate_scores"])
    assert len(request["events"]) >= 2
    assert vim_calls[0] == candidate_count
    assert narayana_calls[0] == candidate_count


def test_transit_charts_match_unique_event_dates_not_candidates(monkeypatch) -> None:
    transit_calls = [0]
    original = event_engine.domain_calculation_service.compute_chart

    def wrapped(payload: dict[str, Any], *args: Any, **kwargs: Any) -> Any:
        # Natal candidates in this fixture also sit at 12:00; transit charts use the event year.
        if payload.get("hour") == 12 and payload.get("minute") == 0 and payload.get("year") != 1990:
            transit_calls[0] += 1
        return original(payload, *args, **kwargs)

    monkeypatch.setattr(event_engine.domain_calculation_service, "compute_chart", wrapped)
    request = _normalized_request()
    scored = score_candidates(request)
    candidate_count = len(scored["candidate_scores"])
    expected_dates = {
        sampled
        for event in scoreable_request(request)["events"]
        for sampled in sample_event_dates(event)
    }
    assert candidate_count >= 2
    assert len(expected_dates) >= 2
    assert transit_calls[0] == len(expected_dates)
    assert transit_calls[0] != candidate_count * len(expected_dates)


def test_year_precision_transits_still_short_circuit() -> None:
    request = {
        "birth_date": "1990-01-01",
        "start_time": "12:00",
        "end_time": "12:00",
        "lat": 39.9,
        "lon": 116.4,
        "tz": 8,
        "events": [{
            "id": "00000000-0000-4000-8000-000000000009",
            "domain": "career",
            "event_kind": "career_entry",
            "date": "2012",
            "precision": "year",
            "summary": "入职",
        }],
    }
    with patch.object(event_engine.domain_calculation_service, "compute_chart") as compute_chart:
        rules = _controlled_transit_rules(request, request["events"][0], 0, (10,))
    assert rules == []
    compute_chart.assert_not_called()


def _probe_request_and_built() -> tuple[dict[str, Any], dict[str, Any], list[str]]:
    body = public_score_request()
    body["events"] = [
        {
            "id": f"00000000-0000-4000-8000-{index:012d}",
            "domain": domain,
            "event_kind": kind,
            "date_start": "2012-01-01",
            "date_end": "2012-12-31",
            "precision": "year",
            "summary": kind,
        }
        for index, (domain, kind) in enumerate(
            (
                ("education", "education_start"),
                ("career", "career_entry"),
                ("relationship", "relationship_start"),
            ),
            start=1,
        )
    ]
    request = normalize_rectification_request(body, today=FROZEN_TODAY)
    times = ["12:00", "12:01", "12:02"]
    event_ids = [event["id"] for event in request["events"]]
    built = {
        "candidate_times": times,
        "matrix": {
            event_id: {
                clock: {"points": 4, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]}
                for clock in times
            }
            for event_id in event_ids
        },
        "date_sensitivity": [],
        "missing_layers": [],
        "static_contexts": [],
    }
    return request, built, times


def test_discriminating_probes_run_once_when_probe_times_equal_grid(monkeypatch) -> None:
    calls = _count_calls(monkeypatch, event_probes, "_discriminating_event_probe_lists")
    request, built, times = _probe_request_and_built()
    packet = build_refinement_packet(
        request,
        built,
        representative_time="12:00",
        candidate_times=times,
    )
    assert calls[0] == 1
    assert packet["candidate_contrast_opportunities"] == [
        opportunity_from_probe(probe) for probe in packet["discriminating_event_probes"]
    ]


def test_discriminating_probes_run_twice_on_refresh_columns(monkeypatch) -> None:
    calls = _count_calls(monkeypatch, event_probes, "_discriminating_event_probe_lists")
    request, built, times = _probe_request_and_built()
    request = {**request, "refresh_probes": True}
    packet = build_refinement_packet(
        request,
        built,
        representative_time="12:00",
        candidate_times=times,
        column_times=["12:00"],
    )
    assert times != ["12:00"]
    assert calls[0] == 2
    assert "discriminating_event_probes" in packet
    assert "candidate_contrast_opportunities" in packet


def test_dead_row_cache_removed_from_scoring_service() -> None:
    token = "_cached_" + "rows"
    assert not hasattr(scoring_service, token)
    hits: list[str] = []
    for folder in ("scripts", "tests"):
        for path in (ROOT / folder).rglob("*.py"):
            text = path.read_text(encoding="utf-8")
            if token in text:
                hits.append(str(path.relative_to(ROOT)).replace("\\", "/"))
    assert hits == []


def test_compute_event_candidate_rows_reuses_static_context_without_mutating_it() -> None:
    request = {
        "birth_date": "1990-01-01",
        "start_time": "12:00",
        "end_time": "12:01",
        "lat": 39.9,
        "lon": 116.4,
        "tz": 8,
        "events": [{
            "id": "00000000-0000-4000-8000-000000000003",
            "domain": "career",
            "event_kind": "career_entry",
            "date": "2012-06-15",
            "precision": "day",
            "summary": "入职",
        }],
    }
    contexts = [
        build_candidate_static_context(request, candidate)
        for candidate in _candidate_datetimes(request)
    ]
    original_keys = [frozenset(context) for context in contexts]
    rows = compute_event_candidate_rows(request, static_contexts=contexts)
    assert len(rows) == 2
    assert [frozenset(context) for context in contexts] == original_keys
    assert all("_transit_chart_cache" not in context for context in contexts)


def test_cached_static_context_matches_uncached_fallback(monkeypatch) -> None:
    request = _event_engine_request()
    cached = [
        build_candidate_static_context(request, candidate)
        for candidate in _candidate_datetimes(request)
    ]
    assert len(cached) >= 2
    assert len(request["events"]) >= 2
    for context in cached:
        for key in CACHE_LAYER_KEYS:
            assert context.get(key) is not None
    uncached = _contexts_with_layer_cache_cleared(cached)
    for context in uncached:
        for key in CACHE_LAYER_KEYS:
            assert context[key] is None
    calls = _count_calls(monkeypatch, event_engine.shadbala, "calc_shadbala")
    rows_cached = compute_event_candidate_rows(request, static_contexts=cached)
    cached_calls = calls[0]
    rows_uncached = compute_event_candidate_rows(request, static_contexts=uncached)
    uncached_calls = calls[0] - cached_calls
    assert rows_cached == rows_uncached
    assert cached_calls == 0
    assert uncached_calls > cached_calls
    assert uncached_calls >= len(cached) * len(request["events"])


def test_poisoned_static_cache_diverges_from_live_rows() -> None:
    request = _event_engine_request()
    cached = [
        build_candidate_static_context(request, candidate)
        for candidate in _candidate_datetimes(request)
    ]
    rows_cached = compute_event_candidate_rows(request, static_contexts=cached)
    poisoned = [dict(context) for context in cached]
    poisoned[0]["shadbala_result"] = {
        "planets": {
            name: {
                "sthana_bala": {"total": 0.0},
                "drik_bala": 0.0,
                "naisargika_bala": 0.0,
            }
            for name in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")
        }
    }
    rows_poisoned = compute_event_candidate_rows(request, static_contexts=poisoned)
    assert rows_poisoned != rows_cached