#!/usr/bin/env python3 """Regression tests for source-pack content contracts beyond visibility.""" from __future__ import annotations import json import subprocess import sys from pathlib import Path from mcp_server import _collect_strict_evidence, _existing_interpretation_source_pack ROOT = Path(__file__).resolve().parents[1] CORE5 = [ "references/prediction-boundary-protocol.md", "references/event_judgment_skeleton.md", "references/planetary-dignity-complete-reference.md", "references/retrograde-combustion-war-guide.md", "references/transit-multi-reference-guide.md", ] PROMOTE_BATCH2 = [ "references/vimshottari_dasha_guide.md", "references/pratyantar-calculation-guide.md", "references/divisional-chart-deep-reading.md", "references/shadbala-complete-methodology.md", "references/ashtakavarga-complete-system.md", "references/tajika-yoga-complete-guide.md", "references/jaimini-complete-system.md", "references/kp-astrology-complete-system.md", "references/argala-complete-guide.md", "references/badhaka-obstacle-planet-guide.md", "references/condition-dasha-complete.md", ] REFERENCE_ONLY = [ "references/dasa-convergence-methodology.md", "references/multi-dasha-convergence-protocol.md", "references/yoga-strength-scoring-system.md", ] NON_RUNTIME = [ "references/varga-system-quick-reference.md", "references/yoga-list-chinese.md", "references/analysis-full-reading-v4.0.md", "references/analysis-full-reading-v1.8-review.md", "references/audit-skill-full-test-2026-05-04.md", "references/feature-gap-matrix-2026.md", "references/kp-practical-event-timing.md", "references/consultation-case-library.md", ] def _base_modules() -> dict: return { "modules": { "chart": { "ascendant": {"sign": "Leo"}, "planets": { "Sun": {"status": "中性(Neutral)"}, "Moon": {"status": "中性(Neutral)"}, "Mercury": {"status": "中性(Neutral)"}, "Venus": {"status": "中性(Neutral)"}, "Jupiter": {"status": "中性(Neutral)"}, "Saturn": {"status": "中性(Neutral)"}, }, }, "varga_full": { "D10_Dasamsa": {"summary": "career varga present"}, "D9_Navamsha": {"summary": "relationship varga present"}, "D2_Hora": {"summary": "wealth varga present"}, "D11_Rudramsha": {"summary": "gains varga present"}, }, "special_lagnas": { "A10_Karma_Pada": {"sign": "Capricorn", "lord": "Saturn"}, "Upapada_Lagna": {"sign": "Libra", "lord": "Venus"}, }, "jaimini": { "karakas": { "Amatyakaraka": {"planet": "Mercury"}, "Atmakaraka": {"planet": "Sun"}, "Darakaraka": {"planet": "Venus"}, }, "karakamsha": {"karakamsha_sign": "Leo", "karakamsha_lord": "Sun"}, }, "dasha": {"current_dasha": {"mahadasha": "Mercury", "antardasha": "Sun"}}, "narayana_dasha": {"current_dasha": {"sign": "Capricorn", "lord": "Saturn"}}, "dasa_convergence": { "domain_activations": { "career_status": {"convergence_level": "L2", "probability": "35-50%"}, "marriage_relationship": {"convergence_level": "L2", "probability": "35-50%"}, "wealth_income": {"convergence_level": "L2", "probability": "35-50%"}, } }, } } def _run_engine(*args: str) -> dict: completed = subprocess.run( [sys.executable, "scripts/jyotish_engine.py", *args], cwd=ROOT, check=False, text=True, capture_output=True, timeout=180, ) assert completed.returncode == 0, completed.stderr[-2000:] or completed.stdout[-2000:] return json.loads(completed.stdout) def test_core5_sources_drive_prediction_boundary_contract_for_all_strict_routes() -> None: for route in ["career", "relationship", "finance"]: strict = _collect_strict_evidence(route, _base_modules()) contract = strict["prediction_boundary_contract"] assert contract["source_refs"] == CORE5 assert contract["event_judgment_skeleton"]["required_sections"] == [ "promise", "activation", "manifestation", "label", ] assert contract["promise"]["status"] == strict["adjudication_stages"]["promise"]["status"] assert contract["activation"]["status"] == strict["adjudication_stages"]["activation"]["status"] assert contract["manifestation"]["status"] == strict["adjudication_stages"]["manifestation"]["status"] assert contract["confidence_boundary"]["mevg_status"] == "blocked" assert contract["confidence_boundary"]["real_case_calibration_status"] == "blocked" assert contract["confidence_boundary"]["unverified_claim_policy"] == "downgrade_or_block" def test_batch2_and_reference_only_layers_are_wired_without_polluting_truth_sources() -> None: source_pack = _existing_interpretation_source_pack() assert source_pack["promote_batch2_topic_layer"]["status"] == "available" assert source_pack["promote_batch2_topic_layer"]["source_refs"] == PROMOTE_BATCH2 assert source_pack["reference_only_conflict_layer"]["status"] == "available" assert source_pack["reference_only_conflict_layer"]["source_refs"] == REFERENCE_ONLY assert source_pack["reference_only_conflict_layer"]["promotion_status"] == "reference_only" runtime_refs = set(source_pack["source_refs"]) for path in PROMOTE_BATCH2 + REFERENCE_ONLY: assert path in runtime_refs for path in NON_RUNTIME: assert path not in runtime_refs inventory = source_pack["interpretation_source_inventory"] assert inventory["layers"]["promote_batch2_topic_sources"]["source_refs"] == PROMOTE_BATCH2 assert inventory["layers"]["reference_only_conflict_sources"]["source_refs"] == REFERENCE_ONLY assert inventory["summary"]["blocked_non_runtime_count"] >= len(NON_RUNTIME) def test_prompt_pack_and_real_reading_regression_expose_content_contracts() -> None: result = _run_engine( "full-reading", "--year", "REDACTED_YEAR", "--month", "4", "--day", "17", "--hour", "14", "--minute", "49", "--lat", "36.466667", "--lon", "114.2", "--tz", "8", "--today", "2026-07-02", "--transit-date", "2026-07-02", ) prompt_pack = result["ai_prompt_pack"] snapshot = prompt_pack["evidence_snapshot"] assert snapshot["prediction_boundary_contract"]["source_refs"] == CORE5 assert snapshot["interpretation_source_pack"]["core_rule_source_refs"] == CORE5 assert snapshot["interpretation_source_pack"]["promote_batch2_source_refs"] == PROMOTE_BATCH2 assert snapshot["interpretation_source_pack"]["reference_only_source_refs"] == REFERENCE_ONLY assert "必须按 promise → activation → manifestation → label 输出" in prompt_pack["prompt_zh"] assert "未完成 MEVG / Real Case Calibration 时必须降级或标 blocked" in prompt_pack["prompt_zh"] docs = prompt_pack["retrieval_plan"]["local_reference_docs"] for path in CORE5 + PROMOTE_BATCH2 + REFERENCE_ONLY: assert path in docs for path in NON_RUNTIME: assert path not in docs for route in ["career", "relationship", "finance"]: contract = snapshot["strict_workflow_contracts"][route] assert contract["prediction_boundary_contract"]["source_refs"] == CORE5 audit = contract["technique_audit_summary"] assert audit["mevg_global_web_evidence"]["status"] == "blocked" assert audit["real_case_calibration"]["status"] == "blocked" assert audit["interpretation_source_pack"]["core_rule_source_refs"] == CORE5