openlatch-client 0.6.21

#!/usr/bin/env python3
"""Prove evaluator readiness locally before client publication, without publishing.

Run from any directory with --binary pointing at the built release artifact.
The module corpus runs from this checkout; the differential, transcript and
side-effect checks all execute that exact supplied binary. No test can be skipped.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import shlex
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parents[1]
TEST_BINARY = "OPENLATCH_EVALUATE_TEST_BINARY"
TEST_FLOORS = {"zone_eval_conformance": 1, "evaluate_transcript": 4, "evaluate_side_effects": 4}


class GateError(Exception):
    """A failed prerequisite retains its process status; invalid evidence exits 1."""

    def __init__(self, message: str, code: int = 1):
        super().__init__(message)
        self.code = code if code >= 0 else 128 - code


def run_command(command: list[str], env: dict[str, str]) -> subprocess.CompletedProcess:
    """Capture both streams without a shell or pipeline that could hide failure."""
    print(f"+ {shlex.join(command)}", flush=True)
    result = subprocess.run(command, cwd=ROOT, env=env, text=True,
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
    print(result.stdout, end="", flush=True)
    return result


def test_counts(output: str, target: str) -> dict:
    """Require completed, nonempty libtest evidence, including formerly ignored tests."""
    running = re.findall(r"^running (\d+) tests?$", output, re.MULTILINE)
    results = re.findall(
        r"^test result: (\w+)\. (\d+) passed; (\d+) failed; (\d+) ignored;", output, re.MULTILINE
    )
    if len(running) != 1 or len(results) != 1:
        raise GateError(f"{target}: missing or ambiguous libtest summary")
    outcome, passed, failed, ignored = results[0]
    counts = {"passed": int(passed), "failed": int(failed), "ignored": int(ignored)}
    counts["minimum_tests"] = TEST_FLOORS[target]
    if (outcome != "ok" or counts["failed"] or counts["ignored"]
            or counts["passed"] < counts["minimum_tests"]
            or int(running[0]) != counts["passed"]):
        raise GateError(f"{target}: failed, ignored, missing or ZERO tests: {counts}")
    return counts


def module_counts(output: str, paths: list[Path]) -> dict:
    """Use the existing corpus tally, never a count of parsed input lines."""
    totals = re.findall(r"^TOTAL\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s*$", output, re.MULTILINE)
    files = re.findall(r"^(\S+\.jsonl)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s*$",
                       output, re.MULTILINE)
    if len(totals) != 1 or sorted(row[0] for row in files) != sorted(p.name for p in paths):
        raise GateError("module corpus: missing totals or incomplete corpus file coverage")
    rows, passed, failed, errors = map(int, totals[0])
    evaluated = passed + failed
    if (not evaluated or failed or errors or rows != evaluated
            or tuple(sum(int(f[i]) for f in files) for i in range(1, 5))
            != (rows, passed, failed, errors)):
        raise GateError(f"module corpus: {evaluated} evaluated, {failed} failed, {errors} errors")
    return {"evaluated_rows": evaluated, "passed": passed, "failed": failed,
            "errors": errors, "files": [p.name for p in paths]}


def check_batch(report: dict, paths: list[Path], expected_rows: int) -> None:
    """Every expected file must be covered in response totals matching the module tally."""
    if not isinstance(report, dict):
        raise GateError("binary corpus: report is not an object")
    files = report.get("files", [])
    if not isinstance(files, list) or not all(isinstance(f, dict) for f in files):
        raise GateError("binary corpus: file evidence is not a list of objects")
    if (report.get("failures") != 0 or type(report.get("evaluated_rows")) is not int
            or report["evaluated_rows"] <= 0 or report["evaluated_rows"] != expected_rows
            or sorted(f.get("path", "") for f in files) != sorted(str(p) for p in paths)
            or any(type(f.get("evaluated_rows")) is not int or f["evaluated_rows"] < 0
                   or f.get("failures") != 0 for f in files)
            or sum(f["evaluated_rows"] for f in files) != expected_rows):
        raise GateError("binary corpus: red, empty or incomplete evaluated-row evidence")


def verify(binary: Path, report: dict) -> None:
    """Run all prerequisite stages; neither fixture rewriting nor silent fallback is allowed."""
    if not binary.is_file() or not os.access(binary, os.X_OK):
        raise GateError(f"binary is missing or not executable: {binary}", 2)
    report["binary_sha256"] = hashlib.sha256(binary.read_bytes()).hexdigest()
    paths = sorted((ROOT / "schemas/conformance").glob("*.jsonl"))
    if not paths:
        raise GateError("ZERO corpus files; readiness cannot be established")
    env = os.environ.copy()
    env["CARGO_TERM_COLOR"] = "never"
    env[TEST_BINARY] = str(binary)
    env.pop("OL_UPDATE_TRANSCRIPT", None)

    def rust_test(target: str) -> str:
        report["stage"] = target
        result = run_command([
            "cargo", "test", "--locked", "--features", "policy", "--test", target,
            "--", "--include-ignored", "--nocapture", "--color", "never",
        ], env)
        report.setdefault("process_status", {})[target] = result.returncode
        if result.returncode:
            raise GateError(f"{target}: process exited {result.returncode}", result.returncode)
        report.setdefault("tests", {})[target] = test_counts(result.stdout, target)
        return result.stdout

    output = rust_test("zone_eval_conformance")
    report["module"] = module_counts(output, paths)
    report["stage"] = "binary_corpus"
    with tempfile.TemporaryDirectory(prefix="evaluate-readiness-") as scratch:
        evidence = Path(scratch) / "batch.json"
        result = run_command([
            sys.executable, str(ROOT / "tools/zone-eval-ref/drt.py"),
            *map(str, paths), "--evaluate-cmd", shlex.join([str(binary), "system", "evaluate"]),
            "--report-json", str(evidence),
        ], env)
        report["process_status"]["binary_corpus"] = result.returncode
        try:
            if evidence.is_file():
                report["binary_corpus"] = json.loads(evidence.read_text(encoding="utf-8"))
        except (OSError, ValueError) as exc:
            report["binary_corpus_report_error"] = str(exc)
        if result.returncode:
            # A missing/malformed report must never replace the process failure.
            raise GateError(f"binary corpus: process exited {result.returncode}", result.returncode)
        check_batch(report.get("binary_corpus", {}), paths, report["module"]["evaluated_rows"])

    rust_test("evaluate_transcript")
    rust_test("evaluate_side_effects")
    if hashlib.sha256(binary.read_bytes()).hexdigest() != report["binary_sha256"]:
        raise GateError("supplied binary changed during readiness verification")
    report["stage"] = "complete"
    report["ready"] = True


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--binary", type=Path, required=True, help="exact native artifact to verify")
    parser.add_argument("--report-json", type=Path, help="write readiness evidence, including failure")
    args = parser.parse_args(argv)
    binary = args.binary.resolve()
    report = {"ready": False, "binary": str(binary), "stage": "preflight"}
    try:
        verify(binary, report)
        code = 0
    except GateError as exc:
        report["error"] = str(exc)
        code = exc.code
    except (OSError, ValueError, TypeError, KeyError) as exc:
        report["error"] = f"unusable readiness evidence: {exc}"
        code = 1
    report["exit_code"] = code
    if args.report_json:
        args.report_json.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(report, indent=2))
    return code


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