harden prashna and tajika calculation boundaries

scripts/jyotish_api_server.py

@@ -4944,6 +4944,27 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
         return calc_kp_analysis(planets, SIGNS[asc_idx])
 
     def _compute_prashna(self, body):
+        try:
+            from prashna_context import PrashnaContextError, build_prashna_context
+        except ModuleNotFoundError:  # pragma: no cover - package import path
+            from scripts.prashna_context import PrashnaContextError, build_prashna_context
+        if "planets" in body or "asc_degree" in body:
+            raise BadRequest("Prashna planets and ascendant are backend-computed; client values are forbidden")
+        try:
+            context = build_prashna_context(body)
+        except PrashnaContextError as exc:
+            raise BadRequest(str(exc)) from exc
+        return {
+            "success": True,
+            "status": "computed",
+            "prashna_context": context,
+            "verdict": {
+                "status": "blocked",
+                "reason": "Prashna adjudication is disabled until Tajika/Saham/Sphuta kernels pass classic golden cases.",
+            },
+        }
+
+        # Legacy client-supplied-chart pipeline below is unreachable pending deletion.
         question_type = body.get('question', 'general')
         if not isinstance(question_type, str):
             raise BadRequest('question must be a string')

scripts/jyotish_engine.py

@@ -5878,6 +5878,33 @@ def cmd_full_reading(args):
 # ============================================================================
 def cmd_prashna(args):
     """Prashna 问事占星：基于提问时刻的即时星盘分析"""
+    try:
+        from prashna_context import PrashnaContextError, build_prashna_context
+    except ImportError:
+        from scripts.prashna_context import PrashnaContextError, build_prashna_context
+    try:
+        context = build_prashna_context({
+            "question_text": args.question_text,
+            "question_timestamp": args.datetime,
+            "lat": args.lat,
+            "lon": args.lon,
+            "timezone": args.timezone,
+            "ayanamsa": args.ayanamsa,
+            "node_mode": args.node_mode,
+            "location_convention": args.location_convention,
+        })
+    except PrashnaContextError as exc:
+        return {"scope": "prashna_context", "status": "blocked", "reason": str(exc)}
+    if args.mode != "chart":
+        return {
+            "scope": "prashna",
+            "status": "blocked",
+            "reason": f"{args.mode} is blocked pending validated Prashna kernel implementation",
+            "prashna_context": context,
+        }
+    return context
+
+    # Legacy fallback below is intentionally unreachable until removed after migration.
     try:
         from prashna import cast_prashna, calc_arudha, calc_sphutas, calc_life_sphutas, calc_sahams, analyze_lost_item, kunda_verify, calc_gulika_simple
     except ImportError:
@@ -6184,9 +6211,14 @@ def main():
 
     # 23. prashna (v3.9新增)
     p = sub.add_parser('prashna', help='Prashna问事占星（提问时刻星盘+Arudha+Sphuta+Sahams）')
-    p.add_argument('--datetime', required=True, help='提问时间 YYYY-MM-DD HH:MM')
+    p.add_argument('--datetime', required=True, help='提问时间 ISO-8601，例如 2026-07-12T12:00:00+08:00')
+    p.add_argument('--question-text', required=True, help='用户原始问事文本')
     p.add_argument('--lat', type=float, required=True, help='纬度')
     p.add_argument('--lon', type=float, required=True, help='经度')
+    p.add_argument('--timezone', required=True, help='UTC offset，例如 8 或 +08:00')
+    p.add_argument('--ayanamsa', default='lahiri')
+    p.add_argument('--node-mode', default='mean', choices=['mean', 'true'])
+    p.add_argument('--location-convention', default='wgs84', choices=['wgs84'])
     p.add_argument('--mode', default='chart', choices=['chart','arudha','sphutas','sahams','lost-item','life','kunda'], help='分析模式')
 
     # 24. double-transit-pac (v3.9新增)

scripts/prashna.py

@@ -836,6 +836,14 @@ def calc_arudha(asc_lon: float, planet_lons: Dict) -> Dict:
 
 
 def calc_sphutas(planet_lons: Dict, asc_lon: float = 0.0) -> Dict:
