Jyotisha/scripts/run_quality_gate.py

#!/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 re
import subprocess
import sys
import tempfile
import time
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",
    # The staging gate runs the quick profile, so a guard absent from this list never runs in CI.
    # These pin the answer-truth contract every product consultation is built on, and their failure
    # mode is silent widening — nothing errors when they regress (BUG-267, BUG-270).
    "tests/test_consultation_consumer_context.py",
    "tests/test_declared_window_chart.py",
    # Staging quick profile never runs `tests/` wholesale. A distinguish probe
    # with empty mapping or non-positive gain must fail this gate (BUG-393).
    "tests/test_candidate_discriminator_contract.py",
    # Auto staging gate is `--profile quick`. The full pytest tree only runs in
    # the manual `release-quality-gate.yml` (after `--profile release`), so a
    # stale window_scan assertion in this glob stayed red on origin/staging
    # until listed here.
    "tests/test_rectification_*.py",
    "tests/test_varga_segments_api.py",
    "tests/test_varga_resolution_research.py",
    # This file regexes frontend source. Home-split and other page.tsx moves must keep it green.
    "tests/test_supabase_user_data_contract.py",
    # Consultation birth-accuracy invariants are pytest-style functions; unittest
    # discover collects zero of them, so the auto gate must list the file itself.
    "tests/test_consultation_birth_accuracy.py",
    # Locks archive as PATCH archived_at, never HTTP DELETE (data-safety).
    "tests/test_session_management_entrypoints.py",
    # Pure source/SQL regex for the birth-time journey; no runtime services.
    "tests/test_birth_time_journey_contract.py",
    # Freeze scripts/jyotish_api_server.py growth; new features must be modules.
    "tests/test_api_server_growth_contract.py",
    # Pins the two closed __new__ forgeries: MCP/report scripts must keep reaching the
    # compute mixins without constructing an HTTP handler (TASK-api-server-backdoor-close).
    "tests/test_offline_compute_mixins.py",
    # Foreground VedAstro snapshot cache + join cancel (BUG-727 / BUG-728).
    "tests/test_vedastro_snapshot_cache.py",
    # Native seven-governors adapter and the three read-only chart endpoints.
    "tests/test_qizheng_chart_engine.py",
    "tests/test_qizheng_api_productization.py",
    "tests/test_readonly_chart_endpoints.py",
    "tests/test_ephemeris_events.py",
    # Birth-sky cover: real alt/az of planets and bright stars (TASK-birth-sky-cover).
    "tests/test_birth_sky.py",
    # Fail-fast heavy-compute concurrency gate (429 + Retry-After, health ungated).
    "tests/test_api_heavy_compute_gate.py",
    # Upstream-sync acceptance regressions must fail the automatic staging gate.
    "tests/test_consultation_workflow_domains.py",
    # Float hour/minute from the API payload must not 500 consultation_workflow (BUG-524).
    "tests/test_consultation_workflow_birth_time_sensitivity.py",
    "tests/test_report_longform_parity.py",
    "tests/test_report_longform_gaps2.py",
    # 22 SVG fences beside each longform chart; skipHtml dropped the HTML (BUG-607).
    "tests/test_report_chart_block.py",
    "tests/test_timing_precision_contract.py",
    "tests/test_flexible_birth_time_report_contract.py",
    "tests/test_mcp_strict_workflow_finance.py",
    "tests/test_mcp_strict_workflow_relationship.py",
    "tests/test_punarphoo_observation.py",
    # Native technique layers exposed to the web consultation (evidence card v2).
    "tests/test_consultation_native_layers.py",
    # BUG-1060: Double Transit PAC D9 targets use the D9 signs their labels name.
    "tests/test_double_transit_d9_targets.py",
    # BUG-1061: D1+D9 cross-layer entries pair targets by structured identity, not name digits.
    "tests/test_double_transit_cross_layer_pairing.py",
    "tests/test_relationship_event_class_evidence.py",
    "tests/test_thematic_dasha_info_source.py",
    "tests/test_report_orchestrator_marriage_timing.py",
    "tests/test_full_reading_conditional_dashas.py",
    "tests/test_capability_evidence_pool.py",
    "tests/test_readme_badges.py",
    "tests/test_upstream_import_plan.py",
    # Tracked-file privacy must run in staging's explicit quick test list.
    "tests/test_repo_privacy_markers.py",
    # Summary lines by default; failure keeps the last 200 lines and a full log (BUG-1230).
    "tests/test_run_quality_gate_output.py",
    "tests/test_upstream_git_import_b9a0ef8f.py",
    "tests/test_interpretation_template_registry.py",
    "tests/test_vedastro_external_technique_evidence.py",
    # Self-hosted heading font slices: 6500-character coverage, no overlap, 120 KB cap.
    "tests/test_serif_font_slices.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",
]


