fix: restore ayanamsa state and keep tests hermetic

scripts/ayanamsa_utils.py

@@ -10,7 +10,8 @@ comparison fixtures must pass an explicit `lahiri` name.
 
 from __future__ import annotations
 
-from typing import Any
+from contextlib import contextmanager
+from typing import Any, Iterator
 
 try:
     import swisseph as swe
@@ -91,6 +92,7 @@ def current_ayanamsa_name(args: Any = None) -> str:
 
 def apply_ayanamsa(name: Any = None, swe_module: Any = None) -> bool:
     """Set Swiss Ephemeris sidereal mode explicitly."""
+    global ACTIVE_AYANAMSA_NAME
     module = swe_module or swe
     if module is None:
         return False
@@ -99,9 +101,24 @@ def apply_ayanamsa(name: Any = None, swe_module: Any = None) -> bool:
         module.set_sid_mode(AYANAMSA_MODES[key])
     except Exception:
         return False
+    ACTIVE_AYANAMSA_NAME = key
     return True
 
 
+@contextmanager
+def temporary_ayanamsa(name: Any = None, swe_module: Any = None) -> Iterator[str]:
+    """Apply a sidereal mode and restore the previous process-wide mode."""
+    previous = ACTIVE_AYANAMSA_NAME
+    target = normalize_ayanamsa_name(name)
+    if not apply_ayanamsa(target, swe_module):
+        raise RuntimeError(f"failed to apply ayanamsa {target}")
+    try:
+        yield target
+    finally:
+        if not apply_ayanamsa(previous, swe_module):
+            raise RuntimeError(f"failed to restore ayanamsa {previous}")
+
+
 def sidereal_flags(swe_module: Any, ayanamsa_name: Any = None) -> int:
     apply_ayanamsa(ayanamsa_name, swe_module)
     return getattr(swe_module, 'FLG_SWIEPH', 2) | getattr(swe_module, 'FLG_SIDEREAL', 65536)

scripts/candidate_time_sensitivity_scan.py

@@ -11,6 +11,10 @@ from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
 
