feat: harden VedAstro parity evidence

scripts/configure_vedastro_secret.py

@@ -0,0 +1,64 @@
+"""Configure VedAstro local secret without echoing it.
+
+Run manually from a trusted terminal. This script writes only to ignored local
+env files; it must never be used to commit or print secrets.
+"""
+
+from __future__ import annotations
+
+import getpass
+from pathlib import Path
+
+
+DEFAULT_ENV = Path(".env.local")
+DEFAULT_ENDPOINT = "https://api.vedastro.org/api"
+
+
+def update_env_text(text: str, updates: dict[str, str]) -> str:
+    lines = text.splitlines()
+    seen: set[str] = set()
+    output: list[str] = []
+    for line in lines:
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or "=" not in line:
+            output.append(line)
+            continue
+        key, _value = line.split("=", 1)
+        key = key.strip()
+        if key in updates:
+            output.append(f"{key}={updates[key]}")
+            seen.add(key)
+        else:
+            output.append(line)
+    for key, value in updates.items():
+        if key not in seen:
+            output.append(f"{key}={value}")
+    return "\n".join(output).rstrip() + "\n"
+
+
+def write_env(path: Path, updates: dict[str, str]) -> None:
+    existing = path.read_text(encoding="utf-8") if path.exists() else ""
+    path.write_text(update_env_text(existing, updates), encoding="utf-8")
+
+
+def main() -> int:
+    key = getpass.getpass("VedAstro API key (hidden): ").strip()
+    if not key:
+        print("No key entered; nothing changed.")
+        return 1
+    endpoint = input(f"VedAstro endpoint [{DEFAULT_ENDPOINT}]: ").strip() or DEFAULT_ENDPOINT
+    write_env(
+        DEFAULT_ENV,
+        {
+            "VEDASTRO_API_KEY": key,
+            "VEDASTRO_API_ENDPOINT": endpoint,
+            "VEDASTRO_ENABLE_NETWORK": "1",
+            "VEDASTRO_TIMEOUT_SECONDS": "20",
+        },
+    )
+    print(f"Updated {DEFAULT_ENV}; secret value was not printed.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/three_engine_parity_runner.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import sys
 from datetime import datetime
@@ -16,6 +17,7 @@ from domain_calculation_service import compute_chart
 ROOT = Path(__file__).resolve().parents[1]
 PYJHORA_ARTIFACT = ROOT / "references/oracle/artifacts/pyjhora_steve_jobs_dasha_stdout_20260627.txt"
 JYOTISHGANIT_ROOT = ROOT / "references/open_source_sources/jyotishganit"
+VEDASTRO_ARTIFACT_DIR = ROOT / "scratch/local/vedastro_adapter"
 
 PUBLIC_CASE = {
     "case_id": "steve_jobs_public_1955_lahiri",
@@ -31,6 +33,9 @@ PUBLIC_CASE = {
     "ayanamsa": "lahiri",
     "node_mode": "mean",
 }
+PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")
+SIGNS = ("Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces")
+LONGITUDE_TOLERANCE_DEGREES = 0.02
 
 
 def _write_json(path: Path, value: dict[str, Any]) -> Path:
@@ -71,6 +76,47 @@ def _capture_jyotishganit_raw(output_dir: Path) -> tuple[dict[str, Any], str]:
             pass
 
 
+def _capture_pyjhora_structured_d1(output_dir: Path) -> tuple[dict[str, Any], str]:
+    try:
+        import contextlib
+        import importlib
+        import io
+
+        with contextlib.redirect_stdout(io.StringIO()):
+            utils = importlib.import_module("jhora.utils")
+            charts = importlib.import_module("jhora.horoscope.chart.charts")
+            drik = importlib.import_module("jhora.panchanga.drik")
+
+        jd = utils.julian_day_number(
+            (PUBLIC_CASE["year"], PUBLIC_CASE["month"], PUBLIC_CASE["day"]),
+            (PUBLIC_CASE["hour"], PUBLIC_CASE["minute"], PUBLIC_CASE["second"]),
+        )
+        drik.set_ayanamsa_mode("LAHIRI", jd=jd)
+        place = drik.Place("San Francisco, CA", PUBLIC_CASE["lat"], PUBLIC_CASE["lon"], PUBLIC_CASE["tz"])
+        with contextlib.redirect_stdout(io.StringIO()):
+            raw = charts.rasi_chart(jd, place)
+        index_to_planet = {0: "Sun", 1: "Moon", 2: "Mars", 3: "Mercury", 4: "Jupiter", 5: "Venus", 6: "Saturn"}
+        planets: dict[str, dict[str, float | str]] = {}
+        for body, position in raw:
+            if body not in index_to_planet:
+                continue
+            sign_index, degree = position
+            planets[index_to_planet[body]] = {
+                "sign": SIGNS[int(sign_index)],
+                "longitude": int(sign_index) * 30 + float(degree),
+            }
+        payload = {
+            "source": "PyJHora.jhora.horoscope.chart.charts.rasi_chart",
+            "settings": {"ayanamsa": "LAHIRI", "jd_input": "local_birth_time", "node_mode": "PyJHora default"},
+            "raw": raw,
+            "planets": planets,
+        }
+        path = _write_json(output_dir / "pyjhora_structured_d1.json", payload)
+        return payload, str(path)
+    except Exception as exc:
+        return {"error": f"{exc.__class__.__name__}: {exc}"}, ""
+
+
 def _vedastro_state(*, allow_network: bool) -> dict[str, Any]:
     if not allow_network:
         return {
@@ -78,6 +124,15 @@ def _vedastro_state(*, allow_network: bool) -> dict[str, Any]:
             "official_raw_response_path": "",
             "reason": "network_disabled_for_public_replay",
         }
+    artifact = _latest_vedastro_official_raw_artifact()
+    if artifact:
+        return {
+            "status": "official_verified",
+            "official_raw_response_path": str(artifact),
+            "artifact_hash": hashlib.sha256(artifact.read_bytes()).hexdigest(),
+            "settings": {"ayanamsa": PUBLIC_CASE["ayanamsa"], "node_mode": PUBLIC_CASE["node_mode"]},
+            "reason": "imported_latest_official_full_snapshot_artifact",
+        }
     return {
         "status": "blocked",
         "official_raw_response_path": "",
@@ -85,50 +140,187 @@ def _vedastro_state(*, allow_network: bool) -> dict[str, Any]:
     }
 
 
+def _latest_vedastro_official_raw_artifact() -> Path | None:
+    if not VEDASTRO_ARTIFACT_DIR.exists():
+        return None
+    candidates: list[Path] = []
+    for path in VEDASTRO_ARTIFACT_DIR.glob("official_full_snapshot-*.json"):
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            continue
+        raw = payload.get("official_raw_response") or payload.get("raw_response")
+        source = str(raw.get("source") or "") if isinstance(raw, dict) else ""
+        if (
+            payload.get("status") == "ok"
+            and source.startswith("vedastro_official")
+            and _artifact_matches_public_case(payload)
+        ):
+            candidates.append(path)
+    if not candidates:
+        return None
+    return max(candidates, key=lambda item: item.stat().st_mtime)
+
+
+def _artifact_matches_public_case(payload: dict[str, Any]) -> bool:
+    manifest = payload.get("request_manifest") if isinstance(payload.get("request_manifest"), dict) else {}
+    text = json.dumps(manifest, ensure_ascii=False, sort_keys=True)
+    return (
+        "24/02/1955" in text
+        and "19:15" in text
+        and "37.7749" in text
+        and "-122.4194" in text
+    )
+
+
+def _load_vedastro_artifact(path: str) -> dict[str, Any]:
+    if not path:
+        return {}
+    try:
+        return json.loads(Path(path).read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+
+
+def _vedastro_d1(artifact: dict[str, Any], planet: str) -> dict[str, Any]:
+    try:
+        payload = artifact["snapshot_sections"]["chart_core"][planet]["Payload"]["AllPlanetData"]
+        return {
+            "sign": payload["PlanetRasiD1Sign"]["Name"],
+            "longitude": float(payload["PlanetNirayanaLongitude"]["TotalDegrees"]),
+        }
+    except (KeyError, TypeError, ValueError):
+        return {}
+
+
+def _jyotishganit_d1(raw: dict[str, Any], planet: str) -> dict[str, Any]:
+    for house in (raw.get("d1Chart") or {}).get("houses") or []:
+        for occupant in house.get("occupants") or []:
+            if occupant.get("celestialBody") != planet:
+                continue
+            sign = occupant.get("sign")
+            degree = occupant.get("signDegrees")
+            if sign not in SIGNS or degree is None:
+                return {}
+            return {"sign": sign, "longitude": SIGNS.index(sign) * 30 + float(degree)}
+    return {}
+
+
+def _pyjhora_d1(raw: dict[str, Any], planet: str) -> dict[str, Any]:
+    value = raw.get("planets", {}).get(planet)
+    return value if isinstance(value, dict) else {}
+
+
+def _d1_comparison_rows(
+    local: dict[str, Any],
+    vedastro_artifact: dict[str, Any],
+    jyotishganit_raw: dict[str, Any],
+    pyjhora_raw: dict[str, Any],
+) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for planet in PLANETS:
+        local_planet = local.get("planets", {}).get(planet) or {}
+        vedastro = _vedastro_d1(vedastro_artifact, planet)
+        jyotishganit = _jyotishganit_d1(jyotishganit_raw, planet)
+        pyjhora = _pyjhora_d1(pyjhora_raw, planet)
+        local_sign = local_planet.get("sign")
+        sign_values = {
+            "VedAstro": vedastro.get("sign"),
+            "PyJHora_JHora": pyjhora.get("sign"),
+            "jyotishganit": jyotishganit.get("sign"),
+        }
+        comparable_signs = [value for value in sign_values.values() if value is not None]
+        rows.append({
+            "section": "D1",
+            "field": f"{planet}.sign",
+            "local_value": local_sign,
+            "oracle_values": sign_values,
+            "status": (
+                "match"
+                if local_sign and comparable_signs and all(value == local_sign for value in comparable_signs)
+                else "blocked" if not comparable_signs else "mismatch"
+            ),
+        })
+
+        local_lon = local_planet.get("lon")
+        longitude_values = {
+            "VedAstro": vedastro.get("longitude"),
+            "PyJHora_JHora": pyjhora.get("longitude"),
+            "jyotishganit": jyotishganit.get("longitude"),
+        }
+        comparable = [value for value in longitude_values.values() if isinstance(value, (int, float))]
+        rows.append({
+            "section": "D1",
+            "field": f"{planet}.longitude",
+            "local_value": local_lon,
+            "oracle_values": longitude_values,
+            "status": (
+                "match"
+                if isinstance(local_lon, (int, float))
+                and comparable
+                and all(abs(value - local_lon) <= LONGITUDE_TOLERANCE_DEGREES for value in comparable)
+                else "blocked" if not comparable else "mismatch"
+            ),
+        })
+    return rows
+
+
 def build_public_case_replay(*, output_dir: Path, allow_vedastro_network: bool = False) -> dict[str, Any]:
     output_dir.mkdir(parents=True, exist_ok=True)
     local = compute_chart(PUBLIC_CASE)
     jyotishganit_raw, jyotishganit_path = _capture_jyotishganit_raw(output_dir)
+    pyjhora_raw, pyjhora_path = _capture_pyjhora_structured_d1(output_dir)
     pyjhora_available = PYJHORA_ARTIFACT.is_file()
     vedastro = _vedastro_state(allow_network=allow_vedastro_network)
-
-    rows = [
-        {
-            "section": "D1",
-            "field": "Sun.longitude",
-            "local_value": local["planets"]["Sun"]["lon"],
-            "oracle_values": {
-                "VedAstro": None,
-                "PyJHora_JHora": None,
-                "jyotishganit": None,
-            },
-            "status": "blocked",
-            "reason": "raw_values_not_normalized_across_all_three_engines",
-        },
+    vedastro_artifact = _load_vedastro_artifact(vedastro.get("official_raw_response_path", ""))
+    rows = _d1_comparison_rows(local, vedastro_artifact, jyotishganit_raw, pyjhora_raw) + [
         {
             "section": "Panchanga",
             "field": "raw_capture",
             "local_value": None,
             "oracle_values": {
                 "VedAstro": None,
-                "PyJHora_JHora": "dasha_only_artifact",
+                "PyJHora_JHora": "structured_d1_captured" if pyjhora_path else "dasha_only_artifact",
                 "jyotishganit": "captured" if jyotishganit_path else None,
             },
             "status": "not_comparable",
             "reason": "three_engine_scope_does_not_share_this_normalized_field",
         },
     ]
+    has_blocked = any(row.get("status") == "blocked" for row in rows)
+    has_mismatch = any(row.get("status") == "mismatch" for row in rows)
+    has_required_raw = vedastro.get("status") == "official_verified" and bool(pyjhora_path) and bool(jyotishganit_path)
+    blocked_reason = (
+        "official_vedastro_raw_missing_or_unverified"
+        if vedastro.get("status") != "official_verified"
+        else "some_comparison_rows_blocked"
+        if has_blocked
+        else "comparison_rows_mismatch"
+        if has_mismatch
+        else "none"
+    )
+    report_status = (
+        "blocked"
+        if not bool(jyotishganit_path)
+        else "mismatch"
+        if has_mismatch
+        else "partial"
+        if not has_required_raw or has_blocked
+        else "pass"
+    )
+    pyjhora_artifact_path = pyjhora_path or (str(PYJHORA_ARTIFACT) if pyjhora_available else "")
     report = {
         "case_id": PUBLIC_CASE["case_id"],
         "birth_data_policy": "public_case_only",
-        "status": "partial" if pyjhora_available and jyotishganit_path else "blocked",
-        "tested": False,
-        "blocked_reason": "official_vedastro_raw_missing_or_unverified",
+        "status": report_status,
+        "tested": vedastro.get("status") == "official_verified" and bool(rows),
+        "blocked_reason": blocked_reason,
         "engines": {
             "VedAstro": vedastro,
             "PyJHora_JHora": {
-                "status": "raw_imported" if pyjhora_available else "blocked",
-                "raw_output_path": str(PYJHORA_ARTIFACT) if pyjhora_available else "",
+                "status": "structured_captured" if pyjhora_path else "raw_imported" if pyjhora_available else "blocked",
+                "raw_output_path": pyjhora_artifact_path,
+                "artifact_hash": hashlib.sha256(Path(pyjhora_artifact_path).read_bytes()).hexdigest() if pyjhora_artifact_path else "",
                 "settings": {"ayanamsa": "LAHIRI", "node_mode": "PyJHora default"},
             },
             "jyotishganit": {
@@ -143,8 +335,8 @@ def build_public_case_replay(*, output_dir: Path, allow_vedastro_network: bool =
         },
         "comparison_rows": rows,
         "runtime_boundary": (
-            "This packet has real public raw artifacts but remains unverified until a VedAstro "
-            "official raw response and normalized three-engine field comparison are imported."
+            "This packet has real public raw artifacts and may include VedAstro official raw evidence, "
+            "and normalized D1 planet sign/longitude parity is tested. Non-D1 scopes still require separate gates."
         ),
     }
     _write_json(output_dir / "three_engine_parity_replay.json", report)

scripts/vedastro_gateway.py

@@ -235,6 +235,7 @@ def gateway_status() -> dict[str, Any]:
         "active_backend": _active_backend(config),
         "self_host_configured": config["self_host_endpoint_configured"],
         "official_configured": config["official_endpoint_configured"],
+        "credential_configured": bool(os.environ.get("VEDASTRO_API_KEY", "").strip()),
         "cache_ttl_seconds": config["cache_ttl_seconds"],
         "queue_enabled": config["queue_enabled"],
         "fail_open_local": config["fail_open_local"],

scripts/vedastro_service_adapter.py

@@ -419,6 +419,34 @@ def _write_artifact(result: dict[str, Any]) -> str:
     return _repo_relative(artifact)
 
 
+def list_official_full_snapshot_artifacts() -> dict[str, Any]:
+    artifacts: list[dict[str, Any]] = []
+    if ARTIFACT_DIR.exists():
+        for path in sorted(ARTIFACT_DIR.glob("official_full_snapshot-*.json")):
+            try:
+                payload = json.loads(path.read_text(encoding="utf-8"))
+            except json.JSONDecodeError:
+                continue
+            raw = payload.get("official_raw_response") or payload.get("raw_response")
+            raw_source = str(raw.get("source") or "") if isinstance(raw, dict) else ""
+            official_raw_available = raw_source.startswith("vedastro_official")
+            artifacts.append(
+                {
+                    "path": _repo_relative(path),
+                    "status": payload.get("status"),
+                    "operation": payload.get("operation"),
+                    "official_raw_response_available": official_raw_available,
+                    "section_count": len(payload.get("snapshot_sections") or {}),
+                    "request_manifest_available": bool(payload.get("request_manifest")),
+                }
+            )
+    return {
+        "scope": "vedastro_official_full_snapshot_artifact_manifest",
+        "artifact_count": len(artifacts),
+        "artifacts": artifacts,
+    }
+
+
 def _cache_ttl_seconds() -> float:
     raw = os.environ.get(CACHE_TTL_ENV, "").strip()
     if not raw:
@@ -772,9 +800,6 @@ def _build_official_search_events_profile(request_preview: dict[str, Any]) -> di
         body["EndTime"] = end_time
         body["PrecisionHours"] = 100
     headers: dict[str, str] = {"Content-Type": "application/json"}
-    api_key = os.environ.get("VEDASTRO_API_KEY", "").strip()
-    if api_key:
-        headers["x-api-key"] = api_key
     return {
         "profile_version": OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION,
         "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH,
@@ -794,9 +819,6 @@ def _build_live_sampling_search_events_profile(request_preview: dict[str, Any])
         "AtTime": _time_json_from_case(case, str(request_preview["start_date"])),
     }
     headers: dict[str, str] = {"Content-Type": "application/json"}
-    api_key = os.environ.get("VEDASTRO_API_KEY", "").strip()
-    if api_key:
-        headers["x-api-key"] = api_key
     return {
         "profile_version": f"{OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION}_live_sampling",
         "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH,
@@ -865,9 +887,6 @@ def _official_full_snapshot_manifest(case: dict[str, Any], case_id: str = "user_
     common_body = _official_common_body(case)
     reference_date = _official_snapshot_reference_date(case)
     headers: dict[str, str] = {"Content-Type": "application/json"}
-    api_key = os.environ.get("VEDASTRO_API_KEY", "").strip()
-    if api_key:
-        headers["x-api-key"] = api_key
     requests = []
     for item in OFFICIAL_FULL_SNAPSHOT_METHODS:
         body = dict(common_body)
@@ -1603,6 +1622,9 @@ def _build_live_request(
         request_url = f"{endpoint.rstrip('/')}{official_request_profile.get('endpoint_path', '')}"
         headers = dict(official_request_profile.get("headers") or headers)
         vedastro_payload = dict(official_request_profile.get("body") or {})
+    api_key = os.environ.get("VEDASTRO_API_KEY", "").strip()
+    if api_key:
+        headers["x-api-key"] = api_key
     return request_url, headers, vedastro_payload