space_app/verification.py

from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any


ROOT_DIR = Path(__file__).resolve().parents[1]
EVALS_SRC_DIR = ROOT_DIR / "evals" / "src"
if str(EVALS_SRC_DIR) not in sys.path:
    sys.path.insert(0, str(EVALS_SRC_DIR))

from evals_verifier import (  # type: ignore
    check_bridges_structural_validity,
    check_galaxies_structural_validity,
    check_undead_structural_validity,
    verify_with_rlp,
)
from puzzles import PUZZLES  # type: ignore


# Puzzle families whose "solved" check runs in the native RLP C library (loaded
# in-process via ctypes). That library can segfault or hang when its puzzle
# objects are reused across many requests in a long-lived server process, which
# crashes the web worker and surfaces as an opaque HTTP 500 (and, sometimes, a
# hung request). We run every native check in a short-lived subprocess so a
# native crash/hang kills only the child and we return a clean "not solved"
# result instead of taking down the worker. (flow_free already shells out to its
# own compiled solver binary, so it is naturally isolated.)
SUBPROCESS_RLP_PUZZLES = {"undead", "loopy", "bridges", "galaxies", "pattern"}
RLP_VERIFIER_TIMEOUT_SECONDS = 20
# flow_free shells out to a compiled C solver with no internal timeout and can
# raise SystemExit on a missing compiler/binary; isolate it too so it can never
# hang or crash the worker. Enumerating all solutions can be slower than the
# RLP solved-check, so it gets a more generous budget.
FLOW_FREE_TIMEOUT_SECONDS = 30


class VerificationService:
    def _structural_validity(
        self, *, puzzle_type: str, board_ascii: str, problem_ascii: str
    ) -> bool:
        # Pure-Python structural pre-checks are safe to run in-process and also
        # cover the "start board was modified" case for bridges/galaxies. Puzzle
        # families without such a check are treated as structurally valid here;
        # the subprocess solved-check is the source of truth for correctness.
        if puzzle_type == "bridges":
            return bool(
                check_bridges_structural_validity(board_ascii, problem_ascii=problem_ascii)
            )
        if puzzle_type == "galaxies":
            return bool(
                check_galaxies_structural_validity(board_ascii, problem_ascii=problem_ascii)
            )
        if puzzle_type == "undead":
            return bool(check_undead_structural_validity(board_ascii))
        return True

    def _verify_rlp_with_subprocess(
        self,
        *,
        puzzle_type: str,
        problem_ascii: str,
        board_ascii: str,
        args: str,
    ) -> dict[str, Any]:
        result = {
            "board_exists": bool(board_ascii),
            "board_valid": False,
            "board_modified": False,
            "correct": False,
        }
        if not board_ascii:
            return result

        result["board_valid"] = self._structural_validity(
            puzzle_type=puzzle_type, board_ascii=board_ascii, problem_ascii=problem_ascii
        )
        if not result["board_valid"]:
            result["board_modified"] = True
            return result

        verifier_path = ROOT_DIR / "submodules" / "rlp" / "verifier.py"
        command = [sys.executable, str(verifier_path), puzzle_type, "--arg", args]
        # The RLP libraries pull in pygame, which noisily probes audio/video on
        # import; force the dummy drivers (and silence warnings) so a headless
        # verifier run does not pollute the captured stderr.
        env = {
            **os.environ,
            "SDL_AUDIODRIVER": "dummy",
            "SDL_VIDEODRIVER": "dummy",
            "PYTHONWARNINGS": "ignore",
        }
        try:
            completed = subprocess.run(
                command,
                input=board_ascii,
                text=True,
                capture_output=True,
                cwd=ROOT_DIR,
                timeout=RLP_VERIFIER_TIMEOUT_SECONDS,
                check=False,
                env=env,
            )
        except subprocess.TimeoutExpired:
            result["error"] = f"{puzzle_type.capitalize()} verification timed out."
            return result

        result["correct"] = completed.returncode == 0 and "SOLVED" in completed.stdout
        if completed.stderr.strip():
            result["stderr"] = completed.stderr.strip()[:500]
        return result

    def _verify_flow_free_with_subprocess(
        self, *, problem_ascii: str, board_ascii: str
    ) -> dict[str, Any]:
        result = {
            "board_exists": bool(board_ascii),
            "board_valid": False,
            "board_modified": False,
            "correct": False,
        }
        if not board_ascii:
            return result

        runner_path = Path(__file__).resolve().parent / "flow_free_runner.py"
        payload = json.dumps({"problem_ascii": problem_ascii, "board_ascii": board_ascii})
        env = {**os.environ, "PYTHONWARNINGS": "ignore"}
        try:
            completed = subprocess.run(
                [sys.executable, str(runner_path)],
                input=payload,
                text=True,
                capture_output=True,
                cwd=ROOT_DIR,
                timeout=FLOW_FREE_TIMEOUT_SECONDS,
                check=False,
                env=env,
            )
        except subprocess.TimeoutExpired:
            result["error"] = "Flow_free verification timed out."
            return result

        if completed.returncode == 0 and completed.stdout.strip():
            try:
                parsed = json.loads(completed.stdout)
                if isinstance(parsed, dict):
                    return parsed
            except json.JSONDecodeError:
                pass
        result["error"] = "Flow_free verification failed."
        if completed.stderr.strip():
            result["stderr"] = completed.stderr.strip()[:500]
        return result

    def verify(
        self,
        *,
        puzzle_type: str,
        problem_ascii: str,
        board_ascii: str,
        args: str,
    ) -> dict[str, Any]:
        spec = PUZZLES[puzzle_type]
        if spec.verifier_type == "flow_free":
            return self._verify_flow_free_with_subprocess(
                problem_ascii=problem_ascii, board_ascii=board_ascii
            )
        if spec.verifier_type in SUBPROCESS_RLP_PUZZLES:
            return self._verify_rlp_with_subprocess(
                puzzle_type=spec.verifier_type,
                problem_ascii=problem_ascii,
                board_ascii=board_ascii,
                args=args,
            )
        return verify_with_rlp(
            puzzle_type=spec.verifier_type,
            problem_ascii=problem_ascii,
            extracted_board=board_ascii,
            args=args,
        )