+try:
+    from scripts.ayanamsa_utils import DEFAULT_AYANAMSA_NAME
+except ModuleNotFoundError:  # pragma: no cover - direct script execution
+    from ayanamsa_utils import DEFAULT_AYANAMSA_NAME
 try:
     from scripts.rectification_input_contract import (
         candidate_input_fingerprint,
@@ -109,7 +113,7 @@ def scan_candidate_times(payload: dict[str, Any], *, uncertainty_minutes: int =
             "version": "rectification-input-v1",
             "center_input_fingerprint": candidate_input_fingerprint(payload),
             "settings": {
-                "ayanamsa": str(payload.get("ayanamsa") or "lahiri").strip().lower(),
+                "ayanamsa": str(payload.get("ayanamsa") or DEFAULT_AYANAMSA_NAME).strip().lower(),
                 "node_mode": str(
                     payload.get("node_mode", payload.get("nodeMode", "mean"))
                 ).strip().lower(),

scripts/cmd_muhurta.py

@@ -17,7 +17,7 @@ import os
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
-from ayanamsa_utils import sidereal_flags
+from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, sidereal_flags
 
 
 def _weekday_to_vara(py_weekday: int) -> int:
@@ -26,7 +26,7 @@ def _weekday_to_vara(py_weekday: int) -> int:
 
 
 def _get_sun_moon_lons(year: int, month: int, day: int,
-                       hour: int = 12, ayanamsa_name: str = 'lahiri') -> tuple:
+                       hour: int = 12, ayanamsa_name: str = DEFAULT_AYANAMSA_NAME) -> tuple:
     """获取太阳/月亮恒星黄经。优先 swisseph，退化为近似算法。"""
     def _primary_longitude(value):
         if isinstance(value, (list, tuple)):
@@ -112,7 +112,7 @@ def cmd_muhurta(args, chart_data: Optional[Dict] = None) -> int:
     hour_from_sunrise = getattr(args, 'hour_from_sunrise', 6.0)
     target_activity = getattr(args, 'activity', None)
     scan_days = getattr(args, 'scan_days', 1)
-    ayanamsa_name = getattr(args, 'ayanamsa', 'lahiri')
+    ayanamsa_name = getattr(args, 'ayanamsa', None) or DEFAULT_AYANAMSA_NAME
 
     activities = list(ACTIVITY_RULES.keys())
     if target_activity and target_activity not in activities:

scripts/cmd_solar_return.py

@@ -28,6 +28,7 @@ def cmd_solar_return(args: Any) -> Dict[str, Any]:
     """
     try:
         sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+        from ayanamsa_utils import DEFAULT_AYANAMSA_NAME
         from solar_return import solar_return_full_report
     except ImportError as e:
         return {"error": f"solar_return 模块导入失败: {e}"}
@@ -41,7 +42,7 @@ def cmd_solar_return(args: Any) -> Dict[str, Any]:
         args.hour, args.minute,
         args.lat, args.lon, args.tz,
         args.target_year,
-        ayanamsa_name=getattr(args, 'ayanamsa', 'lahiri'),
+        ayanamsa_name=getattr(args, 'ayanamsa', None) or DEFAULT_AYANAMSA_NAME,
     )
 
 

scripts/collect_vedastro_official_parity_raw.py

@@ -144,7 +144,7 @@ def collect_chart_core(
     settings = {
         "case_id": case_id,
         "birth": {key: case.get(key) for key in ("year", "month", "day", "hour", "minute", "second", "lat", "lon", "tz")},
-        "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri",
+        "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri",  # 参考口径，勿改为默认
         "node_policy": case.get("node_policy") or case.get("node_mode") or "mean",
         "endpoint_host": adapter._endpoint_host(endpoint),
         "calculator": "AllPlanetData",

scripts/ephemeris_adapter_contract.py

@@ -22,6 +22,7 @@ if str(SCRIPT_DIR) not in sys.path:
     sys.path.insert(0, str(SCRIPT_DIR))
 
 from jyotish_engine import HAS_SWE, compute_chart_data  # noqa: E402
+from ayanamsa_utils import DEFAULT_AYANAMSA_NAME  # noqa: E402
 
 
 BODY_KEYS = ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"]
@@ -40,7 +41,7 @@ class EphemerisAdapterContract:
     lat: float
     lon: float
     tz: float
-    ayanamsa_policy: str = "lahiri"
+    ayanamsa_policy: str = DEFAULT_AYANAMSA_NAME
     node_policy: str = "mean"
     body_list: tuple[str, ...] = tuple(BODY_KEYS)
 

scripts/full_technique_invocation_matrix.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 
+import argparse
 import hashlib
 import json
 import re
@@ -9,7 +10,7 @@ from pathlib import Path
 
 
 ROOT = Path(__file__).resolve().parents[1]
-OUTPUT = ROOT / "references/oracle/full_technique_invocation_matrix_2026_07_22.json"
+OUTPUT_NAME = "full_technique_invocation_matrix_2026_07_22.json"
 
 SCAN_ROOTS = {
     "skill": ["SKILL.md", "skills"],
@@ -339,13 +340,18 @@ def _claim_boundary(tech: Technique, claim_status: str) -> str:
 
 
 def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output-dir", type=Path, default=ROOT / "references/oracle")
+    args = parser.parse_args()
     packet = build_matrix()
     without_hash = dict(packet)
     without_hash["content_hash"] = ""
     digest = hashlib.sha256(json.dumps(without_hash, ensure_ascii=False, sort_keys=True).encode("utf-8")).hexdigest()
     packet["content_hash"] = digest
-    OUTPUT.write_text(json.dumps(packet, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    print(json.dumps({"path": str(OUTPUT.relative_to(ROOT)), "sha256": digest, "summary": packet["summary"]}, ensure_ascii=False, indent=2))
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    output = args.output_dir / OUTPUT_NAME
+    output.write_text(json.dumps(packet, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps({"path": str(output), "sha256": digest, "summary": packet["summary"]}, ensure_ascii=False, indent=2))
 
 
 if __name__ == "__main__":

scripts/gulika.py

@@ -12,9 +12,9 @@ try:
 except ImportError:
     from scripts.saham_daynight import determine_daytime
 try:
-    from ayanamsa_utils import ayanamsa_display_name, normalize_ayanamsa_name, sidereal_flags
+    from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, ayanamsa_display_name, normalize_ayanamsa_name, temporary_ayanamsa
 except ImportError:
-    from scripts.ayanamsa_utils import ayanamsa_display_name, normalize_ayanamsa_name, sidereal_flags
+    from scripts.ayanamsa_utils import DEFAULT_AYANAMSA_NAME, ayanamsa_display_name, normalize_ayanamsa_name, temporary_ayanamsa
 
 
 # Monday=0, matching datetime.weekday(). Values are the end of Saturn's share
@@ -72,10 +72,10 @@ TEMPORAL_UPAGRAHAS = {
 }
 
 
-def _sidereal_ascendant(jd_ut: float, lat: float, lon: float, ayanamsa: str = "raman") -> float:
-    flags = sidereal_flags(swe, ayanamsa)
-    cusps, ascmc = swe.houses_ex(jd_ut, lat, lon, b"P", flags)
-    return float(ascmc[0]) % 360
+def _sidereal_ascendant(jd_ut: float, lat: float, lon: float, ayanamsa: str = DEFAULT_AYANAMSA_NAME) -> float:
+    with temporary_ayanamsa(ayanamsa, swe):
+        _, ascmc = swe.houses_ex(jd_ut, lat, lon, b"P", swe.FLG_SWIEPH | swe.FLG_SIDEREAL)
+        return float(ascmc[0]) % 360
 
 
 def calculate_gulika(
@@ -85,7 +85,7 @@ def calculate_gulika(
     lon: float,
     tz: float,
     method: str = "saturn_part_start",
-    ayanamsa: str = "raman",
+    ayanamsa: str = DEFAULT_AYANAMSA_NAME,
 ) -> dict[str, Any]:
     """Return Gulika from local moment/location using Swiss sunrise and sunset."""
     ayanamsa_name = normalize_ayanamsa_name(ayanamsa)
@@ -140,7 +140,7 @@ def _temporal_upagraha(
     tz: float,
     planet_index: int,
     segment_part: str,
-    ayanamsa: str = "raman",
+    ayanamsa: str = DEFAULT_AYANAMSA_NAME,
 ) -> dict[str, Any]:
     ayanamsa_name = normalize_ayanamsa_name(ayanamsa)
     daynight = determine_daytime(moment, lat=lat, lon=lon, tz=tz)
@@ -198,7 +198,7 @@ def calculate_upagrahas(
     lon: float,
     tz: float,
     sun_longitude: float,
-    ayanamsa: str = "raman",
+    ayanamsa: str = DEFAULT_AYANAMSA_NAME,
 ) -> dict[str, Any]:
     """Calculate the D1 Upagraha packet using local Swiss Ephemeris inputs."""
     ayanamsa_name = normalize_ayanamsa_name(ayanamsa)

scripts/jyotish_api_server.py

@@ -1046,7 +1046,9 @@ def _attach_gulika_observation(chart: dict, birth_payload: dict) -> None:
             int(float(birth_payload.get('minute', 0) or 0)),
             int(float(birth_payload.get('second', 0) or 0)),
         )
-        result = _load_local_module('gulika').calculate_gulika(moment, lat=lat, lon=lon, tz=tz)
+        result = _load_local_module('gulika').calculate_gulika(
+            moment, lat=lat, lon=lon, tz=tz, ayanamsa=_request_ayanamsa(birth_payload)
+        )
         result['claim_boundary'] = 'observation_only; not a verified event claim'
         modules['gulika'] = result
     except Exception as exc:

scripts/local_accuracy_report.py

@@ -227,7 +227,7 @@ def build_report() -> dict[str, Any]:
     }
     return {
         "scope": "local_jyotish_accuracy_report",
-        "reference_ayanamsa": "lahiri",
+        "reference_ayanamsa": "lahiri",  # 参考口径，勿改为默认
         "generated_at": datetime.now(timezone.utc).isoformat(),
         "summary": summary,
         "checks": checks,

scripts/muhurta.py

@@ -18,7 +18,7 @@ from datetime import datetime, timedelta
 import calendar
 import math
 
-from ayanamsa_utils import sidereal_flags
+from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, sidereal_flags
 
 # ── Vara（周日行星）──────────────────────────────────────────────────
 VARA_LORDS = {
@@ -2138,7 +2138,7 @@ def _local_time_from_jd(jd: float, tz: float) -> str:
     return local_dt.strftime('%Y-%m-%d %H:%M')
 
 
-def _swisseph_sun_moon_lon(jd: float, ayanamsa_name: str = 'lahiri') -> Optional[Tuple[float, float]]:
+def _swisseph_sun_moon_lon(jd: float, ayanamsa_name: str = DEFAULT_AYANAMSA_NAME) -> Optional[Tuple[float, float]]:
     try:
         import swisseph as swe
         flags = sidereal_flags(swe, ayanamsa_name)
@@ -2192,7 +2192,7 @@ def calc_panchanga_end_times(
     day: int,
     tz: float,
     local_hours: float,
-    ayanamsa_name: str = 'raman',
+    ayanamsa_name: str = DEFAULT_AYANAMSA_NAME,
 ) -> Optional[Dict]:
     """Calculate current Tithi/Nakshatra/Yoga end times from SwissEph positions."""
     jd_start = _jd_for_local_time(year, month, day, local_hours, tz)
@@ -2903,7 +2903,7 @@ def panchanga_range_report(
     lat: Optional[float] = None,
     lon: Optional[float] = None,
     tz: Optional[float] = None,
-    ayanamsa_name: str = 'raman',
+    ayanamsa_name: str = DEFAULT_AYANAMSA_NAME,
 ) -> Dict:
     """Build a date-range Panchanga calendar with daytime inauspicious windows."""
     start_dt = datetime.strptime(start_date[:10], '%Y-%m-%d')
@@ -3034,7 +3034,7 @@ def muhurta_range_search(
     lon: Optional[float] = None,
     tz: Optional[float] = None,
     avoid_inauspicious_periods: bool = True,
-    ayanamsa_name: str = 'raman',
+    ayanamsa_name: str = DEFAULT_AYANAMSA_NAME,
 ) -> Dict:
     """Search a date range for ranked Muhurta candidates for a selected activity."""
     limit = max(1, min(int(limit or 5), 20))
@@ -3106,7 +3106,7 @@ def build_muhurta_sidecar(
     lat: Optional[float] = None,
     lon: Optional[float] = None,
     tz: Optional[float] = None,
-    ayanamsa_name: str = 'raman',
+    ayanamsa_name: str = DEFAULT_AYANAMSA_NAME,
 ) -> Dict:
     """Compact Muhurta/Panchanga packet for unified workflow consumers."""
     activity = activity if activity in ACTIVITY_RULES else 'business'

scripts/muhurta_factor_only_scoring_packet.py

@@ -2,6 +2,7 @@
 """Capture a Muhurta factor-only scoring observation packet."""
 from __future__ import annotations
 
+import argparse
 import hashlib
 import json
 from pathlib import Path
@@ -11,7 +12,7 @@ from muhurta_factor_probe import build_probe
 
 
 ROOT = Path(__file__).resolve().parents[1]
-OUTPUT = ROOT / "references/oracle/muhurta_factor_only_scoring_packet_2026_07_23.json"
+OUTPUT_NAME = "muhurta_factor_only_scoring_packet_2026_07_23.json"
 
 
 def _stable_json(data: Any) -> str:
@@ -63,8 +64,14 @@ def build() -> dict[str, Any]:
 
 
 def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output-dir", type=Path, default=ROOT / "references/oracle")
+    args = parser.parse_args()
     packet = build()
-    OUTPUT.write_text(json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    (args.output_dir / OUTPUT_NAME).write_text(
+        json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
     print(json.dumps(packet, ensure_ascii=False, sort_keys=True))
     return 0
 

scripts/oracle_boundary_audit.py

@@ -256,7 +256,7 @@ def _namespace_from_template(case: dict[str, Any]) -> argparse.Namespace:
         birth["node_mode"] = settings["node_mode"]
     return _namespace_from_birth(
         birth,
-        ayanamsa=settings.get("ayanamsa", "lahiri"),
+        ayanamsa=settings.get("ayanamsa", "lahiri"),  # 参考口径，勿改为默认
         moon_lon=None,
         nakshatra=None,
         pada=None,

scripts/prashna_context.py

@@ -9,6 +9,10 @@ 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
+try:
+    from scripts.ayanamsa_utils import DEFAULT_AYANAMSA_NAME, normalize_ayanamsa_name
+except ModuleNotFoundError:  # pragma: no cover - CLI execution path
+    from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, normalize_ayanamsa_name
 try:
     from scripts.gulika import calculate_gulika
 except ModuleNotFoundError:  # pragma: no cover - CLI execution path
@@ -39,6 +43,10 @@ def _timezone_offset(value: Any, moment: datetime) -> float:
     raise PrashnaContextError("timezone must be a numeric UTC offset or present in question_timestamp")
 
 
+def _payload_ayanamsa(payload: dict[str, Any]) -> str:
+    return normalize_ayanamsa_name(payload.get("ayanamsa") or DEFAULT_AYANAMSA_NAME)
+
+
 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, "")]
@@ -56,18 +64,21 @@ def build_prashna_context(payload: dict[str, Any]) -> dict[str, Any]:
         if abs(timestamp_tz - tz) > 0.001:
             raise PrashnaContextError("timezone conflicts with question_timestamp offset")
         moment = moment.replace(tzinfo=None)
+    ayanamsa = _payload_ayanamsa(payload)
     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"),
+            "ayanamsa": ayanamsa,
             "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
     try:
-        gulika = calculate_gulika(moment, lat=float(payload["lat"]), lon=float(payload["lon"]), tz=tz)
+        gulika = calculate_gulika(
+            moment, lat=float(payload["lat"]), lon=float(payload["lon"]), tz=tz, ayanamsa=ayanamsa
+        )
     except Exception as exc:
         gulika = {
             "status": "blocked",
@@ -92,7 +103,7 @@ def build_prashna_context(payload: dict[str, Any]) -> dict[str, Any]:
         "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"),
+        "ayanamsa": ayanamsa,
         "node_mode": str(payload.get("node_mode") or "mean"),
         "chart_source": "swiss_ephemeris_backend",
         "ascendant": chart["ascendant"],

scripts/prashna_sphuta_oss_case_probe.py

@@ -37,5 +37,5 @@ def build(date):
     return data
 
 def main():
-    ap=argparse.ArgumentParser(); ap.add_argument('--date',default='2026-07-20'); args=ap.parse_args(); data=build(args.date); OUT.write_text(json.dumps(data,ensure_ascii=False,indent=2,sort_keys=True)+'\n'); print(json.dumps(data,ensure_ascii=False,indent=2,sort_keys=True))
+    ap=argparse.ArgumentParser(); ap.add_argument('--date',default='2026-07-20'); ap.add_argument('--output-dir',type=Path,default=OUT.parent); args=ap.parse_args(); data=build(args.date); args.output_dir.mkdir(parents=True,exist_ok=True); (args.output_dir/OUT.name).write_text(json.dumps(data,ensure_ascii=False,indent=2,sort_keys=True)+'\n'); print(json.dumps(data,ensure_ascii=False,indent=2,sort_keys=True))
 if __name__=='__main__': main()

scripts/pyjhora_multi_case_panchanga_gulika_replay.py

@@ -53,7 +53,7 @@ def build_report() -> dict[str, Any]:
                     "abhijit_muhurta": drik.abhijit_muhurta(local_jd, place),
                 }
                 utc_jd = local_jd - tz / 24
-                sun_moon = _swisseph_sun_moon_lon(utc_jd, "lahiri")
+                sun_moon = _swisseph_sun_moon_lon(utc_jd, "lahiri")  # 参考口径，勿改为默认
                 local = calc_panchanga(*sun_moon, weekday=(datetime(y, m, d).weekday() + 1) % 7)
                 local_gulika = calculate_gulika(datetime(y, m, d, hh, mm, ss), lat=lat, lon=lon, tz=tz)
                 py_gulika_deg = raw["gulika"][0] * 30 + raw["gulika"][1]

scripts/shadbala_dig_source_of_truth_audit.py

@@ -70,7 +70,7 @@ def _namespace_from_template(case: dict[str, Any]) -> Any:
             "lat": birth["lat"],
             "lon": birth["lon"],
             "tz": birth["tz"],
-            "ayanamsa": settings.get("ayanamsa", "lahiri"),
+            "ayanamsa": settings.get("ayanamsa", "lahiri"),  # 参考口径，勿改为默认
             "node_mode": settings.get("node_mode", "mean"),
         },
     )()