Harden Jyotish runtime evidence gates

2026-07-05 18:27:33 +08:00
parent 8520921f40
commit 4764482c73
76 changed files with 5628 additions and 253 deletions
@@ -0,0 +1,154 @@
+#!/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 subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+PYTHON = sys.executable
+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_round25_2026_06_25.md",
+]
+
+
+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,
+) -> str:
+    if not docs_ok or not fragment_ok or not pytest_ok or not external_engine_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)
+    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)
+    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)
+        pytest_result = run([PYTHON, "-m", "pytest", "-q", *FOCUSED_TEST_TARGETS], command_timeout, env=env)
+    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"],
+    )
+    return {
+        "scope": "pre_work_check",
+        "status": status,
+        "must_not_claim_synced": remote_report.get("must_not_claim_synced", True),
+        "docs": docs,
+        "git": {
+            "status_ok": git_status["ok"],
+            "status": git_status["stdout"],
+            "remote_ok": git_remote["ok"],
+            "remote": git_remote["stdout"],
+        },
+        "checks": {
+            "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": {
+            "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())