lean-bench/leanbench/verify.py

"""Step 3, second half: hand the generated Lean to the real compiler and believe it.

Two verification modes, matching the two kinds of item:

``full``       elaborate the declaration as written. If the model supplied a proof,
               this is a genuine proof check.
``statement``  the item is an open problem, so there is no proof to check: we only
               ask whether the *statement* is legal Lean -- it elaborates with
               ``sorry`` standing in for the proof.

``verified`` never means "proved". ``proved`` does, and it is only true when Lean
accepted the declaration with no ``sorry`` anywhere in it.
"""
from __future__ import annotations

import os
import re
import shutil
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path

MODE_FULL = "full"
MODE_STATEMENT = "statement"

_ERROR = re.compile(r"^.*?:\d+:\d+: error:", re.MULTILINE)
_SORRY_WARNING = re.compile(r"declaration uses 'sorry'")
_SORRY_TOKEN = re.compile(r"(?<![A-Za-z0-9_'])sorry(?![A-Za-z0-9_'])")


@dataclass
class VerifyResult:
    ok: bool
    mode: str
    used_sorry: bool
    diagnostics: str
    skipped: bool = False

    @property
    def proved(self) -> bool:
        return self.ok and not self.used_sorry and self.mode == MODE_FULL

    def as_dict(self) -> dict:
        return {
            "ok": self.ok,
            "mode": self.mode,
            "used_sorry": self.used_sorry,
            "proved": self.proved,
            "skipped": self.skipped,
            "diagnostics": self.diagnostics[-4000:],
        }


def contains_sorry(code: str) -> bool:
    return bool(_SORRY_TOKEN.search(code))


@dataclass
class LeanChecker:
    """Runs ``lake env lean`` (or whatever you configure) on a scratch file."""

    command: list[str]
    project: str | None = None
    timeout: float = 120.0
    header: str = "import Mathlib\n"

    @classmethod
    def from_settings(cls, settings) -> LeanChecker:  # noqa: ANN001 - LeanSettings
        return cls(
            command=list(settings.command),
            project=settings.project,
            timeout=settings.timeout,
            header=settings.header,
        )

    def available(self) -> bool:
        if not self.command:
            return False
        if shutil.which(self.command[0]) is None:
            return False
        if self.project and not Path(self.project).is_dir():
            return False
        return True

    def source_for(self, code: str, imports: list[str] | None = None) -> str:
        header = self.header
        for module in imports or []:
            module = module.strip()
            if not module:
                continue
            line = module if module.startswith("import ") else f"import {module}"
            if line not in header:
                header += line + "\n"
        return header.rstrip("\n") + "\n\n" + code.strip() + "\n"

    def check(
        self,
        code: str,
        *,
        mode: str = MODE_FULL,
        imports: list[str] | None = None,
    ) -> VerifyResult:
        if not self.available():
            return VerifyResult(
                ok=False,
                mode=mode,
                used_sorry=contains_sorry(code),
                diagnostics=(
                    f"Lean not available: command {self.command!r}"
                    + (f", project {self.project!r}" if self.project else "")
                ),
                skipped=True,
            )
        source = self.source_for(code, imports)
        workdir = self.project or os.getcwd()
        with tempfile.TemporaryDirectory(prefix="leanbench-", dir=workdir) as directory:
            path = Path(directory) / "LeanBenchCheck.lean"
            path.write_text(source, encoding="utf-8")
            try:
                completed = subprocess.run(  # noqa: S603 - command is operator-supplied
                    [*self.command, str(path)],
                    cwd=workdir,
                    capture_output=True,
                    text=True,
                    timeout=self.timeout,
                )
            except subprocess.TimeoutExpired:
                return VerifyResult(False, mode, contains_sorry(code),
                                    f"Lean timed out after {self.timeout}s")
            except FileNotFoundError as error:
                return VerifyResult(False, mode, contains_sorry(code),
                                    f"Lean not runnable: {error}", skipped=True)
        diagnostics = (completed.stdout or "") + (completed.stderr or "")
        ok = completed.returncode == 0 and not _ERROR.search(diagnostics)
        used_sorry = bool(_SORRY_WARNING.search(diagnostics)) or contains_sorry(code)
        return VerifyResult(ok, mode, used_sorry, diagnostics[-8000:])


class NullChecker:
    """Stands in when Lean is unavailable (unit tests, a machine without elan)."""

    def available(self) -> bool:
        return False

    def check(self, code: str, *, mode: str = MODE_FULL, imports=None) -> VerifyResult:
        return VerifyResult(False, mode, contains_sorry(code), "Lean checking disabled",
                            skipped=True)