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

Golden payload in tests/golden/rectification_engine_memoization_v1.json was
produced from origin/staging @ a8d29d1b before any memoization landed.

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 inspect
import json
import math
from datetime import date, datetime
from pathlib import Path
from typing import Any
from unittest.mock import patch

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_v1.json"
FROZEN_TODAY = date(2026, 9, 16)
TIMING_KEYS = frozenset({"column_compare_ms"})
SOURCE_COMMIT = "a8d29d1b6cc37ff865ddec6c8bccdf9aa889ee53"
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)


def _golden_payload() -> dict[str, Any]:
    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 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_memoization_payloads(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