diff --git a/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py b/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py index 71fc6bd2f..cb0db9d73 100644 --- a/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py +++ b/src/inference_endpoint/evaluation/livecodebench/lcb_serve.py @@ -35,6 +35,7 @@ """ import argparse +import ctypes import json import logging import multiprocessing as mp @@ -43,26 +44,46 @@ from collections.abc import Callable from concurrent.futures import ProcessPoolExecutor, as_completed from functools import lru_cache +from multiprocessing.sharedctypes import Synchronized from pathlib import Path +from typing import cast +import numpy as np import pandas as pd from tqdm import tqdm from .generate import generate_dataset +from .run_lcb_tests import run_test logger = logging.getLogger(__name__) +# Error codes that mean the judge is broken, as opposed to the submitted code +# failing its tests: -5 TestRunnerError, -6 GradingChildDied (child died +# before grading started, e.g. a bad start method). Submission-attributed +# codes stay out: timeouts (-1), sys.exit() (-7 SubmissionExit), submissions +# that kill their own interpreter (-8 SubmissionKilledChild), and -4 (the +# submission's code failed to compile or define the expected function -- +# see the outer except in run_lcb_tests.grade_call_based/grade_stdio callers +# -- which is the submission's fault, not the judge's, even though it's +# labeled "Error during testing"). +_LCB_INFRA_ERROR_CODES = {-5, -6} -def execute_code_single(test_suite_json: str, code: str, timeout_sec: int = 60): + +def execute_code_single( + test_suite_json: str, + code: str, + timeout_sec: int = 60, + *, + started_flag: Synchronized | None = None, +) -> tuple[list, dict]: # Run code with lcb_runner. Note that the lcb_runner has a very rudimentary sandbox # which is extremely easy to bypass, and as such it is recommended to run this both # in an unprivileged container and in a separate process. - import numpy as np - - from .run_lcb_tests import run_test - res, metadata = run_test( - {"input_output": test_suite_json}, test=code, timeout=timeout_sec + {"input_output": test_suite_json}, + test=code, + timeout=timeout_sec, + started_flag=started_flag, ) # LCB results are expected to be plain booleans or error codes. @@ -78,16 +99,36 @@ def execute_code_single(test_suite_json: str, code: str, timeout_sec: int = 60): def execute_code_single_suppressed_errors( - *args, resp_buffer: list | None = None, **kwargs -): + test_suite_json: str, + code: str, + timeout_sec: int = 60, + *, + resp_buffer: list | None = None, + started_flag: Synchronized | None = None, +) -> tuple[list, dict]: """Wrapper around execute code so that all errors are resurfaced as failed tests""" try: - res, metadata = execute_code_single(*args, **kwargs) + # started_flag is flipped inside run_test, once judge-side setup + # (reliability_guard, suite parse) is done and the submission's own + # code is next; see run_lcb_tests.run_test. + res, metadata = execute_code_single( + test_suite_json, code, timeout_sec=timeout_sec, started_flag=started_flag + ) if not isinstance(res, list): raise ValueError(f"Expected boolean result, got {type(res)}") if not isinstance(metadata, dict): raise ValueError(f"Expected metadata to be a dict, got {type(metadata)}") + except SystemExit as e: + # sys.exit() in submitted code is a BaseException, not caught below. + # It's the submission's fault, not the judge's, so give it its own + # error code and keep it out of _LCB_INFRA_ERROR_CODES. + res = [-2] + metadata = { + "error": f"Submission called sys.exit({e.code!r})", + "error_code": -7, + "error_message": "SubmissionExit", + } except Exception: # Magic number (see https://github.com/LiveCodeBench/LiveCodeBench/blob/28fef95ea8c9f7a547c8329f2cd3d32b92c1fa24/lcb_runner/evaluation/compute_code_generation_metrics.py#L65) res = [-2] # LCB internal error code for test runner failed test cases @@ -106,7 +147,7 @@ def run_code_subprocess( test_suite_json: str, code: str, timeout_sec: int = 60, -): +) -> tuple[list, dict]: # Compute global timeout - # https://github.com/LiveCodeBench/LiveCodeBench/blob/28fef95ea8c9f7a547c8329f2cd3d32b92c1fa24/lcb_runner/evaluation/compute_code_generation_metrics.py#L43 @@ -117,37 +158,64 @@ def run_code_subprocess( suite["inputs"] ) + flat_timeout_extension - manager = mp.Manager() - resp_buffer = manager.list() - p = mp.Process( - target=execute_code_single_suppressed_errors, - args=( - test_suite_json, - code, - ), - kwargs={ - "resp_buffer": resp_buffer, - "timeout_sec": timeout_sec, - }, - ) - p.start() - p.join(timeout=global_timeout) + with mp.Manager() as manager: + resp_buffer = manager.list() + # typeshed types ctx.Value() as SynchronizedBase, which lacks .value + started_flag = cast(Synchronized, mp.Value(ctypes.c_bool, False)) + p = mp.Process( + target=execute_code_single_suppressed_errors, + args=( + test_suite_json, + code, + ), + kwargs={ + "resp_buffer": resp_buffer, + "timeout_sec": timeout_sec, + "started_flag": started_flag, + }, + ) + p.start() + p.join(timeout=global_timeout) - if p.is_alive(): - p.kill() + timed_out = p.is_alive() + if timed_out: + p.kill() + p.join() - if len(resp_buffer) == 0: - # Assume timeout - res = [-1] * len(suite["inputs"]) + if len(resp_buffer) > 0: + return resp_buffer[0] + + started = bool(started_flag.value) + exitcode = p.exitcode + + # No result was reported: every test case counts as failed, only the + # attribution differs. + res = [-1] * len(suite["inputs"]) + if timed_out: + # Still running at the deadline: the submitted code took too long. metadata = { "error": "Test suite timeout", "error_code": -1, "error_message": f"Subprocess did not complete in time ({global_timeout}s)", } - return res, metadata + elif started: + # The interpreter died while grading was running (os._exit(), + # segfault, OOM, ...). Grading executes the untrusted submission, so + # this is the submission's fault, not the judge's. + metadata = { + "error": "Grading child killed while executing the submission", + "error_code": -8, + "error_message": f"SubmissionKilledChild (exitcode={exitcode})", + } else: - res, metadata = resp_buffer[0] - return res, metadata + # Died before grading started (e.g. bad start method): the judge is + # broken. + metadata = { + "error": "Grading subprocess died before grading started", + "error_code": -6, + "error_message": f"GradingChildDied (exitcode={exitcode})", + } + return res, metadata class LCBTestLoader: @@ -270,6 +338,7 @@ def __call__( for qid, test_codes in zip(question_ids, codes, strict=False): results[qid] = [False] * len(test_codes) futures = {} + infra_errors = 0 with ProcessPoolExecutor(max_workers=self.n_lcb_workers) as executor: for qid, test_codes in zip(question_ids, codes, strict=False): @@ -290,9 +359,19 @@ def __call__( qid, code_idx = futures[future] res, metadata = future.result() if "error" in metadata: - logger.warning( - f"Test execution error for question {qid}: {metadata}" - ) + if metadata.get("error_code") in _LCB_INFRA_ERROR_CODES: + infra_errors += 1 + logger.error( + f"Test execution error for question {qid}: {metadata}" + ) + else: + # Routine submission-attributed outcomes (timeout, + # sys.exit, os._exit, bad code) -- expected at scale, + # would otherwise flood ERROR and drown out the + # infra signal above. + logger.warning( + f"Test execution error for question {qid}: {metadata}" + ) # LCB uses any result > 0 as a 'pass' since: # Negative numbers indicate error codes @@ -317,6 +396,17 @@ def __call__( exc_info=True, ) + # Every subprocess hitting an infra error means the judge itself is + # broken, not that every code sample failed its tests. Timeouts are + # excluded: a batch where every submission loops forever is a valid + # 0 score, not a broken judge. + if futures and infra_errors == len(futures): + raise RuntimeError( + f"All {len(futures)} grading subprocesses reported " + "infrastructure errors - the LCB judge is broken; refusing " + "to report a 0 score. See the logged error metadata above." + ) + return results @@ -348,6 +438,7 @@ def __init__( if n_workers is None: n_workers = mp.cpu_count() // 2 logger.info("Using %d workers for LCB eval", n_workers) + logger.info("Multiprocessing start method: %s", mp.get_start_method()) self.n_workers = n_workers self.path_to_dataset = ( diff --git a/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py b/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py index 16798f6ff..9ac56b30a 100644 --- a/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py +++ b/src/inference_endpoint/evaluation/livecodebench/run_lcb_tests.py @@ -30,6 +30,7 @@ from decimal import Decimal from enum import Enum from io import StringIO +from multiprocessing.sharedctypes import Synchronized # from pyext import RuntimeModule from types import ModuleType @@ -475,7 +476,7 @@ def grade_stdio( return all_results, {"execution time": total_execution_time} -def run_test(sample, test=None, timeout=6): +def run_test(sample, test=None, timeout=6, started_flag: Synchronized | None = None): """ if test(generated_code) is not None it'll try to run the code. otherwise it'll just return an input and output pair. @@ -506,7 +507,16 @@ def run_test(sample, test=None, timeout=6): elif test is not None: results = [] + if started_flag is not None: + # Judge-side setup (reliability_guard, suite parse) is done; the + # next thing that runs is the submission itself (grade_call_based + # / grade_stdio compile and exec `test`), so a death from here on + # is the submission's doing, not the judge's. + started_flag.value = True + if which_type == CODE_TYPE.call_based: + # method_name is only None for CODE_TYPE.standard_input (above). + assert method_name is not None signal.alarm(timeout) try: results, metadata = grade_call_based( @@ -518,7 +528,11 @@ def run_test(sample, test=None, timeout=6): ) return results, metadata except Exception as e: + # Reached only if the submission's code fails to compile or + # doesn't define the expected function -- grade_call_based's + # own per-test-case loop already handles runtime errors. return [-4], { + "error": repr(e), "error_code": -4, "error_message": f"Error during testing: {e}", } @@ -538,7 +552,10 @@ def run_test(sample, test=None, timeout=6): ) return results, metadata except Exception as e: + # Same as the call_based branch above: a compile/definition + # failure in the submission's own code, not a judge bug. return [-4], { + "error": repr(e), "error_code": -4, "error_message": f"Error during testing: {e}", } diff --git a/tests/unit/evaluation/test_lcb_serve.py b/tests/unit/evaluation/test_lcb_serve.py new file mode 100644 index 000000000..154cc0c80 --- /dev/null +++ b/tests/unit/evaluation/test_lcb_serve.py @@ -0,0 +1,270 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 CoreWeave, Inc. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for lcb_serve.py's grading-child error attribution. + +run_code_subprocess forks a grading child per sample and must classify a +child that reports no result into exactly one of: + -1 timeout (submission ran past the deadline) + -6 GradingChildDied (judge-side: died before or while doing its own + setup -- reliability_guard, suite parse -- i.e. + before the submission's code ever ran) + -8 SubmissionKilledChild (submission-side: died while its own code, or + run_test's dispatch into it, was executing) + +These tests call run_code_subprocess directly (not through _LCBWorker's +pool) so that monkeypatches applied in the test process are inherited by +the forked grading child; the batch-level guard tests below go through the +real pool instead, for the same reason. + +The fault-injection tests below need "fork" specifically -- not just as the +interpreter's default, but pinned explicitly -- because pytest runs this +whole suite in one process, and some other module (see +endpoint_client/worker.py's multiprocessing.set_start_method("spawn"), +called at import time) may have already claimed the process-wide default +before this file's tests run. Pinning fork here, locally, keeps the fault +injection deterministic regardless of import order without deciding +lcb_serve's own production start-method policy, which is being tracked in +a follow-up PR. +""" + +import multiprocessing +import os +from concurrent.futures import ProcessPoolExecutor +from functools import partial + +import pytest +from inference_endpoint.evaluation.livecodebench import lcb_serve, run_lcb_tests + +pytestmark = pytest.mark.unit + + +def _force_fork_context(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin lcb_serve's pool and inner Process/Manager/Value to "fork" for a + single test, regardless of the process-wide multiprocessing default.""" + fork_ctx = multiprocessing.get_context("fork") + monkeypatch.setattr(lcb_serve, "mp", fork_ctx) + monkeypatch.setattr( + lcb_serve, + "ProcessPoolExecutor", + partial(ProcessPoolExecutor, mp_context=fork_ctx), + ) + + +CALL_BASED_SUITE = '{"fn_name": "solve", "inputs": ["1"], "outputs": ["1"]}' +PASSING_CALL_BASED = "def solve(x):\n return x\n" +CALL_SYSEXIT = "def solve(x):\n import sys\n sys.exit(3)\n" +CALL_OS_EXIT = "import os\nos._exit(3)\ndef solve(x):\n return x\n" + +# Keep timeouts short; the grading itself is instant, this only bounds how +# long a genuinely hung test would take to fail. +TIMEOUT_SEC = 5 + + +def test_passing_submission_scores_true(): + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, PASSING_CALL_BASED, timeout_sec=TIMEOUT_SEC + ) + assert res == [True] + assert metadata.get("error_code") is None + + +def test_sys_exit_is_submission_exit_not_infra(): + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, CALL_SYSEXIT, timeout_sec=TIMEOUT_SEC + ) + assert metadata["error_code"] == -7 + assert metadata["error_code"] not in lcb_serve._LCB_INFRA_ERROR_CODES + + +def test_os_exit_in_submission_is_submission_killed_child_not_infra(): + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, CALL_OS_EXIT, timeout_sec=TIMEOUT_SEC + ) + assert metadata["error_code"] == -8 + assert metadata["error_code"] not in lcb_serve._LCB_INFRA_ERROR_CODES + + +def test_death_before_run_test_is_infra_error(monkeypatch): + """A judge process that dies before run_test is even reached (e.g. a + bad start method, or a crash while importing) is -6, not -8.""" + _force_fork_context(monkeypatch) + orig = lcb_serve.execute_code_single_suppressed_errors + + def _die_immediately(*args, **kwargs): + os._exit(1) + + lcb_serve.execute_code_single_suppressed_errors = _die_immediately + try: + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, PASSING_CALL_BASED, timeout_sec=TIMEOUT_SEC + ) + finally: + lcb_serve.execute_code_single_suppressed_errors = orig + + assert metadata["error_code"] == -6 + assert metadata["error_code"] in lcb_serve._LCB_INFRA_ERROR_CODES + + +def test_death_during_judge_setup_is_infra_error_not_submission(monkeypatch): + """A death inside run_test's own setup (reliability_guard, suite parse) + -- after run_test has started but before started_flag is set -- must + still be -6. This is the exact boundary the started_flag fix moved: it + used to flip True on wrapper entry, before this setup ran, which would + have misattributed this case as -8.""" + _force_fork_context(monkeypatch) + orig_guard = run_lcb_tests.reliability_guard + + def _guard_then_die(*args, **kwargs): + orig_guard(*args, **kwargs) + os._exit(1) + + run_lcb_tests.reliability_guard = _guard_then_die + try: + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, PASSING_CALL_BASED, timeout_sec=TIMEOUT_SEC + ) + finally: + run_lcb_tests.reliability_guard = orig_guard + + assert metadata["error_code"] == -6 + assert metadata["error_code"] in lcb_serve._LCB_INFRA_ERROR_CODES + + +def test_compile_error_in_submission_is_not_infra_but_is_logged(): + """A submission that doesn't even compile hits the outer except in + grade_call_based's caller, gets -4, and is the submission's fault, not + the judge's -- but must still carry an "error" key so it's visible.""" + bad_syntax = "def solve(x\n return x\n" # missing closing paren + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, bad_syntax, timeout_sec=TIMEOUT_SEC + ) + assert metadata["error_code"] == -4 + assert "error" in metadata + assert metadata["error_code"] not in lcb_serve._LCB_INFRA_ERROR_CODES + + +def test_timeout_is_not_infra_error(): + # grade_call_based enforces its own per-test-case timeout via + # signal.alarm and returns a normal -3 result well within it, so + # exercising the *outer* process-level timeout (run_code_subprocess's + # p.join(timeout=...)) needs a submission that defeats the alarm too. + ignore_alarm_and_hang = ( + "import signal\n" + "signal.signal(signal.SIGALRM, signal.SIG_IGN)\n" + "def solve(x):\n" + " while True: pass\n" + ) + res, metadata = lcb_serve.run_code_subprocess( + CALL_BASED_SUITE, ignore_alarm_and_hang, timeout_sec=1 + ) + assert metadata["error_code"] == -1 + assert metadata["error_code"] not in lcb_serve._LCB_INFRA_ERROR_CODES + + +class _DictTestLoader(dict): + """Minimal stand-in for LCBTestLoader: _LCBWorker only does + self.test_loader[qid], which dict already supports.""" + + +def test_all_os_exit_batch_scores_zero_without_raising(): + """A batch where every submission kills its own interpreter is a + legitimate 0, not an infra failure -- the guard must not fire.""" + worker = lcb_serve._LCBWorker( + _DictTestLoader(q1=CALL_BASED_SUITE, q2=CALL_BASED_SUITE), + n_lcb_workers=2, + worker_timeout_sec=TIMEOUT_SEC, + ) + results = worker(["q1", "q2"], [[CALL_OS_EXIT], [CALL_OS_EXIT]]) + assert results == {"q1": [False], "q2": [False]} + + +def test_submission_error_logs_at_warning_not_error(caplog): + """Routine submission-attributed outcomes (-8 here) must not log at + ERROR -- that level is reserved for judge-side infra failures, so ops + can alert on it without being flooded by ordinary bad submissions.""" + worker = lcb_serve._LCBWorker( + _DictTestLoader(q1=CALL_BASED_SUITE), + n_lcb_workers=1, + worker_timeout_sec=TIMEOUT_SEC, + ) + with caplog.at_level("WARNING", logger=lcb_serve.logger.name): + results = worker(["q1"], [[CALL_OS_EXIT]]) + + assert results == {"q1": [False]} + levels = [r.levelname for r in caplog.records] + assert "WARNING" in levels + assert "ERROR" not in levels + + +def test_infra_error_logs_at_error_level(monkeypatch, caplog): + """A judge-side death (-6) must log at ERROR so it's distinguishable + from the routine submission-attributed WARNINGs above. + + Forces the pool onto "fork" so the monkeypatch below reaches the pool + workers -- see _force_fork_context. + """ + _force_fork_context(monkeypatch) + + def _die_immediately(*args, **kwargs): + os._exit(1) + + monkeypatch.setattr( + lcb_serve, "execute_code_single_suppressed_errors", _die_immediately + ) + + worker = lcb_serve._LCBWorker( + _DictTestLoader(q1=CALL_BASED_SUITE), + n_lcb_workers=1, + worker_timeout_sec=TIMEOUT_SEC, + ) + with caplog.at_level("WARNING", logger=lcb_serve.logger.name): + # Single-sample batch is also all-infra, so the guard raises; the + # log call happens before that, inside the executor's with-block. + with pytest.raises(RuntimeError, match="infrastructure errors"): + worker(["q1"], [[PASSING_CALL_BASED]]) + + levels = [r.levelname for r in caplog.records] + assert "ERROR" in levels + assert "WARNING" not in levels + + +def test_all_judge_startup_death_batch_raises(monkeypatch): + """A batch where every grading child dies before run_test is reached + means the judge itself is broken, not that every sample failed -- the + all-infra guard must raise instead of silently reporting a 0. + + Forces the pool onto "fork" so the monkeypatch below (applied in this + process) reaches the pool workers -- see _force_fork_context. A + spawn/forkserver worker would re-import the module fresh and not see + it. The guard logic under test lives in _LCBWorker.__call__ and is + identical regardless of pool context. + """ + _force_fork_context(monkeypatch) + + def _die_immediately(*args, **kwargs): + os._exit(1) + + monkeypatch.setattr( + lcb_serve, "execute_code_single_suppressed_errors", _die_immediately + ) + + worker = lcb_serve._LCBWorker( + _DictTestLoader(q1=CALL_BASED_SUITE, q2=CALL_BASED_SUITE), + n_lcb_workers=2, + worker_timeout_sec=TIMEOUT_SEC, + ) + with pytest.raises(RuntimeError, match="infrastructure errors"): + worker(["q1", "q2"], [[PASSING_CALL_BASED], [PASSING_CALL_BASED]])