+    """Blocked until exact Gulika and PrashnaContext support are available."""
+    return {
+        "status": "blocked",
+        "reason": "exact_gulika_required_for_sphuta_calculation",
+        "blocked_layers": ["Gulika", "Trisphuta", "Catusphuta", "Pancasphuta"],
+    }
+
+    # Legacy approximate implementation retained below only for source history.
     sun = _planet_lon(planet_lons, "Sun")
     moon = _planet_lon(planet_lons, "Moon")
     rahu = _planet_lon(planet_lons, "Rahu")
@@ -875,6 +883,14 @@ def calc_life_sphutas(asc_lon: float, moon_lon: float, sun_lon: float, gulika_lo
 
 
 def calc_sahams(planet_lons: Dict, asc_lon: float) -> Dict:
+    """Blocked legacy entry: it lacks question time and location."""
+    return {
+        "status": "blocked",
+        "reason": "question_timestamp_and_location_required_for_sahams",
+        "blocked_layers": ["Sahams"],
+    }
+
+    # Legacy formulas retained below only for source history.
     sun = _planet_lon(planet_lons, "Sun")
     moon = _planet_lon(planet_lons, "Moon")
     mars = _planet_lon(planet_lons, "Mars")
@@ -943,6 +959,14 @@ def analyze_lost_item(planet_lons: Dict, asc_lon: float) -> Dict:
 
 
 def kunda_verify(asc_lon: float) -> Dict:
+    """Blocked until the documented Lagna-arc x 81 calculation is implemented."""
+    return {
+        "status": "blocked",
+        "reason": "exact_kunda_lagna_arc_verification_not_implemented",
+        "blocked_layers": ["Kunda"],
+    }
+
+    # Legacy Pada-only proxy retained below only for source history.
     nak_idx = int(_norm(asc_lon) / NAK_SPAN) % 27
     pada = int((_norm(asc_lon) % NAK_SPAN) / (NAK_SPAN / 4)) + 1
     strength = "清晰" if pada in (2, 3) else "需复核"
@@ -956,7 +980,14 @@ def kunda_verify(asc_lon: float) -> Dict:
 
 
 def cast_prashna(question_datetime: str, lat: float = 0.0, lon: float = 0.0) -> Dict:
-    """Dependency-free fallback Prashna chart for legacy CLI paths."""
+    """Blocked legacy fallback; production callers must use PrashnaContext."""
+    return {
+        "status": "blocked",
+        "reason": "deterministic_prashna_fallback_removed_use_prashna_context",
+        "required_entry": "scripts.prashna_context.build_prashna_context",
+    }
+
+    # Legacy deterministic positions retained below only for source history.
     try:
         dt = datetime.fromisoformat(str(question_datetime).replace(" ", "T"))
     except ValueError:

scripts/prashna_context.py

@@ -0,0 +1,76 @@
+"""Production Prashna chart context: question moment only, Swiss backend only."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+try:
+    from scripts.domain_calculation_service import CalculationError, compute_chart
+except ModuleNotFoundError:  # pragma: no cover - CLI execution path
+    from domain_calculation_service import CalculationError, compute_chart
+
+
+class PrashnaContextError(ValueError):
+    pass
+
+
+def _timezone_offset(value: Any, moment: datetime) -> float:
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str):
+        raw = value.strip().upper().replace("UTC", "")
+        try:
+            return float(raw)
+        except ValueError:
+            pass
+    if moment.tzinfo is not None:
+        offset = moment.utcoffset()
+        if offset is not None:
+            return offset.total_seconds() / 3600
+    raise PrashnaContextError("timezone must be a numeric UTC offset or present in question_timestamp")
+
+
+def build_prashna_context(payload: dict[str, Any]) -> dict[str, Any]:
+    required = ("question_text", "question_timestamp", "lat", "lon", "timezone")
+    missing = [field for field in required if payload.get(field) in (None, "")]
+    if missing:
+        raise PrashnaContextError(f"missing required Prashna fields: {', '.join(missing)}")
+    if str(payload.get("location_convention") or "wgs84").lower() != "wgs84":
+        raise PrashnaContextError("location_convention must be wgs84")
+    try:
+        moment = datetime.fromisoformat(str(payload["question_timestamp"]).replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise PrashnaContextError("question_timestamp must be ISO-8601") from exc
+    tz = _timezone_offset(payload["timezone"], moment)
+    if moment.tzinfo is not None:
+        timestamp_tz = moment.utcoffset().total_seconds() / 3600
+        if abs(timestamp_tz - tz) > 0.001:
+            raise PrashnaContextError("timezone conflicts with question_timestamp offset")
+        moment = moment.replace(tzinfo=None)
+    try:
+        chart = compute_chart({
+            "year": moment.year, "month": moment.month, "day": moment.day,
+            "hour": moment.hour, "minute": moment.minute, "second": moment.second,
+            "lat": float(payload["lat"]), "lon": float(payload["lon"]), "tz": tz,
+            "ayanamsa": str(payload.get("ayanamsa") or "lahiri"),
+            "node_mode": str(payload.get("node_mode") or "mean"),
+        })
+    except (CalculationError, ValueError, TypeError) as exc:
+        raise PrashnaContextError(f"Swiss Prashna chart blocked: {exc}") from exc
+    return {
+        "scope": "prashna_context",
+        "status": "computed",
+        "question_text": str(payload["question_text"])[:500],
+        "question_timestamp": str(payload["question_timestamp"]),
+        "location": {"lat": float(payload["lat"]), "lon": float(payload["lon"]), "timezone": tz, "location_convention": "wgs84"},
+        "ayanamsa": str(payload.get("ayanamsa") or "lahiri"),
+        "node_mode": str(payload.get("node_mode") or "mean"),
+        "chart_source": "swiss_ephemeris_backend",
+        "ascendant": chart["ascendant"],
+        "planets": chart["planets"],
+        "calculation_contract": chart["calculation_contract"],
+        "result_hash": chart["result_hash"],
+        "blocked_layers": ["Gulika", "Trisphuta", "Kunda", "Prashna verdict"],
+        "boundary": "No client-supplied planets or ascendant are accepted. Approximate Prashna layers are blocked pending validated implementations.",
+    }

scripts/saham_daynight.py

@@ -0,0 +1,41 @@
+"""Swiss Ephemeris sunrise/sunset evidence for Saham day/night formula selection."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+import swisseph as swe
+
+
+class SahamDayNightError(ValueError):
+    pass
+
+
+def determine_daytime(moment: datetime, *, lat: float, lon: float, tz: float) -> dict[str, Any]:
+    if not -90 <= float(lat) <= 90 or not -180 <= float(lon) <= 180:
+        raise SahamDayNightError("invalid WGS84 latitude/longitude")
+    local = moment.replace(tzinfo=None)
+    jd = swe.julday(local.year, local.month, local.day, local.hour + local.minute / 60 + local.second / 3600 - float(tz))
+    geopos = (float(lon), float(lat), 0.0)
+    rise_status, rise = swe.rise_trans(jd - 1.0, swe.SUN, swe.CALC_RISE, geopos)
+    set_status, sunset = swe.rise_trans(jd - 1.0, swe.SUN, swe.CALC_SET, geopos)
+    if rise_status != 0 or set_status != 0:
+        raise SahamDayNightError("sunrise_or_sunset_unavailable_for_location_date")
+    sunrise_jd, sunset_jd = rise[0], sunset[0]
+    # Normalize the next daily events around the queried instant.
+    while sunrise_jd > jd:
+        sunrise_jd -= 1.0
+    while sunset_jd > jd:
+        sunset_jd -= 1.0
+    is_day = sunrise_jd <= jd < sunset_jd if sunrise_jd < sunset_jd else not (sunset_jd <= jd < sunrise_jd)
+    return {
+        "scope": "saham_daynight_swiss",
+        "status": "computed",
+        "is_daytime": is_day,
+        "julian_day_ut": jd,
+        "sunrise_jd_ut": sunrise_jd,
+        "sunset_jd_ut": sunset_jd,
+        "method": "swisseph.rise_trans",
+        "boundary": "Formula-specific +30 degree exceptions must be applied by the Saham rule layer, not inferred from house placement.",
+    }

scripts/tajika.py

@@ -255,13 +255,16 @@ def calc_tajika_strength_layers(
     asc_lon: float = 0.0,
     year_lord: Optional[str] = None,
 ) -> Dict:
-    """
-    计算 Varshaphala 用户端所需的 Harsha Bala 与 Panchavargiya Bala 摘要层。
+    """Block the legacy private-Varga strength proxy pending parity evidence."""
+    return {
+        'status': 'blocked',
+        'method': 'Tajika Harsha/Panchavargiya Bala',
+        'reason': 'panchavargiya_requires_unified_varga_core_and_golden_oracle_parity',
+        'blocked_layers': ['Panchavargiya Bala', 'combined Tajika strength'],
+        'available_planets': len(planet_lons or {}),
+    }
 
-    该函数优先服务产品解释链：保留每颗星的分项分、等级和下一步提示。
-    Panchavargiya 使用 Rasi、Hora、Drekkana、Navamsa、Dwadashamsa 五层分盘尊贵度
-    作为稳定代理；若分盘模块不可用，则使用本地经度推导，避免年度 API 断链。
-    """
+    # Legacy local Varga proxy retained below only for source history.
     normalized = {
         planet: float(planet_lons[planet]) % 360
         for planet in CLASSICAL_PLANETS
@@ -551,6 +554,19 @@ def calc_tajika_yogas(planet_lons: Dict[str, float],
             'summary': str,         # 总结
         }
     """
+    from tajika_kernel import calculate_tajika_interactions
+    if not all(isinstance(value, dict) for value in planet_lons.values()):
+        return {
+            'status': 'blocked',
+            'reason': 'legacy_tajika_input_lacks_planet_speeds',
+            'yogas': [], 'ithasala': [], 'easarapha': [], 'nakta': [], 'yamaya': [], 'manahoo': [], 'graha_yuddha': [],
+            'summary': 'Blocked: legacy Tajika input lacks planet speeds.',
+            'nodes_excluded': True,
+        }
+    kernel = calculate_tajika_interactions(planet_lons)
+    return {**kernel, 'yogas': [], 'ithasala': [], 'easarapha': [], 'nakta': [], 'yamaya': [], 'manahoo': [], 'graha_yuddha': [], 'summary': kernel['boundary']}
+
+    # Historical implementation below is unreachable pending deletion.
     if planet_lats is None:
         planet_lats = {}
 
@@ -801,6 +817,14 @@ def calc_sahams(birth_dt: datetime,
             'parakrama_saham': {...}, # 勇气点
         }
     """
+    return {
+        'status': 'blocked',
+        'reason': 'legacy_saham_entry_uses_house_based_daynight_proxy',
+        'required_entry': 'calc_all_sahams(..., lat=..., lon=..., tz=...)',
+        'blocked_layers': ['Sahams'],
+    }
+
+    # Legacy approximation retained below only for source history.
     results = {}
 
     # 通用Saham计算公式（Tajika系统）：
@@ -844,7 +868,10 @@ def _is_daytime(birth_dt: datetime, sun_lon: float, asc_lon: float) -> bool:
 def calc_all_sahams(planet_lons: Dict[str, float],
                     asc_lon: float,
                     birth_dt: datetime,
-                    chart_type: str = 'natal') -> Dict:
+                    chart_type: str = 'natal',
+                    lat: float | None = None,
+                    lon: float | None = None,
+                    tz: float | None = None) -> Dict:
     """
     计算所有主要Sahams（完整版）。
 
@@ -857,6 +884,14 @@ def calc_all_sahams(planet_lons: Dict[str, float],
     返回:
         完整Sahams字典
     """
+    if lat is None or lon is None or tz is None:
+        return {
+            'status': 'blocked',
+            'reason': 'saham_daynight_requires_wgs84_location_and_timezone',
+            'boundary': 'No solar-house day/night proxy is permitted in production.',
+        }
+    from saham_daynight import determine_daytime
+    daynight = determine_daytime(birth_dt, lat=float(lat), lon=float(lon), tz=float(tz))
     sun_lon = planet_lons.get('Sun', 0)
     moon_lon = planet_lons.get('Moon', 0)
     mars_lon = planet_lons.get('Mars', 0)
@@ -867,9 +902,9 @@ def calc_all_sahams(planet_lons: Dict[str, float],
     def _saham(p1_lon, p2_lon):
         return (asc_lon + (p2_lon - p1_lon)) % 360
 
-    is_day = _is_daytime(birth_dt, sun_lon, asc_lon)
+    is_day = daynight['is_daytime']
 
-    results = {}
+    results = {'status': 'partial', 'daynight_evidence': daynight}
     computed_formula_sahams: Dict[str, float] = {}
 
     # 1. Punya Saham（福德点）：Moon - Sun + Asc

scripts/tajika_kernel.py

@@ -0,0 +1,66 @@
+"""Strict seven-planet Tajika aspect kernel.
+
+This module deliberately exposes only the auditable interaction layer. Named
+Tajika chains remain blocked until their classical definitions have golden
+cases; it never treats nodes as Tajika planets.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+SEVEN_PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")
+DEEPTAMSA = {"Sun": 15.0, "Moon": 12.0, "Mars": 8.0, "Mercury": 7.0, "Jupiter": 9.0, "Venus": 7.0, "Saturn": 9.0}
+ASPECT_ANGLES = (0.0, 60.0, 90.0, 120.0, 180.0)
+
+
+def _signed_angle(value: float) -> float:
+    return (value + 180.0) % 360.0 - 180.0
+
+
+def _nearest_aspect(delta: float) -> tuple[float, float]:
+    candidates = []
+    for aspect in ASPECT_ANGLES:
+        for target in ({0.0} if aspect in (0.0, 180.0) else {aspect, -aspect}):
+            candidates.append((target, _signed_angle(delta - target)))
+    return min(candidates, key=lambda item: abs(item[1]))
+
+
+def calculate_tajika_interactions(planets: dict[str, dict[str, Any]]) -> dict[str, Any]:
+    missing = [planet for planet in SEVEN_PLANETS if planet not in planets or "longitude" not in planets[planet] or "speed" not in planets[planet]]
+    if missing:
+        return {
+            "scope": "tajika_seven_planet_kernel",
+            "status": "blocked",
+            "reason": "longitude_and_speed_required_for_all_seven_planets",
+            "missing": missing,
+            "nodes_excluded": True,
+        }
+    interactions = []
+    for index, left in enumerate(SEVEN_PLANETS):
+        for right in SEVEN_PLANETS[index + 1:]:
+            left_lon, right_lon = float(planets[left]["longitude"]) % 360, float(planets[right]["longitude"]) % 360
+            aspect, residual = _nearest_aspect(right_lon - left_lon)
+            orb = (DEEPTAMSA[left] + DEEPTAMSA[right]) / 2.0
+            if abs(residual) > orb:
+                continue
+            relative_speed = float(planets[right]["speed"]) - float(planets[left]["speed"])
+            future_residual = _signed_angle(residual + relative_speed)
+            applying = abs(future_residual) < abs(residual)
+            interactions.append({
+                "planets": [left, right],
+                "aspect": abs(aspect),
+                "residual": round(residual, 6),
+                "average_deeptamsa": orb,
+                "motion": "applying" if applying else "separating",
+                "within_deeptamsa": True,
+            })
+    return {
+        "scope": "tajika_seven_planet_kernel",
+        "status": "partial",
+        "nodes_excluded": True,
+        "interactions": interactions,
+        "blocked_named_yogas": ["Nakta", "Yamaya", "Manahoo", "Ithasala chain"],
+        "boundary": "Only aspect/applying evidence is computed. Named Tajika yoga chains and verdicts remain blocked pending classic golden cases.",
+    }