Add external oracle parity tooling

scripts/external_oracle_raw_import.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Validate a reviewable external raw-oracle artifact before parity replay."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+
+SUPPORTED_ENGINES = {"VedAstro", "PyJHora_JHora", "jyotishganit"}
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def build_raw_oracle_import(engine: str, artifact_path: str | Path, metadata: dict[str, Any]) -> dict[str, Any]:
+    if engine not in SUPPORTED_ENGINES:
+        raise ValueError(f"unsupported oracle engine: {engine}")
+    path = Path(artifact_path).expanduser().resolve()
+    if not path.is_file():
+        raise ValueError("source artifact does not exist")
+    required = ("case_id", "license_boundary", "collection_method", "birth_data_policy")
+    missing = [key for key in required if not metadata.get(key)]
+    if missing:
+        raise ValueError(f"missing raw-oracle metadata: {', '.join(missing)}")
+    if metadata["birth_data_policy"] != "public_case_only":
+        raise ValueError("raw-oracle imports require public_case_only birth data")
+    return {
+        "scope": "external_raw_oracle_import",
+        "schema_version": 1,
+        "engine": engine,
+        "status": "raw_imported_uncompared",
+        "source_artifact": str(path),
+        "source_artifact_sha256": sha256_file(path),
+        "metadata": {
+            key: metadata[key]
+            for key in (*required, "engine_version", "ayanamsa", "node_mode", "captured_at")
+            if metadata.get(key) is not None
+        },
+        "comparison_ready": False,
+        "boundary": "Import integrity only. Parity is external_verified only after normalized field comparison passes.",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--engine", required=True, choices=sorted(SUPPORTED_ENGINES))
+    parser.add_argument("--artifact", required=True)
+    parser.add_argument("--metadata-json", required=True, help="JSON file containing import metadata")
+    args = parser.parse_args()
+    metadata = json.loads(Path(args.metadata_json).read_text(encoding="utf-8"))
+    print(json.dumps(build_raw_oracle_import(args.engine, args.artifact, metadata), ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/pyjhora_parity_summary.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Summarize reviewable PyJHora comparison matrices without overstating coverage."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any
+
+
+REQUIRED_FULL_PARITY = ("D1", "D9", "D10", "D2", "D4", "Vimshottari", "Shadbala", "Ashtakavarga")
+MATRIX_SECTION_MAP = {"ascendant": "D1", "planet": "D1", "dasha": "Vimshottari", "D9": "D9", "D10": "D10"}
+
+
+def summarize_matrix(path: str | Path, *, settings: dict[str, Any]) -> dict[str, Any]:
+    path = Path(path)
+    rows = list(csv.DictReader(path.open(encoding="utf-8")))
+    status_counts = Counter(str(row.get("status") or "unknown") for row in rows)
+    sections: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "match": 0, "mismatch": 0})
+    covered = set()
+    for row in rows:
+        section = MATRIX_SECTION_MAP.get(str(row.get("section") or ""))
+        if not section:
+            continue
+        covered.add(section)
+        sections[section]["total"] += 1
+        if row.get("status") == "match":
+            sections[section]["match"] += 1
+        elif row.get("status") == "mismatch":
+            sections[section]["mismatch"] += 1
+    missing = [field for field in REQUIRED_FULL_PARITY if field not in covered]
+    return {
+        "scope": "pyjhora_same_chart_parity_summary",
+        "matrix_path": str(path),
+        "tested": bool(rows),
+        "settings": settings,
+        "row_counts": dict(status_counts),
+        "coverage": dict(sorted(sections.items())),
+        "covered_outputs": sorted(covered),
+        "missing_required_outputs": missing,
+        "status": "partial_verified" if rows and not status_counts.get("mismatch") else "partial_mismatch",
+        "full_parity_verified": not missing and not status_counts.get("mismatch"),
+        "boundary": "Only covered outputs are compared. This summary cannot promote full parity while required outputs are absent.",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("matrix")
+    parser.add_argument("--ayanamsa", default="lahiri")
+    parser.add_argument("--node-mode", default="mean", choices=["mean", "true"])
+    args = parser.parse_args()
+    print(json.dumps(summarize_matrix(args.matrix, settings={"ayanamsa": args.ayanamsa, "node_mode": args.node_mode}), ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/skill_release_package.py

@@ -110,6 +110,10 @@ REQUIRED_CONTRACTS = [
     "references/oracle/western_oracle_adapter_contract.md",
     "scripts/user_invocation_acceptance_check.py",
     "scripts/diagnose_external_engine_adapters.py",
+    "scripts/external_oracle_raw_import.py",
+    "scripts/three_engine_parity_runner.py",
+    "scripts/three_engine_parity_replay_validator.py",
+    "scripts/pyjhora_parity_summary.py",
     "scripts/western_chart_engine.py",
     "scripts/western_timing_engine.py",
 ]

scripts/three_engine_parity_replay_validator.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 from pathlib import Path
 from typing import Any
@@ -12,6 +13,33 @@ from typing import Any
 REQUIRED_ENGINES = {"VedAstro", "PyJHora_JHora", "jyotishganit"}
 REQUIRED_ROW_FIELDS = {"section", "field", "local_value", "oracle_values", "status"}
 VALID_ROW_STATUSES = {"match", "mismatch", "blocked", "not_comparable"}
+RAW_VERIFIED_STATUSES = {"verified", "official_verified", "imported"}
+
+
+def _artifact_errors(engine: str, payload: Any, manifest_dir: Path) -> list[dict[str, Any]]:
+    if not isinstance(payload, dict):
+        return [{"field": f"engines.{engine}", "error": "not_object"}]
+    if payload.get("status") not in RAW_VERIFIED_STATUSES:
+        return []
+    raw_path = payload.get("official_raw_response_path") or payload.get("raw_output_path")
+    artifact_hash = payload.get("artifact_hash")
+    errors: list[dict[str, Any]] = []
+    if not isinstance(raw_path, str) or not raw_path:
+        errors.append({"field": f"engines.{engine}.raw_output_path", "error": "required_for_verified_status"})
+        return errors
+    if not isinstance(artifact_hash, str) or len(artifact_hash) != 64:
+        errors.append({"field": f"engines.{engine}.artifact_hash", "error": "sha256_required_for_verified_status"})
+        return errors
+    artifact_path = (manifest_dir / raw_path).resolve()
+    if not artifact_path.is_file():
+        errors.append({"field": f"engines.{engine}.raw_output_path", "error": "missing_artifact"})
+        return errors
+    actual_hash = hashlib.sha256(artifact_path.read_bytes()).hexdigest()
+    if actual_hash != artifact_hash:
+        errors.append({"field": f"engines.{engine}.artifact_hash", "error": "hash_mismatch"})
+    if not isinstance(payload.get("settings"), dict):
+        errors.append({"field": f"engines.{engine}.settings", "error": "required_for_verified_status"})
+    return errors
 
 
 def _row_errors(row: Any, index: int) -> list[dict[str, Any]]:
@@ -37,6 +65,8 @@ def validate_manifest(path: str | Path) -> dict[str, Any]:
     missing_engines = sorted(REQUIRED_ENGINES - set(engines))
     for engine in missing_engines:
         errors.append({"field": f"engines.{engine}", "error": "missing"})
+    for engine, payload in engines.items():
+        errors.extend(_artifact_errors(engine, payload, manifest_path.parent))
 
     if not isinstance(rows, list):
         rows = []
@@ -89,9 +119,11 @@ def validate_manifest(path: str | Path) -> dict[str, Any]:
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("manifest", nargs="?", default="references/oracle/three_engine_parity_replay_manifest.json")
+    parser.add_argument("--require-pass", action="store_true", help="Return nonzero unless all comparison rows pass.")
     args = parser.parse_args(argv)
-    print(json.dumps(validate_manifest(args.manifest), ensure_ascii=False, indent=2, sort_keys=True))
-    return 0
+    report = validate_manifest(args.manifest)
+    print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
+    return 0 if not args.require_pass or report["status"] == "pass" else 1
 
 
 if __name__ == "__main__":

scripts/three_engine_parity_runner.py

@@ -0,0 +1,360 @@
+#!/usr/bin/env python3
+"""Capture a public same-chart parity packet without overstating oracle closure."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+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",
+    "year": 1955,
+    "month": 2,
+    "day": 24,
+    "hour": 19,
+    "minute": 15,
+    "second": 0,
+    "lat": 37.7749,
+    "lon": -122.4194,
+    "tz": -8.0,
+    "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:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
+    return path
+
+
+def _capture_jyotishganit_raw(output_dir: Path) -> tuple[dict[str, Any], str]:
+    sys.path.insert(0, str(JYOTISHGANIT_ROOT))
+    try:
+        from jyotishganit import calculate_birth_chart, get_birth_chart_json
+
+        chart = calculate_birth_chart(
+            datetime(
+                PUBLIC_CASE["year"],
+                PUBLIC_CASE["month"],
+                PUBLIC_CASE["day"],
+                PUBLIC_CASE["hour"],
+                PUBLIC_CASE["minute"],
+                PUBLIC_CASE["second"],
+            ),
+            PUBLIC_CASE["lat"],
+            PUBLIC_CASE["lon"],
+            PUBLIC_CASE["tz"],
+            location_name="San Francisco, CA",
+            name="Steve Jobs (public benchmark)",
+        )
+        raw = get_birth_chart_json(chart)
+        path = _write_json(output_dir / "jyotishganit_raw.json", raw)
+        return raw, str(path)
+    except Exception as exc:
+        return {"error": f"{exc.__class__.__name__}: {exc}"}, ""
+    finally:
+        try:
+            sys.path.remove(str(JYOTISHGANIT_ROOT))
+        except ValueError:
+            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 {
+            "status": "blocked",
+            "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": "",
+        "reason": "official_runner_requires_explicit_raw_capture_workflow",
+    }
+
+
+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)
+    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": "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": report_status,
+        "tested": vedastro.get("status") == "official_verified" and bool(rows),
+        "blocked_reason": blocked_reason,
+        "engines": {
+            "VedAstro": vedastro,
+            "PyJHora_JHora": {
+                "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": {
+                "status": "raw_captured" if jyotishganit_path else "blocked",
+                "raw_output_path": jyotishganit_path,
+                "error": jyotishganit_raw.get("error") if isinstance(jyotishganit_raw, dict) else None,
+            },
+        },
+        "local": {
+            "result_hash": local["result_hash"],
+            "calculation_contract": local["calculation_contract"],
+        },
+        "comparison_rows": rows,
+        "runtime_boundary": (
+            "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)
+    return report
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output-dir", default="scratch/local/three_engine_parity")
+    parser.add_argument("--allow-vedastro-network", action="store_true")
+    args = parser.parse_args()
+    report = build_public_case_replay(
+        output_dir=ROOT / args.output_dir,
+        allow_vedastro_network=args.allow_vedastro_network,
+    )
+    print(json.dumps(report, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_external_oracle_raw_import.py

@@ -0,0 +1,35 @@
+from pathlib import Path
+
+import pytest
+
+from scripts.external_oracle_raw_import import build_raw_oracle_import
+
+
+def _metadata(**overrides):
+    value = {
+        "case_id": "public_case",
+        "license_boundary": "external benchmark only",
+        "collection_method": "manual export",
+        "birth_data_policy": "public_case_only",
+    }
+    value.update(overrides)
+    return value
+
+
+def test_raw_import_requires_reviewable_public_case_artifact(tmp_path: Path):
+    artifact = tmp_path / "oracle.json"
+    artifact.write_text('{"raw": true}', encoding="utf-8")
+
+    result = build_raw_oracle_import("VedAstro", artifact, _metadata())
+
+    assert result["status"] == "raw_imported_uncompared"
+    assert result["source_artifact_sha256"]
+    assert result["comparison_ready"] is False
+
+
+def test_raw_import_rejects_non_public_birth_policy(tmp_path: Path):
+    artifact = tmp_path / "oracle.json"
+    artifact.write_text("{}", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="public_case_only"):
+        build_raw_oracle_import("VedAstro", artifact, _metadata(birth_data_policy="private"))

tests/test_three_engine_parity_artifact_contract.py

@@ -0,0 +1,21 @@
+import json
+
+from scripts.three_engine_parity_replay_validator import validate_manifest
+
+
+def test_verified_oracle_requires_raw_artifact_hash_and_settings(tmp_path):
+    manifest = {
+        "engines": {
+            "VedAstro": {"status": "official_verified"},
+            "PyJHora_JHora": {"status": "blocked"},
+            "jyotishganit": {"status": "blocked"},
+        },
+        "comparison_rows": [],
+    }
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps(manifest), encoding="utf-8")
+
+    result = validate_manifest(path)
+
+    assert result["status"] == "invalid"
+    assert any(error["error"] == "required_for_verified_status" for error in result["errors"])

tests/test_three_engine_parity_runner.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import three_engine_parity_runner  # noqa: E402
+from three_engine_parity_runner import build_public_case_replay  # noqa: E402
+
+
+def test_public_same_chart_replay_never_promotes_missing_vedastro_raw(tmp_path: Path) -> None:
+    report = build_public_case_replay(output_dir=tmp_path, allow_vedastro_network=False)
+
+    assert report["case_id"] == "steve_jobs_public_1955_lahiri"
+    assert report["birth_data_policy"] == "public_case_only"
+    assert report["engines"]["PyJHora_JHora"]["status"] == "structured_captured"
+    assert report["engines"]["jyotishganit"]["status"] == "raw_captured"
+    assert report["engines"]["VedAstro"]["status"] == "blocked"
+    assert report["status"] in {"partial", "blocked"}
+    assert report["tested"] is False
+    assert report["comparison_rows"]
+    assert any(row["status"] == "match" for row in report["comparison_rows"])
+    assert all(row["status"] in {"blocked", "match", "not_comparable"} for row in report["comparison_rows"])
+
+
+def test_public_same_chart_replay_imports_verified_vedastro_artifact(tmp_path: Path, monkeypatch) -> None:
+    artifact_root = tmp_path / "vedastro_adapter"
+    artifact_root.mkdir()
+    artifact = artifact_root / "official_full_snapshot-abc-def.json"
+    artifact.write_text(
+        json.dumps(
+            {
+                "status": "ok",
+                "operation": "official_full_snapshot",
+                "raw_response": {"source": "vedastro_official_full_snapshot", "sections": {"chart_core": {}}},
+                "request_manifest": {
+                    "settings": {"ayanamsa": "lahiri", "node_mode": "mean"},
+                    "requests": [
+                        {
+                            "body": {
+                                "BirthTime": {
+                                    "StdTime": "19:15 24/02/1955 -08:00",
+                                    "Location": {"Latitude": 37.7749, "Longitude": -122.4194},
+                                }
+                            }
+                        }
+                    ],
+                },
+                "snapshot_sections": {"chart_core": {"Status": "Pass"}},
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(three_engine_parity_runner, "VEDASTRO_ARTIFACT_DIR", artifact_root)
+
+    report = build_public_case_replay(output_dir=tmp_path / "out", allow_vedastro_network=True)
+
+    assert report["engines"]["VedAstro"]["status"] == "official_verified"
+    assert report["engines"]["VedAstro"]["official_raw_response_path"] == str(artifact)
+    assert len(report["engines"]["VedAstro"]["artifact_hash"]) == 64
+    assert report["blocked_reason"] == "none"
+
+
+def test_public_same_chart_replay_rejects_wrong_vedastro_artifact(tmp_path: Path, monkeypatch) -> None:
+    artifact_root = tmp_path / "vedastro_adapter"
+    artifact_root.mkdir()
+    artifact = artifact_root / "official_full_snapshot-wrong-chart.json"
+    artifact.write_text(
+        json.dumps(
+            {
+                "status": "ok",
+                "raw_response": {"source": "vedastro_official_full_snapshot"},
+                "request_manifest": {
+                    "requests": [
+                        {
+                            "body": {
+                                "BirthTime": {
+                                    "StdTime": "12:00 01/01/1990 +05:30",
+                                    "Location": {"Latitude": 28.6139, "Longitude": 77.209},
+                                }
+                            }
+                        }
+                    ]
+                },
+                "snapshot_sections": {"chart_core": {"Status": "Pass"}},
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(three_engine_parity_runner, "VEDASTRO_ARTIFACT_DIR", artifact_root)
+
+    report = build_public_case_replay(output_dir=tmp_path / "out", allow_vedastro_network=True)
+
+    assert report["engines"]["VedAstro"]["status"] == "blocked"
+    assert report["engines"]["VedAstro"]["reason"] == "official_runner_requires_explicit_raw_capture_workflow"
+
+
+def test_public_same_chart_replay_adds_normalized_d1_rows(tmp_path: Path, monkeypatch) -> None:
+    artifact_root = tmp_path / "vedastro_adapter"
+    artifact_root.mkdir()
+    artifact = artifact_root / "official_full_snapshot-abc-def.json"
+    artifact.write_text(
+        json.dumps(
+            {
+                "status": "ok",
+                "raw_response": {"source": "vedastro_official_full_snapshot"},
+                "request_manifest": {
+                    "requests": [
+                        {
+                            "body": {
+                                "BirthTime": {
+                                    "StdTime": "19:15 24/02/1955 -08:00",
+                                    "Location": {"Latitude": 37.7749, "Longitude": -122.4194},
+                                }
+                            }
+                        }
+                    ]
+                },
+                "snapshot_sections": {
+                    "chart_core": {
+                        "Sun": {
+                            "Payload": {
+                                "AllPlanetData": {
+                                    "PlanetNirayanaLongitude": {"TotalDegrees": "312.5122"},
+                                    "PlanetRasiD1Sign": {"Name": "Aquarius"},
+                                }
+                            }
+                        }
+                    }
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(three_engine_parity_runner, "VEDASTRO_ARTIFACT_DIR", artifact_root)
+
+    report = build_public_case_replay(output_dir=tmp_path / "out", allow_vedastro_network=True)
+    rows = {(row["section"], row["field"]): row for row in report["comparison_rows"]}
+
+    assert rows[("D1", "Sun.sign")]["status"] == "match"
+    assert rows[("D1", "Sun.longitude")]["status"] == "match"