| """Runs one LiveCodeBench problem the way the official harness does (lcb_runner/evaluation/testing_util.py of |
| LiveCodeBench, read 2026-11-06): the same import prelude, stdin programs wrapped into wrapped_function() after removing a |
| final `if __name__ == '__main__':`, the same mocked stdin (MockStdinWithBuffer), stdout capture and comparison (exact |
| line match, else Decimal equality), call-based results compared with == (tuple -> list), the same reliability_guard. |
| Windows has no SIGALRM, so each test's 6 s limit is enforced by a watchdog thread that reports a timeout and ends the |
| process. All tests of one problem run in one process, in order, stopping at the first failure. |
| usage (from lcb_official.py): python -I lcb_official_runner.py payload.json -> one JSON line per test on stdout |
| """ |
| import ast |
| import faulthandler |
| import json |
| import os |
| import platform |
| import sys |
| import threading |
| from decimal import Decimal |
| from io import StringIO |
| from types import ModuleType |
| from unittest.mock import mock_open, patch |
|
|
| import numpy as np |
|
|
| import_string = "from string import *\nfrom re import *\nfrom datetime import *\nfrom collections import *\nfrom heapq import *\nfrom bisect import *\nfrom copy import *\nfrom math import *\nfrom random import *\nfrom statistics import *\nfrom itertools import *\nfrom functools import *\nfrom operator import *\nfrom io import *\nfrom sys import *\nfrom json import *\nfrom builtins import *\nfrom typing import *\nimport string\nimport re\nimport datetime\nimport collections\nimport heapq\nimport bisect\nimport copy\nimport math\nimport random\nimport statistics\nimport itertools\nimport functools\nimport operator\nimport io\nimport sys\nimport json\nsys.setrecursionlimit(50000)\n" |
|
|
|
|
| class Capturing(list): |
| def __enter__(self): |
| self._stdout = sys.stdout |
| sys.stdout = self._stringio = StringIO() |
| self._stringio.close = lambda x: 1 |
| return self |
|
|
| def __exit__(self, *args): |
| self.append(self._stringio.getvalue()) |
| del self._stringio |
| sys.stdout = self._stdout |
|
|
|
|
| class MockBuffer: |
| def __init__(self, inputs: str): |
| self.inputs = inputs.encode("utf-8") |
|
|
| def read(self, *args): |
| return self.inputs |
|
|
| def readline(self, *args): |
| return self.inputs.split(b"\n")[0] + b"\n" |
|
|
|
|
| class MockStdinWithBuffer: |
| def __init__(self, inputs: str): |
| self.inputs = inputs |
| self._stringio = StringIO(inputs) |
| self.buffer = MockBuffer(inputs) |
|
|
| def read(self, *args): |
| return self.inputs |
|
|
| def readline(self, *args): |
| return self._stringio.readline(*args) |
|
|
| def readlines(self, *args): |
| return self.inputs.split("\n") |
|
|
| def __getattr__(self, name): |
| return getattr(self._stringio, name) |
|
|
|
|
| def call_method(method, inputs): |
| if isinstance(inputs, list): |
| inputs = "\n".join(inputs) |
| inputs_line_iterator = iter(inputs.split("\n")) |
| mock_stdin = MockStdinWithBuffer(inputs) |
|
|
| @patch("builtins.open", mock_open(read_data=inputs)) |
| @patch("sys.stdin", mock_stdin) |
| @patch("sys.stdin.readline", lambda *args: next(inputs_line_iterator)) |
| @patch("sys.stdin.readlines", lambda *args: inputs.split("\n")) |
| @patch("sys.stdin.read", lambda *args: inputs) |
| def _inner_call_method(_method): |
| try: |
| return _method() |
| except SystemExit: |
| pass |
|
|
| return _inner_call_method(method) |
|
|
|
|
| def clean_if_name(code: str) -> str: |
| try: |
| astree = ast.parse(code) |
| last_block = astree.body[-1] |
| if isinstance(last_block, ast.If): |
| condition = last_block.test |
| if ast.unparse(condition).strip() == "__name__ == '__main__'": |
| code = ast.unparse(astree.body[:-1]) + "\n" + ast.unparse(last_block.body) |
| except Exception: |
| pass |
| return code |
|
|
|
|
| def make_function(code: str) -> str: |
| try: |
| import_stmts = [] |
| all_other_stmts = [] |
| astree = ast.parse(code) |
| for stmt in astree.body: |
| if isinstance(stmt, (ast.Import, ast.ImportFrom)): |
| import_stmts.append(stmt) |
| else: |
| all_other_stmts.append(stmt) |
| function_ast = ast.FunctionDef( |
| name="wrapped_function", |
| args=ast.arguments(posonlyargs=[], args=[], kwonlyargs=[], kw_defaults=[], defaults=[]), |
| body=all_other_stmts, decorator_list=[], lineno=-1) |
| return import_string + "\n" + ast.unparse(import_stmts) + "\n" + ast.unparse(function_ast) |
| except Exception: |
| return code |
|
|
|
|
| def get_stripped_lines(val: str): |
| val = val.strip() |
| return [val_line.strip() for val_line in val.split("\n")] |
|
|
|
|
| def convert_line_to_decimals(line: str): |
| try: |
| return True, [Decimal(elem) for elem in line.split()] |
| except Exception: |
| return False, [] |
|
|
|
|
| def same_stdio(prediction, gt_out): |
| p, g = get_stripped_lines(prediction), get_stripped_lines(gt_out) |
| if len(p) != len(g): |
| return False |
| for a, b in zip(p, g): |
| if a == b: |
| continue |
| ok1, da = convert_line_to_decimals(a) |
| ok2, db = convert_line_to_decimals(b) |
| if not (ok1 and ok2 and da == db): |
| return False |
| return True |
|
|
|
|
| def reliability_guard(): |
| faulthandler.disable() |
| import builtins |
| builtins.quit = None |
| os.environ["OMP_NUM_THREADS"] = "1" |
| for name in ("kill", "system", "putenv", "remove", "removedirs", "rmdir", "fchdir", "setuid", "fork", "forkpty", |
| "killpg", "rename", "renames", "truncate", "replace", "unlink", "fchmod", "fchown", "chmod", "chown", |
| "chroot", "lchflags", "lchmod", "lchown", "getcwd", "chdir"): |
| setattr(os, name, None) |
| import shutil |
| shutil.rmtree = shutil.move = shutil.chown = None |
| import subprocess |
| subprocess.Popen = None |
| builtins.help = None |
| for m in ("ipdb", "joblib", "resource", "psutil", "tkinter"): |
| sys.modules[m] = None |
|
|
|
|
| def main(): |
| p = json.load(open(sys.argv[1], encoding="utf-8")) |
| res_fd = os.dup(1) |
| null = os.open(os.devnull, os.O_WRONLY) |
| os.dup2(null, 1) |
| _exit, _write = os._exit, os.write |
|
|
| def report(d): |
| _write(res_fd, (json.dumps(d) + "\n").encode()) |
|
|
| def watchdog(k): |
| report({"k": k, "res": "tle"}) |
| _exit(0) |
|
|
| reliability_guard() |
| timeout, mode = p["timeout"], p["mode"] |
| code = import_string + "\n\n" + p["code"] if mode == "functional" else make_function(clean_if_name(p["code"])) |
| t = threading.Timer(timeout, watchdog, args=(0,)) |
| t.start() |
| try: |
| tmp_sol = ModuleType("tmp_sol", "") |
| exec(code, tmp_sol.__dict__) |
| compiled = tmp_sol.Solution() if "class Solution" in code else tmp_sol |
| except BaseException as e: |
| t.cancel() |
| report({"k": 0, "res": "error", "err": "compile: " + repr(e)[:300]}) |
| return |
| t.cancel() |
| method = getattr(compiled, p["fn"] if mode == "functional" else "wrapped_function", None) |
| if method is None: |
| report({"k": 0, "res": "error", "err": "function not found"}) |
| return |
| tests = p["tests"] |
| if mode == "functional": |
| try: |
| args_all = [[json.loads(line) for line in inp.split("\n")] for inp, _ in tests] |
| outs_all = [json.loads(out) for _, out in tests] |
| except Exception as e: |
| report({"k": 0, "res": "error", "err": "decode: " + repr(e)[:200]}) |
| return |
| for k, (inp, exp) in enumerate(tests): |
| t = threading.Timer(timeout, watchdog, args=(k,)) |
| t.start() |
| if mode == "functional": |
| try: |
| pred = method(*args_all[k]) |
| except BaseException as e: |
| t.cancel() |
| report({"k": k, "res": "error", "err": repr(e)[:300]}) |
| return |
| t.cancel() |
| if isinstance(pred, tuple): |
| pred = list(pred) |
| if not pred == outs_all[k]: |
| report({"k": k, "res": "wrong", "got": repr(pred)[:300]}) |
| return |
| else: |
| err = None |
| with Capturing() as captured: |
| try: |
| call_method(method, inp) |
| except BaseException as e: |
| err = repr(e)[:300] |
| t.cancel() |
| if err is not None: |
| report({"k": k, "res": "error", "err": err}) |
| return |
| if not same_stdio(captured[0], exp): |
| report({"k": k, "res": "wrong", "got": captured[0][:300]}) |
| return |
| report({"k": k, "res": "ok"}) |
|
|
|
|
| main() |
|
|
