Bridge functional benefic malefic layer

docs/research/functional_benefic_malefic_bridge_audit_2026_06_28.md

@@ -0,0 +1,42 @@
+# Functional Benefic/Malefic Bridge Audit - 2026-06-28
+
+## Scope
+
+This pass promotes the previously untracked `scripts/oracle_functional_benefics.py`
+fragment into a reusable strict-workflow layer.
+
+## What Changed
+
+- Added `/Users/wuyongnaren/Documents/印度占星/scripts/functional_benefics.py` as the single source for functional benefic/malefic classification by Lagna.
+- Converted `/Users/wuyongnaren/Documents/印度占星/scripts/oracle_functional_benefics.py` into a CLI wrapper around that module.
+- Routed MCP strict workflows through the shared module.
+- Routed full-reading prompt-pack and API prompt-pack snapshots through the same shared module.
+
+## Contract
+
+The bridge requires only a valid ascendant sign.
+
+It returns:
+
+- `status`
+- `ascendant`
+- `functional_benefics`
+- `functional_malefics`
+- `functional_neutrals`
+- `yogakarakas`
+- `owned_houses`
+- `effect_on_confidence`
+- `source`
+
+## Boundary
+
+- This layer classifies functional house-lord roles.
+- It does not replace natural benefic/malefic assessment.
+- It does not directly force event labels.
+- It must appear in strict workflow evidence and Technique Audit Table outputs for high-rigor readings.
+
+## Regression Coverage
+
+- `/Users/wuyongnaren/Documents/印度占星/tests/test_mcp_strict_workflow_functional_layer.py`
+- `/Users/wuyongnaren/Documents/印度占星/tests/test_cli_smoke.py`
+- `/Users/wuyongnaren/Documents/印度占星/tests/test_api_server_security.py::test_chart_ai_prompt_pack_exposes_functional_benefic_malefic_layer`

mcp_server.py

@@ -34,6 +34,7 @@ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, os.path.join(SCRIPT_DIR, "scripts"))
 
 from mcp.server.fastmcp import FastMCP
+from functional_benefics import derive_functional_benefic_malefic
 
 # ============================================================================
 # MCP Server
@@ -452,86 +453,19 @@ def _derive_external_activation_support(modules: Dict[str, Any], domain: str) ->
 def _derive_functional_benefic_malefic(modules: Dict[str, Any]) -> Dict[str, Any]:
     chart = _safe_get(modules, "chart")
     ascendant = _safe_get(chart, "ascendant") if isinstance(chart, dict) else None
-    planets = _safe_get(chart, "planets") if isinstance(chart, dict) else None
-    if not isinstance(ascendant, dict) or not isinstance(planets, dict):
+    if not isinstance(ascendant, dict):
         return {
             "status": "blocked",
             "ascendant": ascendant.get("sign") if isinstance(ascendant, dict) else None,
             "functional_benefics": [],
             "functional_malefics": [],
-            "effect_on_confidence": "Missing chart.ascendant or chart.planets; functional layer blocked.",
+            "functional_neutrals": [],
+            "yogakarakas": [],
+            "owned_houses": {},
+            "effect_on_confidence": "Missing chart.ascendant; functional layer blocked.",
             "source": "strict_functional_benefic_malefic_v1",
         }
