"""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 # noqa: F401 (reliability_guard of the original refers to it) 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 # noqa: F401 (testing_util.py imports numpy before reliability_guard, so solutions can import it) 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) # results go here; the program's own stray writes to the real stdout go to devnull 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()