#!/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())