-    asc_sign = ascendant.get("sign")
-    asc_idx = _SIGN_TO_INDEX.get(asc_sign)
-    if asc_idx is None:
-        return {
-            "status": "blocked",
-            "ascendant": asc_sign,
-            "functional_benefics": [],
-            "functional_malefics": [],
-            "effect_on_confidence": "Functional layer blocked: unknown ascendant sign.",
-            "source": "strict_functional_benefic_malefic_v1",
-        }
-
-    owned_houses: Dict[str, List[int]] = {}
-    for house_num in range(1, 13):
-        sign = _SIGNS[(asc_idx + house_num - 1) % 12]
-        lord = _SIGN_LORDS.get(sign)
-        if lord:
-            owned_houses.setdefault(lord, []).append(house_num)
-
-    trines = {1, 5, 9}
-    kendras = {1, 4, 7, 10}
-    challenging = {3, 6, 8, 11, 12}
-    benefics: set[str] = set()
-    malefics: set[str] = set()
-    yogakarakas: set[str] = set()
-    neutrals: set[str] = set()
-
-    for planet in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"):
-        houses = owned_houses.get(planet, [])
-        if not houses:
-            continue
-        owns_trine = any(house in trines for house in houses)
-        owns_kendra = any(house in kendras for house in houses)
-        owns_challenge = any(house in challenging for house in houses)
-
-        if owns_trine and owns_kendra and planet not in {"Sun", "Moon"}:
-            yogakarakas.add(planet)
-            benefics.add(planet)
-        elif owns_trine:
-            benefics.add(planet)
-        elif owns_challenge and 1 not in houses:
-            malefics.add(planet)
-        elif owns_kendra and planet in {"Jupiter", "Venus", "Mercury", "Moon"}:
-            neutrals.add(planet)
-        else:
-            neutrals.add(planet)
-
-    # Classical softening: Sun/Moon as 8L are not treated as harshly as other 8L.
-    eighth_lord = owned_houses and next(
-        (planet for planet, houses in owned_houses.items() if 8 in houses),
-        None,
-    )
-    if eighth_lord in {"Sun", "Moon"} and eighth_lord in malefics:
-        malefics.remove(eighth_lord)
-        neutrals.add(eighth_lord)
-
-    return {
-        "status": "used",
-        "ascendant": asc_sign,
-        "functional_benefics": sorted(benefics),
-        "functional_malefics": sorted(malefics),
-        "functional_neutrals": sorted(neutrals - benefics - malefics),
-        "yogakarakas": sorted(yogakarakas),
-        "owned_houses": {planet: houses for planet, houses in sorted(owned_houses.items())},
-        "effect_on_confidence": (
-            "High-rigor outputs must combine functional house-lord roles with natural roles; "
-            "conflicts should cap confidence or be explicitly noted."
-        ),
-        "source": "strict_functional_benefic_malefic_v1",
-    }
+    return derive_functional_benefic_malefic(ascendant.get("sign"))
 
 
 def _derive_synastry_relationship_support(modules: Dict[str, Any]) -> Dict[str, Any]:

