#!/usr/bin/env python3 """Run the Jyotish skill quality gate used by local development and CI.""" from __future__ import annotations import argparse import contextlib import json import os import py_compile import subprocess import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) from local_env import load_local_env # noqa: E402 load_local_env(ROOT) APP = ROOT / "frontend" PYTHON = sys.executable os.environ.setdefault("PYTHON", PYTHON) COMPILE_DIRS = [ ROOT / "scripts", ROOT / "jyotish_vedic", ] EXTRA_COMPILE_TARGETS = [ ROOT / "mcp_server.py", ROOT / "scripts" / "audit_fragments.py", ROOT / "scripts" / "character_level_inventory_manifest.py", ROOT / "scripts" / "dasha_reference_audit.py", ROOT / "scripts" / "external_oracle_sanity_closure.py", ROOT / "scripts" / "interpretation_source_inventory_gate.py", ROOT / "scripts" / "oracle_boundary_audit.py", ROOT / "scripts" / "oracle_collection_queue.py", ROOT / "scripts" / "oracle_evidence_validator.py", ROOT / "scripts" / "sync_final_evidence_packet_status.py", ROOT / "tests" / "run_golden_cases.py", ROOT / "tests" / "run_real_case_revalidation.py", ] CORE_PYTEST_TARGETS = [ "tests/test_cli_smoke.py", "tests/test_api_server_security.py", "tests/test_jaimini.py", "tests/test_shadbala_complete.py", "tests/test_transit_trigger.py", "tests/test_oracle_collection_queue.py", "tests/test_oracle_evidence_validator.py", "tests/test_external_oracle_sanity_closure.py", ] RUNTIME_TRUTH_PYTEST_TARGETS = [ "tests/test_api_server_security.py::test_high_rigor_vedastro_official_summary_passes_through_contract_fields", "tests/test_api_server_security.py::test_high_rigor_vedastro_official_summary_exposes_top_reader_contract_from_full_snapshot", "tests/test_vedastro_external_technique_evidence.py::test_strict_workflow_uses_shared_consultation_executor", "tests/test_vedastro_runtime_mode_diagnostics.py", "tests/test_interpretation_source_inventory_gate.py::test_quality_gate_runs_interpretation_source_inventory_gate", "tests/test_interpretation_source_runtime_coverage.py", "tests/test_final_jhora_evidence_packet_acceptance.py", ] RELEASE_CRITICAL_UNTRACKED_PATHS = [ "docs/research/desktop_packaging_spike_2026_06_23.md", "docs/research/ephemeris_abstraction_feasibility_2026_06_23.md", "docs/research/ephemeris_adapter_contract_2026_06_23.md", "docs/research/ephemeris_candidate_adapter_spike_2026_06_23.md", "docs/research/open_source_scan_2026_06_22.md", "docs/research/product_gap_matrix_2026_06_22.md", "docs/research/whole_machine_git_audit_2026_06_23.md", "findings.md", "progress.md", "references/oracle/dasha_shadbala_oracle_cases.json", "scripts/audit_fragments.py", "scripts/deep_varga_avastha.py", "scripts/dasha_reference_audit.py", "scripts/ephemeris_adapter_contract.py", "scripts/ephemeris_backend_probe.py", "scripts/ephemeris_candidate_adapter_spike.py", "scripts/oracle_boundary_audit.py", "scripts/oracle_collection_queue.py", "scripts/oracle_evidence_validator.py", "task_plan.md", "tests/test_api_server_security.py", "tests/test_dasha_reference_audit.py", "tests/test_deep_varga_avastha.py", "tests/test_oracle_boundary_audit.py", "tests/test_oracle_collection_queue.py", "tests/test_oracle_evidence_validator.py", ] QUALITY_GATE_PROFILES = { "quick": { "skip_slow": True, "skip_yoga_logic": True, "skip_frontend_runtime": False, "check_release_hygiene": False, "skip_real_cases": True, "skip_dasha_audit": True, "skip_oracle_audit": True, "skip_local_accuracy_report": True, "skip_vedastro_live": True, }, "browser": { "skip_slow": True, "skip_yoga_logic": True, "skip_frontend_runtime": False, "check_release_hygiene": False, "skip_real_cases": True, "skip_dasha_audit": True, "skip_oracle_audit": True, "skip_local_accuracy_report": True, "skip_vedastro_live": True, }, "release": { "skip_slow": False, "skip_yoga_logic": False, "skip_frontend_runtime": False, "check_release_hygiene": True, "skip_real_cases": False, "skip_dasha_audit": False, "skip_oracle_audit": False, "skip_local_accuracy_report": False, "skip_vedastro_live": True, }, "accuracy": { "skip_slow": True, "skip_yoga_logic": False, "skip_frontend_runtime": True, "check_release_hygiene": False, "skip_real_cases": False, "skip_dasha_audit": False, "skip_oracle_audit": False, "skip_local_accuracy_report": False, "skip_vedastro_live": True, }, "vedastro-live": { "skip_slow": True, "skip_yoga_logic": True, "skip_frontend_runtime": True, "check_release_hygiene": False, "skip_real_cases": True, "skip_dasha_audit": True, "skip_oracle_audit": True, "skip_local_accuracy_report": True, "skip_vedastro_live": False, }, "runtime-truth": { "skip_slow": True, "skip_yoga_logic": True, "skip_frontend_runtime": True, "check_release_hygiene": False, "skip_real_cases": True, "skip_dasha_audit": True, "skip_oracle_audit": True, "skip_local_accuracy_report": True, "skip_vedastro_live": True, }, } DASHA_REFERENCE_AUDIT_CMD = [ PYTHON, "scripts/dasha_reference_audit.py", "--year", "1990", "--month", "4", "--day", "17", "--hour", "14", "--minute", "45", "--second", "20", "--lat", "36.466667", "--lon", "114.2", "--tz", "8", "--target-start-date", "2021-05-18", "--target-source", "synthetic_fixture", ] ORACLE_BOUNDARY_AUDIT_CMD = [ PYTHON, "scripts/oracle_boundary_audit.py", "--oracle-file", "references/oracle/dasha_shadbala_oracle_cases.json", ] EXTERNAL_ORACLE_SANITY_CLOSURE_CMD = [ PYTHON, "scripts/external_oracle_sanity_closure.py", "--format", "json", ] ORACLE_COLLECTION_QUEUE_CMD = [ PYTHON, "scripts/oracle_collection_queue.py", "--oracle-file", "references/oracle/dasha_shadbala_oracle_cases.json", "--format", "json", ] ORACLE_COLLECTION_QUEUE_EXPECTED_FIELDS = ["evidence_packet", "capture_id", "target_fields"] ORACLE_EVIDENCE_VALIDATOR_CMD = [ PYTHON, "scripts/oracle_evidence_validator.py", "--queue-file", "{queue_file}", ] def tail_text(text: str, *, limit: int = 2400) -> str: text = text.strip() if len(text) <= limit: return text return f"...\n{text[-limit:]}" def extract_json_payload(text: str) -> dict: text = text.strip() if not text: return {} for start in [text.find("{"), text.rfind("{")]: if start < 0: continue try: payload = json.loads(text[start:]) except json.JSONDecodeError: continue if isinstance(payload, dict): return payload return {} def format_failure_summary( step: str, cmd: list[str], returncode: int, *, stdout: str = "", stderr: str = "", cwd: Path = ROOT, ) -> str: combined_payload = extract_json_payload(stderr) or extract_json_payload(stdout) reason = combined_payload.get("reason") if isinstance(combined_payload, dict) else None snapshot = combined_payload.get("process_snapshot") if isinstance(combined_payload, dict) else None lines = [ "", "== Quality gate failed ==", f"step: {step}", f"exit code: {returncode}", f"cwd: {cwd.relative_to(ROOT) if cwd != ROOT else '.'}", f"command: {' '.join(cmd)}", ] if reason: lines.append(f"reason: {reason}") if snapshot: lines.append(f"process_snapshot: {json.dumps(snapshot, ensure_ascii=False)}") if stdout.strip(): lines.append(f"stdout tail:\n{tail_text(stdout)}") if stderr.strip(): lines.append(f"stderr tail:\n{tail_text(stderr)}") lines.extend([ "普通用户启动路径:", "1. 本地 API 服务：.venv/bin/python scripts/jyotish_api_server.py --host 127.0.0.1 --port 5200", "2. 网页服务：npm run dev --prefix frontend", "3. Open http://127.0.0.1:3000 and verify /api/health.", "Next action: Run the focused command above, then rerun the affected Python or Next.js check.", "", ]) return "\n".join(lines) def run(cmd: list[str], *, optional: bool = False, step: str | None = None, cwd: Path = ROOT) -> bool: label = step or " ".join(cmd[:2]) print(f"\n$ {' '.join(cmd)}") completed = subprocess.run(cmd, cwd=cwd, text=True) if completed.returncode == 0: return True if optional: print(f"Optional step failed with exit code {completed.returncode}; continuing.") return False print( format_failure_summary(label, cmd, completed.returncode, stdout="", stderr="", cwd=cwd), file=sys.stderr, ) raise SystemExit(completed.returncode) def run_oracle_collection_queue_and_validator() -> None: with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False, encoding="utf-8") as handle: queue_path = Path(handle.name) try: print(f"\n$ {' '.join(ORACLE_COLLECTION_QUEUE_CMD)}") completed = subprocess.run(ORACLE_COLLECTION_QUEUE_CMD, cwd=ROOT, text=True, capture_output=True) if completed.stdout: print(completed.stdout, end="" if completed.stdout.endswith("\n") else "\n") if completed.stderr: print(completed.stderr, end="" if completed.stderr.endswith("\n") else "\n", file=sys.stderr) if completed.returncode != 0: print( format_failure_summary( "oracle_collection_queue", ORACLE_COLLECTION_QUEUE_CMD, completed.returncode, stdout=completed.stdout, stderr=completed.stderr, ), file=sys.stderr, ) raise SystemExit(completed.returncode) queue_path.write_text(completed.stdout, encoding="utf-8") validator_cmd = [part if part != "{queue_file}" else str(queue_path) for part in ORACLE_EVIDENCE_VALIDATOR_CMD] run(validator_cmd, step="oracle_evidence_validator") finally: with contextlib.suppress(FileNotFoundError): queue_path.unlink() def git_untracked_files() -> set[str]: completed = subprocess.run( ["git", "ls-files", "--others", "--exclude-standard"], cwd=ROOT, text=True, capture_output=True, check=False, ) if completed.returncode != 0: raise SystemExit( format_failure_summary( "release_hygiene_check", ["git", "ls-files", "--others", "--exclude-standard"], completed.returncode, stdout=completed.stdout, stderr=completed.stderr, ) ) return {line.strip() for line in completed.stdout.splitlines() if line.strip()} def release_hygiene_check(require_external_parity: bool = False) -> None: print("\n== Release hygiene check ==") untracked = git_untracked_files() critical = [path for path in RELEASE_CRITICAL_UNTRACKED_PATHS if path in untracked] if critical: payload = { "reason": "release_critical_untracked_files", "untracked_count": len(critical), "untracked_files": critical, "next_action": "Stage or commit these product files before running the release profile.", } print(json.dumps(payload, ensure_ascii=False, indent=2), file=sys.stderr) raise SystemExit(1) run([PYTHON, "scripts/public_release_privacy_scan.py", "--json"]) run([PYTHON, "scripts/report_renderer_isolation_poc.py", "--strict"]) parity_command = [PYTHON, "scripts/three_engine_parity_replay_validator.py", "references/oracle/three_engine_parity_replay_manifest.json"] if require_external_parity: parity_command.append("--require-pass") run(parity_command) print("release_hygiene_check ok: no release-critical product files are untracked") def run_vedastro_live_smoke() -> None: print("\n== VedAstro live adapter smoke ==") endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip() network_enabled = os.environ.get("VEDASTRO_ENABLE_NETWORK", "").strip().lower() in {"1", "true", "yes"} if not endpoint or not network_enabled: print(json.dumps({ "status": "blocked", "reason": "vedastro_live_endpoint_or_network_flag_missing", "required_env": { "endpoint": "VEDASTRO_API_ENDPOINT", "network": "VEDASTRO_ENABLE_NETWORK", }, "boundary": "Default CI stays deterministic; configure both env vars to run a real VedAstro live smoke.", }, ensure_ascii=False, indent=2)) return run([ PYTHON, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "career", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2026-12-31", ]) def compile_targets() -> None: print("\n== Compile core Python files ==") targets: list[Path] = [] for directory in COMPILE_DIRS: targets.extend(sorted(directory.glob("*.py"))) targets.extend(EXTRA_COMPILE_TARGETS) seen: set[Path] = set() for target in targets: if target in seen or not target.exists(): continue seen.add(target) print(f"compile {target.relative_to(ROOT)}") py_compile.compile(str(target), doraise=True) def validate_json_files() -> None: print("\n== Validate critical JSON files ==") for relative in [ "references/technique_registry.json", "references/yoga_rules.json", "references/standard_test_charts.json", "references/validation_logic_report.json", "references/oracle/dasha_shadbala_oracle_cases.json", "tests/golden/golden_cases.json", ]: path = ROOT / relative if not path.exists(): continue with path.open("r", encoding="utf-8") as handle: json.load(handle) print(f"json ok {relative}") def run_profile(args: argparse.Namespace) -> dict: profile = dict(QUALITY_GATE_PROFILES[args.profile]) for key in [ "skip_slow", "skip_yoga_logic", "skip_frontend_runtime", "skip_real_cases", "skip_dasha_audit", "skip_oracle_audit", "skip_local_accuracy_report", "skip_vedastro_live", ]: if getattr(args, key): profile[key] = True return profile def main() -> int: parser = argparse.ArgumentParser(description="Run Jyotish skill quality gate") parser.add_argument("--profile", choices=["quick", "browser", "release", "accuracy", "vedastro-live", "runtime-truth"], default="browser", help="Quality gate profile: quick, browser, release, accuracy, vedastro-live, or runtime-truth") parser.add_argument("--skip-slow", action="store_true", help="Skip slow golden-case regressions") parser.add_argument("--skip-yoga-logic", action="store_true", help="Skip Yoga logic comparison report refresh") parser.add_argument("--skip-frontend-runtime", action="store_true", help="Skip Next.js tests, lint, and production build") parser.add_argument("--skip-real-cases", action="store_true", help="Skip public real-person chart revalidation") parser.add_argument("--skip-dasha-audit", action="store_true", help="Skip Dasha reference-drift audit") parser.add_argument("--skip-oracle-audit", action="store_true", help="Skip combined Dasha/Shadbala external oracle boundary audit") parser.add_argument("--skip-local-accuracy-report", action="store_true", help="Skip consolidated local accuracy report") parser.add_argument("--skip-vedastro-live", action="store_true", help="Skip optional VedAstro live endpoint smoke") parser.add_argument("--all-tests", action="store_true", help="Run every pytest file, including optional-dependency suites") parser.add_argument("--require-external-parity", action="store_true", help="Fail the release gate unless the three-engine raw parity manifest passes.") args = parser.parse_args() profile = run_profile(args) os.environ.setdefault("PYTHONPATH", str(ROOT / "scripts")) print(f"\n== Quality gate profile: {args.profile} ==") print(json.dumps(profile, ensure_ascii=False, indent=2)) if args.profile == "runtime-truth": for target in [ ROOT / "scripts" / "jyotish_api_server.py", ROOT / "scripts" / "diagnose_vedastro_mode.py", ROOT / "scripts" / "diagnose_external_engine_adapters.py", ROOT / "scripts" / "interpretation_source_runtime_coverage.py", ROOT / "scripts" / "sync_final_evidence_packet_status.py", ROOT / "scripts" / "external_validation_release_gate.py", ]: py_compile.compile(str(target), doraise=True) print(f"compiled {target.relative_to(ROOT)}") run([PYTHON, "scripts/sync_final_evidence_packet_status.py"]) run([PYTHON, "scripts/interpretation_source_inventory_gate.py"]) run([PYTHON, "scripts/diagnose_vedastro_mode.py", "--json"]) run([PYTHON, "scripts/diagnose_external_engine_adapters.py", "--json"]) run([PYTHON, "scripts/external_validation_release_gate.py", "--require-match"]) else: compile_targets() validate_json_files() run([PYTHON, "scripts/audit_capabilities.py", "--mode", "validate"]) run([PYTHON, "scripts/audit_fragments.py", "--strict"]) run([PYTHON, "scripts/interpretation_source_inventory_gate.py"]) run([PYTHON, "scripts/character_level_inventory_manifest.py", "--scope", "project", "--no-write", "--summary-only"]) if profile["check_release_hygiene"]: release_hygiene_check(require_external_parity=args.require_external_parity) run([PYTHON, "scripts/validate_bphs_invariants.py"]) if args.all_tests: pytest_targets = ["tests"] elif args.profile == "runtime-truth": pytest_targets = RUNTIME_TRUTH_PYTEST_TARGETS else: pytest_targets = CORE_PYTEST_TARGETS run([PYTHON, "-m", "pytest", *pytest_targets]) if not profile["skip_frontend_runtime"]: run(["npm", "test"], optional=False, cwd=APP) run(["npm", "run", "lint"], optional=False, cwd=APP) run(["npm", "run", "build"], optional=False, cwd=APP) if not profile["skip_slow"]: run([PYTHON, "tests/run_golden_cases.py", "--python", PYTHON]) if not profile["skip_real_cases"]: run([PYTHON, "tests/run_real_case_revalidation.py", "--python", PYTHON, "--summary"]) if not profile["skip_dasha_audit"]: run(DASHA_REFERENCE_AUDIT_CMD) if not profile["skip_oracle_audit"]: run(ORACLE_BOUNDARY_AUDIT_CMD) run(EXTERNAL_ORACLE_SANITY_CLOSURE_CMD) run_oracle_collection_queue_and_validator() if not profile["skip_yoga_logic"]: run([PYTHON, "scripts/validate_logic_v2.py"], optional=True) if not profile["skip_local_accuracy_report"]: run([PYTHON, "scripts/local_accuracy_report.py", "--format", "json"]) if not profile["skip_vedastro_live"]: run_vedastro_live_smoke() print("\nQuality gate passed.") return 0 if __name__ == "__main__": raise SystemExit(main())