diff --git a/mcp_server.py b/mcp_server.py index fc2a7f38..21ed10d9 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -448,6 +448,131 @@ def _derive_external_activation_support(modules: Dict[str, Any], domain: str) -> } +def _sign_to_index(sign: str) -> Optional[int]: + try: + return _SIGNS.index(sign) + except ValueError: + return None + + +def _lord_for_house_from_lagna(asc_sign: Optional[str], house_num: int) -> Optional[str]: + asc_idx = _sign_to_index(asc_sign) if asc_sign else None + if asc_idx is None: + return None + sign = _SIGNS[(asc_idx + house_num - 1) % 12] + return _SIGN_LORDS.get(sign) + + +def _extract_dignity_code(status: str) -> Optional[str]: + if "Neecha Bhanga" in status or "落陷取消" in status: + return "NEECHA_BHANGA" + if "Great Enemy" in status or "极敌" in status: + return "GREAT_ENEMY" + return None + + +def _derive_dignity_guardrail(route: str, present: Dict[str, Any]) -> Dict[str, Any]: + base = { + "route": route, + "status": "blocked", + "score_delta": 0, + "source": "chart.planets.status", + "relevant_planets": [], + "ignored_planets": [], + "conflict_flags": [], + "notes": ["Only domain-relevant planets are allowed to affect score."], + } + + chart = present.get("chart") if isinstance(present.get("chart"), dict) else {} + ascendant = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {} + planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {} + asc_sign = ascendant.get("sign") + + if not asc_sign or not isinstance(planets, dict) or not planets: + return base + + relevant_roles: Dict[str, str] = {} + + if route == "relationship": + lord_7 = _lord_for_house_from_lagna(asc_sign, 7) + if not lord_7: + return base + relevant_roles[lord_7] = "7l" + relevant_roles["Venus"] = "relationship_karaka" + relevant_roles["Jupiter"] = "relationship_support" + darakaraka = present.get("darakaraka") + if isinstance(darakaraka, dict) and darakaraka.get("planet"): + relevant_roles[darakaraka["planet"]] = "darakaraka" + elif route == "finance": + lord_2 = _lord_for_house_from_lagna(asc_sign, 2) + lord_11 = _lord_for_house_from_lagna(asc_sign, 11) + if not lord_2 or not lord_11: + return base + relevant_roles[lord_2] = "2l" + relevant_roles[lord_11] = "11l" + relevant_roles["Venus"] = "finance_karaka" + relevant_roles["Jupiter"] = "finance_support" + if present.get("career_convergence"): + lord_10 = _lord_for_house_from_lagna(asc_sign, 10) + if lord_10: + relevant_roles[lord_10] = "10l_career_monetization" + else: + base["status"] = "ok" + return base + + supportive_hits: List[str] = [] + friction_hits: List[str] = [] + + for planet_name, pdata in planets.items(): + if planet_name not in relevant_roles: + base["ignored_planets"].append({ + "planet": planet_name, + "reason": "not_domain_relevant", + }) + continue + if not isinstance(pdata, dict): + return base + status = str(pdata.get("status", "")) + if not status: + return base + dignity_code = _extract_dignity_code(status) + effect = ( + "supportive_recovery" if dignity_code == "NEECHA_BHANGA" + else "high_friction" if dignity_code == "GREAT_ENEMY" + else "none" + ) + base["relevant_planets"].append({ + "planet": planet_name, + "role": relevant_roles[planet_name], + "status": status, + "dignity_code": dignity_code, + "effect": effect, + }) + if dignity_code == "NEECHA_BHANGA": + supportive_hits.append(planet_name) + elif dignity_code == "GREAT_ENEMY": + friction_hits.append(planet_name) + + if supportive_hits and friction_hits: + base["status"] = "conflict" + base["conflict_flags"] = [ + "neecha_bhanga_on_key_significator", + "great_enemy_on_key_significator", + ] + base["score_delta"] = 0 + elif supportive_hits: + base["status"] = "caution" + base["score_delta"] = 5 + elif friction_hits: + base["status"] = "caution" + base["score_delta"] = -5 + else: + base["status"] = "ok" + base["score_delta"] = 0 + + return base + + def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[str]) -> Dict[str, Any]: if route == "relationship": score = 0 @@ -458,17 +583,8 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s score += 10 if present.get("vimshottari_current") else 0 score += 10 if present.get("narayana_current") else 0 score += _convergence_score(present.get("marriage_convergence")) - - # Dignity Adjustments - chart = present.get("chart") if isinstance(present.get("chart"), dict) else {} - planets = chart.get("planets", {}) - for pn, pd in planets.items(): - if isinstance(pd, dict): - st = str(pd.get("status", "")) - if "Neecha Bhanga" in st or "落陷取消" in st: - score += 10 - elif "Great Enemy" in st or "极敌" in st: - score -= 5 + dignity_guardrail = present.get("dignity_guardrail") or {} + score += dignity_guardrail.get("score_delta", 0) if missing: score = min(score, 35) score = min(score, 100) @@ -493,6 +609,12 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s external_activation = present.get("external_activation") or {} if external_activation.get("level") == "moderate": secondary_context.append("external_activation_support") + if dignity_guardrail.get("status") == "conflict": + secondary_context.append("dignity_conflict") + elif dignity_guardrail.get("score_delta") == 5: + secondary_context.append("dignity_supportive_recovery") + elif dignity_guardrail.get("score_delta") == -5: + secondary_context.append("dignity_high_friction") hard_gate_missing = any( key in missing for key in ( @@ -547,17 +669,8 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s _convergence_score(present.get("gains_convergence")), _convergence_score(present.get("career_convergence")), ) - - # Dignity Adjustments - chart = present.get("chart") if isinstance(present.get("chart"), dict) else {} - planets = chart.get("planets", {}) - for pn, pd in planets.items(): - if isinstance(pd, dict): - st = str(pd.get("status", "")) - if "Neecha Bhanga" in st or "落陷取消" in st: - score += 10 - elif "Great Enemy" in st or "极敌" in st: - score -= 5 + dignity_guardrail = present.get("dignity_guardrail") or {} + score += dignity_guardrail.get("score_delta", 0) score -= 5 if isinstance(avayogi_risk, dict) and avayogi_risk.get("risk_level") == "moderate" else 0 public_wealth_lift = ( not missing @@ -606,6 +719,12 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s external_activation = present.get("external_activation") or {} if external_activation.get("level") == "moderate": secondary_context.append("external_activation_support") + if dignity_guardrail.get("status") == "conflict": + secondary_context.append("dignity_conflict") + elif dignity_guardrail.get("score_delta") == 5: + secondary_context.append("dignity_supportive_recovery") + elif dignity_guardrail.get("score_delta") == -5: + secondary_context.append("dignity_high_friction") return { "event_family": "finance", "score": score, @@ -660,15 +779,18 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present["jaimini_timing_support"] = _safe_get(modules, "jaimini", "marriage_timing_support") present["jaimini_marriage_support"] = _derive_jaimini_marriage_support(present) present["external_activation"] = _derive_external_activation_support(modules, "marriage") + present["dignity_guardrail"] = _derive_dignity_guardrail(route, present) missing = [ key for key, value in present.items() - if key not in {"chart", "external_activation", "jaimini_marriage_support", "jaimini_timing_support"} + if key not in {"chart", "external_activation", "dignity_guardrail", "jaimini_marriage_support", "jaimini_timing_support"} and value in (None, {}, [], "") ] convergence = present["marriage_convergence"] or {} confidence_cap = "medium" if missing: confidence_cap = "low" + elif present["dignity_guardrail"].get("status") == "conflict": + confidence_cap = "low" elif convergence.get("convergence_level") in {"L4", "L5"}: confidence_cap = "medium-high" elif convergence.get("convergence_level") == "L3": @@ -716,8 +838,9 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An "avayogi_risk": avayogi_risk, } present["external_activation"] = _derive_external_activation_support(modules, "wealth") + present["dignity_guardrail"] = _derive_dignity_guardrail(route, present) missing = [key for key, value in present.items() if key not in { - "chart", "external_activation", "gains_convergence", "career_convergence", "avayogi_risk" + "chart", "external_activation", "dignity_guardrail", "gains_convergence", "career_convergence", "avayogi_risk" } and value in (None, {}, [], "")] convergence_hits: List[Dict[str, Any]] = [ item for item in [ @@ -730,6 +853,8 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An confidence_cap = "medium" if missing: confidence_cap = "low" + elif present["dignity_guardrail"].get("status") == "conflict": + confidence_cap = "low" elif any(hit.get("convergence_level") in {"L4", "L5"} for hit in convergence_hits): confidence_cap = "medium-high" elif convergence_hits: diff --git a/tests/test_mcp_strict_workflow_finance.py b/tests/test_mcp_strict_workflow_finance.py index bab495b9..0d2cbcce 100644 --- a/tests/test_mcp_strict_workflow_finance.py +++ b/tests/test_mcp_strict_workflow_finance.py @@ -260,6 +260,73 @@ def test_collect_strict_evidence_finance_collects_vedastro_range_scan_as_externa assert "external_activation_support" in strict["event_judgement"]["secondary_context"] +def test_finance_dignity_guardrail_conflict_caps_to_zero_delta() -> None: + result = { + "modules": { + "varga_full": {"D2_Hora": {"summary": "ok"}, "D10_Dasamsa": {"summary": "ok"}}, + "shadbala": {"planets": {"Venus": {"total_rupa": 8.2}}}, + "ashtakavarga": {"house_scores": {"2": 31, "11": 36}}, + "chart": { + "ascendant": {"sign": "Leo"}, + "planets": { + "Mercury": {"status": "落陷取消(Neecha Bhanga)"}, + "Venus": {"status": "极敌(Great Enemy)"}, + "Jupiter": {"status": "中性(Neutral)"}, + }, + }, + "dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Mercury"}}, + "narayana_dasha": {"current_dasha": {"sign": "Taurus", "lord": "Venus"}}, + "dasa_convergence": { + "domain_activations": { + "wealth_family": {"convergence_level": "L1", "probability": "+15-20%"}, + "gains_wishes": {"convergence_level": "L1", "probability": "+15-20%"}, + "career_status": {"convergence_level": "L1", "probability": "+15-20%"}, + } + }, + } + } + + strict = _collect_strict_evidence("finance", result) + + assert strict["present_evidence"]["dignity_guardrail"]["status"] == "conflict" + assert strict["present_evidence"]["dignity_guardrail"]["score_delta"] == 0 + assert "dignity_conflict" in strict["event_judgement"]["secondary_context"] + + +def test_finance_dignity_guardrail_ignores_non_relevant_planets() -> None: + result = { + "modules": { + "varga_full": {"D2_Hora": {"summary": "ok"}, "D10_Dasamsa": {"summary": "ok"}}, + "shadbala": {"planets": {"Venus": {"total_rupa": 8.2}}}, + "ashtakavarga": {"house_scores": {"2": 31, "11": 36}}, + "chart": { + "ascendant": {"sign": "Leo"}, + "planets": { + "Mars": {"status": "落陷取消(Neecha Bhanga)"}, + "Saturn": {"status": "极敌(Great Enemy)"}, + "Venus": {"status": "中性(Neutral)"}, + "Jupiter": {"status": "中性(Neutral)"}, + }, + }, + "dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Mercury"}}, + "narayana_dasha": {"current_dasha": {"sign": "Taurus", "lord": "Venus"}}, + "dasa_convergence": { + "domain_activations": { + "wealth_family": {"convergence_level": "L1", "probability": "+15-20%"}, + "gains_wishes": {"convergence_level": "L1", "probability": "+15-20%"}, + "career_status": {"convergence_level": "L1", "probability": "+15-20%"}, + } + }, + } + } + + strict = _collect_strict_evidence("finance", result) + + assert strict["present_evidence"]["dignity_guardrail"]["score_delta"] == 0 + assert "dignity_supportive_recovery" not in strict["event_judgement"]["secondary_context"] + assert "dignity_high_friction" not in strict["event_judgement"]["secondary_context"] + + def test_collect_strict_evidence_finance_ignores_non_matching_external_activation_domains() -> None: result = { "modules": { diff --git a/tests/test_mcp_strict_workflow_relationship.py b/tests/test_mcp_strict_workflow_relationship.py index 6421525b..abf7996c 100644 --- a/tests/test_mcp_strict_workflow_relationship.py +++ b/tests/test_mcp_strict_workflow_relationship.py @@ -81,3 +81,40 @@ def test_relationship_collects_vedastro_range_scan_as_external_activation_contex assert strict["present_evidence"]["external_activation"]["level"] == "moderate" assert strict["present_evidence"]["external_activation"]["source"] == "vedastro_service_adapter_candidate" assert "external_activation_support" in strict["event_judgement"]["secondary_context"] + + +def test_relationship_dignity_guardrail_ignores_non_relevant_planets() -> None: + result = _base_relationship_result() + result["modules"]["chart"] = { + "ascendant": {"sign": "Leo"}, + "planets": { + "Mars": {"status": "落陷取消(Neecha Bhanga)"}, + "Mercury": {"status": "极敌(Great Enemy)"}, + "Venus": {"status": "中性(Neutral)"}, + "Jupiter": {"status": "中性(Neutral)"}, + }, + } + + strict = _collect_strict_evidence("relationship", result) + + assert strict["present_evidence"]["dignity_guardrail"]["score_delta"] == 0 + assert strict["event_judgement"]["dominant_label"] == "legal_marriage" + assert "dignity_supportive_recovery" not in strict["event_judgement"]["secondary_context"] + assert "dignity_high_friction" not in strict["event_judgement"]["secondary_context"] + + +def test_relationship_dignity_guardrail_uses_relevant_planets_only() -> None: + result = _base_relationship_result() + result["modules"]["chart"] = { + "ascendant": {"sign": "Aries"}, + "planets": { + "Venus": {"status": "落陷取消(Neecha Bhanga)"}, + "Jupiter": {"status": "中性(Neutral)"}, + "Saturn": {"status": "极敌(Great Enemy)"}, + }, + } + + strict = _collect_strict_evidence("relationship", result) + + assert strict["present_evidence"]["dignity_guardrail"]["score_delta"] == 5 + assert "dignity_supportive_recovery" in strict["event_judgement"]["secondary_context"]