def _expand_pytest_targets(targets: list[str]) -> list[str]:
    """Expand glob entries so pytest argv never depends on a shell.

    `subprocess.run` is invoked without `shell=True`. A literal
    `tests/test_rectification_*.py` is then a filename pytest may ignore,
    so a pin that only checks the string is present can stay green while
    the suite never runs.
    """
    expanded: list[str] = []
    for target in targets:
        if any(mark in target for mark in "*?["):
            matches = sorted(
                path.relative_to(ROOT).as_posix()
                for path in ROOT.glob(target)
                if path.is_file()
            )
            if not matches:
                raise SystemExit(f"pytest glob {target!r} matched no files under {ROOT}")
            expanded.extend(matches)
            continue
        expanded.append(target)
    return expanded


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",
    "docs/history/findings.md",
    "docs/history/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",
    "docs/history/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": {
        "test_timeout_seconds": 90,
        "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": {
        "test_timeout_seconds": 120,
        "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": {
        "test_timeout_seconds": 600,
        "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": {
        "test_timeout_seconds": 300,
        "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": {
        "test_timeout_seconds": 120,
        "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": {
        "test_timeout_seconds": 180,
        "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:]}"


# Default output is one line per successful step. `--verbose` restores the
# previous passthrough so a local debugging run still shows child output.
VERBOSE = False
_step_ordinal = 0
_FAILURE_TAIL_LINES = 200


def set_verbose(enabled: bool) -> None:
    global VERBOSE
    VERBOSE = enabled


def reset_output_state() -> None:
    global _step_ordinal
    _step_ordinal = 0


def format_profile_banner(profile_name: str, profile: dict, *, verbose: bool) -> str:
    if verbose:
        body = json.dumps(profile, ensure_ascii=False, indent=2)
        return f"\n== Quality gate profile: {profile_name} ==\n{body}"
    flags = " ".join(f"{key}={str(value).lower()}" for key, value in profile.items())
    return f"quality gate profile={profile_name} {flags}"


def _step_name(cmd: list[str], step: str | None) -> str:
    if step:
        return step
    if len(cmd) >= 2 and str(cmd[1]).endswith(".py"):
        return str(cmd[1]).replace("\\", "/")
    if len(cmd) >= 3 and cmd[1] == "-m":
        return str(cmd[2])
    return " ".join(str(part) for part in cmd[:3])


def _slug(name: str) -> str:
    slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-")
    return slug[:80] or "step"


def _combine_output(stdout: str, stderr: str) -> str:
    parts: list[str] = []
    if stdout:
        parts.append(stdout if stdout.endswith("\n") else f"{stdout}\n")
    if stderr:
        if stdout:
            parts.append("===== stderr =====\n")
        parts.append(stderr if stderr.endswith("\n") else f"{stderr}\n")
    return "".join(parts)


def last_output_lines(text: str, count: int = _FAILURE_TAIL_LINES) -> str:
    lines = text.splitlines()
    if len(lines) <= count:
        return "\n".join(lines)
    return "\n".join(lines[-count:])


def _write_failure_log(name: str, output: str) -> Path:
    directory = ROOT / "gate-logs"
    directory.mkdir(parents=True, exist_ok=True)
    path = directory / f"{_step_ordinal:02d}-{_slug(name)}.log"
    path.write_text(output, encoding="utf-8")
    return path


def _report_captured_failure(
    label: str,
    cmd: list[str],
    returncode: int,
    output: str,
    *,
    cwd: Path,
    optional: bool,
) -> bool:
    log_path = _write_failure_log(label, output)
    relative = log_path.relative_to(ROOT).as_posix()
    print(f"✗ {label} exit={returncode}", file=sys.stderr)
    tail = last_output_lines(output)
    if tail:
        print(tail, file=sys.stderr)
    print(f"full log: {relative}", file=sys.stderr)
    if optional:
        print(f"Optional step failed with exit code {returncode}; continuing.")
        return False
    print(
        format_failure_summary(label, cmd, returncode, stdout="", stderr="", cwd=cwd),
        file=sys.stderr,
    )
    raise SystemExit(returncode)


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 | None = None) -> bool:
    """Run one gate step.

    Compact mode captures stdout and stderr: success prints one line, failure
    prints the last 200 lines and writes the full output under ``gate-logs/``.
    ``--verbose`` inherits the child streams, matching the previous behavior.
    Timeouts are unchanged: this wrapper does not add a ``timeout=`` of its own,
    and ``--test-timeout`` / ``test_timeout_seconds`` stay profile metadata.
    """
    global _step_ordinal
    _step_ordinal += 1
    if cwd is None:
        cwd = ROOT
    label = _step_name(cmd, step)
    if VERBOSE:
        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)

    started = time.perf_counter()
    completed = subprocess.run(cmd, cwd=cwd, text=True, capture_output=True, errors="replace")
    elapsed = time.perf_counter() - started
    if completed.returncode == 0:
        print(f"✓ {label} {elapsed:.1f}s")
        return True
    output = _combine_output(completed.stdout or "", completed.stderr or "")
    return _report_captured_failure(label, cmd, completed.returncode, output, cwd=cwd, optional=optional)


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:
        global _step_ordinal
        _step_ordinal += 1
        label = "oracle_collection_queue"
        if VERBOSE:
            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(
                        label,
                        ORACLE_COLLECTION_QUEUE_CMD,
                        completed.returncode,
                        stdout=completed.stdout,
                        stderr=completed.stderr,
                    ),
                    file=sys.stderr,
                )
                raise SystemExit(completed.returncode)
        else:
            started = time.perf_counter()
            completed = subprocess.run(
                ORACLE_COLLECTION_QUEUE_CMD,
                cwd=ROOT,
                text=True,
                capture_output=True,
                errors="replace",
            )
            elapsed = time.perf_counter() - started
            if completed.returncode != 0:
                output = _combine_output(completed.stdout or "", completed.stderr or "")
                _report_captured_failure(
                    label,
                    ORACLE_COLLECTION_QUEUE_CMD,
                    completed.returncode,
                    output,
                    cwd=ROOT,
                    optional=False,
                )
            print(f"✓ {label} {elapsed:.1f}s")
        queue_path.write_text(completed.stdout or "", 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:
    if VERBOSE:
        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()
    started = time.perf_counter()
    for target in targets:
        if target in seen or not target.exists():
            continue
        seen.add(target)
        if VERBOSE:
            print(f"compile {target.relative_to(ROOT)}")
        py_compile.compile(str(target), doraise=True)
    if not VERBOSE:
        elapsed = time.perf_counter() - started
        print(f"✓ py_compile {len(seen)} files {elapsed:.1f}s")


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(
        "--test-timeout",
        type=float,
        default=None,
        metavar="SECONDS",
        help="Fail an individual pytest call after SECONDS; defaults to the selected profile boundary.",
    )
    parser.add_argument("--require-external-parity", action="store_true", help="Fail the release gate unless the three-engine raw parity manifest passes.")
    parser.add_argument("--verbose", action="store_true", help="Stream every command and its output. The default prints one summary line per successful step.")
    args = parser.parse_args()
    profile = run_profile(args)
    set_verbose(args.verbose)
    reset_output_state()

    os.environ.setdefault("PYTHONPATH", str(ROOT / "scripts"))
    print(format_profile_banner(args.profile, profile, verbose=args.verbose))
    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", *_expand_pytest_targets(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())