Blame for codestar/core/runner.py

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