diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0534de7..07543dd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,10 @@
## [Unreleased]
+- 统一 Schema v1 特殊 Judge 路径栅栏：lint、正式打包和 local Judge 共同拒绝绝对路径、`..`、符号链接、junction/reparse point、非普通文件和题目目录外目标，并保留 `judge.type: checker` 兼容别名。
+- 新增共享 Checker/Interactor Core 运行层，供 local Judge、stress 与数据生成复用；结果分离 verdict、execution status、failure kind、责任方、终止原因、双方资源/流量证据与清理结果，沙箱缓存 Schema 升至 7。
+- Checker feedback 同时保留有界诊断与简短正式失败原因；Interactor 非普通 feedback、资源异常、启动失败、取消和清理失败均返回一致的结构化语义，清理失败不会再被成功 verdict 或取消状态掩盖。
+
## [0.6.5] - 2026-08-04
- Schema v1 题面—Validator 约束对账新增多组数据累计约束检查：保守识别 LaTeX 求和、中文“所有测试用例之和”表述，以及 Validator 中直接 `+=` / `acc = acc + term` 累加和后续 `ensuref` 上限；统一规范化为 `sum:n`、`sum:len:s`、`sum:n+m` 等主体。
diff --git a/probhub/calibration.py b/probhub/calibration.py
index 8fa2d52..3164681 100644
--- a/probhub/calibration.py
+++ b/probhub/calibration.py
@@ -14,7 +14,7 @@
CALIBRATION_SCHEMA_VERSION = 2
CALIBRATION_STRATEGY_VERSION = 2
-SANDBOX_CACHE_SCHEMA_VERSION = 6
+SANDBOX_CACHE_SCHEMA_VERSION = 7
EVIDENCE_FILENAME = "judge-evidence-v2.json"
EVIDENCE_LOCK_FILENAME = "judge-evidence.lock"
DEFAULT_ACCEPTED_TIME_MULTIPLIER = 3.0
diff --git a/probhub/datagen.py b/probhub/datagen.py
index 64e6a87..696d5db 100644
--- a/probhub/datagen.py
+++ b/probhub/datagen.py
@@ -24,7 +24,8 @@
from .errors import ProbHubError
from .io import normalize_newlines as _normalize_newlines, read_yaml, write_json
from .process_control import DEFAULT_PROCESS_LIMIT
-from .stressing import _compare_custom, _prepare_program, _run
+from .special_judges import run_checker_to_files
+from .stressing import _prepare_program, _run
from .transactions import (
TRANSACTION_PHASE_COMMITTED,
TRANSACTION_PHASE_PREPARED,
@@ -694,26 +695,37 @@ def generator_command(relative):
checker_answer = checker_dir / "answer.ans"
checker_input.write_bytes(input_bytes)
checker_answer.write_bytes(answer_bytes)
- checked = _compare_custom(
+ checker_env = os.environ.copy()
+ checker_env["PYTHONIOENCODING"] = "utf-8"
+ checked = run_checker_to_files(
checker_cmd,
checker_input,
checker_answer,
- answer_bytes,
- tool_timeout,
- checker_dir,
- TOOL_MEMORY_LIMIT_MB,
- output_limit,
- process_limit,
+ checker_answer,
+ timeout=tool_timeout,
+ cwd=checker_dir,
+ memory_limit_mb=TOOL_MEMORY_LIMIT_MB,
+ output_limit_bytes=int(output_limit * 1024 * 1024),
+ process_limit=process_limit,
+ env=checker_env,
)
- if checked["status"] != "AC":
+ if checked.get("verdict") != "AC":
failure = {
"case": case,
"stage": "checker",
- "status": checked["status"],
+ "status": checked.get("verdict") or "FAIL",
"message": checked.get("message") or "checker rejected the generated answer",
}
- if checked.get("execution_status"):
- failure["execution_status"] = checked["execution_status"]
+ legacy_execution_status = {
+ "time_limit": "TLE",
+ "memory_limit": "MLE",
+ "output_limit": "OLE",
+ "process_limit": "RE",
+ "start_error": "RE",
+ "output_control_error": "FAIL",
+ }.get(checked.get("execution_status"))
+ if legacy_execution_status:
+ failure["execution_status"] = legacy_execution_status
failures.append(failure)
continue
diff --git a/probhub/linting.py b/probhub/linting.py
index e85e5d0..28136a4 100644
--- a/probhub/linting.py
+++ b/probhub/linting.py
@@ -2,7 +2,7 @@
import json
import math
import re
-from pathlib import Path, PureWindowsPath
+from pathlib import Path
from .builder_fingerprint import (
BUILD_MANIFEST_SCHEMA_VERSION,
@@ -16,6 +16,7 @@
from .errors import ProbHubError
from .hashing import files_under, hash_file, hash_paths
from .metadata import build_meta, normalize_display_name
+from .problem_paths import ProblemPathError, resolve_problem_regular_file
from .solutions import analyze_solution_verification
from .statement import parse_statement
from .statement_consistency import analyze_constraint_consistency, reconcile_constraints
@@ -70,54 +71,17 @@ def _problem_relative_path(problem_dir, value):
return candidate
-def _is_link_like(path):
- try:
- return path.is_symlink() or (
- hasattr(path, "is_junction") and path.is_junction()
- )
- except OSError:
- return False
-
-
def _problem_regular_file_path(problem_dir, value):
- """Resolve a configured problem-local file without accepting links.
-
- Returns ``(path, reason)`` where reason is one of ``invalid``, ``outside``,
- ``symlink``, or ``missing``. The explicit reason lets lint keep stable,
- field-specific diagnostics without ever opening an escaped path.
- """
-
- if not isinstance(value, str) or not value.strip():
- return None, "invalid"
- problem_dir = Path(problem_dir).resolve()
- value = value.strip()
- relative = Path(value)
- if relative.is_absolute() or PureWindowsPath(value).is_absolute():
- return None, "outside"
- if ".." in relative.parts or ".." in PureWindowsPath(value).parts:
- return None, "outside"
- candidate = problem_dir / relative
- current = problem_dir
try:
- for part in relative.parts:
- if part in {"", "."}:
- continue
- current = current / part
- if _is_link_like(current):
- return None, "symlink"
- resolved = candidate.resolve()
- resolved.relative_to(problem_dir)
- except (OSError, RuntimeError, ValueError):
- return None, "outside"
- if not resolved.is_file():
- return None, "missing"
- return resolved, None
+ return resolve_problem_regular_file(problem_dir, value), None
+ except ProblemPathError as exc:
+ return None, exc.reason
def _configured_file_error(field, value, reason, missing_label):
if reason == "outside":
return f"{field} must stay inside the problem directory: {value}"
- if reason == "symlink":
+ if reason == "link":
return f"{field} must be a regular non-symlink file: {value}"
if reason == "invalid":
return f"{field} must be a non-empty relative path"
@@ -390,15 +354,29 @@ def lint_problem(root, workspace, entry):
if judge_type == "custom":
if not checker:
errors.append("judge.checker is required for custom judging")
- elif not (problem_dir / checker).is_file():
- errors.append(f"checker not found: {checker}")
+ else:
+ _, checker_reason = _problem_regular_file_path(problem_dir, checker)
+ if checker_reason:
+ errors.append(_configured_file_error(
+ "judge.checker",
+ checker,
+ checker_reason,
+ "checker",
+ ))
if interactor:
errors.append("judge.interactor is only valid for interactive judging")
elif judge_type == "interactive":
if not interactor:
errors.append("judge.interactor is required for interactive judging")
- elif not (problem_dir / interactor).is_file():
- errors.append(f"interactor not found: {interactor}")
+ else:
+ _, interactor_reason = _problem_regular_file_path(problem_dir, interactor)
+ if interactor_reason:
+ errors.append(_configured_file_error(
+ "judge.interactor",
+ interactor,
+ interactor_reason,
+ "interactor",
+ ))
if checker:
errors.append("judge.checker is not used for interactive judging")
interactive = judge.get("interactive") or {}
diff --git a/probhub/package_tools.py b/probhub/package_tools.py
index 3403a66..ab2478a 100644
--- a/probhub/package_tools.py
+++ b/probhub/package_tools.py
@@ -17,6 +17,7 @@
from .errors import ProbHubError
from .io import write_yaml
+from .problem_paths import ProblemPathError, resolve_problem_regular_file
from .process_control import DEFAULT_PROCESS_LIMIT, ProcessCancelled, run_managed_to_files
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
@@ -280,37 +281,16 @@ def _expected_validation(config):
def _problem_regular_file(problem_dir, relative, label):
- if not isinstance(relative, str) or not relative:
- raise ProbHubError(f"{label} is required", code="unsafe_package_source")
- problem_dir = Path(problem_dir).resolve()
- relative_path = Path(relative)
- if relative_path.is_absolute() or relative_path.drive or ".." in relative_path.parts:
- raise ProbHubError(
- f"{label} must stay inside the problem directory: {relative}",
- code="unsafe_package_source",
- )
- candidate = problem_dir / relative_path
- current = problem_dir
- has_symlink = False
- for part in relative_path.parts:
- current = current / part
- if current.is_symlink():
- has_symlink = True
- break
- if has_symlink or not candidate.is_file():
- raise ProbHubError(
- f"{label} must be a regular file: {relative}",
- code="unsafe_package_source",
- )
- resolved = candidate.resolve()
try:
- resolved.relative_to(problem_dir)
- except ValueError as exc:
- raise ProbHubError(
- f"{label} must stay inside the problem directory: {relative}",
- code="unsafe_package_source",
- ) from exc
- return resolved
+ return resolve_problem_regular_file(problem_dir, relative)
+ except ProblemPathError as exc:
+ if exc.reason == "invalid":
+ message = f"{label} is required"
+ elif exc.reason == "outside":
+ message = f"{label} must stay inside the problem directory: {relative}"
+ else:
+ message = f"{label} must be a regular file: {relative}"
+ raise ProbHubError(message, code="unsafe_package_source") from exc
def _problem_directory(problem_dir, relative, label):
@@ -358,8 +338,6 @@ def prepare_output_validator(problem_dir, config):
return None
source_value = judge.get(source_key)
- if not source_value:
- raise ProbHubError(f"judge.{source_key} is required for {judge_type} judging")
source = _problem_regular_file(problem_dir, source_value, f"judge.{source_key}")
if not TESTLIB_PATH.is_file():
raise ProbHubError(f"testlib.h not found: {TESTLIB_PATH}")
diff --git a/probhub/problem_paths.py b/probhub/problem_paths.py
new file mode 100644
index 0000000..21a2b6a
--- /dev/null
+++ b/probhub/problem_paths.py
@@ -0,0 +1,84 @@
+import os
+import stat
+from pathlib import Path, PurePosixPath, PureWindowsPath
+
+
+_FILE_ATTRIBUTE_REPARSE_POINT = getattr(
+ stat,
+ "FILE_ATTRIBUTE_REPARSE_POINT",
+ 0x400,
+)
+
+
+class ProblemPathError(ValueError):
+ """A configured problem-local path failed a stable safety check."""
+
+ def __init__(self, reason):
+ self.reason = reason
+ super().__init__(reason)
+
+
+def resolve_problem_regular_file(problem_dir, value):
+ """Resolve a problem-local regular file without traversing link-like paths."""
+
+ if not isinstance(value, str) or not value.strip():
+ raise ProblemPathError("invalid")
+
+ normalized = value.strip().replace("\\", "/")
+ posix_path = PurePosixPath(normalized)
+ windows_path = PureWindowsPath(normalized)
+ if posix_path.is_absolute() or windows_path.is_absolute() or windows_path.drive:
+ raise ProblemPathError("outside")
+ if ".." in posix_path.parts:
+ raise ProblemPathError("outside")
+
+ try:
+ access_root = Path(os.path.abspath(os.fspath(problem_dir)))
+ problem_root = access_root.resolve(strict=True)
+ except (TypeError, ValueError) as exc:
+ raise ProblemPathError("invalid") from exc
+ except (OSError, RuntimeError) as exc:
+ raise ProblemPathError("missing") from exc
+ if not problem_root.is_dir():
+ raise ProblemPathError("non_regular")
+
+ parts = posix_path.parts
+ if not parts:
+ raise ProblemPathError("non_regular")
+
+ current = access_root
+ for index, part in enumerate(parts):
+ current = current / part
+ try:
+ info = os.lstat(current)
+ except ValueError as exc:
+ raise ProblemPathError("invalid") from exc
+ except NotADirectoryError as exc:
+ raise ProblemPathError("non_regular") from exc
+ except OSError as exc:
+ raise ProblemPathError("missing") from exc
+
+ attributes = getattr(info, "st_file_attributes", 0)
+ if stat.S_ISLNK(info.st_mode) or attributes & _FILE_ATTRIBUTE_REPARSE_POINT:
+ raise ProblemPathError("link")
+ if index < len(parts) - 1:
+ if not stat.S_ISDIR(info.st_mode):
+ raise ProblemPathError("non_regular")
+ elif not stat.S_ISREG(info.st_mode):
+ raise ProblemPathError("non_regular")
+
+ try:
+ resolved = current.resolve(strict=True)
+ except FileNotFoundError as exc:
+ raise ProblemPathError("missing") from exc
+ except RuntimeError as exc:
+ raise ProblemPathError("link") from exc
+ except OSError as exc:
+ raise ProblemPathError("missing") from exc
+ try:
+ resolved.relative_to(problem_root)
+ except ValueError as exc:
+ raise ProblemPathError("outside") from exc
+ # Keep the caller's absolute spelling (for example Windows 8.3 or SUBST
+ # aliases) after validating the canonical target against problem_root.
+ return current
diff --git a/probhub/special_judges.py b/probhub/special_judges.py
new file mode 100644
index 0000000..ce3307d
--- /dev/null
+++ b/probhub/special_judges.py
@@ -0,0 +1,947 @@
+"""Shared execution primitives for custom and interactive judges."""
+
+import codecs
+import os
+import shutil
+import subprocess
+import tempfile
+import threading
+import time
+from pathlib import Path
+
+from .io import read_bounded_text
+from .process_control import (
+ DEFAULT_PROCESS_LIMIT,
+ OutputBudgetError,
+ ProcessCancelled,
+ cancellation_requested,
+ output_path_size,
+ run_managed_to_files,
+ spawn_managed,
+)
+
+
+MIB = 1024 * 1024
+MAX_CHECKER_DIAGNOSTIC_BYTES = 8 * MIB
+_FEEDBACK_NAMES = ("judgemessage.txt", "teammessage.txt")
+_RESOURCE_LIMIT_REASONS = frozenset(
+ ("time_limit", "memory_limit", "output_limit", "process_limit")
+)
+_CLEANUP_ERROR_BYTES = 4096
+
+
+def _cleanup_error(stage, actor, exc):
+ message = str(exc).encode("utf-8", errors="replace")[:_CLEANUP_ERROR_BYTES]
+ return {
+ "stage": stage,
+ "actor": actor,
+ "message": message.decode("utf-8", errors="replace"),
+ }
+
+
+def _remove_runtime_directory(path):
+ if path is None:
+ return True, 0, None
+ last_error = None
+ for attempt in range(1, 21):
+ try:
+ shutil.rmtree(path)
+ return True, attempt, None
+ except FileNotFoundError:
+ return True, attempt, None
+ except OSError as exc:
+ last_error = exc
+ if attempt < 20:
+ time.sleep(0.05)
+ return False, 20, last_error
+
+
+def _apply_cleanup_failure(result, cleanup):
+ result["cleanup"] = cleanup
+ if cleanup["ok"]:
+ return result
+ result["pre_cleanup_result"] = {
+ key: result.get(key)
+ for key in (
+ "verdict",
+ "execution_status",
+ "failure_kind",
+ "actor",
+ "termination_reason",
+ )
+ }
+ result.update({
+ "verdict": None,
+ "execution_status": "cleanup_error",
+ "failure_kind": "cleanup_failure",
+ "actor": "supervisor",
+ "termination_reason": "cleanup_error",
+ "message": cleanup["errors"][0]["message"] if cleanup["errors"] else "cleanup failed",
+ })
+ if "status" in result:
+ result["status"] = "FAIL"
+ return result
+
+
+def _feedback_message(feedback_dir, fallback="", limit_bytes=MAX_CHECKER_DIAGNOSTIC_BYTES):
+ for name in _FEEDBACK_NAMES:
+ path = Path(feedback_dir) / name
+ try:
+ result = read_bounded_text(path, limit_bytes)
+ except OSError:
+ continue
+ message = result["text"].strip()
+ if message:
+ result["source"] = name
+ return message, result
+ encoded = (fallback or "").encode("utf-8", errors="replace")
+ retained = encoded[: max(int(limit_bytes), 0)]
+ return retained.decode("utf-8", errors="replace").strip(), {
+ "source": "stderr",
+ "observed_bytes": len(encoded),
+ "retained_bytes": len(retained),
+ "truncated": len(retained) < len(encoded),
+ "exists": bool(encoded),
+ }
+
+
+def _failed_checker_result(reason, message, diagnostic_limit_bytes):
+ stderr = str(message).encode("utf-8", errors="replace")
+ retained = stderr[: max(int(diagnostic_limit_bytes), 0)]
+ bounded_message = retained.decode("utf-8", errors="replace").strip()
+ feedback = {
+ "source": "stderr",
+ "observed_bytes": len(stderr),
+ "retained_bytes": len(retained),
+ "truncated": len(retained) < len(stderr),
+ "exists": bool(stderr),
+ }
+ return {
+ "verdict": None,
+ "execution_status": reason,
+ "failure_kind": (
+ "control_failure" if reason == "output_control_error" else "startup_failure"
+ ),
+ "actor": "supervisor" if reason == "output_control_error" else "checker",
+ "termination_reason": reason,
+ "message": bounded_message,
+ "feedback_message": bounded_message,
+ "returncode": None,
+ "time": 0.0,
+ "memory": None,
+ "memory_enforced": False,
+ "process_limit_enforced": False,
+ "output_bytes": 0,
+ "retained_output_bytes": 0,
+ "stdout_retained_bytes": 0,
+ "stderr_retained_bytes": 0,
+ "output_truncated": False,
+ "stdout": b"",
+ "stderr": stderr,
+ "feedback": feedback,
+ }
+
+
+def run_checker_to_files(
+ checker_command,
+ input_path,
+ answer_path,
+ contestant_output_path,
+ *,
+ timeout,
+ cwd=None,
+ memory_limit_mb=256,
+ output_limit_bytes=64 * MIB,
+ process_limit=DEFAULT_PROCESS_LIMIT,
+ diagnostic_limit_bytes=None,
+ env=None,
+):
+ """Run a DOMjudge/testlib Checker and return protocol and execution evidence.
+
+ The Checker receives `` `` as arguments and
+ reads contestant output from stdin. Its stdout, stderr, and both standard
+ feedback files share one bounded output budget.
+ """
+ effective_output_limit = min(
+ max(int(output_limit_bytes), 0),
+ MAX_CHECKER_DIAGNOSTIC_BYTES,
+ )
+ if diagnostic_limit_bytes is None:
+ diagnostic_limit_bytes = effective_output_limit
+ else:
+ diagnostic_limit_bytes = min(
+ max(int(diagnostic_limit_bytes), 0),
+ effective_output_limit,
+ )
+
+ runtime_dir = None
+ result = None
+ cancelled_error = None
+ cleanup = {
+ "ok": True,
+ "checker_tree_termination": "not-started",
+ "runtime_removed": False,
+ "runtime_remove_attempts": 0,
+ "errors": [],
+ }
+ try:
+ runtime_dir = Path(
+ tempfile.mkdtemp(prefix=".probhub-checker-", dir=cwd)
+ )
+ feedback_dir = runtime_dir / "feedback"
+ feedback_dir.mkdir()
+ stdout_path = runtime_dir / "checker.stdout"
+ stderr_path = runtime_dir / "checker.stderr"
+ feedback_paths = tuple(feedback_dir / name for name in _FEEDBACK_NAMES)
+ try:
+ cleanup["checker_tree_termination"] = "pending"
+ execution = run_managed_to_files(
+ [
+ *checker_command,
+ os.fspath(input_path),
+ os.fspath(answer_path),
+ os.fspath(feedback_dir),
+ ],
+ input_path=contestant_output_path,
+ stdout_path=stdout_path,
+ stderr_path=stderr_path,
+ additional_output_paths=feedback_paths,
+ timeout=timeout,
+ memory_limit_mb=memory_limit_mb,
+ output_limit_bytes=effective_output_limit,
+ process_limit=process_limit,
+ cwd=cwd,
+ env=env,
+ )
+ cleanup["checker_tree_termination"] = "completed"
+ except OutputBudgetError as exc:
+ cleanup["checker_tree_termination"] = "completed"
+ result = _failed_checker_result(
+ "output_control_error", str(exc), diagnostic_limit_bytes
+ )
+ return result
+ except OSError as exc:
+ cleanup["checker_tree_termination"] = "not-started"
+ result = _failed_checker_result("start_error", str(exc), diagnostic_limit_bytes)
+ return result
+
+ stdout = stdout_path.read_bytes() if stdout_path.is_file() else b""
+ stderr = stderr_path.read_bytes() if stderr_path.is_file() else b""
+ feedback_message, feedback = _feedback_message(
+ feedback_dir,
+ stderr.decode("utf-8", errors="replace"),
+ diagnostic_limit_bytes,
+ )
+ message = feedback_message
+ termination_reason = execution.get("reason") or "completed"
+ if termination_reason == "cancelled":
+ raise ProcessCancelled(execution.get("message") or "execution cancelled")
+ returncode = execution.get("returncode")
+ verdict = None
+ failure_kind = None
+ actor = "checker"
+ if termination_reason == "completed" and returncode in {0, 42}:
+ verdict = "AC"
+ execution_status = "completed"
+ actor = "session"
+ elif termination_reason == "completed" and returncode in {1, 2, 43}:
+ verdict = "WA"
+ execution_status = "completed"
+ failure_kind = "wrong_answer"
+ actor = "contestant"
+ elif termination_reason == "completed":
+ execution_status = "completed"
+ failure_kind = "judge_failure"
+ message = message or f"checker exited with code {returncode}"
+ elif termination_reason in _RESOURCE_LIMIT_REASONS:
+ execution_status = termination_reason
+ failure_kind = "resource_limit"
+ message = execution.get("message") or termination_reason.replace("_", " ")
+ else:
+ execution_status = "completed"
+ failure_kind = "judge_failure"
+ message = execution.get("message") or termination_reason.replace("_", " ")
+
+ result = {
+ "verdict": verdict,
+ "execution_status": execution_status,
+ "failure_kind": failure_kind,
+ "actor": actor,
+ "termination_reason": termination_reason,
+ "message": message,
+ "feedback_message": feedback_message,
+ "returncode": returncode,
+ "time": execution.get("time", 0.0),
+ "memory": execution.get("memory"),
+ "memory_enforced": execution.get("memory_enforced", False),
+ "process_limit_enforced": execution.get("process_limit_enforced", False),
+ "output_bytes": execution.get("output_bytes", 0),
+ "retained_output_bytes": execution.get("retained_output_bytes", 0),
+ "stdout_retained_bytes": execution.get("stdout_retained_bytes", 0),
+ "stderr_retained_bytes": execution.get("stderr_retained_bytes", 0),
+ "output_truncated": execution.get("output_truncated", False),
+ "stdout": stdout,
+ "stderr": stderr,
+ "feedback": feedback,
+ }
+ return result
+ except ProcessCancelled as exc:
+ cancelled_error = exc
+ result = _failed_checker_result(
+ "cancelled", str(exc), diagnostic_limit_bytes
+ )
+ result.update({
+ "execution_status": "cancelled",
+ "failure_kind": "cancelled",
+ "actor": "supervisor",
+ "termination_reason": "cancelled",
+ })
+ except OutputBudgetError as exc:
+ result = _failed_checker_result(
+ "output_control_error", str(exc), diagnostic_limit_bytes
+ )
+ return result
+ except OSError as exc:
+ result = _failed_checker_result("start_error", str(exc), diagnostic_limit_bytes)
+ result["actor"] = "supervisor"
+ return result
+ finally:
+ removed, attempts, error = _remove_runtime_directory(runtime_dir)
+ cleanup["runtime_removed"] = removed
+ cleanup["runtime_remove_attempts"] = attempts
+ if error is not None:
+ cleanup["errors"].append(
+ _cleanup_error("runtime_remove", "supervisor", error)
+ )
+ cleanup["ok"] = not cleanup["errors"]
+ if result is not None:
+ _apply_cleanup_failure(result, cleanup)
+ if cancelled_error is not None:
+ if result is not None and result.get("failure_kind") == "cleanup_failure":
+ return result
+ raise cancelled_error
+ return result
+
+
+def _command_list(command):
+ if isinstance(command, (str, bytes, os.PathLike)):
+ return [os.fspath(command)]
+ return [os.fspath(item) for item in command]
+
+
+def _protocol_outcome(returncode, message, actor):
+ message = (message or "").strip()
+ if returncode in {0, 42}:
+ return {
+ "status": "AC",
+ "verdict": "AC",
+ "execution_status": "completed",
+ "failure_kind": None,
+ "actor": "session",
+ "termination_reason": "completed",
+ "message": message,
+ }
+ if returncode in {1, 2, 43}:
+ return {
+ "status": "WA",
+ "verdict": "WA",
+ "execution_status": "completed",
+ "failure_kind": "wrong_answer",
+ "actor": "contestant",
+ "termination_reason": "completed",
+ "message": message,
+ }
+ return {
+ "status": "FAIL",
+ "verdict": None,
+ "execution_status": "completed",
+ "failure_kind": "judge_failure",
+ "actor": actor,
+ "termination_reason": "completed",
+ "message": message or f"{actor} exited with code {returncode}",
+ }
+
+
+def _record_interactive_transcript(transcript, direction, payload, decoder, *, final=False):
+ """Reserve the shared byte budget and decode one direction incrementally."""
+ with transcript["_lock"]:
+ remaining = max(transcript["limit"] - transcript["bytes"], 0)
+ saved = payload[:remaining]
+ transcript["bytes"] += len(saved)
+ if len(saved) < len(payload):
+ transcript["truncated"] = True
+ text = decoder.decode(saved, final=final)
+ if text:
+ transcript["entries"].append({"direction": direction, "data": text})
+
+
+def _pump_interactive_stream(source, destination, direction, transcript, activity, traffic):
+ """Forward bytes while recording traffic and a bounded transcript."""
+ decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+ try:
+ while True:
+ chunk = source.read(4096)
+ if not chunk:
+ break
+ now = time.monotonic()
+ with transcript["_lock"]:
+ activity["last"] = max(activity["last"], now)
+ traffic[direction] = traffic.get(direction, 0) + len(chunk)
+ limit_exceeded = traffic[direction] > traffic["limit"]
+ if limit_exceeded:
+ break
+ _record_interactive_transcript(transcript, direction, chunk, decoder)
+ destination.write(chunk)
+ destination.flush()
+ except (BrokenPipeError, OSError, ValueError):
+ pass
+ finally:
+ _record_interactive_transcript(transcript, direction, b"", decoder, final=True)
+ for stream in (destination, source):
+ try:
+ stream.close()
+ except (OSError, ValueError):
+ pass
+
+
+def _interactive_evidence(
+ traffic,
+ solution_stderr,
+ interactor_stderr,
+ feedback_dir=None,
+):
+ solution_stderr_bytes = output_path_size(solution_stderr, required=True)[0]
+ interactor_stderr_bytes = output_path_size(interactor_stderr, required=True)[0]
+ feedback_bytes = 0
+ if feedback_dir:
+ for name in _FEEDBACK_NAMES:
+ feedback_bytes += output_path_size(Path(feedback_dir) / name)[0]
+ lock = traffic.get("_lock")
+ if lock is None:
+ solution_protocol_bytes = traffic.get("solution_to_interactor", 0)
+ interactor_protocol_bytes = traffic.get("interactor_to_solution", 0)
+ limit = int(traffic["limit"])
+ else:
+ with lock:
+ solution_protocol_bytes = traffic.get("solution_to_interactor", 0)
+ interactor_protocol_bytes = traffic.get("interactor_to_solution", 0)
+ limit = int(traffic["limit"])
+ return {
+ "limit_bytes": limit,
+ "solution_to_interactor": int(solution_protocol_bytes),
+ "interactor_to_solution": int(interactor_protocol_bytes),
+ "solution_stderr": int(solution_stderr_bytes),
+ "interactor_stderr": int(interactor_stderr_bytes),
+ "feedback": int(feedback_bytes),
+ }
+
+
+def _interactive_output_classification(evidence, diagnostic_limit_bytes):
+ limit = evidence["limit_bytes"]
+ if evidence["interactor_to_solution"] > limit:
+ return {
+ "status": "FAIL",
+ "verdict": None,
+ "execution_status": "output_limit",
+ "failure_kind": "resource_limit",
+ "actor": "interactor",
+ "termination_reason": "output_limit",
+ "message": "interactor output limit exceeded",
+ }
+ if evidence["interactor_stderr"] + evidence["feedback"] > int(diagnostic_limit_bytes):
+ return {
+ "status": "FAIL",
+ "verdict": None,
+ "execution_status": "output_limit",
+ "failure_kind": "resource_limit",
+ "actor": "interactor",
+ "termination_reason": "output_limit",
+ "message": "interactor diagnostic output limit exceeded",
+ }
+ if evidence["solution_to_interactor"] + evidence["solution_stderr"] > limit:
+ return {
+ "status": "OLE",
+ "verdict": "OLE",
+ "execution_status": "output_limit",
+ "failure_kind": "resource_limit",
+ "actor": "contestant",
+ "termination_reason": "output_limit",
+ "message": "interactive output limit exceeded",
+ }
+ return None
+
+
+def _interactive_result(
+ outcome,
+ *,
+ elapsed,
+ memory,
+ memory_enforced,
+ transcript,
+ traffic_evidence,
+ cleanup,
+ exit_codes,
+ timeout_kind=None,
+ resources=None,
+):
+ lock = transcript.get("_lock")
+ if lock is None:
+ entries = [dict(entry) for entry in transcript["entries"]]
+ transcript_truncated = bool(transcript["truncated"])
+ transcript_bytes = int(transcript.get("bytes", 0))
+ else:
+ with lock:
+ entries = [dict(entry) for entry in transcript["entries"]]
+ transcript_truncated = bool(transcript["truncated"])
+ transcript_bytes = int(transcript.get("bytes", 0))
+ result = dict(outcome)
+ result.update({
+ "time": float(elapsed),
+ "memory": memory,
+ "memory_enforced": bool(memory_enforced),
+ "exit_codes": dict(exit_codes),
+ "traffic": dict(traffic_evidence or {}),
+ "transcript": entries,
+ "transcript_bytes": transcript_bytes,
+ "transcript_truncated": transcript_truncated,
+ "output_bytes": (
+ int(traffic_evidence.get("solution_to_interactor", 0))
+ + int(traffic_evidence.get("solution_stderr", 0))
+ if traffic_evidence else 0
+ ),
+ "cleanup": cleanup,
+ })
+ if resources is not None:
+ result["resources"] = resources
+ if timeout_kind:
+ result["timeout_kind"] = timeout_kind
+ return result
+
+
+def _terminate_interactive_actor(managed, actor, cleanup):
+ if managed is None:
+ return
+ key = f"{actor}_tree_termination"
+ try:
+ managed.terminate()
+ cleanup[key] = "completed"
+ except BaseException as exc:
+ cleanup[key] = "failed"
+ cleanup["errors"].append(_cleanup_error("tree_termination", actor, exc))
+
+
+def _join_interactive_pumps(threads, cleanup):
+ for thread in threads:
+ thread.join(timeout=1)
+ cleanup["pumps_joined"] = all(not thread.is_alive() for thread in threads)
+ if not cleanup["pumps_joined"] and not any(
+ item["stage"] == "pump_join" for item in cleanup["errors"]
+ ):
+ cleanup["errors"].append({
+ "stage": "pump_join",
+ "actor": "supervisor",
+ "message": "interactive pump thread did not stop",
+ })
+
+
+def execute_interactive_session(
+ contestant_command,
+ interactor_command,
+ input_path,
+ answer_path,
+ *,
+ work_dir=None,
+ time_limit=1.0,
+ memory_limit_mb=256,
+ idle_limit=None,
+ transcript_limit=65536,
+ output_limit_bytes=64 * MIB,
+ process_limit=DEFAULT_PROCESS_LIMIT,
+ env=None,
+):
+ """Run one contestant/Interactor session with explicit responsibility evidence."""
+ monotonic_start = time.monotonic()
+ wall_start = time.time()
+ idle_limit = max(
+ float(idle_limit if idle_limit is not None else min(time_limit, 2.0)),
+ 0.1,
+ )
+ state_lock = threading.RLock()
+ transcript = {
+ "entries": [],
+ "bytes": 0,
+ "limit": max(int(transcript_limit), 0),
+ "truncated": False,
+ "_lock": state_lock,
+ }
+ activity = {"last": monotonic_start}
+ traffic = {"limit": max(int(output_limit_bytes), 0), "_lock": state_lock}
+ cleanup = {
+ "ok": True,
+ "contestant_tree_termination": "not-started",
+ "interactor_tree_termination": "not-started",
+ "pumps_joined": True,
+ "runtime_removed": False,
+ "runtime_remove_attempts": 0,
+ "errors": [],
+ }
+ runtime_dir = None
+ solution_stderr = None
+ interactor_stderr = None
+ feedback_dir = None
+ solution_managed = None
+ interactor_managed = None
+ solution_proc = None
+ interactor_proc = None
+ memory_enforced = False
+ peak_memory_mb = None
+ solution_process_limit_enforced = False
+ interactor_memory_enforced = False
+ interactor_peak_memory_mb = None
+ interactor_process_limit_enforced = False
+ threads = []
+ result = None
+ cancelled_error = None
+ spawn_actor = "supervisor"
+ diagnostic_limit_bytes = min(max(int(output_limit_bytes), 0), MAX_CHECKER_DIAGNOSTIC_BYTES)
+
+ def current_exit_codes():
+ return {
+ "contestant": getattr(solution_proc, "returncode", None),
+ "interactor": getattr(interactor_proc, "returncode", None),
+ }
+
+ def current_evidence():
+ if solution_stderr is None or interactor_stderr is None:
+ return {}
+ return _interactive_evidence(
+ traffic,
+ solution_stderr,
+ interactor_stderr,
+ feedback_dir,
+ )
+
+ def finish(outcome, *, timeout_kind=None, traffic_evidence=None):
+ nonlocal result
+ actor = outcome.get("actor")
+ reported_memory = (
+ interactor_peak_memory_mb if actor == "interactor" else peak_memory_mb
+ )
+ reported_memory_enforced = (
+ interactor_memory_enforced if actor == "interactor" else memory_enforced
+ )
+ result = _interactive_result(
+ outcome,
+ elapsed=time.time() - wall_start,
+ memory=reported_memory,
+ memory_enforced=reported_memory_enforced,
+ transcript=transcript,
+ traffic_evidence=(
+ current_evidence() if traffic_evidence is None else traffic_evidence
+ ),
+ cleanup=cleanup,
+ exit_codes=current_exit_codes(),
+ timeout_kind=timeout_kind,
+ resources={
+ "contestant": {
+ "memory": peak_memory_mb,
+ "memory_enforced": bool(memory_enforced),
+ "process_limit_enforced": bool(solution_process_limit_enforced),
+ },
+ "interactor": {
+ "memory": interactor_peak_memory_mb,
+ "memory_enforced": bool(interactor_memory_enforced),
+ "process_limit_enforced": bool(interactor_process_limit_enforced),
+ },
+ },
+ )
+ return result
+
+ try:
+ runtime_dir = Path(tempfile.mkdtemp(prefix=".probhub-interactive-", dir=work_dir))
+ solution_stderr = runtime_dir / "solution.stderr"
+ interactor_stderr = runtime_dir / "interactor.stderr"
+ feedback_dir = runtime_dir / "feedback"
+ feedback_dir.mkdir()
+ with solution_stderr.open("wb") as solution_err, interactor_stderr.open("wb") as interactor_err:
+ spawn_actor = "contestant"
+ cleanup["contestant_tree_termination"] = "pending"
+ solution_managed = spawn_managed(
+ _command_list(contestant_command),
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=solution_err,
+ bufsize=0,
+ memory_limit_mb=memory_limit_mb,
+ process_limit=process_limit,
+ cwd=work_dir,
+ env=env,
+ )
+ solution_proc = solution_managed.proc
+ memory_enforced = solution_managed.memory_enforced
+ solution_process_limit_enforced = bool(
+ getattr(solution_managed, "process_limit_enforced", False)
+ )
+ spawn_actor = "interactor"
+ cleanup["interactor_tree_termination"] = "pending"
+ interactor_managed = spawn_managed(
+ [
+ *_command_list(interactor_command),
+ os.fspath(input_path),
+ os.fspath(answer_path),
+ os.fspath(feedback_dir),
+ ],
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=interactor_err,
+ bufsize=0,
+ memory_limit_mb=memory_limit_mb,
+ process_limit=process_limit,
+ cwd=work_dir,
+ env=env,
+ )
+ interactor_proc = interactor_managed.proc
+ interactor_memory_enforced = interactor_managed.memory_enforced
+ interactor_process_limit_enforced = bool(
+ getattr(interactor_managed, "process_limit_enforced", False)
+ )
+ spawn_actor = "supervisor"
+ threads = [
+ threading.Thread(
+ target=_pump_interactive_stream,
+ args=(
+ solution_proc.stdout,
+ interactor_proc.stdin,
+ "solution_to_interactor",
+ transcript,
+ activity,
+ traffic,
+ ),
+ daemon=True,
+ ),
+ threading.Thread(
+ target=_pump_interactive_stream,
+ args=(
+ interactor_proc.stdout,
+ solution_proc.stdin,
+ "interactor_to_solution",
+ transcript,
+ activity,
+ traffic,
+ ),
+ daemon=True,
+ ),
+ ]
+ cleanup["pumps_joined"] = False
+ for thread in threads:
+ thread.start()
+
+ monotonic_start = time.monotonic()
+ wall_start = time.time()
+ activity["last"] = max(activity["last"], monotonic_start)
+ last_resource_sample = monotonic_start
+ deadline = monotonic_start + float(time_limit)
+ while solution_proc.poll() is None or interactor_proc.poll() is None:
+ if cancellation_requested():
+ raise ProcessCancelled("execution cancelled")
+ now = time.monotonic()
+ evidence = current_evidence()
+ resource_outcome = _interactive_output_classification(
+ evidence, diagnostic_limit_bytes
+ )
+ if now - last_resource_sample >= 0.05:
+ last_resource_sample = now
+ solution_count, solution_memory = solution_managed.sample()
+ peak_memory_mb = solution_managed.peak_memory_mb
+ if solution_memory is not None and solution_memory >= memory_limit_mb:
+ resource_outcome = {
+ "status": "MLE", "verdict": "MLE",
+ "execution_status": "memory_limit", "failure_kind": "resource_limit",
+ "actor": "contestant", "termination_reason": "memory_limit",
+ "message": "memory limit exceeded",
+ }
+ elif solution_count is not None and solution_count > process_limit:
+ resource_outcome = {
+ "status": "RE", "verdict": "RE",
+ "execution_status": "process_limit", "failure_kind": "resource_limit",
+ "actor": "contestant", "termination_reason": "process_limit",
+ "message": "process limit exceeded",
+ }
+ interactor_count, interactor_memory = interactor_managed.sample()
+ interactor_peak_memory_mb = interactor_managed.peak_memory_mb
+ if interactor_memory is not None and interactor_memory >= memory_limit_mb:
+ resource_outcome = {
+ "status": "FAIL", "verdict": None,
+ "execution_status": "memory_limit", "failure_kind": "resource_limit",
+ "actor": "interactor", "termination_reason": "memory_limit",
+ "message": "interactor memory limit exceeded",
+ }
+ elif interactor_count is not None and interactor_count > process_limit:
+ resource_outcome = {
+ "status": "FAIL", "verdict": None,
+ "execution_status": "process_limit", "failure_kind": "resource_limit",
+ "actor": "interactor", "termination_reason": "process_limit",
+ "message": "interactor process limit exceeded",
+ }
+ elif interactor_proc.poll() is not None:
+ protocol = _protocol_outcome(interactor_proc.returncode, "", "interactor")
+ if protocol["status"] == "FAIL":
+ resource_outcome = protocol
+ if resource_outcome:
+ _terminate_interactive_actor(solution_managed, "contestant", cleanup)
+ solution_managed = None
+ _terminate_interactive_actor(interactor_managed, "interactor", cleanup)
+ interactor_managed = None
+ _join_interactive_pumps(threads, cleanup)
+ return finish(resource_outcome)
+
+ timeout_kind = None
+ if now >= deadline:
+ timeout_kind = "total"
+ else:
+ with state_lock:
+ idle_elapsed = now - activity["last"]
+ if idle_elapsed >= idle_limit:
+ timeout_kind = "idle"
+ if timeout_kind:
+ solution_managed.sample()
+ peak_memory_mb = solution_managed.peak_memory_mb
+ _terminate_interactive_actor(solution_managed, "contestant", cleanup)
+ solution_managed = None
+ _terminate_interactive_actor(interactor_managed, "interactor", cleanup)
+ interactor_managed = None
+ _join_interactive_pumps(threads, cleanup)
+ return finish({
+ "status": "TLE",
+ "verdict": "TLE",
+ "execution_status": "time_limit",
+ "failure_kind": "resource_limit",
+ "actor": "session",
+ "termination_reason": "time_limit",
+ "message": (
+ "interactive idle timeout exceeded"
+ if timeout_kind == "idle"
+ else "interactive time limit exceeded"
+ ),
+ }, timeout_kind=timeout_kind)
+ time.sleep(0.005)
+
+ solution_managed.sample()
+ peak_memory_mb = solution_managed.peak_memory_mb
+ interactor_managed.sample()
+ interactor_peak_memory_mb = interactor_managed.peak_memory_mb
+ _terminate_interactive_actor(solution_managed, "contestant", cleanup)
+ solution_managed = None
+ _terminate_interactive_actor(interactor_managed, "interactor", cleanup)
+ interactor_managed = None
+ _join_interactive_pumps(threads, cleanup)
+ evidence = current_evidence()
+ resource_outcome = _interactive_output_classification(
+ evidence, diagnostic_limit_bytes
+ )
+ if (
+ interactor_proc.returncode != 0
+ and interactor_memory_enforced
+ and interactor_peak_memory_mb is not None
+ and interactor_peak_memory_mb >= memory_limit_mb * 0.98
+ ):
+ resource_outcome = {
+ "status": "FAIL", "verdict": None,
+ "execution_status": "memory_limit", "failure_kind": "resource_limit",
+ "actor": "interactor", "termination_reason": "memory_limit",
+ "message": "interactor memory limit exceeded",
+ }
+
+ solution_message = ""
+ interactor_message = ""
+ try:
+ solution_message = read_bounded_text(
+ solution_stderr, max(int(output_limit_bytes), 0)
+ )["text"].strip()
+ except OSError:
+ pass
+ try:
+ interactor_message = read_bounded_text(
+ interactor_stderr, diagnostic_limit_bytes
+ )["text"].strip()
+ except OSError:
+ pass
+ interactor_message = _feedback_message(
+ feedback_dir, interactor_message, diagnostic_limit_bytes
+ )[0]
+ if resource_outcome:
+ return finish(resource_outcome)
+ if solution_proc.returncode != 0:
+ inferred_memory = (
+ memory_enforced
+ and peak_memory_mb is not None
+ and peak_memory_mb >= memory_limit_mb * 0.98
+ )
+ if inferred_memory:
+ outcome = {
+ "status": "MLE", "verdict": "MLE",
+ "execution_status": "memory_limit", "failure_kind": "resource_limit",
+ "actor": "contestant", "termination_reason": "inferred_memory_limit",
+ "message": solution_message,
+ }
+ else:
+ outcome = {
+ "status": "RE", "verdict": "RE",
+ "execution_status": "completed", "failure_kind": "runtime_error",
+ "actor": "contestant", "termination_reason": "completed",
+ "message": solution_message,
+ }
+ return finish(outcome)
+ return finish(_protocol_outcome(
+ interactor_proc.returncode, interactor_message, "interactor"
+ ))
+ except ProcessCancelled as exc:
+ cancelled_error = exc
+ finish({
+ "status": "cancelled",
+ "verdict": None,
+ "execution_status": "cancelled",
+ "failure_kind": "cancelled",
+ "actor": "supervisor",
+ "termination_reason": "cancelled",
+ "message": str(exc),
+ }, traffic_evidence={})
+ except OutputBudgetError as exc:
+ return finish({
+ "status": "FAIL",
+ "verdict": None,
+ "execution_status": "output_control_error",
+ "failure_kind": "control_failure",
+ "actor": "supervisor",
+ "termination_reason": "output_control_error",
+ "message": str(exc),
+ }, traffic_evidence={})
+ except OSError as exc:
+ return finish({
+ "status": "FAIL",
+ "verdict": None,
+ "execution_status": "start_error",
+ "failure_kind": "startup_failure",
+ "actor": spawn_actor,
+ "termination_reason": "start_error",
+ "message": str(exc),
+ }, traffic_evidence={})
+ finally:
+ _terminate_interactive_actor(solution_managed, "contestant", cleanup)
+ _terminate_interactive_actor(interactor_managed, "interactor", cleanup)
+ _join_interactive_pumps(threads, cleanup)
+ removed, attempts, error = _remove_runtime_directory(runtime_dir)
+ cleanup["runtime_removed"] = removed
+ cleanup["runtime_remove_attempts"] = attempts
+ if error is not None:
+ cleanup["errors"].append(
+ _cleanup_error("runtime_remove", "supervisor", error)
+ )
+ cleanup["ok"] = not cleanup["errors"]
+ if result is not None:
+ _apply_cleanup_failure(result, cleanup)
+ if cancelled_error is not None:
+ if result is not None and result.get("failure_kind") == "cleanup_failure":
+ return result
+ raise cancelled_error
+ return result
diff --git a/probhub/stressing.py b/probhub/stressing.py
index b697acc..029ac06 100644
--- a/probhub/stressing.py
+++ b/probhub/stressing.py
@@ -15,7 +15,7 @@
from .build_lock import workspace_build_lock
from .errors import ProbHubError
-from .io import normalize_newlines, read_bounded_text, read_yaml, write_json, write_yaml
+from .io import normalize_newlines, read_yaml, write_json, write_yaml
from .output_compare import compare_standard_output
from .process_control import (
DEFAULT_PROCESS_LIMIT,
@@ -23,6 +23,7 @@
allocate_shared_prefix_bytes,
run_managed_to_files,
)
+from .special_judges import MAX_CHECKER_DIAGNOSTIC_BYTES, run_checker_to_files
from .transactions import (
TRANSACTION_PHASE_COMMITTED,
TRANSACTION_PHASE_PREPARED,
@@ -40,7 +41,6 @@
_CASE_NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
DEFAULT_TOOL_TIMEOUT = 5.0
MIB = 1024 * 1024
-MAX_CHECKER_DIAGNOSTIC_BYTES = 8 * MIB
STRESS_METADATA_BUDGET_BYTES = 64 * 1024
@@ -349,28 +349,6 @@ def _run(
}
-def _feedback_message(feedback_dir, fallback="", limit_bytes=MAX_CHECKER_DIAGNOSTIC_BYTES):
- for name in ("judgemessage.txt", "teammessage.txt"):
- path = Path(feedback_dir) / name
- try:
- result = read_bounded_text(path, limit_bytes)
- except OSError:
- continue
- message = result["text"].strip()
- if message:
- result["source"] = name
- return message, result
- encoded = (fallback or "").encode("utf-8", errors="replace")
- retained = encoded[: max(int(limit_bytes), 0)]
- return retained.decode("utf-8", errors="replace").strip(), {
- "source": "stderr",
- "observed_bytes": len(encoded),
- "retained_bytes": len(retained),
- "truncated": len(retained) < len(encoded),
- "exists": bool(encoded),
- }
-
-
def _compare_custom(
checker_command,
input_path,
@@ -382,28 +360,27 @@ def _compare_custom(
output_limit,
process_limit,
):
- feedback_dir = Path(tempfile.mkdtemp(prefix="feedback-", dir=cwd))
+ contestant_dir = Path(tempfile.mkdtemp(prefix="checker-input-", dir=cwd))
try:
+ contestant_output_path = contestant_dir / "contestant.out"
+ contestant_output_path.write_bytes(brute_output)
diagnostic_limit_bytes = min(int(output_limit * MIB), MAX_CHECKER_DIAGNOSTIC_BYTES)
- feedback_paths = tuple(
- feedback_dir / name for name in ("judgemessage.txt", "teammessage.txt")
- )
- result = _run(
- [*checker_command, str(input_path), str(answer_path), str(feedback_dir)],
- brute_output,
- timeout,
- cwd,
- memory_limit,
- min(output_limit, 8),
- process_limit,
- additional_output_paths=feedback_paths,
+ env = os.environ.copy()
+ env["PYTHONIOENCODING"] = "utf-8"
+ result = run_checker_to_files(
+ checker_command,
+ input_path,
+ answer_path,
+ contestant_output_path,
+ timeout=timeout,
+ cwd=cwd,
+ memory_limit_mb=memory_limit,
+ output_limit_bytes=int(output_limit * MIB),
+ process_limit=process_limit,
+ diagnostic_limit_bytes=diagnostic_limit_bytes,
+ env=env,
)
stderr = result.get("stderr") or b""
- message, feedback = _feedback_message(
- feedback_dir,
- stderr.decode("utf-8", errors="replace"),
- diagnostic_limit_bytes,
- )
output_details = {
key: result.get(key)
for key in (
@@ -414,44 +391,30 @@ def _compare_custom(
"output_truncated",
)
}
- if result.get("reason") != "completed":
- return {
- "status": "FAIL",
- "execution_status": result.get("status"),
- "match": False,
- "message": message or f"checker {result.get('message') or result.get('reason')}",
- "stderr": stderr,
- "feedback": feedback,
- **output_details,
- }
- if result["returncode"] in {0, 42}:
- return {
- "status": "AC",
- "match": True,
- "message": message,
- "stderr": stderr,
- "feedback": feedback,
- **output_details,
- }
- if result["returncode"] in {1, 2, 43}:
- return {
- "status": "WA",
- "match": False,
- "message": message,
- "stderr": stderr,
- "feedback": feedback,
- **output_details,
- }
- return {
- "status": "FAIL",
- "match": False,
- "message": message or f"checker exited with code {result['returncode']}",
+ verdict = result.get("verdict")
+ diagnostic_message = result.get("message") or ""
+ if result.get("failure_kind") != "cleanup_failure":
+ diagnostic_message = result.get("feedback_message") or diagnostic_message
+ comparison = {
+ "status": verdict or "FAIL",
+ "match": verdict == "AC",
+ "message": diagnostic_message,
"stderr": stderr,
- "feedback": feedback,
+ "feedback": result.get("feedback"),
**output_details,
}
+ if verdict is None and result.get("termination_reason") != "completed":
+ comparison["execution_status"] = {
+ "time_limit": "TLE",
+ "memory_limit": "MLE",
+ "output_limit": "OLE",
+ "process_limit": "RE",
+ "start_error": "RE",
+ "output_control_error": "FAIL",
+ }.get(result.get("execution_status"), "RE")
+ return comparison
finally:
- shutil.rmtree(feedback_dir, ignore_errors=True)
+ shutil.rmtree(contestant_dir, ignore_errors=True)
def _comparison(configured, commands, round_dir, input_data, accepted_output, brute_output):
diff --git a/scripts/local_judge.py b/scripts/local_judge.py
index 5282c4d..5a2b599 100644
--- a/scripts/local_judge.py
+++ b/scripts/local_judge.py
@@ -1,4 +1,3 @@
-import codecs
import hashlib
import json
import os
@@ -6,10 +5,8 @@
import re
import shutil
import signal
-import subprocess
import sys
import tempfile
-import threading
import time
import yaml
@@ -29,14 +26,14 @@
from probhub.errors import ProbHubError
from probhub.io import read_bounded_text
from probhub.output_compare import compare_standard_output
+from probhub.problem_paths import ProblemPathError, resolve_problem_regular_file
+from probhub.special_judges import execute_interactive_session, run_checker_to_files
from probhub.process_control import (
DEFAULT_PROCESS_LIMIT,
OutputBudgetError,
ProcessCancelled,
cancellation_requested,
- output_path_size,
run_managed_to_files,
- spawn_managed,
)
from probhub.solutions import (
analyze_solution_verification,
@@ -569,7 +566,10 @@ def resolve_problem_path(prob_dir, entry):
relative = _entry_file(entry)
if not relative:
return None
- return os.path.normpath(os.path.join(prob_dir, str(relative)))
+ try:
+ return os.fspath(resolve_problem_regular_file(prob_dir, str(relative)))
+ except ProblemPathError:
+ return None
def display_problem_path(prob_dir, path):
@@ -984,29 +984,6 @@ def run_sample_answer_testcase(
_remove_file_with_retries(output_file)
-def _checker_result(returncode, message):
- message = (message or "").strip()
- if returncode in {0, 42}:
- return "AC", message
- if returncode in {1, 2, 43}:
- return "WA", message
- return "FAIL", message or f"checker exited with code {returncode}"
-
-
-def _feedback_message(feedback_dir, fallback="", limit_bytes=MAX_TOOL_DIAGNOSTIC_BYTES):
- for name in ("judgemessage.txt", "teammessage.txt"):
- path = os.path.join(feedback_dir, name)
- try:
- result = read_bounded_text(path, limit_bytes)
- except OSError:
- continue
- message = result["text"].strip()
- if message:
- return message
- encoded = (fallback or "").encode("utf-8", errors="replace")
- return encoded[: max(int(limit_bytes), 0)].decode("utf-8", errors="replace").strip()
-
-
def run_custom_testcase(
bin_path,
checker_bin,
@@ -1018,13 +995,8 @@ def run_custom_testcase(
process_limit=DEFAULT_PROCESS_LIMIT,
capture_sample_answer=False,
):
- """Run a DOMjudge/testlib-style output validator.
-
- The validator receives as argv and reads
- contestant output from stdin, matching DOMjudge's custom validation protocol.
- """
+ """Run a contestant and adapt the shared Checker result to the legacy tuple."""
output_file = _temporary_output_path(bin_path)
- feedback_dir = tempfile.mkdtemp(prefix=".probhub-feedback-", dir=os.path.dirname(bin_path))
try:
status, elapsed, memory, memory_enforced, message, details = run_program_to_file(
bin_path, in_file, output_file, time_limit, memory_limit, output_limit, process_limit
@@ -1032,214 +1004,79 @@ def run_custom_testcase(
if capture_sample_answer:
details["sample_answer"] = compare_sample_answer(output_file, ans_file)
if status != "AC":
+ termination_reason = details.get("termination_reason")
+ if not termination_reason:
+ termination_reason = "start_error"
+ details["termination_reason"] = termination_reason
+ details.update({
+ "verdict": status if status != "FAIL" else None,
+ "execution_status": (
+ "memory_limit" if termination_reason == "inferred_memory_limit"
+ else termination_reason
+ if termination_reason in {
+ "time_limit", "memory_limit", "output_limit", "process_limit",
+ "output_control_error", "start_error",
+ }
+ else "completed"
+ ),
+ "failure_kind": (
+ "control_failure" if status == "FAIL"
+ else "startup_failure" if termination_reason == "start_error"
+ else "resource_limit" if status in {"TLE", "MLE", "OLE"}
+ or termination_reason == "process_limit"
+ else "runtime_error"
+ ),
+ "actor": "supervisor" if status == "FAIL" else "contestant",
+ })
return status, elapsed, memory, memory_enforced, message, details
- checker_stdout = output_file + ".checker.out"
- checker_stderr = output_file + ".checker.stderr"
- try:
- diagnostic_limit_bytes = min(
+ checker = run_checker_to_files(
+ [checker_bin],
+ in_file,
+ ans_file,
+ output_file,
+ timeout=max(5.0, float(time_limit)),
+ cwd=os.path.dirname(bin_path),
+ memory_limit_mb=memory_limit,
+ output_limit_bytes=min(
int(output_limit * 1024 * 1024), MAX_TOOL_DIAGNOSTIC_BYTES
- )
- feedback_paths = tuple(
- os.path.join(feedback_dir, name)
- for name in ("judgemessage.txt", "teammessage.txt")
- )
- checker = run_managed_to_files(
- [checker_bin, in_file, ans_file, feedback_dir],
- input_path=output_file,
- stdout_path=checker_stdout,
- stderr_path=checker_stderr,
- additional_output_paths=feedback_paths,
- timeout=max(5.0, float(time_limit)),
- memory_limit_mb=memory_limit,
- output_limit_bytes=min(int(output_limit * 1024 * 1024), 8 * 1024 * 1024),
- process_limit=process_limit,
- )
- try:
- stderr = open(checker_stderr, "r", encoding="utf-8", errors="replace").read()
- except OSError:
- stderr = ""
- checker_message = _feedback_message(
- feedback_dir, stderr, diagnostic_limit_bytes
- )
- if checker["reason"] != "completed":
- detail = checker["message"] or checker["reason"].replace("_", " ")
- return "FAIL", elapsed, memory, memory_enforced, f"checker {detail}", details
- checker_status, checker_message = _checker_result(
- checker["returncode"], checker_message
- )
- return checker_status, elapsed, memory, memory_enforced, checker_message, details
- except OutputBudgetError as exc:
- details = dict(details)
- details["checker_termination_reason"] = "output_control_error"
- return (
- "FAIL",
- elapsed,
- memory,
- memory_enforced,
- f"checker output control failed: {exc}",
- details,
- )
- except OSError as exc:
- return "FAIL", elapsed, memory, memory_enforced, f"checker failed to start: {exc}", details
- finally:
- for file_name in (checker_stdout, checker_stderr):
- _remove_file_with_retries(file_name)
- finally:
- _remove_file_with_retries(output_file)
- shutil.rmtree(feedback_dir, ignore_errors=True)
-
-
-def _record_interactive_transcript(transcript, direction, payload, decoder, *, final=False):
- """Reserve the shared byte budget and decode one direction incrementally."""
- with transcript["_lock"]:
- remaining = max(transcript["limit"] - transcript["bytes"], 0)
- saved = payload[:remaining]
- transcript["bytes"] += len(saved)
- if len(saved) < len(payload):
- transcript["truncated"] = True
- text = decoder.decode(saved, final=final)
- if text:
- transcript["entries"].append({
- "direction": direction,
- "data": text,
- })
-
-
-def _pump_interactive_stream(source, destination, direction, transcript, activity, traffic):
- """Forward bytes while recording a bounded transcript and last activity time."""
- decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
- try:
- while True:
- chunk = source.read(4096)
- if not chunk:
- break
- now = time.monotonic()
- with transcript["_lock"]:
- activity["last"] = max(activity["last"], now)
- traffic[direction] = traffic.get(direction, 0) + len(chunk)
- limit_exceeded = traffic[direction] > traffic["limit"]
- if limit_exceeded:
- break
- _record_interactive_transcript(transcript, direction, chunk, decoder)
- destination.write(chunk)
- destination.flush()
- except (BrokenPipeError, OSError, ValueError):
- pass
- finally:
- _record_interactive_transcript(
- transcript, direction, b"", decoder, final=True
+ ),
+ process_limit=process_limit,
)
- try:
- destination.close()
- except (OSError, ValueError):
- pass
- try:
- source.close()
- except (OSError, ValueError):
- pass
-
-
-def _interactive_solution_output_bytes(traffic, solution_stderr):
- try:
- stderr_size = os.path.getsize(solution_stderr)
- except OSError:
- stderr_size = 0
- lock = traffic.get("_lock")
- if lock is None:
- protocol_bytes = traffic.get("solution_to_interactor", 0)
- else:
- with lock:
- protocol_bytes = traffic.get("solution_to_interactor", 0)
- return int(protocol_bytes + stderr_size)
-
-
-def _interactive_result(
- status,
- elapsed,
- memory,
- memory_enforced,
- message,
- transcript,
- timeout_kind=None,
- traffic=None,
- solution_stderr=None,
- termination_reason=None,
-):
- lock = transcript.get("_lock")
- if lock is None:
- entries = [dict(entry) for entry in transcript["entries"]]
- transcript_truncated = transcript["truncated"]
- else:
- with lock:
- entries = [dict(entry) for entry in transcript["entries"]]
- transcript_truncated = transcript["truncated"]
- details = {
- "transcript": entries,
- "transcript_truncated": transcript_truncated,
- }
- if traffic is not None and solution_stderr is not None:
- details["output_bytes"] = _interactive_solution_output_bytes(
- traffic, solution_stderr
+ checker_status = checker.get("verdict") or "FAIL"
+ actor = checker.get("actor")
+ failure_kind = checker.get("failure_kind")
+ if checker_status == "AC":
+ actor = "session"
+ elif checker_status == "WA":
+ actor = "contestant"
+ failure_kind = "wrong_answer"
+ details.update({
+ "verdict": checker.get("verdict"),
+ "execution_status": checker.get("execution_status"),
+ "failure_kind": failure_kind,
+ "actor": actor,
+ "termination_reason": checker.get("termination_reason"),
+ "checker_termination_reason": checker.get("termination_reason"),
+ "cleanup": checker.get("cleanup"),
+ })
+ checker_message = checker.get("message") or ""
+ if checker_status == "FAIL":
+ if checker.get("execution_status") == "output_control_error":
+ checker_message = f"checker output control failed: {checker_message}"
+ elif checker.get("execution_status") == "start_error":
+ checker_message = f"checker failed to start: {checker_message}"
+ elif checker.get("termination_reason") != "completed":
+ checker_message = f"checker {checker_message}"
+ return (
+ checker_status,
+ elapsed,
+ memory,
+ memory_enforced,
+ checker_message,
+ details,
)
- if termination_reason:
- details["termination_reason"] = termination_reason
- if timeout_kind:
- details["timeout_kind"] = timeout_kind
- return status, elapsed, memory, memory_enforced, message, details
-
-
-def _interactive_output_status(
- traffic,
- solution_stderr,
- interactor_stderr,
- feedback_dir=None,
- diagnostic_limit_bytes=MAX_TOOL_DIAGNOSTIC_BYTES,
-):
- """Classify combined protocol/stderr output after every lifecycle edge."""
- solution_stderr_size = output_path_size(solution_stderr, required=True)[0]
- interactor_stderr_size = output_path_size(interactor_stderr, required=True)[0]
- feedback_size = 0
- if feedback_dir:
- for name in ("judgemessage.txt", "teammessage.txt"):
- feedback_size += output_path_size(
- os.path.join(feedback_dir, name)
- )[0]
-
- lock = traffic.get("_lock")
- if lock is None:
- limit = int(traffic["limit"])
- interactor_protocol_bytes = traffic.get("interactor_to_solution", 0)
- solution_protocol_bytes = traffic.get("solution_to_interactor", 0)
- else:
- with lock:
- limit = int(traffic["limit"])
- interactor_protocol_bytes = traffic.get("interactor_to_solution", 0)
- solution_protocol_bytes = traffic.get("solution_to_interactor", 0)
- interactor_diagnostic_bytes = interactor_stderr_size + feedback_size
- solution_bytes = solution_protocol_bytes + solution_stderr_size
- if interactor_protocol_bytes > limit:
- return "FAIL", "interactor output limit exceeded"
- if interactor_diagnostic_bytes > int(diagnostic_limit_bytes):
- return "FAIL", "interactor diagnostic output limit exceeded"
- if solution_bytes > limit:
- return "OLE", "interactive output limit exceeded"
- return None, ""
-
-
-def _remove_interactive_temp(path):
- """Best-effort cleanup for Windows handles that may close asynchronously."""
- if not path:
- return
- for _ in range(20):
- try:
- shutil.rmtree(path)
- return
- except FileNotFoundError:
- return
- except PermissionError:
- time.sleep(0.05)
- except OSError:
- return
+ finally:
+ _remove_file_with_retries(output_file)
def run_interactive_testcase(
@@ -1255,285 +1092,43 @@ def run_interactive_testcase(
process_limit=DEFAULT_PROCESS_LIMIT,
):
"""Connect contestant and interactor with transcript, idle and total deadlines."""
- start = time.time()
- monotonic_start = time.monotonic()
- idle_limit = max(float(idle_limit if idle_limit is not None else min(time_limit, 2.0)), 0.1)
- state_lock = threading.RLock()
- transcript = {
- "entries": [],
- "bytes": 0,
- "limit": max(int(transcript_limit), 0),
- "truncated": False,
- "_lock": state_lock,
+ result = execute_interactive_session(
+ [bin_path],
+ [interactor_bin],
+ in_file,
+ ans_file,
+ work_dir=os.path.dirname(bin_path),
+ time_limit=time_limit,
+ memory_limit_mb=memory_limit,
+ idle_limit=idle_limit,
+ transcript_limit=transcript_limit,
+ output_limit_bytes=int(output_limit * 1024 * 1024),
+ process_limit=process_limit,
+ )
+ details = {
+ "transcript": result.get("transcript", []),
+ "transcript_truncated": result.get("transcript_truncated", False),
+ "output_bytes": result.get("output_bytes"),
+ "termination_reason": result.get("termination_reason"),
+ "verdict": result.get("verdict"),
+ "execution_status": result.get("execution_status"),
+ "failure_kind": result.get("failure_kind"),
+ "actor": result.get("actor"),
+ "cleanup": result.get("cleanup"),
+ "exit_codes": result.get("exit_codes"),
+ "traffic": result.get("traffic"),
+ "resources": result.get("resources"),
}
- activity = {"last": monotonic_start}
- traffic = {"limit": int(output_limit * 1024 * 1024), "_lock": state_lock}
- last_resource_sample = monotonic_start
- solution_managed = None
- interactor_managed = None
- memory_enforced = False
- peak_memory_mb = None
- threads = []
- runtime_dir = tempfile.mkdtemp(prefix=".probhub-interactive-", dir=os.path.dirname(bin_path))
- solution_stderr = os.path.join(runtime_dir, "solution.stderr")
- interactor_stderr = os.path.join(runtime_dir, "interactor.stderr")
- feedback_dir = os.path.join(runtime_dir, "feedback")
- os.mkdir(feedback_dir)
- diagnostic_limit_bytes = min(
- int(output_limit * 1024 * 1024), MAX_TOOL_DIAGNOSTIC_BYTES
+ if result.get("timeout_kind"):
+ details["timeout_kind"] = result["timeout_kind"]
+ return (
+ result.get("status") or "FAIL",
+ float(result.get("time") or 0.0),
+ result.get("memory"),
+ bool(result.get("memory_enforced")),
+ result.get("message") or "",
+ details,
)
- try:
- with open(solution_stderr, "wb") as solution_err, open(interactor_stderr, "wb") as interactor_err:
- solution_managed = spawn_managed(
- [bin_path],
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=solution_err,
- bufsize=0,
- memory_limit_mb=memory_limit,
- process_limit=process_limit,
- )
- memory_enforced = solution_managed.memory_enforced
- solution = solution_managed.proc
- interactor_managed = spawn_managed(
- [interactor_bin, in_file, ans_file, feedback_dir],
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=interactor_err,
- bufsize=0,
- memory_limit_mb=memory_limit,
- process_limit=process_limit,
- )
- interactor = interactor_managed.proc
-
- threads = [
- threading.Thread(
- target=_pump_interactive_stream,
- args=(solution.stdout, interactor.stdin, "solution_to_interactor", transcript, activity, traffic),
- daemon=True,
- ),
- threading.Thread(
- target=_pump_interactive_stream,
- args=(interactor.stdout, solution.stdin, "interactor_to_solution", transcript, activity, traffic),
- daemon=True,
- ),
- ]
- for thread in threads:
- thread.start()
- # Process creation and Windows Job setup are infrastructure time, not
- # contestant execution or protocol idleness. Start both clocks once
- # both pumps can observe traffic so a slow runner cannot consume the
- # formal TL before the interactive session is ready.
- monotonic_start = time.monotonic()
- start = time.time()
- activity["last"] = max(activity["last"], monotonic_start)
- last_resource_sample = monotonic_start
-
- deadline = monotonic_start + float(time_limit)
- while solution.poll() is None or interactor.poll() is None:
- if cancellation_requested():
- raise ProcessCancelled("execution cancelled")
- now = time.monotonic()
- resource_status, resource_message = _interactive_output_status(
- traffic,
- solution_stderr,
- interactor_stderr,
- feedback_dir,
- diagnostic_limit_bytes,
- )
- if now - last_resource_sample >= 0.05:
- last_resource_sample = now
- solution_count, solution_memory = solution_managed.sample()
- peak_memory_mb = solution_managed.peak_memory_mb
- if solution_memory is not None and solution_memory >= memory_limit:
- resource_status, resource_message = "MLE", "memory limit exceeded"
- elif solution_count is not None and solution_count > process_limit:
- resource_status, resource_message = "RE", "process limit exceeded"
- interactor_count, interactor_memory = interactor_managed.sample()
- if interactor_memory is not None and interactor_memory >= memory_limit:
- resource_status, resource_message = "FAIL", "interactor memory limit exceeded"
- elif interactor_count is not None and interactor_count > process_limit:
- resource_status, resource_message = "FAIL", "interactor process limit exceeded"
- elif interactor.poll() is not None:
- interactor_status, interactor_message = _checker_result(
- interactor.returncode, ""
- )
- if interactor_status == "FAIL":
- resource_status = "FAIL"
- resource_message = interactor_message
- if resource_status:
- solution_managed.terminate()
- solution_managed = None
- interactor_managed.terminate()
- interactor_managed = None
- for thread in threads:
- thread.join(timeout=1)
- return _interactive_result(
- resource_status,
- time.time() - start,
- peak_memory_mb,
- memory_enforced,
- resource_message,
- transcript,
- traffic=traffic,
- solution_stderr=solution_stderr,
- termination_reason=(
- "memory_limit" if resource_status == "MLE"
- else "output_limit" if resource_status == "OLE"
- else None
- ),
- )
- timeout_kind = None
- if now >= deadline:
- timeout_kind = "total"
- else:
- with state_lock:
- idle_elapsed = now - activity["last"]
- if idle_elapsed >= idle_limit:
- timeout_kind = "idle"
- if timeout_kind:
- solution_managed.sample()
- peak_memory_mb = solution_managed.peak_memory_mb
- solution_managed.terminate()
- solution_managed = None
- interactor_managed.terminate()
- interactor_managed = None
- for thread in threads:
- thread.join(timeout=1)
- message = (
- "interactive idle timeout exceeded"
- if timeout_kind == "idle"
- else "interactive time limit exceeded"
- )
- elapsed = min(time.time() - start, float(time_limit))
- return _interactive_result(
- "TLE",
- elapsed,
- peak_memory_mb,
- memory_enforced,
- message,
- transcript,
- timeout_kind=timeout_kind,
- traffic=traffic,
- solution_stderr=solution_stderr,
- termination_reason="time_limit",
- )
- time.sleep(0.005)
-
- # Both direct processes may exit while descendants keep their pipes
- # or diagnostic files open. Preserve telemetry, terminate both full
- # trees, then classify the stable final byte counts.
- solution_managed.sample()
- peak_memory_mb = solution_managed.peak_memory_mb
- interactor_managed.sample()
- interactor_peak_memory = interactor_managed.peak_memory_mb
- solution_managed.terminate()
- interactor_managed.terminate()
- for thread in threads:
- thread.join(timeout=1)
- resource_status, resource_message = _interactive_output_status(
- traffic,
- solution_stderr,
- interactor_stderr,
- feedback_dir,
- diagnostic_limit_bytes,
- )
- if (
- interactor.returncode != 0
- and interactor_managed.memory_enforced
- and interactor_peak_memory is not None
- and interactor_peak_memory >= memory_limit * 0.98
- ):
- resource_status = "FAIL"
- resource_message = "interactor memory limit exceeded"
-
- elapsed = time.time() - start
- solution_message = ""
- interactor_message = ""
- try:
- solution_message = read_bounded_text(
- solution_stderr, int(output_limit * 1024 * 1024)
- )["text"].strip()
- except OSError:
- pass
- try:
- interactor_message = read_bounded_text(
- interactor_stderr, diagnostic_limit_bytes
- )["text"].strip()
- except OSError:
- pass
- interactor_message = _feedback_message(
- feedback_dir, interactor_message, diagnostic_limit_bytes
- )
-
- if resource_status:
- return _interactive_result(
- resource_status,
- elapsed,
- peak_memory_mb,
- memory_enforced,
- resource_message,
- transcript,
- traffic=traffic,
- solution_stderr=solution_stderr,
- termination_reason=(
- "memory_limit" if resource_status == "MLE"
- else "output_limit" if resource_status == "OLE"
- else None
- ),
- )
-
- if solution.returncode != 0:
- status = _failed_status(
- solution.returncode,
- solution_message,
- memory_enforced,
- peak_memory_mb,
- memory_limit,
- )
- return _interactive_result(
- status,
- elapsed,
- peak_memory_mb,
- memory_enforced,
- solution_message,
- transcript,
- traffic=traffic,
- solution_stderr=solution_stderr,
- termination_reason=(
- "inferred_memory_limit" if status == "MLE" else None
- ),
- )
- interactor_status, message = _checker_result(interactor.returncode, interactor_message)
- return _interactive_result(
- interactor_status,
- elapsed,
- peak_memory_mb,
- memory_enforced,
- message,
- transcript,
- traffic=traffic,
- solution_stderr=solution_stderr,
- )
- except OSError as exc:
- return _interactive_result(
- "FAIL",
- time.time() - start,
- peak_memory_mb,
- memory_enforced,
- str(exc),
- transcript,
- traffic=traffic,
- solution_stderr=solution_stderr,
- )
- finally:
- if solution_managed is not None:
- solution_managed.terminate()
- if interactor_managed is not None:
- interactor_managed.terminate()
- for thread in threads:
- thread.join(timeout=1)
- _remove_interactive_temp(runtime_dir)
def run_testcase(
bin_path,
@@ -2683,6 +2278,11 @@ def _main_unlocked():
transcript_truncated=judge_details.get("transcript_truncated", False),
output_bytes=judge_details.get("output_bytes"),
termination_reason=judge_details.get("termination_reason"),
+ verdict=judge_details.get("verdict"),
+ execution_status=judge_details.get("execution_status"),
+ failure_kind=judge_details.get("failure_kind"),
+ actor=judge_details.get("actor"),
+ cleanup=judge_details.get("cleanup"),
cached=cached,
)
case_results.append({
@@ -2695,6 +2295,11 @@ def _main_unlocked():
"memory_enforced": memory_enforced,
"output_bytes": judge_details.get("output_bytes"),
"termination_reason": judge_details.get("termination_reason"),
+ "verdict": judge_details.get("verdict"),
+ "execution_status": judge_details.get("execution_status"),
+ "failure_kind": judge_details.get("failure_kind"),
+ "actor": judge_details.get("actor"),
+ "cleanup": judge_details.get("cleanup"),
"cached": cached,
})
if requires_sample_answer:
diff --git a/tests/test_calibration.py b/tests/test_calibration.py
index 5015c30..93d1df8 100644
--- a/tests/test_calibration.py
+++ b/tests/test_calibration.py
@@ -21,6 +21,7 @@
from probhub.cli import _ensure_local_gitignore
from probhub.io import write_yaml
from probhub.linting import lint_problem, problem_status
+from probhub.special_judges import _interactive_result
ROOT = Path(__file__).resolve().parents[1]
@@ -512,20 +513,27 @@ def test_interactive_result_records_solution_output_bytes(self):
with tempfile.TemporaryDirectory() as temp:
stderr = Path(temp) / "solution.stderr"
stderr.write_bytes(b"12345")
- transcript = {"entries": [], "truncated": False}
- result = LOCAL_JUDGE._interactive_result(
- "OLE",
- 0.1,
- 1,
- True,
- "output limit exceeded",
- transcript,
- traffic={"solution_to_interactor": 7},
- solution_stderr=str(stderr),
- termination_reason="output_limit",
+ transcript = {"entries": [], "bytes": 0, "truncated": False}
+ result = _interactive_result(
+ {
+ "status": "OLE",
+ "verdict": "OLE",
+ "execution_status": "output_limit",
+ "failure_kind": "resource_limit",
+ "actor": "contestant",
+ "termination_reason": "output_limit",
+ "message": "output limit exceeded",
+ },
+ elapsed=0.1,
+ memory=1,
+ memory_enforced=True,
+ transcript=transcript,
+ traffic_evidence={"solution_to_interactor": 7, "solution_stderr": 5},
+ cleanup={"ok": True},
+ exit_codes={},
)
- self.assertEqual(result[-1]["output_bytes"], 12)
- self.assertEqual(result[-1]["termination_reason"], "output_limit")
+ self.assertEqual(result["output_bytes"], 12)
+ self.assertEqual(result["termination_reason"], "output_limit")
def test_tle_probe_distinguishes_exact_and_censored_measurements(self):
with tempfile.TemporaryDirectory() as temp:
diff --git a/tests/test_core.py b/tests/test_core.py
index 69e3542..da79ed2 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -502,6 +502,19 @@ def test_lint_validates_custom_checker_and_interactor_configuration(self):
(problem / "code/checker.cpp").write_text("int main(){}\n", encoding="utf-8")
self.assertTrue(lint_workspace(root, workspace)["ok"])
+ config["judge"] = {
+ "type": "checker",
+ "validator": "code/validator.cpp",
+ "checker": "../outside-checker.cpp",
+ }
+ write_yaml(config_path, config)
+ result = lint_workspace(root, workspace)
+ self.assertFalse(result["ok"])
+ self.assertIn(
+ "judge.checker must stay inside the problem directory: ../outside-checker.cpp",
+ result["problems"][0]["errors"],
+ )
+
config["judge"] = {
"type": "interactive",
"validator": "code/validator.cpp",
@@ -515,6 +528,58 @@ def test_lint_validates_custom_checker_and_interactor_configuration(self):
(problem / "code/interactor.cpp").write_text("int main(){}\n", encoding="utf-8")
self.assertTrue(lint_workspace(root, workspace)["ok"])
+ config["judge"]["interactor"] = r"..\outside-interactor.cpp"
+ write_yaml(config_path, config)
+ result = lint_workspace(root, workspace)
+ self.assertFalse(result["ok"])
+ self.assertIn(
+ r"judge.interactor must stay inside the problem directory: ..\outside-interactor.cpp",
+ result["problems"][0]["errors"],
+ )
+
+ checker_link = problem / "code/checker-link.cpp"
+ try:
+ checker_link.symlink_to(problem / "code/checker.cpp")
+ except (NotImplementedError, OSError):
+ checker_link = None
+ if checker_link is not None:
+ config["judge"] = {
+ "type": "checker",
+ "validator": "code/validator.cpp",
+ "checker": "code/checker-link.cpp",
+ }
+ write_yaml(config_path, config)
+ result = lint_workspace(root, workspace)
+ self.assertFalse(result["ok"])
+ self.assertIn(
+ "judge.checker must be a regular non-symlink file: code/checker-link.cpp",
+ result["problems"][0]["errors"],
+ )
+
+ interactor_link = problem / "code/interactor-link.cpp"
+ try:
+ interactor_link.symlink_to(problem / "code/interactor.cpp")
+ except (NotImplementedError, OSError):
+ interactor_link = None
+ if interactor_link is not None:
+ config["judge"] = {
+ "type": "interactive",
+ "validator": "code/validator.cpp",
+ "interactor": "code/interactor-link.cpp",
+ }
+ write_yaml(config_path, config)
+ result = lint_workspace(root, workspace)
+ self.assertFalse(result["ok"])
+ self.assertIn(
+ "judge.interactor must be a regular non-symlink file: code/interactor-link.cpp",
+ result["problems"][0]["errors"],
+ )
+
+ config["judge"] = {
+ "type": "interactive",
+ "validator": "code/validator.cpp",
+ "interactor": "code/interactor.cpp",
+ }
config["judge"]["interactive"] = {"idle_limit": 0, "transcript_limit": -1}
write_yaml(config_path, config)
result = lint_workspace(root, workspace)
diff --git a/tests/test_datagen.py b/tests/test_datagen.py
index 17b234a..3c7bf72 100644
--- a/tests/test_datagen.py
+++ b/tests/test_datagen.py
@@ -500,14 +500,12 @@ def run(command, _input, *_args, **_kwargs):
return {"status": "AC", "stdout": stdout, "stderr": b"", "message": ""}
outcomes = (
- ({"status": "WA", "match": False, "message": "rejected", "stderr": b""}, "WA", None),
- ({"status": "FAIL", "match": False, "message": "checker failed", "stderr": b""}, "FAIL", None),
+ ({"verdict": "WA", "execution_status": "completed", "message": "rejected"}, "WA", None),
+ ({"verdict": None, "execution_status": "completed", "message": "checker failed"}, "FAIL", None),
({
- "status": "FAIL",
- "execution_status": "TLE",
- "match": False,
+ "verdict": None,
+ "execution_status": "time_limit",
"message": "checker timed out",
- "stderr": b"",
}, "FAIL", "TLE"),
)
for comparison, expected_status, execution_status in outcomes:
@@ -516,7 +514,7 @@ def run(command, _input, *_args, **_kwargs):
with (
mock.patch("probhub.datagen._prepare_program", side_effect=prepare),
mock.patch("probhub.datagen._run", side_effect=run),
- mock.patch("probhub.datagen._compare_custom", return_value=comparison),
+ mock.patch("probhub.datagen.run_checker_to_files", return_value=comparison),
):
result = generate_problem_data(problem, config, apply_changes=True)
self.assertFalse(result["ok"])
@@ -529,18 +527,25 @@ def run(command, _input, *_args, **_kwargs):
with tempfile.TemporaryDirectory() as temp:
problem, config = fixture(Path(temp))
- accepted = {"status": "AC", "match": True, "message": "", "stderr": b""}
+ accepted = {"verdict": "AC", "execution_status": "completed", "message": ""}
+
+ def accept_checker(*args, **_kwargs):
+ self.assertEqual(Path(args[3]).read_bytes(), b"2\n")
+ return accepted
+
with (
mock.patch("probhub.datagen._prepare_program", side_effect=prepare),
mock.patch("probhub.datagen._run", side_effect=run),
- mock.patch("probhub.datagen._compare_custom", return_value=accepted) as checker,
+ mock.patch(
+ "probhub.datagen.run_checker_to_files", side_effect=accept_checker
+ ) as checker,
):
result = generate_problem_data(problem, config, apply_changes=True)
self.assertTrue(result["ok"])
self.assertTrue(result["applied"])
self.assertEqual((problem / "data/secret/gen01.ans").read_bytes(), b"2\n")
self.assertTrue((problem / GEN_MANIFEST_PATH).is_file())
- self.assertEqual(checker.call_args.args[3], b"2\n")
+ self.assertEqual(checker.call_args.args[2], checker.call_args.args[3])
def test_lint_reports_recipe_errors_and_coverage_warnings(self):
with tempfile.TemporaryDirectory() as temp:
diff --git a/tests/test_local_judge_layout.py b/tests/test_local_judge_layout.py
index e55766b..20a4acd 100644
--- a/tests/test_local_judge_layout.py
+++ b/tests/test_local_judge_layout.py
@@ -88,6 +88,30 @@ def test_schema_paths_resolve_inside_code_directory(self):
cases = MODULE.collect_testcases(problem, config)
self.assertEqual([case["case"] for case in cases], ["sample/1"])
+ def test_schema_execution_paths_use_the_shared_problem_fence(self):
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ problem = root / "A"
+ code = problem / "code"
+ code.mkdir(parents=True)
+ outside = root / "outside.cpp"
+ outside.write_text("int main(){}\n", encoding="utf-8")
+ (code / "checker.cpp").write_text("int main(){}\n", encoding="utf-8")
+
+ self.assertIsNone(MODULE.resolve_problem_path(problem, "../outside.cpp"))
+ self.assertIsNone(MODULE.resolve_problem_path(problem, str(outside)))
+ self.assertEqual(
+ Path(MODULE.resolve_problem_path(problem, "code/checker.cpp")),
+ (code / "checker.cpp").absolute(),
+ )
+
+ link = code / "linked-checker.cpp"
+ try:
+ link.symlink_to(outside)
+ except OSError:
+ return
+ self.assertIsNone(MODULE.resolve_problem_path(problem, "code/linked-checker.cpp"))
+
def test_legacy_workspace_prefers_code_directory_when_present(self):
with tempfile.TemporaryDirectory() as temp:
diff --git a/tests/test_package_tools.py b/tests/test_package_tools.py
index ce9ffe2..e219ccf 100644
--- a/tests/test_package_tools.py
+++ b/tests/test_package_tools.py
@@ -587,6 +587,43 @@ def create_checker_problem(self, root):
"judge": {"type": "checker", "checker": "code/checker.cpp"},
}
+ def test_output_validator_rejects_unsafe_checker_sources(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp) / "A"
+ config = self.create_checker_problem(problem)
+ for source in (None, "", " "):
+ with self.subTest(source=source):
+ config["judge"]["checker"] = source
+ with self.assertRaises(ProbHubError) as raised:
+ validate_output_validator_source(problem, config)
+ self.assertEqual(raised.exception.code, "unsafe_package_source")
+ self.assertEqual(str(raised.exception), "judge.checker is required")
+
+ for source in ("../outside.cpp", r"..\outside.cpp", r"C:relative.cpp"):
+ with self.subTest(source=source):
+ config["judge"]["checker"] = source
+ with self.assertRaises(ProbHubError) as raised:
+ validate_output_validator_source(problem, config)
+ self.assertEqual(raised.exception.code, "unsafe_package_source")
+ self.assertIn(
+ "judge.checker must stay inside the problem directory",
+ str(raised.exception),
+ )
+
+ checker_link = problem / "code/checker-link.cpp"
+ try:
+ checker_link.symlink_to(problem / "code/checker.cpp")
+ except (NotImplementedError, OSError):
+ return
+ config["judge"]["checker"] = "code/checker-link.cpp"
+ with self.assertRaises(ProbHubError) as raised:
+ validate_output_validator_source(problem, config)
+ self.assertEqual(raised.exception.code, "unsafe_package_source")
+ self.assertEqual(
+ str(raised.exception),
+ "judge.checker must be a regular file: code/checker-link.cpp",
+ )
+
def test_output_validator_compile_uses_managed_limits(self):
with tempfile.TemporaryDirectory() as temp:
problem = Path(temp) / "A"
diff --git a/tests/test_problem_paths.py b/tests/test_problem_paths.py
new file mode 100644
index 0000000..469c02e
--- /dev/null
+++ b/tests/test_problem_paths.py
@@ -0,0 +1,155 @@
+import os
+import stat
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+from probhub.problem_paths import ProblemPathError, resolve_problem_regular_file
+
+
+class ProblemPathTests(unittest.TestCase):
+ def assert_reason(self, problem_dir, value, reason):
+ with self.assertRaises(ProblemPathError) as raised:
+ resolve_problem_regular_file(problem_dir, value)
+ self.assertEqual(raised.exception.reason, reason)
+
+ def test_resolves_nested_file_with_either_separator(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp) / "problem"
+ source = problem / "code" / "checker.cpp"
+ source.parent.mkdir(parents=True)
+ source.write_text("int main() {}\n", encoding="utf-8")
+
+ expected = source.absolute()
+ self.assertEqual(
+ resolve_problem_regular_file(problem, "code/checker.cpp"),
+ expected,
+ )
+ self.assertEqual(
+ resolve_problem_regular_file(problem, r"code\checker.cpp"),
+ expected,
+ )
+
+ def test_preserves_access_spelling_after_canonical_containment_check(self):
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ access_problem = root / "access" / "problem"
+ canonical_problem = root / "canonical" / "problem"
+ access_source = access_problem / "code" / "checker.cpp"
+ canonical_source = canonical_problem / "code" / "checker.cpp"
+ for source in (access_source, canonical_source):
+ source.parent.mkdir(parents=True)
+ source.write_text("int main() {}\n", encoding="utf-8")
+
+ path_type = type(access_problem)
+ real_resolve = path_type.resolve
+
+ def canonicalized(path, strict=False):
+ if path == access_problem:
+ return canonical_problem
+ if path == access_source:
+ return canonical_source
+ return real_resolve(path, strict=strict)
+
+ with mock.patch.object(
+ path_type,
+ "resolve",
+ autospec=True,
+ side_effect=canonicalized,
+ ):
+ self.assertEqual(
+ resolve_problem_regular_file(
+ access_problem, "code/checker.cpp"
+ ),
+ access_source,
+ )
+
+ def test_rejects_invalid_values(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp)
+ for value in (None, "", " ", Path("code/checker.cpp")):
+ with self.subTest(value=value):
+ self.assert_reason(problem, value, "invalid")
+
+ def test_rejects_host_and_cross_platform_absolute_or_driven_paths(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp) / "problem"
+ problem.mkdir()
+ outside = str((problem.parent / "outside.cpp").resolve())
+ for value in (
+ outside,
+ "/absolute/checker.cpp",
+ r"C:\absolute\checker.cpp",
+ r"\\server\share\checker.cpp",
+ r"C:relative.cpp",
+ ):
+ with self.subTest(value=value):
+ self.assert_reason(problem, value, "outside")
+
+ def test_rejects_parent_components_with_either_separator(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp) / "problem"
+ problem.mkdir()
+ for value in ("../outside.cpp", r"..\outside.cpp", "code/../checker.cpp"):
+ with self.subTest(value=value):
+ self.assert_reason(problem, value, "outside")
+
+ def test_distinguishes_missing_and_non_regular_targets(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp) / "problem"
+ code = problem / "code"
+ code.mkdir(parents=True)
+
+ self.assert_reason(problem, "code/missing.cpp", "missing")
+ self.assert_reason(problem, "code", "non_regular")
+ blocker = problem / "blocker"
+ blocker.write_text("not a directory\n", encoding="utf-8")
+ self.assert_reason(problem, "blocker/checker.cpp", "non_regular")
+
+ def test_rejects_symlinks_to_inside_and_outside(self):
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ problem = root / "problem"
+ code = problem / "code"
+ code.mkdir(parents=True)
+ inside = code / "checker.cpp"
+ inside.write_text("int main() {}\n", encoding="utf-8")
+ outside = root / "outside.cpp"
+ outside.write_text("int main() {}\n", encoding="utf-8")
+ inside_link = code / "inside-link.cpp"
+ outside_link = code / "outside-link.cpp"
+ try:
+ inside_link.symlink_to(inside)
+ outside_link.symlink_to(outside)
+ except (NotImplementedError, OSError) as exc:
+ self.skipTest(f"symlink creation is unavailable: {exc}")
+
+ self.assert_reason(problem, "code/inside-link.cpp", "link")
+ self.assert_reason(problem, "code/outside-link.cpp", "link")
+
+ def test_rejects_mocked_reparse_point_component(self):
+ with tempfile.TemporaryDirectory() as temp:
+ problem = Path(temp) / "problem"
+ code = problem / "code"
+ code.mkdir(parents=True)
+ (code / "checker.cpp").write_text("int main() {}\n", encoding="utf-8")
+ real_lstat = os.lstat
+ reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)
+
+ def mocked_lstat(path):
+ info = real_lstat(path)
+ if Path(path) == code:
+ return SimpleNamespace(
+ st_mode=info.st_mode,
+ st_file_attributes=reparse_flag,
+ )
+ return info
+
+ with mock.patch("probhub.problem_paths.os.lstat", side_effect=mocked_lstat):
+ self.assert_reason(problem, "code/checker.cpp", "link")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_special_judges.py b/tests/test_special_judges.py
index d33c68f..1196a99 100644
--- a/tests/test_special_judges.py
+++ b/tests/test_special_judges.py
@@ -16,6 +16,7 @@
import yaml
+import probhub.special_judges as SPECIAL_JUDGES
from probhub.process_control import (
process_alive,
spawn_managed,
@@ -65,7 +66,7 @@ def test_custom_checker_output_control_failure_has_precise_reason(self):
"run_program_to_file",
return_value=("AC", 0.01, 1, True, "", {"output_bytes": 0}),
), mock.patch.object(
- JUDGE_MODULE,
+ SPECIAL_JUDGES,
"run_managed_to_files",
side_effect=JUDGE_MODULE.OutputBudgetError("feedback is not regular"),
):
@@ -81,6 +82,45 @@ def test_custom_checker_output_control_failure_has_precise_reason(self):
self.assertNotIn("failed to start", message)
self.assertEqual(details["checker_termination_reason"], "output_control_error")
+ def test_custom_contestant_failure_fields_are_normalized_before_checker(self):
+ root = Path(tempfile.mkdtemp())
+ self.addCleanup(shutil.rmtree, root, True)
+ cases = (
+ (
+ ("MLE", 0.01, 255, True, "", {
+ "output_bytes": 0,
+ "termination_reason": "inferred_memory_limit",
+ }),
+ "memory_limit",
+ "resource_limit",
+ "inferred_memory_limit",
+ ),
+ (
+ ("RE", 0.0, None, False, "missing", {"output_bytes": 0}),
+ "start_error",
+ "startup_failure",
+ "start_error",
+ ),
+ )
+ for program_result, execution_status, failure_kind, termination_reason in cases:
+ with self.subTest(execution_status=execution_status), mock.patch.object(
+ JUDGE_MODULE, "run_program_to_file", return_value=program_result
+ ), mock.patch.object(
+ JUDGE_MODULE, "run_checker_to_files"
+ ) as checker:
+ result = JUDGE_MODULE.run_custom_testcase(
+ str(root / "solution"),
+ str(root / "checker"),
+ str(root / "case.in"),
+ str(root / "case.ans"),
+ )
+ details = result[5]
+ self.assertEqual(details["execution_status"], execution_status)
+ self.assertEqual(details["failure_kind"], failure_kind)
+ self.assertEqual(details["termination_reason"], termination_reason)
+ self.assertEqual(details["actor"], "contestant")
+ checker.assert_not_called()
+
def test_interactive_final_classification_follows_tree_termination(self):
class FakeProcess:
def __init__(self):
@@ -118,7 +158,7 @@ def fake_spawn(*_args, **kwargs):
return managed
with tempfile.TemporaryDirectory() as temp, mock.patch.object(
- JUDGE_MODULE, "spawn_managed", side_effect=fake_spawn
+ SPECIAL_JUDGES, "spawn_managed", side_effect=fake_spawn
):
root = Path(temp)
result = JUDGE_MODULE.run_interactive_testcase(
@@ -145,7 +185,7 @@ def test_interactive_transcript_decodes_utf8_across_chunks(self):
traffic = {"limit": 1024, "_lock": lock}
destination = self._NonClosingBytesIO()
- JUDGE_MODULE._pump_interactive_stream(
+ SPECIAL_JUDGES._pump_interactive_stream(
self._ChunkedSource([b"\xe4\xbd", b"\xa0\n"]),
destination,
"interactor_to_solution",
@@ -190,7 +230,7 @@ def __getitem__(self, key):
destinations = [self._NonClosingBytesIO(), self._NonClosingBytesIO()]
threads = [
threading.Thread(
- target=JUDGE_MODULE._pump_interactive_stream,
+ target=SPECIAL_JUDGES._pump_interactive_stream,
args=(
self._ChunkedSource([b"abcdefgh"]),
destinations[index],
@@ -234,7 +274,7 @@ def spawn(command, **kwargs):
raise OSError("interactor containment failed")
with tempfile.TemporaryDirectory() as temp, mock.patch.object(
- JUDGE_MODULE, "spawn_managed", side_effect=spawn
+ SPECIAL_JUDGES, "spawn_managed", side_effect=spawn
):
root = Path(temp)
result = JUDGE_MODULE.run_interactive_testcase(
@@ -262,11 +302,14 @@ def test_interactive_output_status_combines_protocol_and_stderr(self):
"solution_to_interactor": 700,
"interactor_to_solution": 0,
}
- status, message = JUDGE_MODULE._interactive_output_status(
+ evidence = SPECIAL_JUDGES._interactive_evidence(
traffic, solution_stderr, interactor_stderr
)
- self.assertEqual(status, "OLE")
- self.assertIn("output limit", message)
+ outcome = SPECIAL_JUDGES._interactive_output_classification(
+ evidence, SPECIAL_JUDGES.MAX_CHECKER_DIAGNOSTIC_BYTES
+ )
+ self.assertEqual(outcome["status"], "OLE")
+ self.assertIn("output limit", outcome["message"])
def test_interactive_feedback_uses_bounded_diagnostic_budget(self):
with tempfile.TemporaryDirectory() as temp:
@@ -283,38 +326,347 @@ def test_interactive_feedback_uses_bounded_diagnostic_budget(self):
"solution_to_interactor": 0,
"interactor_to_solution": 0,
}
- status, message = JUDGE_MODULE._interactive_output_status(
+ evidence = SPECIAL_JUDGES._interactive_evidence(
traffic,
solution_stderr,
interactor_stderr,
feedback,
- diagnostic_limit_bytes=1000,
)
- self.assertEqual(status, "FAIL")
- self.assertIn("diagnostic output limit", message)
+ outcome = SPECIAL_JUDGES._interactive_output_classification(evidence, 1000)
+ self.assertEqual(outcome["status"], "FAIL")
+ self.assertIn("diagnostic output limit", outcome["message"])
(feedback / "judgemessage.txt").unlink()
(feedback / "judgemessage.txt").mkdir()
with self.assertRaisesRegex(
JUDGE_MODULE.OutputBudgetError, "not a regular file"
):
- JUDGE_MODULE._interactive_output_status(
+ SPECIAL_JUDGES._interactive_evidence(
traffic,
solution_stderr,
interactor_stderr,
feedback,
- diagnostic_limit_bytes=1000,
)
+ def test_interactive_non_regular_feedback_is_structured_control_failure(self):
+ class FakeProcess:
+ def __init__(self):
+ self.stdin = io.BytesIO()
+ self.stdout = io.BytesIO()
+ self.returncode = 0
+
+ def poll(self):
+ return self.returncode
+
+ class FakeManaged:
+ memory_enforced = True
+ process_limit_enforced = True
+ peak_memory_mb = 1
+
+ def __init__(self):
+ self.proc = FakeProcess()
+
+ def sample(self):
+ return 1, 1
+
+ def terminate(self):
+ pass
+
+ spawned = []
+
+ def fake_spawn(command, **_kwargs):
+ managed = FakeManaged()
+ if spawned:
+ Path(command[-1], "judgemessage.txt").mkdir()
+ spawned.append(managed)
+ return managed
+
+ with tempfile.TemporaryDirectory() as temp, mock.patch.object(
+ SPECIAL_JUDGES, "spawn_managed", side_effect=fake_spawn
+ ):
+ root = Path(temp)
+ result = SPECIAL_JUDGES.execute_interactive_session(
+ [str(root / "solution")],
+ [str(root / "interactor")],
+ root / "case.in",
+ root / "case.ans",
+ work_dir=root,
+ )
+ self.assertEqual(result["status"], "FAIL")
+ self.assertEqual(result["execution_status"], "output_control_error")
+ self.assertEqual(result["failure_kind"], "control_failure")
+ self.assertEqual(result["actor"], "supervisor")
+ self.assertTrue(result["cleanup"]["ok"])
+
def test_interactive_temp_cleanup_retries_permission_error(self):
with mock.patch.object(
- JUDGE_MODULE.shutil,
+ SPECIAL_JUDGES.shutil,
"rmtree",
side_effect=[PermissionError("busy"), None],
- ) as rmtree, mock.patch.object(JUDGE_MODULE.time, "sleep"):
- JUDGE_MODULE._remove_interactive_temp("temporary")
+ ) as rmtree, mock.patch.object(SPECIAL_JUDGES.time, "sleep"):
+ removed, attempts, error = SPECIAL_JUDGES._remove_runtime_directory("temporary")
+ self.assertTrue(removed)
+ self.assertEqual(attempts, 2)
+ self.assertIsNone(error)
self.assertEqual(rmtree.call_count, 2)
+ def test_checker_cleanup_failure_is_structured_infrastructure_failure(self):
+ execution = {
+ "reason": "completed",
+ "returncode": 0,
+ "time": 0.01,
+ "memory": 1,
+ "memory_enforced": True,
+ "process_limit_enforced": True,
+ "output_bytes": 0,
+ "retained_output_bytes": 0,
+ "stdout_retained_bytes": 0,
+ "stderr_retained_bytes": 0,
+ "output_truncated": False,
+ }
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ for name in ("case.in", "case.ans", "contestant.out"):
+ (root / name).write_bytes(b"")
+ with mock.patch.object(
+ SPECIAL_JUDGES, "run_managed_to_files", return_value=execution
+ ), mock.patch.object(
+ SPECIAL_JUDGES.shutil,
+ "rmtree",
+ side_effect=PermissionError("runtime busy"),
+ ) as rmtree, mock.patch.object(SPECIAL_JUDGES.time, "sleep"):
+ result = SPECIAL_JUDGES.run_checker_to_files(
+ ["checker"],
+ root / "case.in",
+ root / "case.ans",
+ root / "contestant.out",
+ timeout=1,
+ cwd=root,
+ )
+ self.assertIsNone(result["verdict"])
+ self.assertEqual(result["actor"], "supervisor")
+ self.assertEqual(result["execution_status"], "cleanup_error")
+ self.assertEqual(result["failure_kind"], "cleanup_failure")
+ self.assertFalse(result["cleanup"]["ok"])
+ self.assertEqual(result["cleanup"]["runtime_remove_attempts"], 20)
+ self.assertEqual(rmtree.call_count, 20)
+
+ def test_checker_cancel_cleanup_failure_overrides_cancellation(self):
+ execution = {
+ "reason": "cancelled",
+ "message": "execution cancelled",
+ "returncode": None,
+ }
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ for name in ("case.in", "case.ans", "contestant.out"):
+ (root / name).write_bytes(b"")
+ with mock.patch.object(
+ SPECIAL_JUDGES, "run_managed_to_files", return_value=execution
+ ), mock.patch.object(
+ SPECIAL_JUDGES.shutil,
+ "rmtree",
+ side_effect=PermissionError("runtime busy"),
+ ), mock.patch.object(SPECIAL_JUDGES.time, "sleep"):
+ result = SPECIAL_JUDGES.run_checker_to_files(
+ ["checker"],
+ root / "case.in",
+ root / "case.ans",
+ root / "contestant.out",
+ timeout=1,
+ cwd=root,
+ )
+ self.assertEqual(result["execution_status"], "cleanup_error")
+ self.assertEqual(result["failure_kind"], "cleanup_failure")
+ self.assertEqual(
+ result["pre_cleanup_result"]["execution_status"], "cancelled"
+ )
+
+ def test_checker_cancel_with_successful_cleanup_propagates(self):
+ execution = {
+ "reason": "cancelled",
+ "message": "execution cancelled",
+ "returncode": None,
+ }
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ for name in ("case.in", "case.ans", "contestant.out"):
+ (root / name).write_bytes(b"")
+ with mock.patch.object(
+ SPECIAL_JUDGES, "run_managed_to_files", return_value=execution
+ ):
+ with self.assertRaisesRegex(
+ JUDGE_MODULE.ProcessCancelled, "execution cancelled"
+ ):
+ SPECIAL_JUDGES.run_checker_to_files(
+ ["checker"],
+ root / "case.in",
+ root / "case.ans",
+ root / "contestant.out",
+ timeout=1,
+ cwd=root,
+ )
+
+ def test_interactive_cleanup_failure_does_not_skip_other_cleanup(self):
+ class FakeProcess:
+ def __init__(self):
+ self.stdin = io.BytesIO()
+ self.stdout = io.BytesIO()
+ self.returncode = 0
+
+ def poll(self):
+ return self.returncode
+
+ class FakeManaged:
+ memory_enforced = True
+ process_limit_enforced = True
+ peak_memory_mb = 1
+
+ def __init__(self, fail=False):
+ self.proc = FakeProcess()
+ self.fail = fail
+ self.terminate_calls = 0
+
+ def sample(self):
+ return 1, 1
+
+ def terminate(self):
+ self.terminate_calls += 1
+ if self.fail:
+ raise OSError("tree cleanup failed")
+
+ managed = [FakeManaged(fail=True), FakeManaged()]
+ with tempfile.TemporaryDirectory() as temp, mock.patch.object(
+ SPECIAL_JUDGES, "spawn_managed", side_effect=managed
+ ):
+ root = Path(temp)
+ result = SPECIAL_JUDGES.execute_interactive_session(
+ [str(root / "solution")],
+ [str(root / "interactor")],
+ root / "case.in",
+ root / "case.ans",
+ work_dir=root,
+ time_limit=1,
+ )
+ self.assertEqual(result["status"], "FAIL")
+ self.assertEqual(result["actor"], "supervisor")
+ self.assertEqual(result["execution_status"], "cleanup_error")
+ self.assertEqual(result["failure_kind"], "cleanup_failure")
+ self.assertFalse(result["cleanup"]["ok"])
+ self.assertEqual(managed[0].terminate_calls, 1)
+ self.assertEqual(managed[1].terminate_calls, 1)
+ self.assertTrue(result["cleanup"]["runtime_removed"])
+
+ def test_interactive_cancel_cleanup_failure_overrides_cancellation(self):
+ class FakeProcess:
+ def __init__(self):
+ self.stdin = io.BytesIO()
+ self.stdout = io.BytesIO()
+ self.returncode = None
+
+ def poll(self):
+ return self.returncode
+
+ class FakeManaged:
+ memory_enforced = True
+ process_limit_enforced = True
+ peak_memory_mb = 1
+
+ def __init__(self):
+ self.proc = FakeProcess()
+
+ def sample(self):
+ return 1, 1
+
+ def terminate(self):
+ self.proc.returncode = -1
+
+ with tempfile.TemporaryDirectory() as temp, mock.patch.object(
+ SPECIAL_JUDGES, "spawn_managed", side_effect=[FakeManaged(), FakeManaged()]
+ ), mock.patch.object(
+ SPECIAL_JUDGES, "cancellation_requested", return_value=True
+ ), mock.patch.object(
+ SPECIAL_JUDGES.shutil,
+ "rmtree",
+ side_effect=PermissionError("runtime busy"),
+ ), mock.patch.object(SPECIAL_JUDGES.time, "sleep"):
+ root = Path(temp)
+ result = SPECIAL_JUDGES.execute_interactive_session(
+ [str(root / "solution")],
+ [str(root / "interactor")],
+ root / "case.in",
+ root / "case.ans",
+ work_dir=root,
+ )
+ self.assertEqual(result["execution_status"], "cleanup_error")
+ self.assertEqual(result["failure_kind"], "cleanup_failure")
+ self.assertEqual(
+ result["pre_cleanup_result"]["execution_status"], "cancelled"
+ )
+
+ def test_interactive_preparation_failure_is_structured_start_error(self):
+ with tempfile.TemporaryDirectory() as temp, mock.patch.object(
+ SPECIAL_JUDGES.Path,
+ "mkdir",
+ side_effect=OSError("feedback directory unavailable"),
+ ):
+ root = Path(temp)
+ result = SPECIAL_JUDGES.execute_interactive_session(
+ [str(root / "solution")],
+ [str(root / "interactor")],
+ root / "case.in",
+ root / "case.ans",
+ work_dir=root,
+ )
+ self.assertEqual(result["status"], "FAIL")
+ self.assertEqual(result["execution_status"], "start_error")
+ self.assertEqual(result["failure_kind"], "startup_failure")
+ self.assertEqual(result["actor"], "supervisor")
+ self.assertTrue(result["cleanup"]["ok"])
+
+ def test_interactor_memory_failure_reports_interactor_memory(self):
+ class FakeProcess:
+ def __init__(self, returncode):
+ self.stdin = io.BytesIO()
+ self.stdout = io.BytesIO()
+ self.returncode = returncode
+
+ def poll(self):
+ return self.returncode
+
+ class FakeManaged:
+ memory_enforced = True
+ process_limit_enforced = True
+
+ def __init__(self, memory, returncode):
+ self.peak_memory_mb = memory
+ self.proc = FakeProcess(returncode)
+
+ def sample(self):
+ return 1, self.peak_memory_mb
+
+ def terminate(self):
+ pass
+
+ managed = [FakeManaged(5, 0), FakeManaged(100, 1)]
+ with tempfile.TemporaryDirectory() as temp, mock.patch.object(
+ SPECIAL_JUDGES, "spawn_managed", side_effect=managed
+ ):
+ root = Path(temp)
+ result = SPECIAL_JUDGES.execute_interactive_session(
+ [str(root / "solution")],
+ [str(root / "interactor")],
+ root / "case.in",
+ root / "case.ans",
+ work_dir=root,
+ memory_limit_mb=50,
+ )
+ self.assertEqual(result["actor"], "interactor")
+ self.assertEqual(result["execution_status"], "memory_limit")
+ self.assertEqual(result["memory"], 100)
+ self.assertEqual(result["resources"]["contestant"]["memory"], 5)
+ self.assertEqual(result["resources"]["interactor"]["memory"], 100)
+
def test_interactive_deadlines_exclude_process_setup(self):
class FakeProcess:
def __init__(self):
@@ -343,7 +695,7 @@ def slow_spawn(*_args, **_kwargs):
return FakeManaged()
with tempfile.TemporaryDirectory() as temp, mock.patch.object(
- JUDGE_MODULE, "spawn_managed", side_effect=slow_spawn
+ SPECIAL_JUDGES, "spawn_managed", side_effect=slow_spawn
):
root = Path(temp)
result = JUDGE_MODULE.run_interactive_testcase(
@@ -395,11 +747,14 @@ def write_problem(self, root, judge, accepted_source, wrong_source, extra_source
)
return problem
- def run_judge(self, problem):
+ def run_judge(self, problem, *, no_cache=True):
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
+ command = [sys.executable, str(LOCAL_JUDGE), str(problem), "--jsonl"]
+ if no_cache:
+ command.append("--no-cache")
result = subprocess.run(
- [sys.executable, str(LOCAL_JUDGE), str(problem), "--jsonl", "--no-cache"],
+ command,
cwd=ROOT,
capture_output=True,
text=True,
@@ -481,11 +836,38 @@ def test_custom_checker_accepts_non_unique_output_and_kills_wrong_solution(self)
self.assertTrue(std_cases and all(event["status"] == "AC" for event in std_cases))
self.assertTrue(wrong_cases and all(event["status"] == "WA" for event in wrong_cases))
self.assertTrue(all(event["judge_type"] == "custom" for event in std_cases + wrong_cases))
+ self.assertTrue(all(event["actor"] == "session" for event in std_cases))
+ self.assertTrue(all(event["execution_status"] == "completed" for event in std_cases))
+ self.assertTrue(all(event["failure_kind"] is None for event in std_cases))
+ self.assertTrue(all(event["actor"] == "contestant" for event in wrong_cases))
+ self.assertTrue(all(event["failure_kind"] == "wrong_answer" for event in wrong_cases))
+ self.assertTrue(all(event["cleanup"]["ok"] for event in std_cases + wrong_cases))
sample_check = next(
event for event in events if event.get("type") == "sample_check"
)
self.assertTrue(sample_check["matches"], sample_check)
+ cached_result, cached_events = self.run_judge(problem, no_cache=False)
+ self.assertEqual(cached_result.returncode, 0, cached_result.stderr + cached_result.stdout)
+ cached_cases = [
+ event for event in cached_events if event.get("type") == "case"
+ ]
+ self.assertTrue(cached_cases and all(event["cached"] for event in cached_cases))
+ fields = (
+ "status", "judge_type", "verdict", "actor", "execution_status",
+ "failure_kind", "termination_reason", "cleanup",
+ )
+ first_by_key = {
+ (event["kind"], event["case"]): event
+ for event in std_cases + wrong_cases
+ }
+ for event in cached_cases:
+ original = first_by_key[(event["kind"], event["case"])]
+ self.assertEqual(
+ {field: event.get(field) for field in fields},
+ {field: original.get(field) for field in fields},
+ )
+
def test_custom_checker_feedback_flood_is_infrastructure_fail(self):
with tempfile.TemporaryDirectory() as temp:
problem = self.write_problem(
@@ -515,6 +897,10 @@ def test_custom_checker_feedback_flood_is_infrastructure_fail(self):
)
self.assertEqual(case["status"], "FAIL", case)
self.assertIn("output limit", case["message"])
+ self.assertEqual(case["actor"], "checker", case)
+ self.assertEqual(case["execution_status"], "output_limit", case)
+ self.assertEqual(case["failure_kind"], "resource_limit", case)
+ self.assertEqual(case["termination_reason"], "output_limit", case)
def test_custom_checker_ac_cannot_hide_sample_answer_mismatch(self):
with tempfile.TemporaryDirectory() as temp:
@@ -598,6 +984,11 @@ def test_interactor_runs_bidirectional_protocol_and_kills_wrong_solution(self):
self.assertTrue(std_cases and all(event["status"] == "AC" for event in std_cases))
self.assertTrue(wrong_cases and all(event["status"] == "WA" for event in wrong_cases))
self.assertTrue(all(event["judge_type"] == "interactive" for event in std_cases + wrong_cases))
+ self.assertTrue(all(event["actor"] == "session" for event in std_cases))
+ self.assertTrue(all(event["failure_kind"] is None for event in std_cases))
+ self.assertTrue(all(event["actor"] == "contestant" for event in wrong_cases))
+ self.assertTrue(all(event["failure_kind"] == "wrong_answer" for event in wrong_cases))
+ self.assertTrue(all(event["cleanup"]["ok"] for event in std_cases + wrong_cases))
transcripts = [event for event in events if event.get("type") == "transcript"]
self.assertTrue(transcripts)
directions = {
@@ -635,6 +1026,10 @@ def test_interactive_fast_exit_is_submission_re_not_infrastructure_fail(self):
if event.get("type") == "case" and event.get("kind") == "std"
)
self.assertEqual(case["status"], "RE", case)
+ self.assertEqual(case["actor"], "contestant", case)
+ self.assertEqual(case["execution_status"], "completed", case)
+ self.assertEqual(case["failure_kind"], "runtime_error", case)
+ self.assertEqual(case["termination_reason"], "completed", case)
def test_interactive_fast_output_flood_is_ole_after_process_exit(self):
with tempfile.TemporaryDirectory() as temp:
@@ -666,6 +1061,10 @@ def test_interactive_fast_output_flood_is_ole_after_process_exit(self):
if event.get("type") == "case" and event.get("kind") == "std"
)
self.assertEqual(case["status"], "OLE", case)
+ self.assertEqual(case["actor"], "contestant", case)
+ self.assertEqual(case["execution_status"], "output_limit", case)
+ self.assertEqual(case["failure_kind"], "resource_limit", case)
+ self.assertEqual(case["termination_reason"], "output_limit", case)
def test_interactor_memory_limit_is_infrastructure_fail(self):
with tempfile.TemporaryDirectory() as temp:
@@ -709,6 +1108,19 @@ def test_interactor_memory_limit_is_infrastructure_fail(self):
if event.get("type") == "case" and event.get("kind") == "std"
)
self.assertEqual(case["status"], "FAIL", case)
+ self.assertEqual(case["actor"], "interactor", case)
+ if case["execution_status"] == "memory_limit":
+ self.assertEqual(case["failure_kind"], "resource_limit", case)
+ self.assertEqual(case["termination_reason"], "memory_limit", case)
+ else:
+ # RLIMIT_AS can reject a large allocation before it becomes
+ # resident. Without measured memory evidence, stderr alone must
+ # not turn an Interactor failure into a claimed MLE.
+ self.assertNotEqual(platform.system(), "Windows", case)
+ self.assertEqual(case["execution_status"], "completed", case)
+ self.assertEqual(case["failure_kind"], "judge_failure", case)
+ self.assertEqual(case["termination_reason"], "completed", case)
+ self.assertTrue(case["memory_enforced"], case)
@unittest.skipUnless(platform.system() == "Linux", "Linux /proc peak-memory regression")
def test_interactive_linux_peak_memory_survives_process_exit(self):
@@ -792,6 +1204,10 @@ def test_interactor_idle_timeout_is_structured_and_preserves_transcript(self):
case = next(event for event in events if event.get("type") == "case")
self.assertEqual(case["status"], "TLE")
self.assertEqual(case["timeout_kind"], "idle")
+ self.assertEqual(case["actor"], "session", case)
+ self.assertEqual(case["execution_status"], "time_limit", case)
+ self.assertEqual(case["failure_kind"], "resource_limit", case)
+ self.assertEqual(case["termination_reason"], "time_limit", case)
transcript = next(event for event in events if event.get("type") == "transcript")
self.assertTrue(transcript["entries"])
self.assertEqual(transcript["entries"][0]["direction"], "interactor_to_solution")
diff --git a/tests/test_stress.py b/tests/test_stress.py
index a2778a7..512f9d5 100644
--- a/tests/test_stress.py
+++ b/tests/test_stress.py
@@ -1,4 +1,5 @@
import json
+import sys
import tempfile
import time
import unittest
@@ -11,6 +12,7 @@
from probhub.io import write_yaml
from probhub.linting import compute_source_hash, lint_workspace
from probhub.process_control import process_alive
+from probhub.special_judges import run_checker_to_files
from probhub.stressing import expand_generator_args, stress_problem
from probhub.workspace import load_problem, load_workspace, problem_entries
@@ -32,6 +34,75 @@ def test_output_budget_enforcement_failure_is_infrastructure(self):
self.assertEqual(result["reason"], "output_control_error")
self.assertIn("cannot enforce", result["message"])
+ def test_checker_core_separates_protocol_verdict_and_execution(self):
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ input_path = root / "case.in"
+ answer_path = root / "case.ans"
+ output_path = root / "contestant.out"
+ checker_path = root / "checker.py"
+ input_path.write_bytes(b"1\n")
+ answer_path.write_bytes(b"2\n")
+ output_path.write_bytes(b"contestant\n")
+ checker_path.write_text(
+ "import pathlib,sys\n"
+ "assert sys.stdin.buffer.read() == b'contestant\\n'\n"
+ "pathlib.Path(sys.argv[-1], 'judgemessage.txt').write_text("
+ "'checker feedback', encoding='utf-8')\n"
+ "raise SystemExit(int(sys.argv[1]))\n",
+ encoding="utf-8",
+ )
+
+ for returncode, verdict, execution_status, failure_kind, actor in (
+ (0, "AC", "completed", None, "session"),
+ (42, "AC", "completed", None, "session"),
+ (1, "WA", "completed", "wrong_answer", "contestant"),
+ (2, "WA", "completed", "wrong_answer", "contestant"),
+ (43, "WA", "completed", "wrong_answer", "contestant"),
+ (7, None, "completed", "judge_failure", "checker"),
+ ):
+ with self.subTest(returncode=returncode):
+ result = run_checker_to_files(
+ [sys.executable, str(checker_path), str(returncode)],
+ input_path,
+ answer_path,
+ output_path,
+ timeout=2,
+ cwd=root,
+ output_limit_bytes=1024 * 1024,
+ )
+ self.assertEqual(result["verdict"], verdict)
+ self.assertEqual(result["execution_status"], execution_status)
+ self.assertEqual(result["failure_kind"], failure_kind)
+ self.assertEqual(result["actor"], actor)
+ self.assertEqual(result["termination_reason"], "completed")
+ self.assertEqual(result["message"], "checker feedback")
+ self.assertEqual(result["feedback_message"], "checker feedback")
+
+ def test_checker_core_output_control_failure_is_structured(self):
+ with tempfile.TemporaryDirectory() as temp, mock.patch(
+ "probhub.special_judges.run_managed_to_files",
+ side_effect=stressing.OutputBudgetError("feedback is not regular"),
+ ):
+ root = Path(temp)
+ for name in ("case.in", "case.ans", "contestant.out"):
+ (root / name).write_bytes(b"")
+ result = run_checker_to_files(
+ ["checker"],
+ root / "case.in",
+ root / "case.ans",
+ root / "contestant.out",
+ timeout=1,
+ cwd=root,
+ output_limit_bytes=1024,
+ )
+ self.assertIsNone(result["verdict"])
+ self.assertEqual(result["execution_status"], "output_control_error")
+ self.assertEqual(result["failure_kind"], "control_failure")
+ self.assertEqual(result["actor"], "supervisor")
+ self.assertEqual(result["termination_reason"], "output_control_error")
+ self.assertIn("not regular", result["message"])
+
def create_workspace(self, root, judge_type="standard"):
write_yaml(root / ".probhub/workspace.yaml", {
"schema_version": 1,