"""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. """ from __future__ import annotations import inspect import json from datetime import date, datetime from pathlib import Path from typing import Any from unittest.mock import patch 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" 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 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 actual["candidate_scores"] == expected["candidate_scores"] assert actual["decision_receipt"] == expected["decision_receipt"] 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)