Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import json | |
| import math | |
| import os | |
| import secrets | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import time | |
| from pathlib import Path | |
| from typing import Dict, Any, List, Optional | |
| from codestar.core.schemas import ( | |
| JudgeResult, | |
| JudgeStatus, | |
| CaseResult, | |
| ProblemMeta, | |
| ) | |
| # Python interpreter to use for running harness | |
| PYTHON_EXE = ( | |
| r"C:\Users\hj\.workbuddy\binaries\python\envs\default\Scripts\python.exe" | |
| if os.path.exists(r"C:\Users\hj\.workbuddy\binaries\python\envs\default\Scripts\python.exe") | |
| else sys.executable | |
| ) | |
| HARNESS_TEMPLATE = """# -*- coding: utf-8 -*- | |
| import sys | |
| import time | |
| __HARNESS_T0__ = time.time() | |
| import json | |
| import math | |
| import traceback | |
| import tracemalloc | |
| import _thread | |
| import threading | |
| from typing import Optional, List, Tuple, Dict, Any | |
| # Setup environment & seeds | |
| try: | |
| import torch | |
| torch.manual_seed(42) | |
| if torch.cuda.is_available(): | |
| torch.cuda.manual_seed_all(42) | |
| except ImportError: | |
| pass | |
| try: | |
| import numpy as np | |
| np.random.seed(42) | |
| except ImportError: | |
| pass | |
| # Import comparator from codestar | |
| from codestar.core.comparator import compare_outputs, summarize_obj | |
| # Argument cloning to prevent in-place mutation side-effects between ref and user code | |
| def _clone_arg(arg): | |
| if "torch" in sys.modules and isinstance(arg, sys.modules["torch"].Tensor): | |
| return arg.clone() | |
| elif "numpy" in sys.modules and isinstance(arg, sys.modules["numpy"].ndarray): | |
| return arg.copy() | |
| elif isinstance(arg, list): | |
| return [_clone_arg(x) for x in arg] | |
| elif isinstance(arg, tuple): | |
| return tuple(_clone_arg(x) for x in arg) | |
| elif isinstance(arg, dict): | |
| return {k: _clone_arg(v) for k, v in arg.items()} | |
| return arg | |
| def _clone_args(args, kwargs): | |
| c_args = tuple(_clone_arg(a) for a in args) if args else () | |
| c_kwargs = {k: _clone_arg(v) for k, v in kwargs.items()} if kwargs else {} | |
| return c_args, c_kwargs | |
| def _safe_wrap(obj): | |
| if isinstance(obj, type): | |
| class WrappedClass(obj): | |
| __wrapped__ = obj | |
| def __init__(self, *args, **kwargs): | |
| c_args, c_kwargs = _clone_args(args, kwargs) | |
| super().__init__(*c_args, **c_kwargs) | |
| def __call__(self, *args, **kwargs): | |
| c_args, c_kwargs = _clone_args(args, kwargs) | |
| return super().__call__(*c_args, **c_kwargs) | |
| def forward(self, *args, **kwargs): | |
| c_args, c_kwargs = _clone_args(args, kwargs) | |
| return super().forward(*c_args, **c_kwargs) | |
| WrappedClass.__name__ = getattr(obj, "__name__", "WrappedClass") | |
| WrappedClass.__qualname__ = getattr(obj, "__qualname__", "WrappedClass") | |
| return WrappedClass | |
| elif callable(obj): | |
| def wrapped(*args, **kwargs): | |
| c_args, c_kwargs = _clone_args(args, kwargs) | |
| return obj(*c_args, **c_kwargs) | |
| wrapped.__name__ = getattr(obj, "__name__", "wrapped") | |
| wrapped.__wrapped__ = obj | |
| return wrapped | |
| return obj | |
| # Load Reference Implementation | |
| ref_ns = {} | |
| try: | |
| exec(__REF_CODE_STR__, ref_ns) | |
| except MemoryError: | |
| msg = "Memory Limit Exceeded: Out of Memory during tensor allocation" | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "MLE", "message": msg, "traceback": msg})) | |
| sys.exit(0) | |
| except Exception as e: | |
| err = traceback.format_exc() | |
| lines = [line.strip() for line in err.splitlines() if line.strip()] | |
| concise = lines[-1] if lines else str(e) | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "RE", "message": concise, "traceback": err})) | |
| sys.exit(0) | |
| # Load User Implementation | |
| user_ns = {} | |
| try: | |
| exec(__USER_CODE_STR__, user_ns) | |
| except MemoryError: | |
| msg = "Memory Limit Exceeded: Out of Memory during tensor allocation" | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "MLE", "message": msg, "traceback": msg})) | |
| sys.exit(0) | |
| except Exception as e: | |
| err = traceback.format_exc() | |
| lines = [line.strip() for line in err.splitlines() if line.strip()] | |
| concise = lines[-1] if lines else str(e) | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "RE", "message": concise, "traceback": err})) | |
| sys.exit(0) | |
| entry_point = "__ENTRY_POINT__" | |
| if entry_point not in user_ns: | |
| msg = f"Entry point '{entry_point}' not found in user submission." | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "RE", "message": msg, "traceback": msg})) | |
| sys.exit(0) | |
| raw_user_obj = user_ns[entry_point] | |
| raw_ref_obj = ref_ns[entry_point] | |
| user_obj = _safe_wrap(raw_user_obj) | |
| ref_obj = _safe_wrap(raw_ref_obj) | |
| atol = __ATOL__ | |
| rtol = __RTOL__ | |
| # Start memory tracking | |
| tracemalloc.start() | |
| # Run problem-specific test runner or custom testcase | |
| cases_results = [] | |
| all_passed = True | |
| custom_input = __CUSTOM_TESTCASE_STR__ | |
| if custom_input.strip(): | |
| t0 = time.time() | |
| safe_builtins = { | |
| "abs": abs, "all": all, "any": any, "bool": bool, "dict": dict, | |
| "enumerate": enumerate, "filter": filter, "float": float, "int": int, | |
| "isinstance": isinstance, "len": len, "list": list, "map": map, | |
| "max": max, "min": min, "range": range, "reversed": reversed, | |
| "round": round, "set": set, "slice": slice, "sorted": sorted, | |
| "str": str, "sum": sum, "tuple": tuple, "zip": zip, | |
| "True": True, "False": False, "None": None, | |
| "__import__": __import__, | |
| } | |
| custom_ns = { | |
| "__builtins__": safe_builtins, | |
| "torch": sys.modules.get("torch"), | |
| "math": sys.modules.get("math"), | |
| "time": time, | |
| "np": sys.modules.get("numpy"), | |
| "List": list, | |
| "Tuple": tuple, | |
| "Dict": dict, | |
| "Optional": Optional, | |
| "user_obj": user_obj, | |
| "ref_obj": ref_obj, | |
| } | |
| if "torch" in sys.modules and hasattr(sys.modules["torch"], "nn"): | |
| custom_ns["nn"] = sys.modules["torch"].nn | |
| custom_ns["F"] = getattr(sys.modules["torch"].nn, "functional", None) | |
| input_text = custom_input.strip() | |
| args = None | |
| kwargs = {} | |
| try: | |
| # First try: direct expression evaluation (e.g. "0.1" or "torch.randn(2, 4)") | |
| try: | |
| val = eval(input_text, custom_ns) | |
| if isinstance(val, tuple): | |
| args = val | |
| else: | |
| args = (val,) | |
| except Exception: | |
| # Second try: statement block with 'args = ...' or local vars | |
| exec(input_text, custom_ns) | |
| if "args" in custom_ns: | |
| args = custom_ns["args"] if isinstance(custom_ns["args"], tuple) else (custom_ns["args"],) | |
| elif "kwargs" in custom_ns: | |
| kwargs = custom_ns["kwargs"] | |
| args = custom_ns.get("args", ()) | |
| else: | |
| excluded = {"torch", "math", "time", "np", "List", "Tuple", "Dict", "Optional", "user_obj", "ref_obj", "nn", "F", "__builtins__"} | |
| user_vars = [v for k, v in custom_ns.items() if not k.startswith("_") and k not in excluded] | |
| if user_vars: | |
| args = tuple(user_vars) | |
| else: | |
| raise ValueError(f"无法解析自定义用例输入: {input_text}") | |
| # Helper to coerce raw Python lists into torch.Tensor / np.ndarray when annotation expects it | |
| import inspect | |
| def _is_all_bools(seq): | |
| if isinstance(seq, bool): | |
| return True | |
| if isinstance(seq, (list, tuple)) and len(seq) > 0: | |
| return all(_is_all_bools(x) for x in seq) | |
| return False | |
| def _is_all_ints(seq): | |
| if isinstance(seq, bool): | |
| return False | |
| if isinstance(seq, int): | |
| return True | |
| if isinstance(seq, (list, tuple)) and len(seq) > 0: | |
| return all(_is_all_ints(x) for x in seq) | |
| return False | |
| def _coerce_val(val, param): | |
| if param is None or val is None: | |
| return val | |
| ann_str = str(param.annotation) if param.annotation != inspect.Parameter.empty else "" | |
| p_name = param.name.lower() | |
| if "Tensor" in ann_str and isinstance(val, (list, tuple)) and "torch" in sys.modules: | |
| _torch = sys.modules["torch"] | |
| if _is_all_bools(val): | |
| return _torch.tensor(val, dtype=_torch.bool) | |
| if _is_all_ints(val) and any(k in p_name for k in ("idx", "index", "indices", "token", "target", "label", "id", "pos", "seq")): | |
| return _torch.tensor(val, dtype=_torch.long) | |
| return _torch.tensor(val, dtype=_torch.float32) | |
| if "ndarray" in ann_str and isinstance(val, (list, tuple)) and "numpy" in sys.modules: | |
| _np = sys.modules["numpy"] | |
| return _np.asarray(val, dtype=_np.float64 if not _is_all_ints(val) else _np.int64) | |
| return val | |
| def _coerce_args_for_callable(fn_obj, raw_args, raw_kwargs): | |
| try: | |
| sig = inspect.signature(fn_obj) | |
| sig_params = [p for p in sig.parameters.values() if p.name not in ("self", "args", "kwargs")] | |
| new_args = [] | |
| for idx_a, a_val in enumerate(raw_args): | |
| p_obj = sig_params[idx_a] if idx_a < len(sig_params) else None | |
| new_args.append(_coerce_val(a_val, p_obj)) | |
| param_map = {p.name: p for p in sig_params} | |
| new_kwargs = {k: _coerce_val(v, param_map.get(k)) for k, v in raw_kwargs.items()} | |
| return tuple(new_args), new_kwargs | |
| except Exception: | |
| return raw_args, raw_kwargs | |
| # Check if entry point is a class (e.g. nn.Module) or a function | |
| is_class = isinstance(ref_obj, type) | |
| if is_class: | |
| target_cls = getattr(ref_obj, "__wrapped__", ref_obj) | |
| if not hasattr(target_cls, "forward") and hasattr(target_cls, "split_heads"): | |
| args, kwargs = _coerce_args_for_callable(target_cls.split_heads, args, kwargs) | |
| exp_s = ref_obj.split_heads(*args, **kwargs) | |
| act_s = user_obj.split_heads(*args, **kwargs) | |
| exp = ref_obj.merge_heads(exp_s) | |
| act = user_obj.merge_heads(act_s) | |
| else: | |
| init_sig = inspect.signature(target_cls.__init__) | |
| init_params = [ | |
| p for p in init_sig.parameters.values() | |
| if p.name not in ("self", "args", "kwargs") | |
| ] | |
| req_init_params = [p for p in init_params if p.default == inspect.Parameter.empty] | |
| req_init_count = len(req_init_params) | |
| # If args has only 1 element (or matches forward signature) and is a list when forward expects Tensor, coerce it first | |
| if hasattr(target_cls, "forward") and len(args) > 0: | |
| fwd_sig = inspect.signature(target_cls.forward) | |
| fwd_params = [p for p in fwd_sig.parameters.values() if p.name not in ("self", "args", "kwargs")] | |
| if len(args) <= len(fwd_params) and isinstance(args[0], (list, tuple)): | |
| args, kwargs = _coerce_args_for_callable(target_cls.forward, args, kwargs) | |
| has_tensor_arg = len(args) > 0 and ( | |
| ("torch" in sys.modules and isinstance(args[0], sys.modules["torch"].Tensor)) or | |
| ("numpy" in sys.modules and isinstance(args[0], sys.modules["numpy"].ndarray)) | |
| ) | |
| if has_tensor_arg: | |
| x0 = args[0] | |
| last_dim = x0.shape[-1] if hasattr(x0, "shape") and len(x0.shape) > 0 else 16 | |
| inferred_init = [] | |
| if req_init_count > 0: | |
| for p in req_init_params: | |
| p_name = p.name.lower() | |
| if any(k in p_name for k in ("dim", "size", "features", "shape", "channel")): | |
| inferred_init.append(last_dim) | |
| elif any(k in p_name for k in ("rank", "r")): | |
| inferred_init.append(min(4, last_dim)) | |
| elif "kv" in p_name and "head" in p_name: | |
| inferred_init.append(1 if last_dim % 2 != 0 else 2) | |
| elif any(k in p_name for k in ("head",)): | |
| inferred_init.append(4 if last_dim % 4 == 0 else (2 if last_dim % 2 == 0 else 1)) | |
| else: | |
| inferred_init.append(last_dim) | |
| elif len(init_params) > 0 and any(k in init_params[0].name.lower() for k in ("dim", "size", "features")): | |
| inferred_init.append(last_dim) | |
| init_slice = tuple(inferred_init) | |
| fwd_slice = args | |
| elif len(args) > req_init_count and len(init_params) > 0 and not has_tensor_arg: | |
| split_at = min(len(init_params), len(args) - 1) | |
| init_slice = args[:split_at] | |
| fwd_slice = args[split_at:] | |
| if hasattr(target_cls, "forward"): | |
| fwd_slice, _ = _coerce_args_for_callable(target_cls.forward, fwd_slice, {}) | |
| else: | |
| init_slice = args | |
| fwd_slice = () | |
| try: | |
| ref_inst = ref_obj(*init_slice) | |
| user_inst = user_obj(*init_slice) | |
| if hasattr(ref_inst, "state_dict") and hasattr(user_inst, "load_state_dict"): | |
| try: | |
| user_inst.load_state_dict(ref_inst.state_dict(), strict=False) | |
| except Exception: | |
| pass | |
| elif hasattr(ref_inst, "weight") and hasattr(user_inst, "weight"): | |
| try: | |
| user_inst.weight.data.copy_(ref_inst.weight.data) | |
| except Exception: | |
| pass | |
| if fwd_slice: | |
| if "torch" in sys.modules: | |
| sys.modules["torch"].manual_seed(42) | |
| exp = ref_inst(*fwd_slice) | |
| if "torch" in sys.modules: | |
| sys.modules["torch"].manual_seed(42) | |
| act = user_inst(*fwd_slice) | |
| else: | |
| exp = ref_inst | |
| act = user_inst | |
| except Exception: | |
| exp = ref_obj(*args, **kwargs) | |
| act = user_obj(*args, **kwargs) | |
| else: | |
| target_fn = getattr(ref_obj, "__wrapped__", ref_obj) | |
| args, kwargs = _coerce_args_for_callable(target_fn, args, kwargs) | |
| exp = ref_obj(*args, **kwargs) | |
| act = user_obj(*args, **kwargs) | |
| match, err, max_diff = compare_outputs(act, exp, atol=atol, rtol=rtol) | |
| passed, msg = match, err | |
| except MemoryError: | |
| passed, msg = False, "Memory Limit Exceeded: Out of Memory during tensor allocation" | |
| act, exp, max_diff = None, None, None | |
| except Exception as e: | |
| passed, msg = False, f"自定义用例运行异常: {str(e)}" | |
| act, exp, max_diff = None, None, None | |
| cases_results.append({ | |
| "case_id": 1, | |
| "case_name": "自定义用例自测 (Custom Input)", | |
| "passed": passed, | |
| "time_ms": round((time.time() - t0) * 1000, 2), | |
| "message": msg, | |
| "user_output_summary": summarize_obj(act) if act is not None else None, | |
| "expected_summary": summarize_obj(exp) if exp is not None else None, | |
| "max_diff": max_diff, | |
| "input_summary": input_text[:300], | |
| }) | |
| else: | |
| # Execute user-provided or standard test suite | |
| test_ns = { | |
| "user_obj": user_obj, | |
| "ref_obj": ref_obj, | |
| "compare_outputs": compare_outputs, | |
| "summarize_obj": summarize_obj, | |
| "atol": atol, | |
| "rtol": rtol, | |
| "cases_results": cases_results, | |
| "time": time, | |
| } | |
| __limit = float(__TIME_LIMIT__) | |
| __elapsed_init = time.time() - __HARNESS_T0__ | |
| __budget = max(0.6, min(__limit, 5.2 - __elapsed_init)) if __limit <= 3.5 else __limit | |
| _watchdog = threading.Timer(__budget, _thread.interrupt_main) | |
| _watchdog.daemon = True | |
| _watchdog.start() | |
| try: | |
| exec(__TEST_CODE_STR__, test_ns) | |
| except KeyboardInterrupt: | |
| msg = f"Time Limit Exceeded (> {__TIME_LIMIT__}s)" | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "TLE", "message": msg, "traceback": msg})) | |
| sys.exit(0) | |
| except MemoryError: | |
| msg = "Memory Limit Exceeded: Out of Memory during tensor allocation" | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "MLE", "message": msg, "traceback": msg})) | |
| sys.exit(0) | |
| except Exception as e: | |
| err = traceback.format_exc() | |
| lines = [line.strip() for line in err.splitlines() if line.strip()] | |
| concise = lines[-1] if lines else str(e) | |
| print("__ERROR_DELIM__") | |
| print(json.dumps({"error": "RE", "message": concise, "traceback": err})) | |
| sys.exit(0) | |
| finally: | |
| _watchdog.cancel() | |
| # Stop memory tracking | |
| _, peak_bytes = tracemalloc.get_traced_memory() | |
| tracemalloc.stop() | |
| peak_memory_mb = round(peak_bytes / (1024 * 1024), 2) | |
| # RFC 8259 Sanitization | |
| def _sanitize_for_json(val): | |
| if isinstance(val, float): | |
| if math.isnan(val) or math.isinf(val): | |
| return 0.0 | |
| return val | |
| elif isinstance(val, dict): | |
| return {k: _sanitize_for_json(v) for k, v in val.items()} | |
| elif isinstance(val, list): | |
| return [_sanitize_for_json(v) for v in val] | |
| elif isinstance(val, tuple): | |
| return [_sanitize_for_json(v) for v in val] | |
| return val | |
| clean_cases = _sanitize_for_json(cases_results) | |
| clean_peak = _sanitize_for_json(peak_memory_mb) | |
| # Output final result | |
| print("__START_DELIM__") | |
| print(json.dumps({"cases": clean_cases, "peak_memory_mb": clean_peak})) | |
| print("__END_DELIM__") | |
| """ | |
| def execute_submission( | |
| problem: ProblemMeta, | |
| user_code: str, | |
| custom_testcase: Optional[str] = None, | |
| python_exe: str = PYTHON_EXE, | |
| ) -> JudgeResult: | |
| """ | |
| Executes a submission against a problem using an isolated subprocess harness. | |
| """ | |
| # Check custom_testcase AST if provided | |
| if custom_testcase and custom_testcase.strip(): | |
| from codestar.core.ast_checker import check_code_ast | |
| c_safe, c_msg, c_status = check_code_ast( | |
| code=custom_testcase, | |
| extra_forbidden_modules=problem.forbidden_modules, | |
| extra_forbidden_calls=problem.forbidden_calls, | |
| ) | |
| if not c_safe: | |
| return JudgeResult( | |
| status=c_status or JudgeStatus.CE, | |
| message=f"Custom testcase error: {c_msg}", | |
| passed_count=0, | |
| total_count=0, | |
| execution_time_ms=0.0, | |
| ) | |
| delim_token = secrets.token_hex(8) | |
| start_delim = f"---JUDGE_OUTPUT_START_{delim_token}---" | |
| end_delim = f"---JUDGE_OUTPUT_END_{delim_token}---" | |
| error_delim = f"---JUDGE_ERROR_{delim_token}---" | |
| harness_content = ( | |
| HARNESS_TEMPLATE | |
| .replace("__REF_CODE_STR__", repr(problem.reference_code)) | |
| .replace("__USER_CODE_STR__", repr(user_code)) | |
| .replace("__ENTRY_POINT__", problem.entry_point) | |
| .replace("__TEST_CODE_STR__", repr(problem.test_code)) | |
| .replace("__CUSTOM_TESTCASE_STR__", repr(custom_testcase or "")) | |
| .replace("__ATOL__", str(problem.atol)) | |
| .replace("__RTOL__", str(problem.rtol)) | |
| .replace("__TIME_LIMIT__", str(problem.time_limit)) | |
| .replace("__START_DELIM__", start_delim) | |
| .replace("__END_DELIM__", end_delim) | |
| .replace("__ERROR_DELIM__", error_delim) | |
| ) | |
| # Write harness to temp file | |
| with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as f: | |
| harness_path = f.name | |
| f.write(harness_content) | |
| t0 = time.time() | |
| try: | |
| # Clean subprocess environment: filter out sensitive environment variables | |
| clean_env = {} | |
| keep_keys = { | |
| "PATH", "PATHEXT", "SYSTEMROOT", "SYSTEMDRIVE", "TEMP", "TMP", | |
| "USERPROFILE", "HOMEDRIVE", "HOMEPATH", "WINDIR", "COMSPEC", | |
| "APPDATA", "LOCALAPPDATA", "PROGRAMDATA", "PROGRAMFILES", | |
| "PROGRAMFILES(X86)", "COMMONPROGRAMFILES", "COMMONPROGRAMFILES(X86)", | |
| "VIRTUAL_ENV", "PYTHONHOME", "CONDA_PREFIX", | |
| } | |
| for k, v in os.environ.items(): | |
| if k.upper() in keep_keys: | |
| clean_env[k] = v | |
| if "VIRTUAL_ENV" not in clean_env and sys.prefix: | |
| clean_env["VIRTUAL_ENV"] = sys.prefix | |
| repo_root = str(Path(__file__).resolve().parents[2]) | |
| clean_env["PYTHONPATH"] = repo_root + (os.pathsep + clean_env["PYTHONPATH"] if "PYTHONPATH" in clean_env else "") | |
| clean_env["PYTHONIOENCODING"] = "utf-8" | |
| clean_env["PYTHONUTF8"] = "1" | |
| # Subprocess timeout: outer safety ceiling allowing sufficient bootstrap overhead on Windows | |
| subproc_timeout = max(problem.time_limit * 2.5 + 8.0, 16.0) | |
| proc = subprocess.Popen( | |
| [python_exe, harness_path], | |
| stdin=subprocess.DEVNULL, | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, | |
| text=True, | |
| cwd=repo_root, | |
| env=clean_env, | |
| encoding="utf-8", | |
| errors="replace", | |
| ) | |
| try: | |
| stdout, stderr = proc.communicate(timeout=subproc_timeout) | |
| except subprocess.TimeoutExpired: | |
| try: | |
| proc.kill() | |
| except Exception: | |
| pass | |
| try: | |
| proc.communicate(timeout=1.0) | |
| except Exception: | |
| pass | |
| return JudgeResult( | |
| status=JudgeStatus.TLE, | |
| message=f"Time Limit Exceeded (> {problem.time_limit}s)", | |
| execution_time_ms=round(problem.time_limit * 1000, 2), | |
| ) | |
| exec_time_ms = round((time.time() - t0) * 1000, 2) | |
| # Check for harness judge output | |
| if start_delim in stdout: | |
| # Extract raw user stdout prior to start_delim | |
| raw_user_stdout = stdout.split(start_delim)[0] | |
| if len(raw_user_stdout) > 8000: | |
| user_stdout = raw_user_stdout[:8000] + "\n[输出已截断: 超过 8000 字符限制]" | |
| else: | |
| user_stdout = raw_user_stdout | |
| user_stdout_val = user_stdout.strip() if user_stdout.strip() else None | |
| parts = stdout.split(start_delim)[1].split(end_delim)[0].strip() | |
| data = json.loads(parts) | |
| raw_cases = data.get("cases", []) | |
| peak_memory_mb = float(data.get("peak_memory_mb", 0.0)) | |
| if math.isnan(peak_memory_mb) or math.isinf(peak_memory_mb): | |
| peak_memory_mb = 0.0 | |
| cases = [] | |
| for c in raw_cases: | |
| if not c.get("input_summary"): | |
| c["input_summary"] = c.get("case_name") or f"Case #{c.get('case_id', 1)}" | |
| if c.get("max_diff") is not None: | |
| if math.isnan(c["max_diff"]) or math.isinf(c["max_diff"]): | |
| c["max_diff"] = 0.0 | |
| cases.append(CaseResult(**c)) | |
| passed_count = sum(1 for c in cases if c.passed) | |
| total_count = len(cases) | |
| # Check if actual user code evaluation time exceeded problem limit | |
| actual_cases_time_s = sum(c.time_ms for c in cases) / 1000.0 | |
| if actual_cases_time_s > problem.time_limit: | |
| return JudgeResult( | |
| status=JudgeStatus.TLE, | |
| message=f"Time Limit Exceeded: computation took {actual_cases_time_s:.2f}s > {problem.time_limit}s", | |
| passed_count=passed_count, | |
| total_count=total_count, | |
| execution_time_ms=exec_time_ms, | |
| peak_memory_mb=peak_memory_mb, | |
| cases=cases, | |
| stdout=user_stdout_val, | |
| ) | |
| if total_count > 0 and passed_count == total_count: | |
| status = JudgeStatus.AC | |
| message = f"All {total_count} test cases passed." | |
| else: | |
| first_failed = next((c for c in cases if not c.passed), None) | |
| fail_msg = first_failed.message if first_failed else "Assertion mismatch" | |
| if fail_msg and ("Memory Limit Exceeded" in fail_msg or "MemoryError" in fail_msg): | |
| status = JudgeStatus.MLE | |
| message = "Memory Limit Exceeded: Out of Memory during tensor allocation" | |
| elif fail_msg and ("Runtime Exception:" in fail_msg or "RuntimeError:" in fail_msg): | |
| status = JudgeStatus.RE | |
| clean_msg = fail_msg.replace("Runtime Exception:", "").strip() | |
| message = f"Runtime Error on case #{first_failed.case_id if first_failed else '?'}: {clean_msg}" | |
| else: | |
| status = JudgeStatus.WA | |
| message = f"Failed on case #{first_failed.case_id if first_failed else '?'}: {fail_msg}" | |
| return JudgeResult( | |
| status=status, | |
| message=message, | |
| passed_count=passed_count, | |
| total_count=total_count, | |
| execution_time_ms=exec_time_ms, | |
| peak_memory_mb=peak_memory_mb, | |
| cases=cases, | |
| stdout=user_stdout_val, | |
| ) | |
| # Check for harness runtime error delimiter | |
| if error_delim in stdout: | |
| raw_user_stdout = stdout.split(error_delim)[0] | |
| if len(raw_user_stdout) > 8000: | |
| user_stdout = raw_user_stdout[:8000] + "\n[输出已截断: 超过 8000 字符限制]" | |
| else: | |
| user_stdout = raw_user_stdout | |
| user_stdout_val = user_stdout.strip() if user_stdout.strip() else None | |
| payload_raw = stdout.split(error_delim)[1].strip() | |
| payload = {} | |
| for line in payload_raw.splitlines(): | |
| line = line.strip() | |
| if line.startswith("{") and line.endswith("}"): | |
| try: | |
| payload = json.loads(line) | |
| break | |
| except Exception: | |
| pass | |
| err_type = payload.get("error") | |
| msg = payload.get("message") or "Runtime Error" | |
| tb = payload.get("traceback") or msg | |
| if err_type == "TLE" or "Time Limit Exceeded" in msg: | |
| return JudgeResult( | |
| status=JudgeStatus.TLE, | |
| message=msg or f"Time Limit Exceeded (> {problem.time_limit}s)", | |
| error_detail=tb, | |
| stdout=user_stdout_val, | |
| execution_time_ms=exec_time_ms, | |
| ) | |
| if err_type == "MLE" or "Memory Limit Exceeded" in msg or "MemoryError" in tb: | |
| return JudgeResult( | |
| status=JudgeStatus.MLE, | |
| message=msg or "Memory Limit Exceeded: Out of Memory during tensor allocation", | |
| error_detail=tb, | |
| stdout=user_stdout_val, | |
| execution_time_ms=exec_time_ms, | |
| ) | |
| return JudgeResult( | |
| status=JudgeStatus.RE, | |
| message=msg, | |
| error_detail=tb, | |
| stdout=user_stdout_val, | |
| execution_time_ms=exec_time_ms, | |
| ) | |
| # Check if legacy error JSON was emitted | |
| for line in stdout.splitlines(): | |
| line = line.strip() | |
| if line.startswith("{") and line.endswith("}"): | |
| try: | |
| payload = json.loads(line) | |
| if payload.get("error") in ("RE", "MLE"): | |
| st = JudgeStatus.MLE if payload.get("error") == "MLE" else JudgeStatus.RE | |
| return JudgeResult( | |
| status=st, | |
| message=payload.get("message", "Runtime Error"), | |
| error_detail=payload.get("traceback"), | |
| execution_time_ms=exec_time_ms, | |
| ) | |
| except Exception: | |
| pass | |
| # Subprocess failed with return code or unhandled exception | |
| err_msg = stderr.strip() or stdout.strip() or f"Process exited with return code {proc.returncode}" | |
| lines = [l.strip() for l in err_msg.splitlines() if l.strip()] | |
| concise = lines[-1] if lines else err_msg | |
| user_stdout_val = stdout.strip()[:8000] if stdout.strip() else None | |
| # Check for Out of Memory | |
| status = JudgeStatus.RE | |
| if any(oom_sig in err_msg for oom_sig in ("MemoryError", "std::bad_alloc", "OutOfMemoryError")) or proc.returncode in (-9, 137): | |
| status = JudgeStatus.MLE | |
| concise = "Memory Limit Exceeded: Process terminated due to out-of-memory" | |
| return JudgeResult( | |
| status=status, | |
| message=concise[:500], | |
| error_detail=err_msg[:8000], | |
| stdout=user_stdout_val, | |
| execution_time_ms=exec_time_ms, | |
| ) | |
| except Exception as e: | |
| import traceback | |
| return JudgeResult( | |
| status=JudgeStatus.RE, | |
| message=f"Runner execution exception: {str(e)}", | |
| error_detail=traceback.format_exc(), | |
| execution_time_ms=round((time.time() - t0) * 1000, 2), | |
| ) | |
| finally: | |
| if os.path.exists(harness_path): | |
| try: | |
| os.remove(harness_path) | |
| except OSError: | |
| pass | |
| # Alias ExecutionResult to JudgeResult for backward compatibility and test verification | |
| ExecutionResult = JudgeResult | |