scripts/functional_benefics.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Functional benefic/malefic classification by ascendant."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+SIGNS = [
+    "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
+    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
+]
+SIGN_TO_INDEX = {name: idx for idx, name in enumerate(SIGNS)}
+SIGN_LORDS = {
+    "Aries": "Mars",
+    "Taurus": "Venus",
+    "Gemini": "Mercury",
+    "Cancer": "Moon",
+    "Leo": "Sun",
+    "Virgo": "Mercury",
+    "Libra": "Venus",
+    "Scorpio": "Mars",
+    "Sagittarius": "Jupiter",
+    "Capricorn": "Saturn",
+    "Aquarius": "Saturn",
+    "Pisces": "Jupiter",
+}
+PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")
+SOURCE = "strict_functional_benefic_malefic_v1"
+
+
+def normalize_sign(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    lowered = value.strip().lower()
+    for sign in SIGNS:
+        if sign.lower() == lowered:
+            return sign
+    return None
+
+
+def derive_functional_benefic_malefic(ascendant: Any) -> dict[str, Any]:
+    """Return functional benefics, malefics and ownership roles for a Lagna."""
+    asc_sign = normalize_sign(ascendant)
+    if asc_sign is None:
+        return {
+            "status": "blocked",
+            "ascendant": ascendant if isinstance(ascendant, str) else None,
+            "functional_benefics": [],
+            "functional_malefics": [],
+            "functional_neutrals": [],
+            "yogakarakas": [],
+            "owned_houses": {},
+            "effect_on_confidence": "Functional layer blocked: unknown ascendant sign.",
+            "source": SOURCE,
+        }
+
+    asc_idx = SIGN_TO_INDEX[asc_sign]
+    owned_houses: dict[str, list[int]] = {}
+    for house_num in range(1, 13):
+        sign = SIGNS[(asc_idx + house_num - 1) % 12]
+        lord = SIGN_LORDS.get(sign)
+        if lord:
+            owned_houses.setdefault(lord, []).append(house_num)
+
+    trines = {1, 5, 9}
+    kendras = {1, 4, 7, 10}
+    challenging = {3, 6, 8, 11, 12}
+    benefics: set[str] = set()
+    malefics: set[str] = set()
+    yogakarakas: set[str] = set()
+    neutrals: set[str] = set()
+
+    for planet in PLANETS:
+        houses = owned_houses.get(planet, [])
+        if not houses:
+            continue
+        owns_trine = any(house in trines for house in houses)
+        owns_kendra = any(house in kendras for house in houses)
+        owns_challenge = any(house in challenging for house in houses)
+
+        if owns_trine and owns_kendra and planet not in {"Sun", "Moon"}:
+            yogakarakas.add(planet)
+            benefics.add(planet)
+        elif owns_trine:
+            benefics.add(planet)
+        elif owns_challenge and 1 not in houses:
+            malefics.add(planet)
+        elif owns_kendra and planet in {"Jupiter", "Venus", "Mercury", "Moon"}:
+            neutrals.add(planet)
+        else:
+            neutrals.add(planet)
+
+    eighth_lord = next((planet for planet, houses in owned_houses.items() if 8 in houses), None)
+    if eighth_lord in {"Sun", "Moon"} and eighth_lord in malefics:
+        malefics.remove(eighth_lord)
+        neutrals.add(eighth_lord)
+
+    return {
+        "status": "used",
+        "ascendant": asc_sign,
+        "functional_benefics": sorted(benefics),
+        "functional_malefics": sorted(malefics),
+        "functional_neutrals": sorted(neutrals - benefics - malefics),
+        "yogakarakas": sorted(yogakarakas),
+        "owned_houses": {planet: houses for planet, houses in sorted(owned_houses.items())},
+        "effect_on_confidence": (
+            "高严谨模式下必须叠加功能性宫主吉凶与自然吉凶；"
+            "若功能属性与自然属性冲突，应降低置信度或显式标记冲突。"
+        ),
+        "source": SOURCE,
+    }

scripts/jyotish_api_server.py

@@ -9,6 +9,7 @@
 
 import argparse
 import base64
+import html as html_lib
 import io
 import json, sys, os, math
 import importlib.util
@@ -441,6 +442,42 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             raise BadRequest('report html cannot include active content')
         return html
 
+    def _inject_functional_benefic_malefic_summary(self, html, snapshot):
+        if not isinstance(snapshot, dict):
+            return html
+        if snapshot.get('status') in {None, 'blocked', 'not_used'}:
+            return html
+        benefics = snapshot.get('functional_benefics')
+        malefics = snapshot.get('functional_malefics')
+        if not isinstance(benefics, list) or not isinstance(malefics, list):
+            return html
+
+        def _escape(value):
+            return html_lib.escape(str(value or ''))
+
+        ascendant = _escape(snapshot.get('ascendant') or snapshot.get('asc_sign') or 'Unknown')
+        benefic_text = _escape(', '.join(str(item) for item in benefics) or 'None')
+        malefic_text = _escape(', '.join(str(item) for item in malefics) or 'None')
+        confidence_text = _escape(snapshot.get('effect_on_confidence') or 'Functional role layer was used in the final judgement.')
+        source_text = _escape(snapshot.get('source') or 'strict_functional_benefic_malefic_v1')
+
+        summary = (
+            '<section data-functional-role-summary="true" '
+            'style="margin:24px 0;padding:16px;border:1px solid #d9dde8;border-radius:8px;'
+            'background:#f7f9fc;font-family:-apple-system,BlinkMacSystemFont,Segoe UI,sans-serif;">'
+            '<h2 style="margin:0 0 12px;font-size:20px;">Functional Benefic/Malefic</h2>'
+            f'<p style="margin:0 0 8px;"><strong>Ascendant:</strong> {ascendant}</p>'
+            f'<p style="margin:0 0 8px;"><strong>Functional Benefics:</strong> {benefic_text}</p>'
+            f'<p style="margin:0 0 8px;"><strong>Functional Malefics:</strong> {malefic_text}</p>'
+            f'<p style="margin:0 0 8px;"><strong>Confidence Impact:</strong> {confidence_text}</p>'
+            f'<p style="margin:0;color:#5b6472;font-size:13px;"><strong>Source:</strong> {source_text}</p>'
+            '</section>'
+        )
+        body_close = re.search(r'</body\s*>', html, re.IGNORECASE)
+        if body_close:
+            return html[:body_close.start()] + summary + html[body_close.start():]
+        return html + summary
+
     def _artifact_base64(self, path):
         size = os.path.getsize(path)
         if size > MAX_REPORT_BASE64_BYTES:
@@ -505,6 +542,10 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
 
     def _compute_report_artifact(self, body):
         html = self._validate_report_html(body.get('html'))
+        html = self._inject_functional_benefic_malefic_summary(
+            html,
+            body.get('functional_benefic_malefic'),
+        )
         fmt = body.get('format', 'html')
         if fmt not in {'html', 'pdf'}:
             raise BadRequest('format must be html or pdf')
@@ -1921,30 +1962,20 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
 
     def _functional_benefic_malefic_snapshot(self, planets, ascendant):
         try:
-            from yoga_engine import YogaContext
+            from functional_benefics import derive_functional_benefic_malefic
             asc_sign = ascendant.get('sign')
-            if not asc_sign or not isinstance(planets, dict):
-                raise ValueError('missing ascendant sign or planets')
-            context = YogaContext(planets, asc_sign)
-            benefics = context.functional_benefics()
-            malefics = context.functional_malefics()
-            return {
-                'status': 'used',
-                'ascendant': asc_sign,
-                'functional_benefics': benefics,
-                'functional_malefics': malefics,
-                'effect_on_confidence': (
-                    '高严谨模式下必须叠加功能性吉凶星；若与自然吉凶属性冲突，'
-                    '应降低置信度并在 Technique Audit Table 中显式说明。'
-                ),
-            }
+            return derive_functional_benefic_malefic(asc_sign)
         except Exception as exc:
             return {
                 'status': 'blocked',
                 'ascendant': ascendant.get('sign'),
                 'functional_benefics': [],
                 'functional_malefics': [],
+                'functional_neutrals': [],
+                'yogakarakas': [],
+                'owned_houses': {},
                 'effect_on_confidence': f'未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}',
+                'source': 'strict_functional_benefic_malefic_v1',
             }
 
     def _detect_yogas(self, planets, asc_idx):

scripts/jyotish_engine.py

@@ -859,30 +859,20 @@ def _oracle_progress_snapshot():
 
 def _functional_benefic_malefic_snapshot(planets, ascendant):
     try:
-        from yoga_engine import YogaContext
+        from functional_benefics import derive_functional_benefic_malefic
         asc_sign = ascendant.get('sign') if isinstance(ascendant, dict) else None
-        if not asc_sign or not isinstance(planets, dict):
-            raise ValueError('missing ascendant sign or planets')
-        context = YogaContext(planets, asc_sign)
-        benefics = context.functional_benefics()
-        malefics = context.functional_malefics()
-        return {
-            'status': 'used',
-            'ascendant': asc_sign,
-            'functional_benefics': benefics,
-            'functional_malefics': malefics,
-            'effect_on_confidence': (
-                '高严谨模式下必须叠加功能性吉凶星；若与自然吉凶属性冲突，'
-                '应降低置信度并在 Technique Audit Table 中显式说明。'
-            ),
-        }
+        return derive_functional_benefic_malefic(asc_sign)
     except Exception as exc:
         return {
             'status': 'blocked',
             'ascendant': ascendant.get('sign') if isinstance(ascendant, dict) else None,
             'functional_benefics': [],
             'functional_malefics': [],
+            'functional_neutrals': [],
+            'yogakarakas': [],
+            'owned_houses': {},
             'effect_on_confidence': f'未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}',
+            'source': 'strict_functional_benefic_malefic_v1',
         }
 
 

scripts/oracle_functional_benefics.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""CLI wrapper for functional benefic/malefic classification."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from typing import Any
+
+from functional_benefics import PLANETS, derive_functional_benefic_malefic
+
+
+def _render_text(report: dict[str, Any]) -> str:
+    lines = [
+        "=== Oracle stdout ===",
+        f"Ascendant: {report.get('ascendant')}",
+        "--------------------------------------------------",
+    ]
+    owned_houses = report.get("owned_houses", {})
+    benefics = set(report.get("functional_benefics", []))
+    malefics = set(report.get("functional_malefics", []))
+    yogakarakas = set(report.get("yogakarakas", []))
+    for planet in PLANETS:
+        owned = owned_houses.get(planet, [])
+        houses = " & ".join(map(str, owned)) if owned else "-"
+        if planet in yogakarakas:
+            lines.append(f"{planet}: Yogakaraka (Lord of {houses}) -> Highly Auspicious")
+        elif planet in benefics:
+            lines.append(f"{planet}: Functional Benefic (Lord of {houses}) -> Auspicious")
+        elif planet in malefics:
+            lines.append(f"{planet}: Functional Malefic (Lord of {houses}) -> Destructive / Obstacle")
+        else:
+            lines.append(f"{planet}: Neutral / Mixed (Lord of {houses}) -> Depends on placement")
+    return "\n".join(lines)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Oracle CLI: Functional Benefic/Malefic Determiner")
+    parser.add_argument("--ascendant", required=True, type=str, help="The ascendant sign, e.g. Leo")
+    parser.add_argument("--format", choices=["text", "json"], default="text")
+    args = parser.parse_args()
+
+    report = derive_functional_benefic_malefic(args.ascendant)
+    if args.format == "json":
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+    else:
+        print(_render_text(report))
+    return 0 if report.get("status") == "used" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_api_server_security.py

@@ -8,6 +8,7 @@ import json
 import os
 import sys
 from io import BytesIO
+from pathlib import Path
 
 import pytest
 
@@ -433,6 +434,30 @@ def test_report_artifact_generates_html_fallback_artifact() -> None:
     assert result['delivery']['next_action']
 
 
+def test_report_artifact_can_render_functional_benefic_malefic_summary() -> None:
+    handler = _handler()
+    result = handler._compute_report_artifact({
+        'format': 'html',
+        'name': 'functional-role-report',
+        'html': '<!doctype html><html><body><h1>Jyotish</h1></body></html>',
+        'functional_benefic_malefic': {
+            'status': 'used',
+            'ascendant': 'Leo',
+            'functional_benefics': ['Sun', 'Mars', 'Jupiter'],
+            'functional_malefics': ['Venus', 'Saturn'],
+            'effect_on_confidence': '高严谨模式下必须叠加功能性吉凶星。',
+        },
+    })
+
+    assert result['success'] is True
+    html = Path(result['html_path']).read_text(encoding='utf-8')
+    assert 'Functional Benefic/Malefic' in html
+    assert 'Leo' in html
+    assert 'Sun, Mars, Jupiter' in html
+    assert 'Venus, Saturn' in html
+    assert '高严谨模式下必须叠加功能性吉凶星。' in html
+
+
 def test_report_artifact_pdf_fallback_exposes_user_visible_delivery(monkeypatch) -> None:
     class BrokenReportBuilder:
         @staticmethod

tests/test_mcp_strict_workflow_functional_layer.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 from mcp_server import _collect_strict_evidence
+from scripts.functional_benefics import derive_functional_benefic_malefic
 
 
 def _base_relationship_result() -> dict:
@@ -48,3 +49,28 @@ def test_relationship_strict_workflow_exposes_functional_benefic_malefic_layer()
     assert "Venus" in functional["functional_malefics"]
     assert "functional_benefic_malefic_used" in strict["event_judgement"]["secondary_context"]
 
+
+def test_functional_benefic_module_calculates_lagna_roles_without_planets() -> None:
+    functional = derive_functional_benefic_malefic("Leo")
+
+    assert functional["status"] == "used"
+    assert functional["ascendant"] == "Leo"
+    assert functional["owned_houses"]["Sun"] == [1]
+    assert functional["owned_houses"]["Venus"] == [3, 10]
+    assert "Sun" in functional["functional_benefics"]
+    assert "Venus" in functional["functional_malefics"]
+    assert functional["source"] == "strict_functional_benefic_malefic_v1"
+
+
+def test_relationship_functional_layer_uses_ascendant_even_when_planets_missing() -> None:
+    result = _base_relationship_result()
+    result["modules"]["chart"] = {"ascendant": {"sign": "Leo"}}
+
+    strict = _collect_strict_evidence("relationship", result)
+
+    functional = strict["present_evidence"]["functional_benefic_malefic"]
+    assert functional["status"] == "used"
+    assert functional["ascendant"] == "Leo"
+    assert "Sun" in functional["functional_benefics"]
+    assert "Venus" in functional["functional_malefics"]
+    assert "functional_benefic_malefic_used" in strict["event_judgement"]["secondary_context"]