Jyotisha/scripts/pre_work_check.py

#!/usr/bin/env python3
"""One-command pre-work governance check.

Runs the lightweight guardrails that should happen before substantial work:
ledger/docs presence, git status visibility, fragment scan, remote visibility,
and focused governance tests.
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
MINIMUM_PYTHON_VERSION = (3, 11)
PYTHON_OVERRIDE_ENV = "JYOTISH_PRE_WORK_PYTHON"
DEFAULT_COMMAND_TIMEOUT_SECONDS = 45
DEFAULT_FRAGMENT_TIMEOUT_SECONDS = 90
FOCUSED_TEST_TARGETS = [
    "tests/test_runtime_import_boundaries.py",
    "tests/test_project_fragment_governance.py",
    "tests/test_preflight_fragment_scan.py",
    "tests/test_remote_repo_visibility_check.py",
    "tests/test_pre_work_check.py",
]
EXTERNAL_ENGINE_DIAGNOSTIC_TARGET = "scripts/diagnose_external_engine_adapters.py"
PRE_WORK_DOCS = [
    "AGENTS.md",
    "docs/research/pre_work_error_ledger.md",
    "docs/research/whole_machine_fragment_sweep_2026_07_05.md",
    "docs/research/whole_machine_fragment_sweep_2026_07_14.md",
    "docs/research/whole_machine_fragment_sweep_round25_2026_06_25.md",
]


def python_candidates(
    *,
    root: Path = ROOT,
    environ: dict[str, str] | None = None,
    current_executable: str | None = None,
) -> list[str]:
    """Return Python candidates in deterministic preference order.

    An explicit override is strict: a broken override must fail closed rather
    than silently run the checks under a different environment.
    """
    env = os.environ if environ is None else environ
    override = env.get(PYTHON_OVERRIDE_ENV, "").strip()
    if override:
        return [override]

    candidates = [
        str(root / ".venv" / "bin" / "python"),
        str(root / ".venv" / "Scripts" / "python.exe"),
        current_executable or sys.executable,
    ]
    candidates.extend(path for name in ("python3.13", "python3.12", "python3.11") if (path := shutil.which(name)))
    return list(dict.fromkeys(candidate for candidate in candidates if candidate))


def probe_python(candidate: str) -> dict[str, Any]:
    probe = (
        "import importlib.util,json,sys;"
        "print(json.dumps({'version':list(sys.version_info[:3]),"
        "'pytest_available':importlib.util.find_spec('pytest') is not None}))"
    )
    try:
        completed = subprocess.run(
            [candidate, "-c", probe],
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=10,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"candidate": candidate, "ok": False, "error": str(exc)}

    if completed.returncode != 0:
        return {
            "candidate": candidate,
            "ok": False,
            "error": (completed.stderr or completed.stdout).strip() or f"probe exited {completed.returncode}",
        }
    try:
        metadata = json.loads(completed.stdout)
        version = tuple(int(part) for part in metadata["version"])
        pytest_available = metadata["pytest_available"] is True
    except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
        return {"candidate": candidate, "ok": False, "error": f"invalid probe output: {exc}"}

    requirements = []
    if version < MINIMUM_PYTHON_VERSION:
        requirements.append(f"Python {version[0]}.{version[1]} is below required 3.11")
    if not pytest_available:
        requirements.append("pytest is not installed")
    return {
        "candidate": candidate,
        "ok": not requirements,
        "version": ".".join(str(part) for part in version),
        "pytest_available": pytest_available,
        "error": "; ".join(requirements),
    }


def select_python(
    candidates: list[str] | None = None,
    probe: Callable[[str], dict[str, Any]] = probe_python,
) -> dict[str, Any]:
    attempts = []
    for candidate in candidates or python_candidates():
        result = probe(candidate)
        attempts.append(result)
        if result.get("ok"):
            return {"ok": True, "executable": candidate, "attempts": attempts, **result}
    return {
        "ok": False,
        "executable": "",
        "attempts": attempts,
        "error": (
            "No usable project Python found. Create .venv with Python >=3.11 and install requirements-dev.txt, "
            f"or set {PYTHON_OVERRIDE_ENV} to a compatible interpreter."
        ),
    }


def run(args: list[str], timeout: int, env: dict[str, str] | None = None) -> dict[str, Any]:
    try:
        completed = subprocess.run(
            args,
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=timeout,
            check=False,
            env=env,
        )
    except subprocess.TimeoutExpired as exc:
        return {
            "ok": False,
            "returncode": None,
            "stdout": exc.stdout or "",
            "stderr": exc.stderr or "",
            "error": f"timeout after {timeout}s",
        }
    return {
        "ok": completed.returncode == 0,
        "returncode": completed.returncode,
        "stdout": completed.stdout,
        "stderr": completed.stderr,
        "error": "" if completed.returncode == 0 else (completed.stderr or completed.stdout).strip(),
    }


def classify_status(
    docs_ok: bool,
    fragment_ok: bool,
    pytest_ok: bool,
    remote_status: str,
    external_engine_ok: bool = True,
    python_ok: bool = True,
) -> str:
    if not docs_ok or not fragment_ok or not pytest_ok or not external_engine_ok or not python_ok:
        return "fail"
    if remote_status == "verified":
        return "pass"
    return "pass_with_remote_blocked"


def build_report(remote_timeout: int, command_timeout: int, fragment_timeout: int, skip_tests: bool = False) -> dict[str, Any]:
    docs = {path: (ROOT / path).exists() for path in PRE_WORK_DOCS}
    git_status = run(["git", "status", "--short", "--branch"], command_timeout)
    git_remote = run(["git", "remote", "-v"], command_timeout)
    python_runtime = select_python()
    python = str(python_runtime.get("executable") or "")
    if python_runtime["ok"]:
        fragment = run([python, "scripts/preflight_fragment_scan.py"], fragment_timeout)
        external_engine = run([python, EXTERNAL_ENGINE_DIAGNOSTIC_TARGET, "--json"], command_timeout)
        remote = run([python, "scripts/remote_repo_visibility_check.py", "--timeout", str(remote_timeout)], command_timeout)
    else:
        unavailable = {
            "ok": False,
            "returncode": None,
            "stdout": "",
            "stderr": "",
            "error": python_runtime["error"],
        }
        fragment = dict(unavailable)
        external_engine = dict(unavailable)
        remote = dict(unavailable)
    remote_report: dict[str, Any] = {}
    if remote["ok"]:
        try:
            remote_report = json.loads(remote["stdout"])
        except json.JSONDecodeError as exc:
            remote_report = {"status": "blocked", "must_not_claim_synced": True, "parse_error": str(exc)}
    pytest_result = {"ok": True, "stdout": "skipped", "stderr": "", "error": ""}
    if not skip_tests:
        env = dict(os.environ)
        if fragment["ok"] and fragment.get("stdout"):
            cache = Path(tempfile.gettempdir()) / "jyotish_preflight_fragment_scan_report.json"
            cache.write_text(fragment["stdout"], encoding="utf-8")
            env["PREFLIGHT_FRAGMENT_SCAN_REPORT"] = str(cache)
        if python_runtime["ok"]:
            pytest_result = run([python, "-m", "pytest", "-q", *FOCUSED_TEST_TARGETS], command_timeout, env=env)
        else:
            pytest_result = {"ok": False, "stdout": "", "stderr": "", "error": python_runtime["error"]}
    remote_status = str(remote_report.get("status") or "blocked")
    status = classify_status(
        docs_ok=all(docs.values()),
        fragment_ok=fragment["ok"],
        pytest_ok=pytest_result["ok"],
        remote_status=remote_status,
        external_engine_ok=external_engine["ok"],
        python_ok=python_runtime["ok"],
    )
    return {
        "scope": "pre_work_check",
        "status": status,
        "must_not_claim_synced": remote_report.get("must_not_claim_synced", True),
        "python": python_runtime,
        "docs": docs,
        "git": {
            "status_ok": git_status["ok"],
            "status": git_status["stdout"],
            "remote_ok": git_remote["ok"],
            "remote": git_remote["stdout"],
        },
        "checks": {
            "python_runtime_ok": python_runtime["ok"],
            "fragment_scan_ok": fragment["ok"],
            "external_engine_adapters_ok": external_engine["ok"],
            "remote_visibility_status": remote_status,
            "remote_visibility_ok": remote["ok"],
            "focused_tests_ok": pytest_result["ok"],
        },
        "errors": {
            "python_runtime": python_runtime.get("error", ""),
            "fragment_scan": fragment["error"],
            "external_engine_adapters": external_engine["error"],
            "remote_visibility": remote["error"],
            "focused_tests": pytest_result["error"],
        },
        "focused_test_targets": FOCUSED_TEST_TARGETS,
        "external_engine_diagnostic_target": EXTERNAL_ENGINE_DIAGNOSTIC_TARGET,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--remote-timeout", type=int, default=8)
    parser.add_argument("--command-timeout", type=int, default=DEFAULT_COMMAND_TIMEOUT_SECONDS)
    parser.add_argument("--fragment-timeout", type=int, default=DEFAULT_FRAGMENT_TIMEOUT_SECONDS)
    parser.add_argument("--skip-tests", action="store_true")
    args = parser.parse_args()
    report = build_report(args.remote_timeout, args.command_timeout, args.fragment_timeout, args.skip_tests)
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 1 if report["status"] == "fail" else 0


if __name__ == "__main__":
    raise SystemExit(main())