from __future__ import annotations import unittest from datetime import date from unittest.mock import patch from uuid import UUID from scripts.active_rectification_event_engine import _candidate_datetimes from scripts.jyotish_api_server import ( API_COMMAND_MAP, TECHNIQUE_EXAMPLE_ENDPOINTS, BadRequest, JyotishAPIHandler, ) from scripts.rectification.api_service import diagnostics, score_candidates from scripts.rectification.contracts import normalize_rectification_request from scripts.rectification.scoring_service import ( build_event_contribution_matrix, calculation_spec, sample_event_dates, score_from_matrix, sha256, ) EVENT_ID = "00000000-0000-4000-8000-000000000001" def event(index: int, domain: str, event_kind: str, *, precision: str = "day"): return { "id": f"00000000-0000-4000-8000-{index:012d}", "domain": domain, "event_kind": event_kind, "date_start": "2016-09-15", "date_end": "2016-09-15", "precision": precision, "summary": event_kind, } def request( *, precision: str = "month", event_kind: str = "education_milestone", domain: str = "education", start_time: str = "05:13", end_time: str = "05:15", date_start: str = "2016-09-01", date_end: str = "2016-09-30", ): return { "birth_date": "1997-08-08", "start_time": start_time, "end_time": end_time, "lat": 36.419, "lon": 114.213, "tz": 8, "events": [{ "id": EVENT_ID, "domain": domain, "event_kind": event_kind, "date_start": date_start, "date_end": date_end, "precision": precision, "summary": "大学入学", }], } class RectificationV5ServicesTest(unittest.TestCase): def test_event_contract_v2_accepts_native_kinds_and_retains_background_events(self): supported = { "education": ("education_start", "education_completion", "education_interruption", "education_change"), "career": ("career_entry", "career_change", "promotion", "career_pressure", "career_exit", "business_start"), "relationship": ("relationship_start", "relationship_commitment", "relationship_separation", "relationship_end"), "relocation": ("relocation", "foreign_move", "return", "home_change"), "finance": ("finance_gain", "finance_loss", "income_change", "asset_change"), "health": ("self_health_event", "pressure_period"), "family": ("family_event",), "other": ("other",), } for domain, kinds in supported.items(): for event_kind in kinds: normalized = normalize_rectification_request( request(domain=domain, event_kind=event_kind), today=date(2026, 7, 28), ) self.assertEqual(normalized["events"][0]["domain"], domain) self.assertEqual(normalized["events"][0]["event_kind"], event_kind) def test_event_contract_v2_preserves_server_provenance_and_enforces_self_scoring(self): body = request(domain="career", event_kind="career_entry") body["events"][0].update({ "source_turn_id": "33333333-3333-4333-8333-333333333333", "subject": "self", "date_source": "user_quote", "date_reliability": "month_exact", "date_corroboration": "劳动合同", "date_conflict_status": "none", }) normalized = normalize_rectification_request(body, today=date(2026, 7, 28)) event_value = normalized["events"][0] self.assertEqual(event_value["source_turn_id"], "33333333-3333-4333-8333-333333333333") self.assertEqual(event_value["subject"], "self") self.assertEqual(event_value["date_source"], "user_quote") self.assertEqual(event_value["date_reliability"], "month_exact") self.assertEqual(event_value["date_corroboration"], "劳动合同") self.assertEqual(event_value["date_conflict_status"], "none") body["events"][0]["subject"] = "family" with self.assertRaisesRegex(ValueError, "subject must be self for scoreable events"): normalize_rectification_request(body, today=date(2026, 7, 28)) def test_background_only_events_are_retained_without_invoking_the_scoring_engine(self): body = request() body["events"] = [ event(1, "family", "family_event"), event(2, "other", "other", precision="year"), ] normalized = normalize_rectification_request(body, today=date(2026, 7, 28)) def fail_if_called(_request): self.fail("background events must not reach the scoring engine") built = build_event_contribution_matrix(normalized, row_provider=fail_if_called) self.assertEqual(built["candidate_times"], []) self.assertEqual(built["matrix"], {}) self.assertEqual(built["date_sensitivity"], []) def test_declared_date_precision_changes_real_contribution_weight(self): def rows(value): event = value["events"][0] return [{ "time": "05:13", "score": 10, "evidence": [{ "event_id": event["id"], "domain": event["domain"], "candidate_time": "05:13", "rule_ids": ["vim_md_domain_house"], "points": 10, }], "missing_layers": [], }] cases = { "day": request(precision="day", event_kind="education_start", date_start="2016-09-15", date_end="2016-09-15"), "month": request(precision="month", event_kind="education_start"), "year": request( precision="year", event_kind="education_start", date_start="2016-01-01", date_end="2016-12-31", ), } points = {} for precision, body in cases.items(): normalized = normalize_rectification_request(body, today=date(2026, 7, 28)) built = build_event_contribution_matrix(normalized, row_provider=rows) points[precision] = built["matrix"][EVENT_ID]["05:13"]["points"] self.assertEqual(points, {"day": 10, "month": 8, "year": 5}) def test_calculation_spec_hash_matches_typescript_for_integral_timezone(self): normalized = normalize_rectification_request(request(), today=date(2026, 7, 28)) self.assertEqual( sha256(calculation_spec(normalized)), "f05fe0f56ef9ba2b18ec3c6c54f1649f06f1ae5a926491a5c5f676d718d92865", ) def test_cross_midnight_range_preserves_next_day_datetimes_and_typescript_hash(self): normalized = normalize_rectification_request( request(start_time="23:00", end_time="03:59"), today=date(2026, 7, 28), ) candidates = _candidate_datetimes(normalized) self.assertEqual(len(candidates), 300) self.assertEqual(candidates[0].isoformat(), "1997-08-08T23:00:00") self.assertEqual(candidates[60].isoformat(), "1997-08-09T00:00:00") self.assertEqual(candidates[-1].isoformat(), "1997-08-09T03:59:00") self.assertEqual( sha256(calculation_spec(normalized)), "b0d5c5ec7f56edbfa2b2e1041b4aa3b648c6cb0681f3f502c7b7910c2894b205", ) def test_candidate_range_boundaries_stay_bounded_and_equal_is_one_minute(self): full_day = normalize_rectification_request( request(start_time="00:00", end_time="23:59"), today=date(2026, 7, 28), ) equal = normalize_rectification_request( request(start_time="05:13", end_time="05:13"), today=date(2026, 7, 28), ) self.assertEqual(len(_candidate_datetimes(full_day)), 1_440) self.assertEqual(len(_candidate_datetimes(equal)), 1) with self.assertRaisesRegex(ValueError, "end_time must be HH:MM"): normalize_rectification_request( request(start_time="00:00", end_time="24:00"), today=date(2026, 7, 28), ) def test_daytime_candidate_range_remains_on_birth_date(self): normalized = normalize_rectification_request(request(), today=date(2026, 7, 28)) candidates = _candidate_datetimes(normalized) self.assertEqual([value.isoformat() for value in candidates], [ "1997-08-08T05:13:00", "1997-08-08T05:14:00", "1997-08-08T05:15:00", ]) def test_shared_validator_retains_background_events_and_rejects_non_self_health(self): background = normalize_rectification_request( request(domain="family", event_kind="family_event"), today=date(2026, 7, 28) ) self.assertEqual(background["events"][0]["event_kind"], "family_event") with self.assertRaisesRegex(ValueError, "event_kind does not match domain"): normalize_rectification_request( request(domain="family", event_kind="family_bereavement"), today=date(2026, 7, 28) ) with self.assertRaisesRegex(ValueError, "event_kind does not match domain"): normalize_rectification_request(request(domain="health_pressure", event_kind="family_health_event"), today=date(2026, 7, 28)) normalized = normalize_rectification_request(request(domain="health_pressure", event_kind="self_health_event"), today=date(2026, 7, 28)) self.assertEqual(normalized["events"][0]["event_kind"], "self_health_event") def test_relationship_end_is_native_and_uses_pressure_semantics(self): normalized = normalize_rectification_request( request(domain="relationship", event_kind="relationship_end", precision="day"), today=date(2026, 7, 28), ) def rows(value): event_value = value["events"][0] return [{ "time": "05:13", "score": 10, "evidence": [{ "event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:13", "rule_ids": ["controlled_transit_saturn_domain_house", f"event_kind:{event_value['event_kind']}"], "points": 10, }], "missing_layers": [], }] built = build_event_contribution_matrix(normalized, row_provider=rows) contribution = built["matrix"][EVENT_ID]["05:13"] self.assertIn("event_kind_profile:relationship_end:pressure", contribution["rule_ids"]) self.assertGreater(contribution["points"], 10) def test_relationship_start_and_change_use_distinct_rule_conditioned_profiles(self): def relationship_rows(_value): return [ { "time": "05:13", "score": 10, "evidence": [{ "event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:13", "rule_ids": ["vim_md_domain_house", "vim_ad_functional_benefic_auxiliary"], "points": 10, }], "missing_layers": [], }, { "time": "05:14", "score": 9, "evidence": [{ "event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:14", "rule_ids": ["controlled_transit_saturn_domain_house", "vim_ad_functional_malefic_auxiliary"], "points": 9, }], "missing_layers": [], }, ] start = normalize_rectification_request( request(domain="relationship", event_kind="relationship_start"), today=date(2026, 7, 28) ) change = normalize_rectification_request( request(domain="relationship", event_kind="relationship_change"), today=date(2026, 7, 28) ) start_matrix = build_event_contribution_matrix(start, row_provider=relationship_rows) change_matrix = build_event_contribution_matrix(change, row_provider=relationship_rows) self.assertGreater(start_matrix["matrix"][EVENT_ID]["05:13"]["points"], start_matrix["matrix"][EVENT_ID]["05:14"]["points"] ) self.assertGreater(change_matrix["matrix"][EVENT_ID]["05:14"]["points"], change_matrix["matrix"][EVENT_ID]["05:13"]["points"] ) self.assertEqual(change_matrix["date_sensitivity"][0]["sample_winners"], ["05:14", "05:14", "05:14"]) self.assertNotIn("event_kind_profile", change_matrix["matrix"][EVENT_ID]["05:14"]["technique_layers"]) def test_relationship_kind_profile_does_not_create_points_without_activation(self): normalized = normalize_rectification_request( request(domain="relationship", event_kind="relationship_change", precision="day"), today=date(2026, 7, 28), ) def rows(_value): return [{ "time": "05:13", "score": 0, "evidence": [{ "event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:13", "rule_ids": ["no_domain_activation", "event_kind:relationship_change"], "points": 0, }], "missing_layers": [], }] built = build_event_contribution_matrix(normalized, row_provider=rows) self.assertEqual(built["matrix"][EVENT_ID]["05:13"]["points"], 0) def test_date_sampling_preserves_declared_range_and_uses_bounded_samples(self): base = request()["events"][0] self.assertEqual(sample_event_dates({**base, "precision": "month"}), ["2016-09-01", "2016-09-15", "2016-09-30"]) year = {**base, "precision": "year", "date_start": "2016-01-01", "date_end": "2016-12-31"} self.assertEqual(len(sample_event_dates(year)), 12) ranged = {**base, "precision": "range", "date_start": "2015-01-01", "date_end": "2016-12-31"} self.assertLessEqual(len(sample_event_dates(ranged)), 12) def test_contribution_matrix_and_leave_out_diagnostics_use_matrix_math(self): normalized = normalize_rectification_request(request(), today=date(2026, 7, 28)) def rows(value): sampled = value["events"][0]["date"] shift = {"2016-09-01": 0, "2016-09-15": 1, "2016-09-30": 2}[sampled] return [{ "time": candidate, "score": points + shift, "evidence": [{ "event_id": EVENT_ID, "domain": "education", "candidate_time": candidate, "rule_ids": ["D24:test"], "points": points + shift, }], "missing_layers": ["KP_cusps"], } for candidate, points in [("05:13", 9), ("05:14", 10), ("05:15", 8)]] built = build_event_contribution_matrix(normalized, row_provider=rows) scored = score_from_matrix(normalized, built) self.assertEqual(built["matrix"][EVENT_ID]["05:14"]["points"], 8.8) self.assertEqual(scored[1]["score"], 8.8) self.assertEqual(built["missing_layers"], ["KP_cusps"]) def test_formal_score_and_diagnostics_endpoints_share_the_service_bundle(self): normalized = normalize_rectification_request(request(), today=date(2026, 7, 28)) built = { "candidate_times": ["05:13", "05:14"], "matrix": {EVENT_ID: { "05:13": {"points": 10, "rule_ids": ["D24:a"], "technique_layers": ["D24"]}, "05:14": {"points": 8, "rule_ids": ["D24:b"], "technique_layers": ["D24"]}, }}, "date_sensitivity": [{ "event_id": EVENT_ID, "declared_date_range": {"start": "2016-09-01", "end": "2016-09-30", "precision": "month"}, "sample_dates": ["2016-09-01", "2016-09-15", "2016-09-30"], "winner_retention_rate": 1, "score_variance": 1, "sample_winners": ["05:13", "05:13", "05:13"], }], "missing_layers": ["KP_cusps"], "static_contexts": [{"feature": {"time": "05:13"}}, {"feature": {"time": "05:14"}}], } feature = { "calculation_spec_hash": "0" * 64, "algorithm_version": "rectification-v5-matrix-scoring-2", "candidate_count": 2, "feature_hash": "1" * 64, "features": [{"time": "05:13"}, {"time": "05:14"}], } with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch( "scripts.rectification.api_service.build_candidate_feature_snapshot", return_value=feature ): scored = score_candidates(normalized) diagnostic_result = diagnostics(normalized) self.assertFalse(scored["can_confirm_exact_minute"]) self.assertIn("event_contribution_matrix", scored) self.assertEqual(diagnostic_result["diagnostics"]["leave_one_event_out_retention_rate"], 1) self.assertFalse(diagnostic_result["can_confirm_exact_minute"]) def test_score_endpoint_returns_candidate_decision_receipt_v2_and_real_execution_ledger(self): body = request() body["events"] = [ event(1, "education", "education_start"), event(2, "career", "promotion", precision="month"), event(3, "finance", "finance_gain"), event(4, "family", "family_event", precision="year"), ] normalized = normalize_rectification_request(body, today=date(2026, 7, 28)) scored_ids = [item["id"] for item in normalized["events"][:3]] built = { "candidate_times": ["05:13", "05:14", "05:15"], "matrix": { event_id: { "05:13": {"points": 4, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]}, "05:14": {"points": 2, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]}, "05:15": {"points": 1, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]}, } for event_id in scored_ids }, "date_sensitivity": [ { "event_id": event_id, "declared_date_range": {"start": "2016-09-15", "end": "2016-09-15", "precision": "day"}, "sample_dates": ["2016-09-15"], "winner_retention_rate": 1, "score_variance": 0, "sample_winners": ["05:13"], } for event_id in scored_ids ], "missing_layers": [], "static_contexts": [], } diagnostic_values = { "primary_cluster_retention_rate": 1, "leave_one_event_out_retention_rate": .9, "leave_one_domain_out_retention_rate": .9, "date_sensitivity_retention_rate": .9, "neighbor_support_minutes": 1, "primary_secondary_margin_percent": 30, "event_date_sensitivity": built["date_sensitivity"], } feature = { "calculation_spec_hash": "0" * 64, "algorithm_version": "rectification-v5-matrix-scoring-2", "candidate_count": 3, "feature_hash": "1" * 64, "features": [], } with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch( "scripts.rectification.api_service.build_candidate_feature_snapshot", return_value=feature ), patch("scripts.rectification.api_service.run_diagnostics", return_value=diagnostic_values): first = score_candidates(normalized) second = score_candidates(normalized) self.assertEqual(first["event_contract_version"], "rectification-event-contract-v2") self.assertEqual(first["candidate_decisions"], second["candidate_decisions"]) self.assertEqual(sum(item["relative_support"] for item in first["candidate_decisions"]), 100) for rank, candidate in enumerate(first["candidate_decisions"], start=1): self.assertEqual(set(candidate), {"candidate_id", "rank", "time", "relative_support", "tied_minute_count"}) self.assertEqual(candidate["rank"], rank) UUID(candidate["candidate_id"]) receipt = first["candidate_decision_receipt"] self.assertEqual(receipt["receipt_version"], "candidate-decision-receipt-v2") self.assertTrue(receipt["selection_allowed"]) self.assertTrue(receipt["acceptance_allowed"]) self.assertFalse(receipt["confirmation_allowed"]) self.assertTrue(receipt["gates"]["domain_diversity"]["passed"]) self.assertTrue(receipt["gates"]["date_quality"]["passed"]) self.assertTrue(receipt["gates"]["exact_confirmation"]["fail_closed"]) self.assertEqual(first["decision_receipt"], receipt) self.assertEqual(first["decision_policy_version"], "rectification-candidate-policy-v2") self.assertTrue(receipt["display_allowed"]) self.assertTrue(receipt["accept_allowed"]) self.assertFalse(receipt["confirm_allowed"]) self.assertIsNotNone(receipt["representative_candidate_id"]) self.assertEqual(receipt["representative_time"], "05:13") entries = first["execution_ledger"] background = next(item for item in entries if item.get("event_id") == normalized["events"][3]["id"]) self.assertEqual(background["status"], "retained_not_scored") executed = next(item for item in entries if item.get("event_id") == normalized["events"][0]["id"]) self.assertEqual(executed["technique_layers"], ["vim_md_domain_house"]) def test_diagnostics_endpoint_mirrors_candidate_decision_policy_fields(self): normalized = normalize_rectification_request(request(), today=date(2026, 7, 28)) scored = { "result_id": "00000000-0000-4000-8000-000000000099", "algorithm_version": "rectification-v5-matrix-scoring-2", "event_contract_version": "rectification-event-contract-v2", "decision_policy_version": "rectification-candidate-policy-v2", "calculation_spec_hash": "0" * 64, "candidate_decisions": [], "candidate_decision_receipt": {}, "decision_receipt": {}, "execution_ledger_version": "rectification-execution-ledger-v2", "execution_ledger": [], "diagnostics": {}, "missing_layers": [], "display_allowed": False, "selection_allowed": False, "acceptance_allowed": False, "confirmation_allowed": False, "representative_candidate_id": None, "representative_time": None, "overall_confidence": "low", "margin_percent": 0.0, } with patch("scripts.rectification.api_service.score_candidates", return_value=scored): result = diagnostics(normalized) for field in ( "display_allowed", "selection_allowed", "acceptance_allowed", "confirmation_allowed", "representative_candidate_id", "representative_time", "overall_confidence", "margin_percent", ): self.assertEqual(result[field], scored[field]) self.assertFalse(result["can_confirm_exact_minute"]) def test_single_domain_or_quantized_top_tie_blocks_candidate_acceptance(self): body = request() body["events"] = [ event(1, "career", "career_entry"), event(2, "career", "promotion"), event(3, "career", "career_change"), ] normalized = normalize_rectification_request(body, today=date(2026, 7, 28)) built = { "candidate_times": ["05:13", "05:14"], "matrix": { item["id"]: { "05:13": {"points": 3.333346, "rule_ids": ["D10:test"], "technique_layers": ["D10"]}, "05:14": {"points": 3.333333, "rule_ids": ["D10:test"], "technique_layers": ["D10"]}, } for item in normalized["events"] }, "date_sensitivity": [], "missing_layers": [], "static_contexts": [], } diagnostic_values = { "primary_cluster_retention_rate": 1, "leave_one_event_out_retention_rate": 1, "leave_one_domain_out_retention_rate": 1, "date_sensitivity_retention_rate": 1, "neighbor_support_minutes": 1, "primary_secondary_margin_percent": 50, } with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch( "scripts.rectification.api_service.build_candidate_feature_snapshot", return_value={} ), patch("scripts.rectification.api_service.run_diagnostics", return_value=diagnostic_values): result = score_candidates(normalized) receipt = result["candidate_decision_receipt"] self.assertFalse(receipt["acceptance_allowed"]) self.assertIn("insufficient_domain_diversity", receipt["reasons"]) self.assertIn("tied_top_score", receipt["reasons"]) self.assertEqual(result["candidate_decisions"][0]["tied_minute_count"], 2) self.assertEqual(receipt["tie_policy"]["score_quantum"], .0001) self.assertEqual(receipt["tie_policy"]["absolute_tolerance"], .0001) def test_low_date_quality_blocks_acceptance_even_with_multiple_domains_and_a_unique_candidate(self): body = request() body["events"] = [ event(1, "education", "education_start", precision="year"), event(2, "career", "promotion", precision="year"), event(3, "finance", "finance_gain", precision="year"), ] normalized = normalize_rectification_request(body, today=date(2026, 7, 28)) built = { "candidate_times": ["05:13", "05:14"], "matrix": { item["id"]: { "05:13": {"points": 4, "rule_ids": ["test"], "technique_layers": ["test"]}, "05:14": {"points": 1, "rule_ids": ["test"], "technique_layers": ["test"]}, } for item in normalized["events"] }, "date_sensitivity": [], "missing_layers": [], "static_contexts": [], } diagnostic_values = { "leave_one_event_out_retention_rate": 1, "leave_one_domain_out_retention_rate": 1, "date_sensitivity_retention_rate": 1, "primary_secondary_margin_percent": 50, } with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch( "scripts.rectification.api_service.build_candidate_feature_snapshot", return_value={} ), patch("scripts.rectification.api_service.run_diagnostics", return_value=diagnostic_values): result = score_candidates(normalized) receipt = result["candidate_decision_receipt"] self.assertTrue(result["candidate_decisions"]) self.assertFalse(receipt["acceptance_allowed"]) self.assertFalse(receipt["gates"]["date_quality"]["passed"]) self.assertIn("low_date_quality", receipt["reasons"]) self.assertFalse(result["can_confirm_exact_minute"]) def test_http_registry_exposes_all_v5_endpoints(self): expected = { "rectification-v5-candidate-features": "/api/rectification/v5/candidate-features", "rectification-v5-score": "/api/rectification/v5/score", "rectification-v5-diagnostics": "/api/rectification/v5/diagnostics", } for command, endpoint in expected.items(): self.assertEqual(API_COMMAND_MAP[command], endpoint) self.assertIn(endpoint, TECHNIQUE_EXAMPLE_ENDPOINTS) def test_http_handler_enforces_subject_and_event_kind_boundaries(self): handler = object.__new__(JyotishAPIHandler) retained = handler._rectification_v5_request(request(domain="family", event_kind="family_event")) self.assertEqual(retained["events"][0]["event_kind"], "family_event") with self.assertRaisesRegex(BadRequest, "event_kind does not match domain"): handler._rectification_v5_request(request(domain="family", event_kind="family_bereavement")) with self.assertRaisesRegex(BadRequest, "event_kind does not match domain"): handler._rectification_v5_request(request(domain="health_pressure", event_kind="family_health_event")) normalized = handler._rectification_v5_request( request(domain="health_pressure", event_kind="self_health_event") ) self.assertEqual(normalized["events"][0]["event_kind"], "self_health_event") def test_v4_compatibility_and_v5_score_handlers_share_the_v5_service(self): handler = object.__new__(JyotishAPIHandler) result = {"result_id": "00000000-0000-4000-8000-000000000099", "can_confirm_exact_minute": False} with patch("scripts.rectification.api_service.score_candidates", return_value=result) as scorer: v5 = handler._compute_rectification_v5_score(request()) v4 = handler._compute_active_rectification_events_v4(request()) self.assertEqual(scorer.call_count, 2) self.assertEqual(v5["endpoint"], "rectification_v5_score") self.assertEqual(v4["endpoint"], "active_rectification_events_v4") self.assertEqual(v5["result_id"], v4["result_id"]) self.assertFalse(v5["can_confirm_exact_minute"]) self.assertFalse(v4["can_confirm_exact_minute"]) if __name__ == "__main__": unittest.main()