From 36ff30f497d627595c9f5eadb7581d504062f286 Mon Sep 17 00:00:00 2001 From: greenthree <1395214327@qq.com> Date: Wed, 5 Aug 2026 18:12:29 +0800 Subject: [PATCH 1/2] add isolated judge QA execution --- probhub/cli.py | 4 + probhub/hashing.py | 10 +- probhub/judge_qa.py | 76 +- probhub/judge_qa_runtime.py | 1065 +++++++++++++++++++++++++++ probhub/linting.py | 56 +- probhub/process_control.py | 17 +- probhub/special_judges.py | 41 +- scripts/local_judge.py | 9 +- tests/test_calibration.py | 4 + tests/test_judge_qa_execution.py | 1169 ++++++++++++++++++++++++++++++ tests/test_judge_qa_schema.py | 19 + tests/test_process_control.py | 58 ++ tests/test_special_judges.py | 214 +++++- 13 files changed, 2694 insertions(+), 48 deletions(-) create mode 100644 probhub/judge_qa_runtime.py create mode 100644 tests/test_judge_qa_execution.py diff --git a/probhub/cli.py b/probhub/cli.py index f654c5f..71157e4 100644 --- a/probhub/cli.py +++ b/probhub/cli.py @@ -94,6 +94,10 @@ def _ensure_local_gitignore(root): "**/.probhub/judge-evidence-v1.json.*.tmp", "**/.probhub/judge-evidence-v2.json", "**/.probhub/judge-evidence-v2.json.*.tmp", + "**/.probhub/judge-qa-evidence.lock", + "**/.probhub/judge-qa-evidence-v1.json", + "**/.probhub/judge-qa-evidence-v1.json.*.tmp", + "**/.probhub/judge-qa-tmp/", "**/.probhub/sandbox-cache-v1.json", "**/.probhub/sandbox-cache-v1.json.tmp", "**/.probhub/stress/", diff --git a/probhub/hashing.py b/probhub/hashing.py index cb4dc9b..3ebcfea 100644 --- a/probhub/hashing.py +++ b/probhub/hashing.py @@ -13,7 +13,7 @@ def hash_file(path): return digest.hexdigest() -def hash_paths(root, paths, *, normalize_lf_suffixes=()): +def hash_paths(root, paths, *, normalize_lf_suffixes=(), check=None): root = Path(root) normalize_lf_suffixes = { str(suffix).lower() for suffix in normalize_lf_suffixes @@ -21,6 +21,8 @@ def hash_paths(root, paths, *, normalize_lf_suffixes=()): digest = hashlib.sha256() existing = [] for path in sorted({Path(p) for p in paths}, key=lambda p: p.as_posix()): + if check is not None: + check() full = path if path.is_absolute() else root / path if not full.is_file(): continue @@ -33,6 +35,8 @@ def hash_paths(root, paths, *, normalize_lf_suffixes=()): observed_size = 0 with full.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): + if check is not None: + check() observed_size += len(chunk) digest.update(chunk) if observed_size != expected_size: @@ -49,12 +53,14 @@ def hash_paths(root, paths, *, normalize_lf_suffixes=()): return digest.hexdigest(), existing -def files_under(path, suffixes=None): +def files_under(path, suffixes=None, *, check=None): path = Path(path) if not path.exists(): return [] result = [] for item in path.rglob("*"): + if check is not None: + check() if item.is_file() and (suffixes is None or item.suffix in suffixes): result.append(item) return result diff --git a/probhub/judge_qa.py b/probhub/judge_qa.py index a49a296..4eb8145 100644 --- a/probhub/judge_qa.py +++ b/probhub/judge_qa.py @@ -20,6 +20,8 @@ _INTERACTOR_STATUSES = {"AC", "WA", "RE", "TLE", "MLE", "OLE"} _INTERACTOR_BEHAVIORS = {"early-eof", "idle", "output-flood"} _ROBUSTNESS_PROBES = {"empty", "truncated", "extra-token", "oversized"} +_JUDGE_QA_OFFICIAL_SOURCE_SUFFIXES = {".cpp"} +_JUDGE_QA_CONTESTANT_SOURCE_SUFFIXES = {".cpp", ".py"} _PATH_REASON_CODES = { "invalid": "judge_qa_fixture_path_invalid", "outside": "judge_qa_fixture_path_outside", @@ -53,18 +55,20 @@ def _under_prefix(relative, prefix): return path_parts[:len(prefix_parts)] == prefix_parts and len(path_parts) > len(prefix_parts) -def judge_fixture_tree_paths(problem_dir): +def judge_fixture_tree_paths(problem_dir, *, check=None): """Return regular non-link files stored under judge-fixtures/.""" - return _judge_qa_tree_paths(problem_dir, "judge-fixtures") + return _judge_qa_tree_paths(problem_dir, "judge-fixtures", check=check) -def _judge_qa_tree_paths(problem_dir, relative_root): +def _judge_qa_tree_paths(problem_dir, relative_root, *, check=None): root = Path(problem_dir).resolve() / relative_root if not root.is_dir() or root.is_symlink(): return [] paths = [] for candidate in root.rglob("*"): + if check is not None: + check() try: relative = candidate.relative_to(Path(problem_dir).resolve()).as_posix() resolved = resolve_problem_regular_file(problem_dir, relative) @@ -74,10 +78,12 @@ def _judge_qa_tree_paths(problem_dir, relative_root): return paths -def _hash_fixture_files(files): +def _hash_fixture_files(files, *, check=None): digest = hashlib.sha256() digest.update(b"probhub-judge-qa-fixtures-v1\0") for item in sorted(files, key=lambda value: value["path"]): + if check is not None: + check() relative = item["path"].encode("utf-8") expected_size = item["size"] digest.update(len(relative).to_bytes(4, "big")) @@ -86,6 +92,8 @@ def _hash_fixture_files(files): observed_size = 0 with item["absolute"].open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): + if check is not None: + check() observed_size += len(chunk) digest.update(chunk) after = item["absolute"].stat() @@ -107,10 +115,12 @@ def _hash_fixture_files(files): return digest.hexdigest() -def inspect_judge_qa(problem_dir, config): +def inspect_judge_qa(problem_dir, config, *, check=None): """Parse Judge QA Schema v1 without executing any fixture.""" problem_dir = Path(problem_dir).resolve() + if check is not None: + check() judge = config.get("judge") if not isinstance(judge, dict) or "qa" not in judge: return { @@ -157,6 +167,8 @@ def unknown_fields(value, allowed, field, *, case_id=None): if not isinstance(value, dict): return for key in sorted(set(value) - set(allowed), key=str): + if check is not None: + check() add( "judge_qa_unknown_field", f"{field} contains unsupported field: {key}", @@ -264,9 +276,24 @@ def resolve_file(field, value, *, required_prefix=None, case_id=None): field="judge.type", judge_type=judge_type, ) + required_programs = ["validator"] + required_programs.append("checker" if judge_type == "custom" else "interactor") + for program_key in required_programs: + program = judge.get(program_key) + if ( + isinstance(program, str) + and PurePosixPath(program.replace("\\", "/")).suffix.casefold() + not in _JUDGE_QA_OFFICIAL_SOURCE_SUFFIXES + ): + add( + "judge_qa_program_type_unsupported", + f"judge.{program_key} must be a reproducible .cpp source for Judge QA", + field=f"judge.{program_key}", + path=program, + ) for tree_name in ("judge-fixtures", "code/judge-qa"): - for path in _judge_qa_tree_paths(problem_dir, tree_name): + for path in _judge_qa_tree_paths(problem_dir, tree_name, check=check): register_file("judge.qa.fixture_tree", path) raw_cases = raw_qa.get("cases") @@ -287,6 +314,8 @@ def resolve_file(field, value, *, required_prefix=None, case_id=None): seen_ids = {} expected_by_id = {} for index, raw_case in enumerate(raw_cases[:MAX_JUDGE_QA_CASES]): + if check is not None: + check() field = f"judge.qa.cases[{index}]" if not isinstance(raw_case, dict): add("judge_qa_case_invalid", f"{field} must be a mapping", field=field) @@ -519,14 +548,27 @@ def resolve_file(field, value, *, required_prefix=None, case_id=None): **({"case_id": case_id} if case_id else {}), ) elif source is not None: + resolved_source = resolve_file( + f"{field}.contestant.source", + source, + required_prefix="code/judge-qa", + case_id=case_id, + ) parsed["contestant"] = { - "source": resolve_file( - f"{field}.contestant.source", - source, - required_prefix="code/judge-qa", - case_id=case_id, - ), + "source": resolved_source, } + if ( + isinstance(resolved_source, str) + and PurePosixPath(resolved_source).suffix.casefold() + not in _JUDGE_QA_CONTESTANT_SOURCE_SUFFIXES + ): + add( + "judge_qa_program_type_unsupported", + f"{field}.contestant.source must be a reproducible .cpp or .py source", + field=f"{field}.contestant.source", + path=resolved_source, + **({"case_id": case_id} if case_id else {}), + ) elif ( not isinstance(behavior, str) or behavior not in _INTERACTOR_BEHAVIORS @@ -658,7 +700,7 @@ def resolve_file(field, value, *, required_prefix=None, case_id=None): fixture_hash = None if not diagnostics: try: - fixture_hash = _hash_fixture_files(resolved_files.values()) + fixture_hash = _hash_fixture_files(resolved_files.values(), check=check) except OSError as exc: add( "judge_qa_fixture_changed", @@ -684,3 +726,11 @@ def resolve_file(field, value, *, required_prefix=None, case_id=None): }, "diagnostics": diagnostics, } + + +def judge_qa_problem(*args, **kwargs): + """Execute Judge QA through a lazy import to avoid linting import cycles.""" + + from .judge_qa_runtime import judge_qa_problem as execute + + return execute(*args, **kwargs) diff --git a/probhub/judge_qa_runtime.py b/probhub/judge_qa_runtime.py new file mode 100644 index 0000000..a864c40 --- /dev/null +++ b/probhub/judge_qa_runtime.py @@ -0,0 +1,1065 @@ +import hashlib +import math +import os +import platform +import shutil +import sys +import tempfile +import time +from datetime import datetime, timezone +from pathlib import Path + +from . import __version__ +from .build_lock import workspace_file_lock +from .calibration import SANDBOX_CACHE_SCHEMA_VERSION +from .errors import ProbHubError +from .hashing import files_under +from .io import atomic_write_json, normalize_newlines, read_bounded_text, read_yaml +from .judge_qa import inspect_judge_qa +from .linting import compute_data_hash, compute_source_hash, problem_source_paths +from .problem_paths import ProblemPathError, resolve_problem_regular_file +from .process_control import ( + CANCEL_FILE_ENV, + DEFAULT_PROCESS_LIMIT, + OutputBudgetError, + ProcessCancelled, + VALIDATOR_MEMORY_LIMIT_MB, + VALIDATOR_OUTPUT_LIMIT_BYTES, + VALIDATOR_TIMEOUT_SECONDS, + cancellation_requested, + run_managed_to_files, +) +from .special_judges import execute_interactive_session, run_checker_to_files + + +JUDGE_QA_EVIDENCE_SCHEMA_VERSION = 1 +JUDGE_QA_POLICY_VERSION = 1 +JUDGE_QA_EVIDENCE_FILENAME = "judge-qa-evidence-v1.json" +JUDGE_QA_EVIDENCE_LOCK_FILENAME = "judge-qa-evidence.lock" +DEFAULT_JUDGE_QA_TIMEOUT_SECONDS = 600.0 +MAX_JUDGE_QA_TIMEOUT_SECONDS = 3600.0 +DEFAULT_TOOL_TIMEOUT_SECONDS = 10.0 +MIN_CHECKER_TIMEOUT_SECONDS = 5.0 +COMPILER_TIMEOUT_SECONDS = 60.0 +COMPILER_MEMORY_LIMIT_MB = 2048 +COMPILER_OUTPUT_LIMIT_BYTES = 8 * 1024 * 1024 +DIAGNOSTIC_LIMIT_BYTES = 4096 +ROBUSTNESS_OVERSIZED_BYTES = 1024 * 1024 + 1 +MAX_JUDGE_QA_OUTPUT_BYTES = 8 * 1024 * 1024 +MAX_JUDGE_QA_TRANSCRIPT_BYTES = 1024 * 1024 +MIN_JUDGE_QA_IDLE_SECONDS = 0.1 + + +class _OverallTimeout(Exception): + pass + + +class _RunCancellation: + def __init__(self, timeout, cancel_check): + self.deadline = time.monotonic() + max(float(timeout), 0.0) + self.cancel_check = cancel_check + self.cause = None + + def requested(self): + try: + requested = cancellation_requested() + if self.cancel_check is not None: + requested = requested or bool(self.cancel_check()) + except ProcessCancelled: + raise + except Exception as exc: + raise ProbHubError( + f"Judge QA cancellation check failed: {_bounded_message(exc)}", + code="judge_qa_cancel_check_failed", + ) from exc + if requested: + self.cause = "cancelled" + return True + if time.monotonic() >= self.deadline: + self.cause = "overall_timeout" + return True + return False + + def remaining(self, maximum=None): + if self.requested(): + if self.cause == "overall_timeout": + raise _OverallTimeout("Judge QA overall deadline exceeded") + raise ProcessCancelled("Judge QA cancelled") + remaining = self.deadline - time.monotonic() + if maximum is not None: + remaining = min(remaining, float(maximum)) + return max(remaining, 0.001) + + +def _result(status, *, ok=False, applicable=True, code=None, message=None, **payload): + return { + "ok": bool(ok), + "applicable": bool(applicable), + "status": status, + **({"code": code} if code else {}), + **({"message": str(message)} if message else {}), + **payload, + } + + +def _bounded_message(value): + text = str(value or "").strip() + if len(text.encode("utf-8", errors="replace")) <= DIAGNOSTIC_LIMIT_BYTES: + return text + retained = text.encode("utf-8", errors="replace")[:DIAGNOSTIC_LIMIT_BYTES] + return retained.decode("utf-8", errors="replace") + "..." + + +def _cleanup_summary(cleanup): + cleanup = cleanup if isinstance(cleanup, dict) else {} + states = { + key: value + for key, value in cleanup.items() + if key.endswith("_termination") + or key in {"runtime_removed", "runtime_remove_attempts", "pumps_joined"} + } + errors = cleanup.get("errors") if isinstance(cleanup.get("errors"), list) else [] + return { + "ok": bool(cleanup.get("ok", not errors)), + **states, + "error_count": len(errors), + "errors": [ + { + "stage": item.get("stage"), + "actor": item.get("actor"), + "message": _bounded_message(item.get("message")), + } + for item in errors[:8] + if isinstance(item, dict) + ], + } + + +def _file_identity(path): + info = path.stat() + return (info.st_dev, info.st_ino, info.st_mtime_ns, info.st_size, info.st_mode) + + +def _copy_regular_file(source, destination, problem_dir, cancellation=None): + relative = source.relative_to(problem_dir).as_posix() + try: + verified = resolve_problem_regular_file(problem_dir, relative) + except ProblemPathError as exc: + raise ProbHubError( + f"Judge QA snapshot source is unsafe: {relative} ({exc.reason})", + code="snapshot_failed", + ) from exc + before = _file_identity(verified) + destination.parent.mkdir(parents=True, exist_ok=True) + with verified.open("rb") as source_stream, destination.open("wb") as target_stream: + while True: + if cancellation is not None: + cancellation.remaining() + chunk = source_stream.read(1024 * 1024) + if not chunk: + break + target_stream.write(chunk) + if os.name != "nt": + os.chmod(destination, before[4] & 0o777) + after = _file_identity(verified) + if before != after or destination.stat().st_size != before[3]: + raise ProbHubError( + f"Judge QA input changed while creating the snapshot: {relative}", + code="inputs_changed", + ) + + +def _snapshot_paths(problem_dir, config, report, cancellation): + problem_dir = Path(problem_dir).resolve() + paths = set() + for path in problem_source_paths( + problem_dir, config, check=cancellation.remaining + ): + if path.exists(): + paths.add(path.resolve()) + for item in report.get("files") or []: + cancellation.remaining() + paths.add(resolve_problem_regular_file(problem_dir, item.get("path")).resolve()) + data = config.get("data") if isinstance(config.get("data"), dict) else {} + for key, default in (("sample_dir", "data/sample"), ("secret_dir", "data/secret")): + directory = problem_dir / str(data.get(key, default)) + for path in files_under( + directory, {".in", ".ans"}, check=cancellation.remaining + ): + paths.add(path.resolve()) + cancellation.remaining() + return sorted(paths, key=lambda path: path.relative_to(problem_dir).as_posix()) + + +def _create_snapshot(problem_dir, destination, config, report, identity, cancellation): + problem_dir = Path(problem_dir).resolve() + destination = Path(destination) + destination.mkdir(parents=True, exist_ok=False) + for source in _snapshot_paths(problem_dir, config, report, cancellation): + _copy_regular_file( + source, + destination / source.relative_to(problem_dir), + problem_dir, + cancellation, + ) + cancellation.remaining() + copied_config = read_yaml(destination / "probhub.yaml") + cancellation.remaining() + copied_report = inspect_judge_qa( + destination, copied_config, check=cancellation.remaining + ) + copied_identity = ( + compute_source_hash(destination, copied_config, check=cancellation.remaining), + compute_data_hash(destination, copied_config, check=cancellation.remaining), + copied_report.get("fixture_hash"), + ) + if not copied_report.get("ok") or copied_identity != identity: + raise ProbHubError( + "Judge QA inputs changed while creating the snapshot", + code="inputs_changed", + ) + return copied_config, copied_report + + +def _remove_snapshot(path): + shutil.rmtree(path) + + +def _compiler_command(source, output, include_dir, role): + command = [ + "g++", + os.fspath(source), + "-o", + os.fspath(output), + "-O2", + "-std=c++17", + "-I", + os.fspath(include_dir), + ] + if platform.system() == "Windows": + command.append("-static") + if role == "validator": + command.append("-DFOR_LINUX") + return command + + +def _compiler_identity(cancellation): + with tempfile.TemporaryDirectory(prefix="probhub-judge-qa-compiler-") as temp: + stdout_path = Path(temp) / "stdout" + stderr_path = Path(temp) / "stderr" + try: + execution = run_managed_to_files( + ["g++", "--version"], + stdout_path=stdout_path, + stderr_path=stderr_path, + timeout=cancellation.remaining(DEFAULT_TOOL_TIMEOUT_SECONDS), + memory_limit_mb=COMPILER_MEMORY_LIMIT_MB, + output_limit_bytes=64 * 1024, + process_limit=DEFAULT_PROCESS_LIMIT, + cancel_check=cancellation.requested, + ) + except (OSError, OutputBudgetError) as exc: + raise ProbHubError( + f"failed to identify the Judge QA compiler: {_bounded_message(exc)}", + code="compiler_identity_failed", + ) from exc + if execution.get("reason") != "completed" or execution.get("returncode") != 0: + raise ProbHubError( + "failed to identify the Judge QA compiler: " + + _bounded_message(execution.get("message") or execution.get("reason")), + code="compiler_identity_failed", + ) + text = read_bounded_text(stdout_path, 64 * 1024).get("text", "") + identity = (text.splitlines() or [""])[0].strip() + if not identity: + raise ProbHubError( + "Judge QA compiler identity output is empty", + code="compiler_identity_failed", + ) + return identity + + +def _compile_program(source, build_dir, role, cancellation): + source = Path(source) + try: + display_source = source.relative_to(Path(build_dir).parent).as_posix() + except ValueError: + display_source = source.name + suffix = source.suffix.lower() + if suffix == ".py": + if role != "contestant": + raise ProbHubError( + f"Judge QA {role} must be a reproducible .cpp source: {display_source}", + code="judge_qa_program_type_unsupported", + ) + return [sys.executable, "-I", str(source)], { + "role": role, + "source": display_source, + "kind": "python", + "interpreter": platform.python_implementation(), + "version": platform.python_version(), + } + if suffix != ".cpp": + raise ProbHubError( + f"Judge QA programs must be reproducible .cpp or .py sources: {display_source}", + code="judge_qa_program_type_unsupported", + ) + + digest = hashlib.sha256(f"{role}\0{source}".encode("utf-8")).hexdigest()[:16] + output = Path(build_dir) / (f"{role}-{digest}.exe" if os.name == "nt" else f"{role}-{digest}") + include_dir = Path(build_dir) / "include" + include_dir.mkdir(parents=True, exist_ok=True) + shutil.copyfile( + Path(__file__).resolve().parents[1] / "references" / "testlib.h", + include_dir / "testlib.h", + ) + source_dir = source.parent + command = _compiler_command( + os.path.relpath(source, source_dir), + os.path.relpath(output, source_dir), + os.path.relpath(include_dir, source_dir), + role, + ) + compile_dir = Path(build_dir) / "compile-output" + compile_dir.mkdir(parents=True, exist_ok=True) + stdout_path = compile_dir / f"{digest}.stdout" + stderr_path = compile_dir / f"{digest}.stderr" + try: + execution = run_managed_to_files( + command, + stdout_path=stdout_path, + stderr_path=stderr_path, + timeout=cancellation.remaining(COMPILER_TIMEOUT_SECONDS), + memory_limit_mb=COMPILER_MEMORY_LIMIT_MB, + output_limit_bytes=COMPILER_OUTPUT_LIMIT_BYTES, + process_limit=DEFAULT_PROCESS_LIMIT, + cwd=source_dir, + cancel_check=cancellation.requested, + ) + except OutputBudgetError as exc: + raise ProbHubError(str(exc), code="compile_output_control_failed") from exc + stderr = read_bounded_text(stderr_path, DIAGNOSTIC_LIMIT_BYTES).get("text", "") + if execution.get("reason") != "completed" or execution.get("returncode") != 0: + detail = stderr.strip() or execution.get("message") or execution.get("reason") + raise ProbHubError( + f"Judge QA {role} failed to compile: {_bounded_message(detail)}", + code="compile_failed", + ) + return [str(output)], { + "role": role, + "source": display_source, + "kind": "cpp17", + "compiler": "g++", + "compiler_path": shutil.which("g++"), + } + + +def _program_path(problem_dir, config, key): + judge = config.get("judge") if isinstance(config.get("judge"), dict) else {} + value = judge.get(key) + try: + return resolve_problem_regular_file(problem_dir, value) + except ProblemPathError as exc: + raise ProbHubError( + f"judge.{key} must be a problem-local regular file ({exc.reason})", + code="judge_qa_program_invalid", + ) from exc + + +def _runtime_limits(config): + raw = config.get("limits") if isinstance(config.get("limits"), dict) else {} + time_limit = raw.get("time") + memory_limit = raw.get("memory") + output_limit = raw.get("output", 64) + process_limit = raw.get("processes", DEFAULT_PROCESS_LIMIT) + if isinstance(time_limit, bool) or not isinstance(time_limit, int) or time_limit <= 0: + raise ProbHubError( + "limits.time must be a positive integer before running Judge QA", + code="judge_qa_limits_invalid", + ) + if ( + isinstance(memory_limit, bool) + or not isinstance(memory_limit, int) + or memory_limit < 256 + or memory_limit & (memory_limit - 1) + ): + raise ProbHubError( + "limits.memory must be a power of two and at least 256 before running Judge QA", + code="judge_qa_limits_invalid", + ) + if isinstance(output_limit, bool) or not isinstance(output_limit, int) or output_limit <= 0: + raise ProbHubError( + "limits.output must be a positive integer before running Judge QA", + code="judge_qa_limits_invalid", + ) + if isinstance(process_limit, bool) or not isinstance(process_limit, int) or process_limit <= 0: + raise ProbHubError( + "limits.processes must be a positive integer before running Judge QA", + code="judge_qa_limits_invalid", + ) + configured_output_bytes = output_limit * 1024 * 1024 + return { + "time": float(time_limit), + "memory": memory_limit, + "output_bytes": min(configured_output_bytes, MAX_JUDGE_QA_OUTPUT_BYTES), + "configured_output_bytes": configured_output_bytes, + "processes": process_limit, + } + + +def _interactive_limits(config, limits): + interactive = (config.get("judge") or {}).get("interactive") or {} + if not isinstance(interactive, dict): + raise ProbHubError( + "judge.interactive must be a mapping before running Judge QA", + code="judge_qa_limits_invalid", + ) + idle_limit = interactive.get("idle_limit", min(limits["time"], 2.0)) + if ( + isinstance(idle_limit, bool) + or not isinstance(idle_limit, (int, float)) + or not math.isfinite(float(idle_limit)) + or float(idle_limit) <= 0 + ): + raise ProbHubError( + "judge.interactive.idle_limit must be a positive finite number before running Judge QA", + code="judge_qa_limits_invalid", + ) + transcript_limit = interactive.get("transcript_limit", 65536) + if ( + isinstance(transcript_limit, bool) + or not isinstance(transcript_limit, int) + or transcript_limit < 0 + ): + raise ProbHubError( + "judge.interactive.transcript_limit must be a non-negative integer before running Judge QA", + code="judge_qa_limits_invalid", + ) + return { + "idle_limit": max(float(idle_limit), MIN_JUDGE_QA_IDLE_SECONDS), + "configured_idle_limit": float(idle_limit), + "transcript_limit": min( + transcript_limit, + MAX_JUDGE_QA_TRANSCRIPT_BYTES, + ), + "configured_transcript_limit": transcript_limit, + } + + +def _validate_inputs(validator_command, problem_dir, cases, cancellation): + seen = set() + results = [] + runtime = Path(problem_dir) / ".judge-qa-validator" + runtime.mkdir(parents=True, exist_ok=True) + try: + for case in cases: + relative = case.get("input") + if not relative or relative.casefold() in seen: + continue + seen.add(relative.casefold()) + input_path = resolve_problem_regular_file(problem_dir, relative) + stdout_path = runtime / f"{len(results)}.stdout" + stderr_path = runtime / f"{len(results)}.stderr" + execution = run_managed_to_files( + validator_command, + input_data=normalize_newlines(input_path.read_bytes()), + stdout_path=stdout_path, + stderr_path=stderr_path, + timeout=cancellation.remaining(VALIDATOR_TIMEOUT_SECONDS), + memory_limit_mb=VALIDATOR_MEMORY_LIMIT_MB, + output_limit_bytes=VALIDATOR_OUTPUT_LIMIT_BYTES, + process_limit=DEFAULT_PROCESS_LIMIT, + cwd=problem_dir, + cancel_check=cancellation.requested, + ) + message = read_bounded_text(stderr_path, DIAGNOSTIC_LIMIT_BYTES).get("text", "") + summary = { + "input": relative, + "ok": execution.get("reason") == "completed" and execution.get("returncode") == 0, + "execution_status": execution.get("reason"), + "returncode": execution.get("returncode"), + "time": execution.get("time"), + "message": _bounded_message(message or execution.get("message")), + } + results.append(summary) + if not summary["ok"]: + raise ProbHubError( + f"Validator rejected Judge QA input {relative}: {summary['message']}", + code="judge_qa_validator_failed", + ) + finally: + shutil.rmtree(runtime, ignore_errors=True) + return results + + +def _checker_summary(case, result): + actual = result.get("verdict") + infrastructure = actual not in {"AC", "WA"} or not (result.get("cleanup") or {}).get("ok", True) + expected = case.get("expected") or {} + matched = not infrastructure and actual == expected.get("status") + return { + "id": case.get("id"), + "purpose": case.get("purpose"), + "expected": expected, + "actual": { + "status": actual, + "actor": result.get("actor"), + "execution_status": result.get("execution_status"), + "failure_kind": result.get("failure_kind"), + "termination_reason": result.get("termination_reason"), + }, + "matched": matched, + "infrastructure_failed": infrastructure, + "message": _bounded_message(result.get("message")), + "resources": { + key: result.get(key) + for key in ( + "time", "memory", "memory_enforced", "process_limit_enforced", + "output_bytes", "retained_output_bytes", "output_truncated", + ) + }, + "cleanup": _cleanup_summary(result.get("cleanup")), + } + + +def _interactor_summary(case, result): + infrastructure = ( + result.get("status") == "FAIL" + or result.get("actor") in {"interactor", "supervisor"} + and result.get("failure_kind") not in {None, "cancelled"} + or not (result.get("cleanup") or {}).get("ok", True) + ) + expected = case.get("expected") or {} + actual = { + "status": result.get("status"), + "timeout_kind": result.get("timeout_kind"), + "termination_reason": result.get("termination_reason"), + "actor": result.get("actor"), + "execution_status": result.get("execution_status"), + "failure_kind": result.get("failure_kind"), + } + matched = not infrastructure and all(actual.get(key) == value for key, value in expected.items()) + return { + "id": case.get("id"), + "purpose": case.get("purpose"), + "expected": expected, + "actual": actual, + "matched": matched, + "infrastructure_failed": infrastructure, + "message": _bounded_message(result.get("message")), + "resources": result.get("resources") or {}, + "traffic": result.get("traffic") or {}, + "transcript": { + "bytes": result.get("transcript_bytes", 0), + "limit": None, + "truncated": bool(result.get("transcript_truncated")), + }, + "cleanup": _cleanup_summary(result.get("cleanup")), + } + + +def _checker_probe_payload(name, baseline): + if name == "empty": + return b"" + if name == "truncated": + stripped = baseline.rstrip() + if not stripped: + return b"" + split = stripped.rfind(b" ") + return stripped[:split] if split > 0 else stripped[:-1] + if name == "extra-token": + return baseline + (b"" if baseline.endswith(b"\n") else b"\n") + b"probhub-extra-token\n" + if name == "oversized": + seed = baseline + (b"" if baseline.endswith(b"\n") else b"\n") + repetitions = (ROBUSTNESS_OVERSIZED_BYTES - len(seed) + 1) // 2 + return (seed + b"0 " * max(repetitions, 0))[:ROBUSTNESS_OVERSIZED_BYTES] + raise ProbHubError(f"unsupported Judge QA probe: {name}", code="judge_qa_probe_invalid") + + +def _run_checker_qa(problem_dir, report, checker_command, limits, cancellation): + checker_timeout = max(MIN_CHECKER_TIMEOUT_SECONDS, float(limits["time"])) + limits["checker_timeout"] = checker_timeout + summaries = [] + case_by_id = {} + for case in report["cases"]: + result = run_checker_to_files( + checker_command, + resolve_problem_regular_file(problem_dir, case["input"]), + resolve_problem_regular_file(problem_dir, case["jury_answer"]), + resolve_problem_regular_file(problem_dir, case["contestant_output"]), + timeout=cancellation.remaining(checker_timeout), + cwd=problem_dir, + memory_limit_mb=limits["memory"], + output_limit_bytes=limits["output_bytes"], + process_limit=limits["processes"], + cancel_check=cancellation.requested, + ) + summary = _checker_summary(case, result) + summaries.append(summary) + case_by_id[str(case.get("id", "")).casefold()] = case + if summary["infrastructure_failed"]: + return summaries, [], "infrastructure-failed" + + probes = [] + robustness = report.get("robustness") + if robustness: + baseline = case_by_id.get(str(robustness.get("baseline", "")).casefold()) + if baseline is None: + raise ProbHubError("Judge QA robustness baseline is missing", code="judge_qa_probe_invalid") + baseline_output = resolve_problem_regular_file( + problem_dir, baseline["contestant_output"] + ).read_bytes() + probe_dir = Path(problem_dir) / ".judge-qa-probes" + probe_dir.mkdir(parents=True, exist_ok=True) + try: + for index, name in enumerate(robustness.get("probes") or []): + output_path = probe_dir / f"{index}.out" + output_path.write_bytes(_checker_probe_payload(name, baseline_output)) + raw = run_checker_to_files( + checker_command, + resolve_problem_regular_file(problem_dir, baseline["input"]), + resolve_problem_regular_file(problem_dir, baseline["jury_answer"]), + output_path, + timeout=cancellation.remaining(checker_timeout), + cwd=problem_dir, + memory_limit_mb=limits["memory"], + output_limit_bytes=limits["output_bytes"], + process_limit=limits["processes"], + cancel_check=cancellation.requested, + ) + summary = _checker_summary( + {"id": name, "purpose": "robustness-probe", "expected": {}}, raw + ) + summary["matched"] = not summary["infrastructure_failed"] + summary["manual_review_required"] = raw.get("verdict") == "AC" + probes.append(summary) + if summary["infrastructure_failed"]: + return summaries, probes, "infrastructure-failed" + finally: + shutil.rmtree(probe_dir, ignore_errors=True) + if any(not item["matched"] for item in summaries): + return summaries, probes, "expectation-failed" + return summaries, probes, "passed" + + +def _builtin_contestant_command(behavior, limits): + if behavior == "early-eof": + code = "raise SystemExit(0)" + elif behavior == "idle": + code = f"import time; time.sleep({max(float(limits['time']) * 4, 10.0)!r})" + elif behavior == "output-flood": + code = "import os; b=b'0'*65536\nwhile True: os.write(1,b)" + else: + raise ProbHubError( + f"unsupported built-in contestant behavior: {behavior}", + code="judge_qa_behavior_invalid", + ) + return [sys.executable, "-I", "-c", code] + + +def _run_interactor_qa( + problem_dir, + report, + interactor_command, + contestant_commands, + config, + limits, + cancellation, +): + interactive_limits = ( + { + "idle_limit": limits["idle_limit"], + "configured_idle_limit": limits["configured_idle_limit"], + "transcript_limit": limits["transcript_limit"], + "configured_transcript_limit": limits["configured_transcript_limit"], + } + if "transcript_limit" in limits + else _interactive_limits(config, limits) + ) + idle_limit = interactive_limits["idle_limit"] + transcript_limit = interactive_limits["transcript_limit"] + summaries = [] + for case in report["cases"]: + contestant = case.get("contestant") or {} + if contestant.get("source"): + command = contestant_commands[contestant["source"].casefold()] + else: + command = _builtin_contestant_command(contestant.get("behavior"), limits) + result = execute_interactive_session( + command, + interactor_command, + resolve_problem_regular_file(problem_dir, case["input"]), + resolve_problem_regular_file(problem_dir, case["jury_answer"]), + work_dir=problem_dir, + time_limit=min(limits["time"], cancellation.remaining()), + memory_limit_mb=limits["memory"], + idle_limit=idle_limit, + transcript_limit=transcript_limit, + output_limit_bytes=limits["output_bytes"], + process_limit=limits["processes"], + cancel_check=cancellation.requested, + ) + summary = _interactor_summary(case, result) + summary["transcript"]["limit"] = transcript_limit + summaries.append(summary) + if summary["infrastructure_failed"]: + return summaries, "infrastructure-failed" + if any(not item["matched"] for item in summaries): + return summaries, "expectation-failed" + return summaries, "passed" + + +def _live_identity(problem_dir, cancellation=None): + check = cancellation.remaining if cancellation is not None else None + if check is not None: + check() + config = read_yaml(Path(problem_dir) / "probhub.yaml") + if check is not None: + check() + report = inspect_judge_qa(problem_dir, config, check=check) + if not report.get("ok"): + raise ProbHubError("Judge QA configuration changed or became invalid", code="inputs_changed") + return config, report, ( + compute_source_hash(problem_dir, config, check=check), + compute_data_hash(problem_dir, config, check=check), + report.get("fixture_hash"), + ) + + +def _evidence_path(problem_dir): + return Path(problem_dir) / ".probhub" / JUDGE_QA_EVIDENCE_FILENAME + + +def _publish_evidence(problem_dir, evidence, expected_identity, cancellation): + with workspace_file_lock( + problem_dir, + Path(".probhub") / JUDGE_QA_EVIDENCE_LOCK_FILENAME, + busy_code="judge_qa_evidence_busy", + busy_message="another Judge QA evidence publisher is running", + wait_timeout=0, + no_follow=True, + ): + cancellation.remaining() + _, _, current_identity = _live_identity(problem_dir, cancellation) + if current_identity != expected_identity: + raise ProbHubError( + "Judge QA inputs changed before evidence publication", + code="inputs_changed", + ) + cancellation.remaining() + atomic_write_json(_evidence_path(problem_dir), evidence) + + +def _build_evidence( + identity, + limits, + validators, + compilers, + cases, + probes, + elapsed, + overall_timeout, +): + now = datetime.now(timezone.utc).isoformat() + return { + "schema_version": JUDGE_QA_EVIDENCE_SCHEMA_VERSION, + "policy_version": JUDGE_QA_POLICY_VERSION, + "sandbox_schema_version": SANDBOX_CACHE_SCHEMA_VERSION, + "probhub_version": __version__, + "source_hash": identity[0], + "data_hash": identity[1], + "fixture_hash": identity[2], + "published_at": now, + "measurement": { + "platform": platform.system(), + "platform_release": platform.release(), + "machine": platform.machine(), + "python": platform.python_version(), + "target_guarantee": False, + }, + "limits": { + "time": limits["time"], + "memory": limits["memory"], + "output_bytes": limits["output_bytes"], + "configured_output_bytes": limits["configured_output_bytes"], + "processes": limits["processes"], + "overall_timeout": float(overall_timeout), + "validator": { + "timeout": VALIDATOR_TIMEOUT_SECONDS, + "memory": VALIDATOR_MEMORY_LIMIT_MB, + "output_bytes": VALIDATOR_OUTPUT_LIMIT_BYTES, + "processes": DEFAULT_PROCESS_LIMIT, + }, + **( + {"checker_timeout": limits["checker_timeout"]} + if "checker_timeout" in limits + else {} + ), + **( + { + "idle_limit": limits["idle_limit"], + "configured_idle_limit": limits["configured_idle_limit"], + "transcript_limit": limits["transcript_limit"], + "configured_transcript_limit": limits[ + "configured_transcript_limit" + ], + } + if "transcript_limit" in limits + else {} + ), + }, + "elapsed": elapsed, + "compilers": compilers, + "validators": validators, + "cases": cases, + "probes": probes, + "cleanup": {"ok": True, "snapshot_removed": True}, + } + + +def _execute_snapshot(snapshot_problem, config, report, cancellation): + limits = _runtime_limits(config) + build_dir = Path(snapshot_problem) / ".judge-qa-build" + build_dir.mkdir(parents=True, exist_ok=True) + compiled = [] + validator_command, compile_info = _compile_program( + _program_path(snapshot_problem, config, "validator"), build_dir, "validator", cancellation + ) + compiled.append(compile_info) + validators = _validate_inputs( + validator_command, snapshot_problem, report["cases"], cancellation + ) + + if report["judge_type"] == "custom": + checker_command, compile_info = _compile_program( + _program_path(snapshot_problem, config, "checker"), build_dir, "checker", cancellation + ) + compiled.append(compile_info) + cases, probes, status = _run_checker_qa( + snapshot_problem, report, checker_command, limits, cancellation + ) + else: + limits.update(_interactive_limits(config, limits)) + interactor_command, compile_info = _compile_program( + _program_path(snapshot_problem, config, "interactor"), build_dir, "interactor", cancellation + ) + compiled.append(compile_info) + contestant_commands = {} + for case in report["cases"]: + source = (case.get("contestant") or {}).get("source") + if not source or source.casefold() in contestant_commands: + continue + command, compile_info = _compile_program( + resolve_problem_regular_file(snapshot_problem, source), + build_dir, + "contestant", + cancellation, + ) + contestant_commands[source.casefold()] = command + compiled.append(compile_info) + cases, status = _run_interactor_qa( + snapshot_problem, + report, + interactor_command, + contestant_commands, + config, + limits, + cancellation, + ) + probes = [] + if any(item.get("kind") == "cpp17" for item in compiled): + compiler_identity = _compiler_identity(cancellation) + for item in compiled: + if item.get("kind") == "cpp17": + item["compiler_identity"] = compiler_identity + return limits, validators, compiled, cases, probes, status + + +def judge_qa_problem( + root, + problem_dir, + *, + timeout=DEFAULT_JUDGE_QA_TIMEOUT_SECONDS, + cancel_check=None, +): + """Execute configured Checker/Interactor fixtures without mutating formal artifacts.""" + + del root # Reserved for the P2 CLI/workspace adapter. + problem_dir = Path(problem_dir).resolve() + started = time.monotonic() + try: + timeout = float(timeout) + except (TypeError, ValueError, OverflowError) as exc: + return _result( + "infrastructure-failed", + code="judge_qa_timeout_invalid", + message=exc, + ) + if ( + not math.isfinite(timeout) + or timeout < 0 + or timeout > MAX_JUDGE_QA_TIMEOUT_SECONDS + ): + return _result( + "infrastructure-failed", + code="judge_qa_timeout_invalid", + message=( + "Judge QA timeout must be a finite non-negative number no greater than " + f"{MAX_JUDGE_QA_TIMEOUT_SECONDS:g} seconds" + ), + ) + cancellation = _RunCancellation(timeout, cancel_check) + try: + with workspace_file_lock( + problem_dir, + ".probhub/judge.lock", + busy_code="judge_busy", + busy_message="another Judge is already running for this problem", + no_follow=True, + ): + cancellation.remaining() + config = read_yaml(problem_dir / "probhub.yaml") + cancellation.remaining() + report = inspect_judge_qa( + problem_dir, config, check=cancellation.remaining + ) + if not report.get("configured"): + return _result( + "not-configured", + ok=True, + applicable=False, + judge_type=report.get("judge_type"), + ) + cancellation.remaining() + if not report.get("ok"): + return _result( + "infrastructure-failed", + code="judge_qa_schema_invalid", + message="Judge QA configuration is invalid", + diagnostics=report.get("diagnostics") or [], + ) + identity = ( + compute_source_hash( + problem_dir, config, check=cancellation.remaining + ), + compute_data_hash( + problem_dir, config, check=cancellation.remaining + ), + report.get("fixture_hash"), + ) + temporary_root = Path(tempfile.mkdtemp(prefix="probhub-judge-qa-")) + snapshot_problem = temporary_root / "problem" + execution = None + execution_error = None + cleanup_error = None + try: + copied_config, copied_report = _create_snapshot( + problem_dir, + snapshot_problem, + config, + report, + identity, + cancellation, + ) + execution = _execute_snapshot( + snapshot_problem, copied_config, copied_report, cancellation + ) + _, _, snapshot_identity = _live_identity( + snapshot_problem, cancellation + ) + if snapshot_identity != identity: + raise ProbHubError( + "Judge QA snapshot inputs changed during execution", + code="inputs_changed", + ) + except Exception as exc: + execution_error = exc + finally: + try: + _remove_snapshot(temporary_root) + except Exception as exc: + cleanup_error = exc + if cleanup_error is not None: + return _result( + "infrastructure-failed", + code="judge_qa_snapshot_cleanup_failed", + message=cleanup_error, + ) + if execution_error is not None: + if not isinstance(execution_error, (ProcessCancelled, _OverallTimeout)): + _, _, current_identity = _live_identity( + problem_dir, cancellation + ) + if current_identity != identity: + return _result( + "infrastructure-failed", + code="inputs_changed", + message="Judge QA inputs changed during execution", + ) + raise execution_error + limits, validators, compilers, cases, probes, status = execution + cancellation.remaining() + _, _, current_identity = _live_identity(problem_dir, cancellation) + if current_identity != identity: + return _result( + "infrastructure-failed", + code="inputs_changed", + message="Judge QA inputs changed during execution", + cases=cases, + probes=probes, + ) + if status != "passed": + return _result( + status, + code=( + "judge_qa_infrastructure_failed" + if status == "infrastructure-failed" + else "judge_qa_expectation_failed" + ), + cases=cases, + probes=probes, + validators=validators, + ) + elapsed = time.monotonic() - started + evidence = _build_evidence( + identity, + limits, + validators, + compilers, + cases, + probes, + elapsed, + timeout, + ) + _publish_evidence(problem_dir, evidence, identity, cancellation) + return _result( + "passed", + ok=True, + code="judge_qa_passed", + cases=cases, + probes=probes, + validators=validators, + evidence=evidence, + evidence_path=str(_evidence_path(problem_dir)), + evidence_published=True, + ) + except ProcessCancelled as exc: + if cancellation.cause == "overall_timeout": + return _result( + "infrastructure-failed", + code="judge_qa_overall_timeout", + message="Judge QA overall deadline exceeded", + ) + return _result("cancelled", code="cancelled", message=exc) + except _OverallTimeout as exc: + return _result( + "infrastructure-failed", + code="judge_qa_overall_timeout", + message=exc, + ) + except (ProbHubError, OSError, OutputBudgetError, ValueError, TypeError) as exc: + return _result( + "infrastructure-failed", + code=getattr(exc, "code", None) or "judge_qa_infrastructure_failed", + message=exc, + ) diff --git a/probhub/linting.py b/probhub/linting.py index 8bebd3a..a306c35 100644 --- a/probhub/linting.py +++ b/probhub/linting.py @@ -90,10 +90,12 @@ def _configured_file_error(field, value, reason, missing_label): return f"{missing_label} not found: {value}" -def problem_statement_asset_paths(problem_dir): +def problem_statement_asset_paths(problem_dir, *, check=None): problem_dir = Path(problem_dir).resolve() paths = [] for path in problem_dir.rglob("*"): + if check is not None: + check() if path.is_symlink() or not path.is_file(): continue relative = path.relative_to(problem_dir) @@ -113,8 +115,10 @@ def compute_statement_assets_hash(problem_dir): return hash_paths(problem_dir, relative_paths)[0] -def problem_source_paths(problem_dir, config): +def problem_source_paths(problem_dir, config, *, check=None): problem_dir = Path(problem_dir).resolve() + if check is not None: + check() paths = [problem_dir / "probhub.yaml"] statement_config = config.get("statement") or {} statement_source = ( @@ -154,30 +158,32 @@ def problem_source_paths(problem_dir, config): paths.append(candidate) code_dir = problem_dir / "code" if code_dir.is_dir(): - paths.extend( - path - for path in code_dir.rglob("*") - if path.is_file() - and not path.is_symlink() - and not any( - part.casefold() in CODE_HASH_IGNORED_DIRS - for part in path.relative_to(code_dir).parts[:-1] - ) - and path.suffix.lower() not in CODE_HASH_IGNORED_SUFFIXES - ) - paths.extend(judge_fixture_tree_paths(problem_dir)) - paths.extend(problem_statement_asset_paths(problem_dir)) + for path in code_dir.rglob("*"): + if check is not None: + check() + if ( + path.is_file() + and not path.is_symlink() + and not any( + part.casefold() in CODE_HASH_IGNORED_DIRS + for part in path.relative_to(code_dir).parts[:-1] + ) + and path.suffix.lower() not in CODE_HASH_IGNORED_SUFFIXES + ): + paths.append(path) + paths.extend(judge_fixture_tree_paths(problem_dir, check=check)) + paths.extend(problem_statement_asset_paths(problem_dir, check=check)) return paths -def compute_source_hash(problem_dir, config): +def compute_source_hash(problem_dir, config, *, check=None): problem_dir = Path(problem_dir).resolve() relative_paths = [ path.relative_to(problem_dir) - for path in problem_source_paths(problem_dir, config) + for path in problem_source_paths(problem_dir, config, check=check) if path.exists() ] - return hash_paths(problem_dir, relative_paths)[0] + return hash_paths(problem_dir, relative_paths, check=check)[0] def compute_workspace_hash(root, workspace): @@ -220,12 +226,20 @@ def compute_collection_hash(root, workspace, loaded_problems=None): return hashlib.sha256(payload).hexdigest() -def compute_data_hash(problem_dir, config): +def compute_data_hash(problem_dir, config, *, check=None): data = config.get("data") or {} paths = [] for key, default in (("sample_dir", "data/sample"), ("secret_dir", "data/secret")): - paths.extend(files_under(problem_dir / data.get(key, default), {".in", ".ans"})) - return hash_paths(problem_dir, [path.relative_to(problem_dir) for path in paths])[0] + paths.extend(files_under( + problem_dir / data.get(key, default), + {".in", ".ans"}, + check=check, + )) + return hash_paths( + problem_dir, + [path.relative_to(problem_dir) for path in paths], + check=check, + )[0] def lint_problem(root, workspace, entry): diff --git a/probhub/process_control.py b/probhub/process_control.py index 0925f42..8834ef4 100644 --- a/probhub/process_control.py +++ b/probhub/process_control.py @@ -14,6 +14,9 @@ DEFAULT_PROCESS_LIMIT = 32 +VALIDATOR_TIMEOUT_SECONDS = 5.0 +VALIDATOR_MEMORY_LIMIT_MB = 512 +VALIDATOR_OUTPUT_LIMIT_BYTES = 8 * 1024 * 1024 DEFAULT_POLL_INTERVAL = 0.005 CANCEL_FILE_ENV = "PROBHUB_CANCEL_FILE" WINDOWS_CREATE_SUSPENDED = 0x00000004 @@ -91,6 +94,11 @@ def cancellation_requested(): return False +def _cancellation_requested(cancel_check): + check = cancellation_requested if cancel_check is None else cancel_check + return bool(check()) + + def _same_executable(left, right): try: return os.path.samefile(left, right) @@ -899,6 +907,7 @@ def wait_managed( optional_output_paths=(), output_limit_bytes=None, poll_interval=DEFAULT_POLL_INTERVAL, + cancel_check=None, ): """Wait for a managed process and always clean the complete process tree.""" started = time.perf_counter() @@ -917,7 +926,7 @@ def wait_managed( try: while managed.proc.poll() is None: now = time.perf_counter() - if cancellation_requested(): + if _cancellation_requested(cancel_check): reason, message = "cancelled", "execution cancelled" break count = current_memory = None @@ -952,7 +961,7 @@ def wait_managed( # A short process can exit before the first polling iteration. Recheck # file size and deadline after exit so fast output floods cannot bypass OLE. observed_output_bytes = _files_size(output_paths, optional_output_paths) - if cancellation_requested(): + if _cancellation_requested(cancel_check): reason, message = "cancelled", "execution cancelled" elif output_limit_bytes is not None and observed_output_bytes > int(output_limit_bytes): reason, message = "output_limit", "output limit exceeded" @@ -1008,11 +1017,12 @@ def run_managed_to_files( process_limit=DEFAULT_PROCESS_LIMIT, cwd=None, env=None, + cancel_check=None, ): input_stream = None temporary_input = None try: - if cancellation_requested(): + if _cancellation_requested(cancel_check): raise ProcessCancelled("execution cancelled") if input_path is not None: input_stream = open(input_path, "rb") @@ -1038,6 +1048,7 @@ def run_managed_to_files( output_paths=(stdout_path, stderr_path), optional_output_paths=tuple(additional_output_paths or ()), output_limit_bytes=output_limit_bytes, + cancel_check=cancel_check, ) if result.get("reason") == "cancelled": raise ProcessCancelled(result.get("message") or "execution cancelled") diff --git a/probhub/special_judges.py b/probhub/special_judges.py index ce3307d..ba3ca8a 100644 --- a/probhub/special_judges.py +++ b/probhub/special_judges.py @@ -155,6 +155,7 @@ def run_checker_to_files( process_limit=DEFAULT_PROCESS_LIMIT, diagnostic_limit_bytes=None, env=None, + cancel_check=None, ): """Run a DOMjudge/testlib Checker and return protocol and execution evidence. @@ -212,6 +213,7 @@ def run_checker_to_files( process_limit=process_limit, cwd=cwd, env=env, + cancel_check=cancel_check, ) cleanup["checker_tree_termination"] = "completed" except OutputBudgetError as exc: @@ -471,6 +473,14 @@ def _interactive_output_classification(evidence, diagnostic_limit_bytes): return None +def _is_interactor_failure(outcome): + return bool( + outcome + and outcome.get("status") == "FAIL" + and outcome.get("actor") == "interactor" + ) + + def _interactive_result( outcome, *, @@ -558,6 +568,7 @@ def execute_interactive_session( output_limit_bytes=64 * MIB, process_limit=DEFAULT_PROCESS_LIMIT, env=None, + cancel_check=None, ): """Run one contestant/Interactor session with explicit responsibility evidence.""" monotonic_start = time.monotonic() @@ -604,6 +615,7 @@ def execute_interactive_session( cancelled_error = None spawn_actor = "supervisor" diagnostic_limit_bytes = min(max(int(output_limit_bytes), 0), MAX_CHECKER_DIAGNOSTIC_BYTES) + is_cancelled = cancellation_requested if cancel_check is None else cancel_check def current_exit_codes(): return { @@ -658,6 +670,8 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): return result try: + if is_cancelled(): + raise ProcessCancelled("execution cancelled") runtime_dir = Path(tempfile.mkdtemp(prefix=".probhub-interactive-", dir=work_dir)) solution_stderr = runtime_dir / "solution.stderr" interactor_stderr = runtime_dir / "interactor.stderr" @@ -682,6 +696,8 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): solution_process_limit_enforced = bool( getattr(solution_managed, "process_limit_enforced", False) ) + if is_cancelled(): + raise ProcessCancelled("execution cancelled") spawn_actor = "interactor" cleanup["interactor_tree_termination"] = "pending" interactor_managed = spawn_managed( @@ -742,7 +758,7 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): 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(): + if is_cancelled(): raise ProcessCancelled("execution cancelled") now = time.monotonic() evidence = current_evidence() @@ -785,8 +801,18 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): } elif interactor_proc.poll() is not None: protocol = _protocol_outcome(interactor_proc.returncode, "", "interactor") - if protocol["status"] == "FAIL": + if protocol["status"] == "FAIL" and not _is_interactor_failure( + resource_outcome + ): resource_outcome = protocol + if interactor_proc.poll() is not None: + protocol = _protocol_outcome( + interactor_proc.returncode, "", "interactor" + ) + if protocol["status"] == "FAIL" and not _is_interactor_failure( + resource_outcome + ): + resource_outcome = protocol if resource_outcome: _terminate_interactive_actor(solution_managed, "contestant", cleanup) solution_managed = None @@ -869,6 +895,13 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): interactor_message = _feedback_message( feedback_dir, interactor_message, diagnostic_limit_bytes )[0] + if resource_outcome and resource_outcome.get("status") == "FAIL": + return finish(resource_outcome) + interactor_protocol = _protocol_outcome( + interactor_proc.returncode, interactor_message, "interactor" + ) + if interactor_protocol["status"] == "FAIL": + return finish(interactor_protocol) if resource_outcome: return finish(resource_outcome) if solution_proc.returncode != 0: @@ -892,9 +925,7 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): "message": solution_message, } return finish(outcome) - return finish(_protocol_outcome( - interactor_proc.returncode, interactor_message, "interactor" - )) + return finish(interactor_protocol) except ProcessCancelled as exc: cancelled_error = exc finish({ diff --git a/scripts/local_judge.py b/scripts/local_judge.py index 5a2b599..08f0fe9 100644 --- a/scripts/local_judge.py +++ b/scripts/local_judge.py @@ -32,6 +32,9 @@ DEFAULT_PROCESS_LIMIT, OutputBudgetError, ProcessCancelled, + VALIDATOR_MEMORY_LIMIT_MB, + VALIDATOR_OUTPUT_LIMIT_BYTES, + VALIDATOR_TIMEOUT_SECONDS, cancellation_requested, run_managed_to_files, ) @@ -1182,7 +1185,7 @@ def run_testcase( ) return status, elapsed, memory, memory_enforced, message, details -def run_validator(val_bin, in_file, timeout=5.0): +def run_validator(val_bin, in_file, timeout=VALIDATOR_TIMEOUT_SECONDS): """Validate LF-normalized input under the shared process-tree controller.""" try: with open(in_file, "rb") as fin: @@ -1196,8 +1199,8 @@ def run_validator(val_bin, in_file, timeout=5.0): stdout_path=stdout_path, stderr_path=stderr_path, timeout=timeout, - memory_limit_mb=512, - output_limit_bytes=8 * 1024 * 1024, + memory_limit_mb=VALIDATOR_MEMORY_LIMIT_MB, + output_limit_bytes=VALIDATOR_OUTPUT_LIMIT_BYTES, process_limit=DEFAULT_PROCESS_LIMIT, ) try: diff --git a/tests/test_calibration.py b/tests/test_calibration.py index 93d1df8..2a4ae53 100644 --- a/tests/test_calibration.py +++ b/tests/test_calibration.py @@ -446,6 +446,10 @@ def test_old_workspace_gitignore_is_migrated_for_evidence_and_lock(self): self.assertIn("**/.probhub/judge-evidence-v1.json.*.tmp", content) self.assertIn("**/.probhub/judge-evidence-v2.json", content) self.assertIn("**/.probhub/judge-evidence-v2.json.*.tmp", content) + self.assertIn("**/.probhub/judge-qa-evidence.lock", content) + self.assertIn("**/.probhub/judge-qa-evidence-v1.json", content) + self.assertIn("**/.probhub/judge-qa-evidence-v1.json.*.tmp", content) + self.assertIn("**/.probhub/judge-qa-tmp/", content) class LocalJudgeCalibrationTests(unittest.TestCase): diff --git a/tests/test_judge_qa_execution.py b/tests/test_judge_qa_execution.py new file mode 100644 index 0000000..1311da7 --- /dev/null +++ b/tests/test_judge_qa_execution.py @@ -0,0 +1,1169 @@ +import json +import os +import shutil +import sys +import tempfile +import threading +import time +import unittest +from collections import Counter +from pathlib import Path +from unittest import mock + +from probhub.errors import ProbHubError +from probhub.build_lock import workspace_file_lock +from probhub.io import write_yaml +from probhub.judge_qa import judge_qa_problem +from probhub.process_control import process_alive +import probhub.judge_qa_runtime as runtime +from tests.fixture_support import FIXTURE_ROOT, copy_workspace_fixture + + +class JudgeQAExecutionTests(unittest.TestCase): + def copy_fixture(self, name): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + return copy_workspace_fixture(name, temporary.name) + + @staticmethod + def evidence_path(fixture): + return fixture.problem / ".probhub" / runtime.JUDGE_QA_EVIDENCE_FILENAME + + def seed_old_evidence(self, fixture): + path = self.evidence_path(fixture) + path.parent.mkdir(parents=True, exist_ok=True) + content = b'{"sentinel":"last-known-good"}\n' + path.write_bytes(content) + return path, content + + @staticmethod + def seed_formal_artifacts(fixture): + sentinels = { + fixture.problem / "meta.json": b"old meta\n", + fixture.problem / "problem.yaml": b"old problem config\n", + fixture.problem / "domjudge-problem.ini": b"old domjudge config\n", + fixture.problem / "problem.pdf": b"old problem pdf\n", + fixture.problem / f"{fixture.problem.name}.zip": b"old package\n", + fixture.problem / ".probhub" / "build-manifest.json": b"old manifest\n", + fixture.root / "main.pdf": b"old contest pdf\n", + } + for path, content in sentinels.items(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + + @staticmethod + def formal_workspace_bytes(fixture): + ignored = { + runtime.JUDGE_QA_EVIDENCE_FILENAME, + runtime.JUDGE_QA_EVIDENCE_LOCK_FILENAME, + "judge.lock", + } + return { + path.relative_to(fixture.root).as_posix(): path.read_bytes() + for path in fixture.root.rglob("*") + if path.is_file() + and not ( + path.parent.name == ".probhub" + and ( + path.name in ignored + or path.name.startswith( + runtime.JUDGE_QA_EVIDENCE_FILENAME + "." + ) + ) + ) + } + + @staticmethod + def fake_success_execution(config): + limits = runtime._runtime_limits(config) + return limits, [], [], [], [], "passed" + + def assert_old_evidence_preserved(self, fixture, expected): + self.assertEqual(self.evidence_path(fixture).read_bytes(), expected) + + def assert_evidence_has_no_stream_content(self, evidence): + def visit(value): + if isinstance(value, dict): + self.assertNotIn("stdout", value) + self.assertNotIn("stderr", value) + self.assertNotIn("entries", value) + if "transcript" in value: + self.assertLessEqual( + set(value["transcript"]), + {"bytes", "limit", "truncated"}, + ) + for child in value.values(): + visit(child) + elif isinstance(value, list): + for child in value: + visit(child) + + visit(evidence) + serialized = json.dumps(evidence, ensure_ascii=False).lower() + self.assertNotIn("transcript_entries", serialized) + self.assertNotIn("stdout_text", serialized) + self.assertNotIn("stderr_text", serialized) + + def run_with_compile_count(self, fixture): + calls = [] + original = runtime._compile_program + + def counted(source, build_dir, role, cancellation): + calls.append((Path(source).name, role)) + return original(source, build_dir, role, cancellation) + + with mock.patch.object(runtime, "_compile_program", side_effect=counted): + result = judge_qa_problem(fixture.root, fixture.problem) + return result, calls + + def assert_successful_evidence(self, fixture, result): + self.assertTrue(result["ok"], result) + self.assertEqual(result["status"], "passed", result) + self.assertEqual(result["code"], "judge_qa_passed", result) + self.assertTrue(result["evidence_published"], result) + evidence_path = self.evidence_path(fixture) + self.assertEqual(Path(result["evidence_path"]), evidence_path) + self.assertEqual( + json.loads(evidence_path.read_text(encoding="utf-8")), + result["evidence"], + ) + self.assertTrue(result["evidence"]["cleanup"]["snapshot_removed"]) + self.assert_evidence_has_no_stream_content(result["evidence"]) + self.assertEqual( + list(evidence_path.parent.glob(f"{evidence_path.name}.*.tmp")), + [], + ) + + def test_checker_fixture_executes_and_publishes_bounded_evidence(self): + fixture = self.copy_fixture("checker-qa") + self.seed_formal_artifacts(fixture) + before = self.formal_workspace_bytes(fixture) + result, compile_calls = self.run_with_compile_count(fixture) + + self.assert_successful_evidence(fixture, result) + self.assertEqual( + Counter(role for _, role in compile_calls), + Counter({"validator": 1, "checker": 1}), + ) + self.assertEqual(len(compile_calls), len(set(compile_calls)), compile_calls) + self.assertEqual( + {case["id"] for case in result["cases"]}, + {"accepts-alternative", "rejects-extra-token"}, + ) + self.assertTrue(all(case["matched"] for case in result["cases"]), result) + self.assertEqual( + {probe["id"] for probe in result["probes"]}, + {"empty", "truncated", "extra-token", "oversized"}, + ) + self.assertEqual( + result["evidence"]["limits"]["checker_timeout"], + runtime.MIN_CHECKER_TIMEOUT_SECONDS, + ) + self.assertEqual( + result["evidence"]["limits"]["validator"], + { + "timeout": runtime.VALIDATOR_TIMEOUT_SECONDS, + "memory": runtime.VALIDATOR_MEMORY_LIMIT_MB, + "output_bytes": runtime.VALIDATOR_OUTPUT_LIMIT_BYTES, + "processes": runtime.DEFAULT_PROCESS_LIMIT, + }, + ) + self.assertTrue(all(probe["matched"] for probe in result["probes"]), result) + self.assertTrue(all( + "manual_review_required" in probe for probe in result["probes"] + )) + self.assertEqual(self.formal_workspace_bytes(fixture), before) + + def test_cpp_checker_qa_runs_from_unicode_and_space_path(self): + with tempfile.TemporaryDirectory() as temp: + workspace = Path(temp) / "中文 workspace" + fixture = copy_workspace_fixture("checker-qa", workspace) + temporary_root = fixture.root / "中文 snapshot root" + original_mkdtemp = tempfile.mkdtemp + calls = 0 + + def make_temporary(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + temporary_root.mkdir() + return str(temporary_root) + return original_mkdtemp(*args, **kwargs) + + with mock.patch.object( + runtime.tempfile, + "mkdtemp", + side_effect=make_temporary, + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assert_successful_evidence(fixture, result) + self.assertFalse(temporary_root.exists()) + + def test_probe_ac_is_preserved_as_manual_review_not_automatic_safety(self): + fixture = self.copy_fixture("checker-qa") + report = runtime.inspect_judge_qa(fixture.problem, fixture.config()) + for case in report["cases"]: + case["expected"]["status"] = "AC" + checker_ac = { + "verdict": "AC", + "execution_status": "completed", + "failure_kind": None, + "actor": "session", + "termination_reason": "completed", + "returncode": 0, + "time": 0.001, + "memory": 1.0, + "memory_enforced": True, + "process_limit_enforced": True, + "output_bytes": 0, + "retained_output_bytes": 0, + "output_truncated": False, + "cleanup": {"ok": True, "errors": []}, + } + cancellation = runtime._RunCancellation(10, None) + with mock.patch.object( + runtime, "run_checker_to_files", return_value=checker_ac + ): + cases, probes, status = runtime._run_checker_qa( + fixture.problem, + report, + ["checker"], + runtime._runtime_limits(fixture.config()), + cancellation, + ) + self.assertEqual(status, "passed", (cases, probes)) + self.assertTrue(probes) + self.assertTrue(all(item["manual_review_required"] for item in probes)) + + def test_resource_and_control_failures_cannot_satisfy_checker_expectations(self): + case = {"id": "fault", "purpose": "fault", "expected": {"status": "WA"}} + for execution_status, failure_kind in ( + ("time_limit", "resource_limit"), + ("memory_limit", "resource_limit"), + ("output_limit", "resource_limit"), + ("process_limit", "resource_limit"), + ("completed", "judge_failure"), + ("start_error", "startup_failure"), + ("output_control_error", "control_failure"), + ): + summary = runtime._checker_summary(case, { + "verdict": None, + "execution_status": execution_status, + "failure_kind": failure_kind, + "actor": "checker", + "termination_reason": execution_status, + "cleanup": {"ok": True, "errors": []}, + }) + self.assertTrue(summary["infrastructure_failed"], summary) + self.assertFalse(summary["matched"], summary) + + def test_checker_timeout_matches_formal_judge_policy(self): + fixture = self.copy_fixture("checker-qa") + report = runtime.inspect_judge_qa(fixture.problem, fixture.config()) + report["cases"] = [report["cases"][0]] + report["robustness"] = None + checker_ac = { + "verdict": "AC", + "execution_status": "completed", + "failure_kind": None, + "actor": "session", + "termination_reason": "completed", + "cleanup": {"ok": True, "errors": []}, + } + + for configured_time, expected_timeout in ((1, 5.0), (20, 20.0)): + with self.subTest(configured_time=configured_time): + config = fixture.config() + config["limits"]["time"] = configured_time + limits = runtime._runtime_limits(config) + with mock.patch.object( + runtime, + "run_checker_to_files", + return_value=checker_ac, + ) as checker: + cases, probes, status = runtime._run_checker_qa( + fixture.problem, + report, + ["checker"], + limits, + runtime._RunCancellation(60, None), + ) + self.assertEqual(status, "passed", (cases, probes)) + actual_timeout = checker.call_args.kwargs["timeout"] + self.assertGreater(actual_timeout, expected_timeout - 0.1) + self.assertLessEqual(actual_timeout, expected_timeout) + self.assertEqual(limits["checker_timeout"], expected_timeout) + + def test_interactor_failures_remain_infrastructure_but_contestant_limits_do_not(self): + case = {"id": "fault", "purpose": "fault", "expected": {"status": "OLE"}} + for actor in ("interactor", "supervisor"): + summary = runtime._interactor_summary(case, { + "status": "FAIL", + "actor": actor, + "execution_status": "output_limit", + "failure_kind": "resource_limit", + "termination_reason": "output_limit", + "cleanup": {"ok": True, "errors": []}, + }) + self.assertTrue(summary["infrastructure_failed"], summary) + self.assertFalse(summary["matched"], summary) + contestant = runtime._interactor_summary(case, { + "status": "OLE", + "actor": "contestant", + "execution_status": "output_limit", + "failure_kind": "resource_limit", + "termination_reason": "output_limit", + "cleanup": {"ok": True, "errors": []}, + }) + self.assertFalse(contestant["infrastructure_failed"], contestant) + self.assertTrue(contestant["matched"], contestant) + + def test_interactor_fixture_executes_and_compiles_unique_contestant_once(self): + fixture = self.copy_fixture("interactor-qa") + result, compile_calls = self.run_with_compile_count(fixture) + + self.assert_successful_evidence(fixture, result) + self.assertEqual( + Counter(role for _, role in compile_calls), + Counter({"validator": 1, "interactor": 1, "contestant": 1}), + ) + self.assertEqual(len(compile_calls), len(set(compile_calls)), compile_calls) + self.assertEqual( + {case["id"] for case in result["cases"]}, + { + "normal-protocol", + "early-eof-player", + "idle-player", + "output-flood-player", + }, + ) + self.assertTrue(all(case["matched"] for case in result["cases"]), result) + self.assertEqual( + result["evidence"]["limits"]["output_bytes"], + 1024 * 1024, + ) + self.assertTrue(all( + case["transcript"]["limit"] <= runtime.MAX_JUDGE_QA_TRANSCRIPT_BYTES + for case in result["cases"] + )) + + def test_interactor_fixture_executes_python_contestant_source(self): + fixture = self.copy_fixture("interactor-qa") + python_source = fixture.problem / "code" / "judge-qa" / "normal.py" + python_source.write_text( + "import sys\n" + "value = int(sys.stdin.readline())\n" + "print(2 * value, flush=True)\n", + encoding="utf-8", + ) + config = fixture.config() + config["judge"]["qa"]["cases"] = [config["judge"]["qa"]["cases"][0]] + config["judge"]["qa"]["cases"][0]["contestant"]["source"] = ( + "code/judge-qa/normal.py" + ) + write_yaml(fixture.config_path, config) + + result, compile_calls = self.run_with_compile_count(fixture) + + self.assert_successful_evidence(fixture, result) + self.assertIn(("normal.py", "contestant"), compile_calls) + contestant = next( + item for item in result["evidence"]["compilers"] + if item["role"] == "contestant" + ) + self.assertEqual(contestant["kind"], "python", contestant) + + def test_runtime_caps_output_and_transcript_without_changing_problem_limits(self): + fixture = self.copy_fixture("interactor-qa") + config = fixture.config() + config["limits"]["output"] = 1024 + config["judge"]["interactive"]["idle_limit"] = 0.01 + config["judge"]["interactive"]["transcript_limit"] = 1024 * 1024 * 1024 + limits = runtime._runtime_limits(config) + self.assertEqual(limits["output_bytes"], runtime.MAX_JUDGE_QA_OUTPUT_BYTES) + self.assertEqual(limits["configured_output_bytes"], 1024 * 1024 * 1024) + + captured = [] + + def fake_session(*_args, **kwargs): + captured.append(kwargs) + return { + "status": "AC", + "actor": "session", + "execution_status": "completed", + "failure_kind": None, + "termination_reason": "completed", + "cleanup": {"ok": True, "errors": []}, + "transcript_bytes": 0, + "transcript_truncated": False, + } + + report = runtime.inspect_judge_qa(fixture.problem, fixture.config()) + report["cases"] = [report["cases"][0]] + with mock.patch.object( + runtime, "execute_interactive_session", side_effect=fake_session + ): + cases, status = runtime._run_interactor_qa( + fixture.problem, + report, + ["interactor"], + {"code/judge-qa/normal.cpp": ["contestant"]}, + config, + limits, + runtime._RunCancellation(10, None), + ) + self.assertEqual(status, "passed", cases) + self.assertEqual(captured[0]["output_limit_bytes"], runtime.MAX_JUDGE_QA_OUTPUT_BYTES) + self.assertEqual( + captured[0]["transcript_limit"], runtime.MAX_JUDGE_QA_TRANSCRIPT_BYTES + ) + self.assertEqual(captured[0]["idle_limit"], runtime.MIN_JUDGE_QA_IDLE_SECONDS) + effective_limits = dict(limits) + effective_limits.update(runtime._interactive_limits(config, limits)) + evidence = runtime._build_evidence( + ("source", "data", "fixture"), + effective_limits, + [], + [], + cases, + [], + 0.1, + 12.5, + ) + self.assertEqual(evidence["limits"]["configured_idle_limit"], 0.01) + self.assertEqual( + evidence["limits"]["idle_limit"], runtime.MIN_JUDGE_QA_IDLE_SECONDS + ) + self.assertEqual( + evidence["limits"]["configured_transcript_limit"], + 1024 * 1024 * 1024, + ) + self.assertEqual( + evidence["limits"]["transcript_limit"], + runtime.MAX_JUDGE_QA_TRANSCRIPT_BYTES, + ) + self.assertEqual(evidence["limits"]["overall_timeout"], 12.5) + + def test_direct_api_rejects_invalid_runtime_limits_without_publishing(self): + mutations = ( + ("checker-qa", lambda config: config["limits"].update(time=-1)), + ("checker-qa", lambda config: config["limits"].update(memory=128)), + ("checker-qa", lambda config: config["limits"].update(output=0)), + ("checker-qa", lambda config: config["limits"].update(processes=0)), + ( + "interactor-qa", + lambda config: config["judge"]["interactive"].update(idle_limit=-1), + ), + ( + "interactor-qa", + lambda config: config["judge"]["interactive"].update( + transcript_limit=-1 + ), + ), + ) + for name, mutate in mutations: + with self.subTest(name=name, mutation=mutate): + fixture = self.copy_fixture(name) + config = fixture.config() + mutate(config) + write_yaml(fixture.config_path, config) + _, old = self.seed_old_evidence(fixture) + result = judge_qa_problem(fixture.root, fixture.problem) + self.assertFalse(result["ok"], result) + self.assertEqual(result["code"], "judge_qa_limits_invalid", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_not_configured_is_successful_and_does_not_publish(self): + fixture = self.copy_fixture("custom") + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertTrue(result["ok"], result) + self.assertFalse(result["applicable"], result) + self.assertEqual(result["status"], "not-configured", result) + self.assertFalse(self.evidence_path(fixture).exists()) + + def test_expectation_mismatch_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + config["judge"]["qa"]["cases"][1]["expected"]["status"] = "AC" + write_yaml(fixture.config_path, config) + _, old = self.seed_old_evidence(fixture) + self.seed_formal_artifacts(fixture) + before = self.formal_workspace_bytes(fixture) + + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "expectation-failed", result) + self.assertEqual(result["code"], "judge_qa_expectation_failed", result) + mismatched = {case["id"]: case for case in result["cases"]} + self.assertFalse(mismatched["rejects-extra-token"]["matched"], result) + self.assert_old_evidence_preserved(fixture, old) + self.assertEqual(self.formal_workspace_bytes(fixture), before) + + def test_validator_rejection_is_infrastructure_failure_and_preserves_evidence(self): + fixture = self.copy_fixture("checker-qa") + rejected = fixture.problem / "code" / "validator-reject.cpp" + shutil.copy2(FIXTURE_ROOT / "faults" / "validator-reject.cpp", rejected) + config = fixture.config() + config["judge"]["validator"] = "code/validator-reject.cpp" + write_yaml(fixture.config_path, config) + _, old = self.seed_old_evidence(fixture) + + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_validator_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_validator_execution_matches_formal_judge_policy(self): + fixture = self.copy_fixture("checker-qa") + report = runtime.inspect_judge_qa(fixture.problem, fixture.config()) + validator_ok = { + "reason": "completed", + "returncode": 0, + "time": 0.001, + "message": "", + } + + with mock.patch.object( + runtime, + "run_managed_to_files", + return_value=validator_ok, + ) as execute: + results = runtime._validate_inputs( + ["validator"], + fixture.problem, + [report["cases"][0]], + runtime._RunCancellation(60, None), + ) + + self.assertTrue(results[0]["ok"], results) + call = execute.call_args.kwargs + self.assertGreater( + call["timeout"], runtime.VALIDATOR_TIMEOUT_SECONDS - 0.1 + ) + self.assertLessEqual(call["timeout"], runtime.VALIDATOR_TIMEOUT_SECONDS) + self.assertEqual(call["memory_limit_mb"], runtime.VALIDATOR_MEMORY_LIMIT_MB) + self.assertEqual( + call["output_limit_bytes"], runtime.VALIDATOR_OUTPUT_LIMIT_BYTES + ) + self.assertEqual(call["process_limit"], runtime.DEFAULT_PROCESS_LIMIT) + + def test_checker_fail_is_infrastructure_failure_and_preserves_evidence(self): + fixture = self.copy_fixture("checker-qa") + failing = fixture.problem / "code" / "checker-fail.cpp" + shutil.copy2(FIXTURE_ROOT / "faults" / "checker-fail.cpp", failing) + config = fixture.config() + config["judge"]["checker"] = "code/checker-fail.cpp" + write_yaml(fixture.config_path, config) + _, old = self.seed_old_evidence(fixture) + + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_infrastructure_failed", result) + self.assertTrue(result["cases"][0]["infrastructure_failed"], result) + self.assert_old_evidence_preserved(fixture, old) + + def test_cancel_before_execution_preserves_old_evidence(self): + fixture = self.copy_fixture("interactor-qa") + _, old = self.seed_old_evidence(fixture) + self.seed_formal_artifacts(fixture) + before = self.formal_workspace_bytes(fixture) + + result = judge_qa_problem( + fixture.root, + fixture.problem, + cancel_check=lambda: True, + ) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "cancelled", result) + self.assertEqual(result["code"], "cancelled", result) + self.assert_old_evidence_preserved(fixture, old) + self.assertEqual(self.formal_workspace_bytes(fixture), before) + + def test_running_cpp_checker_cancel_file_cleans_tree_and_snapshot(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + parent_pid = fixture.root / "checker-parent.pid" + child_pid = fixture.root / "checker-child.pid" + cancel_file = fixture.root / "cancel.requested" + checker = fixture.problem / "code" / "checker-cancellable.cpp" + checker.write_text( + '#include "testlib.h"\n' + "#include \n" + "#include \n" + "#include \n" + "#include \n" + "#include \n" + "#ifdef _WIN32\n" + "#include \n" + "#define GETPID _getpid\n" + "#else\n" + "#include \n" + "#include \n" + "#define GETPID getpid\n" + "#endif\n" + f'const char* parent_path = R"PID({parent_pid.as_posix()})PID";\n' + f'const char* child_path = R"PID({child_pid.as_posix()})PID";\n' + "int main(int argc, char** argv) {\n" + ' if (argc == 2 && std::string(argv[1]) == "--child") {\n' + " std::ofstream(child_path) << GETPID();\n" + " std::this_thread::sleep_for(std::chrono::seconds(30));\n" + " return 0;\n" + " }\n" + " std::ofstream(parent_path) << GETPID();\n" + " registerTestlibCmd(argc, argv);\n" + "#ifdef _WIN32\n" + " return static_cast(_spawnl(_P_WAIT, argv[0], argv[0], \"--child\", nullptr));\n" + "#else\n" + " pid_t child = fork();\n" + " if (child == 0) {\n" + " execl(argv[0], argv[0], \"--child\", nullptr);\n" + " _exit(127);\n" + " }\n" + " if (child < 0) return 1;\n" + " int status = 0;\n" + " return waitpid(child, &status, 0) < 0 ? 1 : status;\n" + "#endif\n" + "}\n", + encoding="utf-8", + ) + config = fixture.config() + config["judge"]["checker"] = "code/checker-cancellable.cpp" + write_yaml(fixture.config_path, config) + temporary_root = fixture.root / "controlled-running-cancel-snapshot" + original_mkdtemp = tempfile.mkdtemp + calls = 0 + + def make_temporary(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + temporary_root.mkdir() + return str(temporary_root) + return original_mkdtemp(*args, **kwargs) + + def request_cancel(): + deadline = time.time() + 60 + while time.time() < deadline: + if parent_pid.is_file(): + child_deadline = time.time() + 3 + while time.time() < child_deadline and not child_pid.is_file(): + time.sleep(0.02) + cancel_file.write_text("cancel\n", encoding="utf-8") + return + time.sleep(0.02) + + requester = threading.Thread(target=request_cancel, daemon=True) + requester.start() + with ( + mock.patch.dict( + os.environ, + {runtime.CANCEL_FILE_ENV: str(cancel_file)}, + ), + mock.patch.object( + runtime.tempfile, + "mkdtemp", + side_effect=make_temporary, + ), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + requester.join(timeout=60) + + self.assertFalse(requester.is_alive(), "cancel requester did not finish") + self.assertEqual(result["status"], "cancelled", result) + self.assertEqual(result["code"], "cancelled", result) + self.assertTrue(parent_pid.is_file(), "checker did not start") + self.assertTrue(child_pid.is_file(), "checker descendant did not start") + for path in (parent_pid, child_pid): + pid = int(path.read_text(encoding="utf-8")) + deadline = time.time() + 10 + while time.time() < deadline and process_alive(pid): + time.sleep(0.05) + self.assertFalse(process_alive(pid), f"process {pid} survived cancellation") + self.assertFalse(temporary_root.exists()) + self.assert_old_evidence_preserved(fixture, old) + + def test_judge_and_evidence_locks_fail_closed_without_publishing(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + + with workspace_file_lock(fixture.problem, ".probhub/judge.lock"): + judge_busy = judge_qa_problem(fixture.root, fixture.problem) + self.assertEqual(judge_busy["code"], "judge_busy", judge_busy) + self.assert_old_evidence_preserved(fixture, old) + + config = fixture.config() + with ( + workspace_file_lock( + fixture.problem, + Path(".probhub") / runtime.JUDGE_QA_EVIDENCE_LOCK_FILENAME, + no_follow=True, + ), + mock.patch.object( + runtime, + "_execute_snapshot", + return_value=self.fake_success_execution(config), + ), + ): + evidence_busy = judge_qa_problem(fixture.root, fixture.problem) + self.assertEqual( + evidence_busy["code"], "judge_qa_evidence_busy", evidence_busy + ) + self.assert_old_evidence_preserved(fixture, old) + + def test_default_cancel_file_is_honored_and_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + cancel_file = fixture.root / "cancel.requested" + cancel_file.write_text("cancel\n", encoding="utf-8") + + with mock.patch.dict( + os.environ, + {runtime.CANCEL_FILE_ENV: str(cancel_file)}, + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "cancelled", result) + self.assertEqual(result["code"], "cancelled", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_cancel_callback_failure_is_structured_and_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + + def fail_cancel_check(): + raise RuntimeError("injected cancellation callback failure") + + result = judge_qa_problem( + fixture.root, + fixture.problem, + cancel_check=fail_cancel_check, + ) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_cancel_check_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_cancel_after_execution_but_before_publish_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + _, old = self.seed_old_evidence(fixture) + checks = 0 + + def cancel_at_publish(): + nonlocal checks + checks += 1 + return checks >= 3 + + with mock.patch.object( + runtime, + "_execute_snapshot", + return_value=self.fake_success_execution(config), + ): + result = judge_qa_problem( + fixture.root, + fixture.problem, + cancel_check=cancel_at_publish, + ) + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "cancelled", result) + self.assertGreaterEqual(checks, 3) + self.assert_old_evidence_preserved(fixture, old) + + def test_overall_timeout_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + + result = judge_qa_problem(fixture.root, fixture.problem, timeout=0) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_overall_timeout", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_running_overall_timeout_cleans_descendants_and_snapshot(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + parent_pid = fixture.root / "timeout-parent.pid" + child_pid = fixture.root / "timeout-child.pid" + checker_script = fixture.root / "timeout-checker.py" + child_code = ( + "import os,pathlib,sys,time;" + "pathlib.Path(sys.argv[1]).write_text(str(os.getpid()), encoding='utf-8');" + "time.sleep(30)" + ) + checker_script.write_text( + "import os, pathlib, subprocess, sys, time\n" + f"pathlib.Path({str(parent_pid)!r}).write_text(str(os.getpid()), encoding='utf-8')\n" + f"subprocess.Popen([sys.executable, '-I', '-c', {child_code!r}, {str(child_pid)!r}])\n" + "time.sleep(30)\n", + encoding="utf-8", + ) + controlled_snapshot = fixture.root / "controlled-timeout-snapshot" + original_mkdtemp = tempfile.mkdtemp + calls = 0 + + def make_temporary(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + controlled_snapshot.mkdir() + return str(controlled_snapshot) + return original_mkdtemp(*args, **kwargs) + + def fake_compile(_source, _build_dir, role, _cancellation): + if role == "validator": + return [sys.executable, "-I", "-c", "import sys; sys.stdin.buffer.read()"], { + "role": role, + "source": "test-validator.cpp", + "kind": "test-double", + } + if role == "checker": + return [sys.executable, "-I", str(checker_script)], { + "role": role, + "source": "test-checker.cpp", + "kind": "test-double", + } + self.fail(f"unexpected compile role: {role}") + + with ( + mock.patch.object(runtime, "_compile_program", side_effect=fake_compile), + mock.patch.object(runtime.tempfile, "mkdtemp", side_effect=make_temporary), + ): + result = judge_qa_problem( + fixture.root, + fixture.problem, + timeout=1.5, + ) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_overall_timeout", result) + self.assertTrue(parent_pid.is_file(), "checker did not start before deadline") + self.assertTrue(child_pid.is_file(), "checker descendant did not start before deadline") + for path in (parent_pid, child_pid): + pid = int(path.read_text(encoding="utf-8")) + deadline = time.time() + 10 + while time.time() < deadline and process_alive(pid): + time.sleep(0.05) + self.assertFalse(process_alive(pid), f"process {pid} survived overall timeout") + self.assertFalse(controlled_snapshot.exists()) + self.assert_old_evidence_preserved(fixture, old) + + def test_timeout_above_supported_maximum_is_rejected(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + + result = judge_qa_problem( + fixture.root, + fixture.problem, + timeout=runtime.MAX_JUDGE_QA_TIMEOUT_SECONDS + 1, + ) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_timeout_invalid", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_snapshot_is_removed_after_success(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + temporary_root = fixture.root / "controlled-judge-qa-tmp" + + def make_temporary(*_args, **_kwargs): + temporary_root.mkdir() + return str(temporary_root) + + with ( + mock.patch.object(runtime.tempfile, "mkdtemp", side_effect=make_temporary), + mock.patch.object( + runtime, + "_execute_snapshot", + return_value=self.fake_success_execution(config), + ), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertTrue(result["ok"], result) + self.assertFalse(temporary_root.exists()) + + def test_snapshot_cleanup_failure_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + _, old = self.seed_old_evidence(fixture) + original_remove = runtime._remove_snapshot + + def remove_then_fail(path): + original_remove(path) + raise OSError("injected cleanup failure") + + with ( + mock.patch.object( + runtime, + "_execute_snapshot", + return_value=self.fake_success_execution(config), + ), + mock.patch.object(runtime, "_remove_snapshot", side_effect=remove_then_fail), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_snapshot_cleanup_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_snapshot_cleanup_failure_overrides_execution_cancellation(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + original_remove = runtime._remove_snapshot + + def remove_then_fail(path): + original_remove(path) + raise OSError("injected cleanup failure") + + with ( + mock.patch.object( + runtime, + "_create_snapshot", + side_effect=runtime.ProcessCancelled("injected cancellation"), + ), + mock.patch.object(runtime, "_remove_snapshot", side_effect=remove_then_fail), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_snapshot_cleanup_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_snapshot_includes_qa_files_excluded_from_general_code_hash(self): + fixture = self.copy_fixture("interactor-qa") + ignored_suffix = fixture.problem / "code" / "judge-qa" / "helper.exe" + ignored_suffix.write_bytes(b"prebuilt identity sentinel") + config = fixture.config() + report = runtime.inspect_judge_qa(fixture.problem, config) + identity = ( + runtime.compute_source_hash(fixture.problem, config), + runtime.compute_data_hash(fixture.problem, config), + report["fixture_hash"], + ) + destination = fixture.root / "snapshot-with-ignored-suffix" + self.addCleanup(lambda: shutil.rmtree(destination, ignore_errors=True)) + runtime._create_snapshot( + fixture.problem, + destination, + config, + report, + identity, + runtime._RunCancellation(10, None), + ) + copied = destination / "code" / "judge-qa" / "helper.exe" + self.assertEqual(copied.read_bytes(), b"prebuilt identity sentinel") + + def test_snapshot_copy_checks_cancellation_between_chunks(self): + fixture = self.copy_fixture("checker-qa") + source = fixture.problem / "judge-fixtures" / "checker" / "large.bin" + source.write_bytes(b"x" * (2 * 1024 * 1024)) + destination = fixture.root / "cancelled-copy.bin" + + class CancelDuringCopy: + calls = 0 + + def remaining(self): + self.calls += 1 + if self.calls >= 2: + raise runtime.ProcessCancelled("cancelled during copy") + return 1.0 + + cancellation = CancelDuringCopy() + with self.assertRaises(runtime.ProcessCancelled): + runtime._copy_regular_file( + source, + destination, + fixture.problem.resolve(), + cancellation, + ) + self.assertGreaterEqual(cancellation.calls, 2) + + def test_snapshot_failure_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + + with mock.patch.object( + runtime, + "_create_snapshot", + side_effect=ProbHubError("injected snapshot failure", code="snapshot_failed"), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["code"], "snapshot_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_compile_failure_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + with mock.patch.object( + runtime, + "_compile_program", + side_effect=ProbHubError("injected compile failure", code="compile_failed"), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + self.assertFalse(result["ok"], result) + self.assertEqual(result["code"], "compile_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_compiler_identity_failure_is_fail_closed(self): + cancellation = runtime._RunCancellation(10, None) + with mock.patch.object(runtime, "run_managed_to_files", return_value={ + "reason": "start_error", + "returncode": None, + "message": "injected identity failure", + }): + with self.assertRaises(ProbHubError) as caught: + runtime._compiler_identity(cancellation) + self.assertEqual(caught.exception.code, "compiler_identity_failed") + + def test_hash_fence_rejects_live_change_and_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + _, old = self.seed_old_evidence(fixture) + + def mutate_live_then_succeed(*_args, **_kwargs): + statement = fixture.problem / "problem.md" + statement.write_text( + statement.read_text(encoding="utf-8") + "\nchanged during Judge QA\n", + encoding="utf-8", + ) + return self.fake_success_execution(config) + + with mock.patch.object( + runtime, + "_execute_snapshot", + side_effect=mutate_live_then_succeed, + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "inputs_changed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_hash_fence_overrides_snapshot_expectation_failure(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + _, old = self.seed_old_evidence(fixture) + + def mutate_live_then_fail(*_args, **_kwargs): + statement = fixture.problem / "problem.md" + statement.write_text( + statement.read_text(encoding="utf-8") + "\nchanged during failure\n", + encoding="utf-8", + ) + limits = runtime._runtime_limits(config) + return limits, [], [], [{"id": "failed"}], [], "expectation-failed" + + with mock.patch.object( + runtime, + "_execute_snapshot", + side_effect=mutate_live_then_fail, + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "inputs_changed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_hash_fence_overrides_snapshot_execution_error(self): + fixture = self.copy_fixture("checker-qa") + _, old = self.seed_old_evidence(fixture) + + def mutate_live_then_raise(*_args, **_kwargs): + statement = fixture.problem / "problem.md" + statement.write_text( + statement.read_text(encoding="utf-8") + "\nchanged before error\n", + encoding="utf-8", + ) + raise ProbHubError("old snapshot failed", code="compile_failed") + + with mock.patch.object( + runtime, + "_execute_snapshot", + side_effect=mutate_live_then_raise, + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "inputs_changed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_publish_failure_preserves_old_evidence(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + _, old = self.seed_old_evidence(fixture) + + with ( + mock.patch.object( + runtime, + "_execute_snapshot", + return_value=self.fake_success_execution(config), + ), + mock.patch.object( + runtime, + "atomic_write_json", + side_effect=OSError("injected publish failure"), + ), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + + self.assertFalse(result["ok"], result) + self.assertEqual(result["status"], "infrastructure-failed", result) + self.assertEqual(result["code"], "judge_qa_infrastructure_failed", result) + self.assert_old_evidence_preserved(fixture, old) + + def test_publish_lock_hash_fence_rechecks_live_identity(self): + fixture = self.copy_fixture("checker-qa") + config = fixture.config() + _, old = self.seed_old_evidence(fixture) + original_identity = runtime._live_identity + live_checks = 0 + + def change_only_at_publish(path, cancellation=None): + nonlocal live_checks + result = original_identity(path, cancellation) + if Path(path).resolve() == fixture.problem.resolve(): + live_checks += 1 + if live_checks == 2: + identity = result[2] + return result[0], result[1], ("changed", identity[1], identity[2]) + return result + + with ( + mock.patch.object( + runtime, + "_execute_snapshot", + return_value=self.fake_success_execution(config), + ), + mock.patch.object( + runtime, "_live_identity", side_effect=change_only_at_publish + ), + ): + result = judge_qa_problem(fixture.root, fixture.problem) + self.assertFalse(result["ok"], result) + self.assertEqual(result["code"], "inputs_changed", result) + self.assertEqual(live_checks, 2) + self.assert_old_evidence_preserved(fixture, old) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_judge_qa_schema.py b/tests/test_judge_qa_schema.py index bc9e5cc..69d2298 100644 --- a/tests/test_judge_qa_schema.py +++ b/tests/test_judge_qa_schema.py @@ -126,6 +126,25 @@ def test_interactor_schema_resolves_source_and_builtin_behaviors(self): self.assertIn({"behavior": "idle"}, contestants) self.assertIn({"behavior": "output-flood"}, contestants) + def test_official_qa_programs_require_cpp_and_contestants_allow_python(self): + _, _, _, _, problem, config = self.copy_fixture("interactor-qa") + config["judge"]["interactor"] = "code/interactor.py" + config["judge"]["qa"]["cases"][0]["contestant"]["source"] = ( + "code/judge-qa/player.exe" + ) + (problem / "code" / "interactor.py").write_text("pass\n", encoding="utf-8") + (problem / "code" / "judge-qa" / "player.exe").write_bytes(b"prebuilt") + report = inspect_judge_qa(problem, config) + diagnostics = [ + item for item in report["diagnostics"] + if item["code"] == "judge_qa_program_type_unsupported" + ] + self.assertEqual( + {item["field"] for item in diagnostics}, + {"judge.interactor", "judge.qa.cases[0].contestant.source"}, + diagnostics, + ) + def test_missing_qa_is_compatible_and_not_reported_as_passed(self): _, root, workspace, _, _, _ = self.copy_fixture("custom") result = lint_workspace(root, workspace) diff --git a/tests/test_process_control.py b/tests/test_process_control.py index 2c799e8..9528914 100644 --- a/tests/test_process_control.py +++ b/tests/test_process_control.py @@ -964,6 +964,64 @@ def cancel_after_process_start(): self.assertTrue(started_pid, "managed process did not start") self.wait_until_dead(started_pid[0]) + def test_custom_cancel_check_stops_managed_process_and_descendant(self): + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + parent_pid_file = root / "parent.pid" + child_pid_file = root / "child.pid" + child = ( + "import os,time,pathlib;" + f"pathlib.Path({str(child_pid_file)!r}).write_text(str(os.getpid()), encoding='utf-8');" + "time.sleep(30)" + ) + parent = ( + "import os,pathlib,subprocess,sys,time;" + f"pathlib.Path({str(parent_pid_file)!r}).write_text(str(os.getpid()), encoding='utf-8');" + f"subprocess.Popen([sys.executable, '-c', {child!r}]);" + "time.sleep(30)" + ) + + def cancel_check(): + return parent_pid_file.is_file() and child_pid_file.is_file() + + with self.assertRaisesRegex(ProcessCancelled, "execution cancelled"): + run_managed_to_files( + [sys.executable, "-c", parent], + input_data=b"", + stdout_path=root / "stdout.txt", + stderr_path=root / "stderr.txt", + timeout=10, + memory_limit_mb=256, + output_limit_bytes=1024, + process_limit=8, + cwd=root, + cancel_check=cancel_check, + ) + + self.assertTrue(parent_pid_file.is_file(), "managed process did not start") + self.assertTrue(child_pid_file.is_file(), "managed descendant did not start") + self.wait_until_dead(int(parent_pid_file.read_text(encoding="utf-8"))) + self.wait_until_dead(int(child_pid_file.read_text(encoding="utf-8"))) + + def test_explicit_cancel_check_replaces_default_cancel_source(self): + with tempfile.TemporaryDirectory() as temp, mock.patch.object( + process_control, "cancellation_requested", return_value=True + ) as default_cancel: + root = Path(temp) + result = run_managed_to_files( + [sys.executable, "-c", "print('ok')"], + stdout_path=root / "stdout.txt", + stderr_path=root / "stderr.txt", + timeout=5, + memory_limit_mb=256, + output_limit_bytes=1024, + process_limit=8, + cwd=root, + cancel_check=lambda: False, + ) + self.assertEqual(result["reason"], "completed", result) + default_cancel.assert_not_called() + def test_external_force_cancel_kills_detached_descendant(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) diff --git a/tests/test_special_judges.py b/tests/test_special_judges.py index 1196a99..5165405 100644 --- a/tests/test_special_judges.py +++ b/tests/test_special_judges.py @@ -45,6 +45,13 @@ class _NonClosingBytesIO(io.BytesIO): def close(self): pass + def assert_processes_dead(self, pids, timeout=10): + deadline = time.time() + timeout + while time.time() < deadline and any(process_alive(pid) for pid in pids): + time.sleep(0.05) + for pid in pids: + self.assertFalse(process_alive(pid), f"process {pid} survived cancellation") + def test_output_budget_enforcement_failure_is_infrastructure_fail(self): with tempfile.TemporaryDirectory() as temp: output = str(Path(temp) / "output.txt") @@ -507,6 +514,50 @@ def test_checker_cancel_with_successful_cleanup_propagates(self): cwd=root, ) + def test_checker_custom_cancel_check_cleans_tree_and_runtime(self): + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + for name in ("case.in", "case.ans", "contestant.out"): + (root / name).write_bytes(b"") + checker_pid_file = root / "checker.pid" + child_pid_file = root / "checker-child.pid" + child = ( + "import os,time,pathlib;" + f"pathlib.Path({str(child_pid_file)!r}).write_text(str(os.getpid()), encoding='utf-8');" + "time.sleep(30)" + ) + checker = ( + "import os,pathlib,subprocess,sys,time;" + f"pathlib.Path({str(checker_pid_file)!r}).write_text(str(os.getpid()), encoding='utf-8');" + f"subprocess.Popen([sys.executable, '-c', {child!r}]);" + "time.sleep(30)" + ) + runtime_dirs = set(root.glob(".probhub-checker-*")) + + def cancel_check(): + return checker_pid_file.is_file() and child_pid_file.is_file() + + with self.assertRaisesRegex( + JUDGE_MODULE.ProcessCancelled, "execution cancelled" + ): + SPECIAL_JUDGES.run_checker_to_files( + [sys.executable, "-c", checker], + root / "case.in", + root / "case.ans", + root / "contestant.out", + timeout=10, + cwd=root, + cancel_check=cancel_check, + ) + + self.assertEqual(set(root.glob(".probhub-checker-*")), runtime_dirs) + self.assertTrue(checker_pid_file.is_file(), "checker did not start") + self.assertTrue(child_pid_file.is_file(), "checker descendant did not start") + self.assert_processes_dead([ + int(checker_pid_file.read_text(encoding="utf-8")), + int(child_pid_file.read_text(encoding="utf-8")), + ]) + def test_interactive_cleanup_failure_does_not_skip_other_cleanup(self): class FakeProcess: def __init__(self): @@ -584,7 +635,7 @@ def terminate(self): 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 + SPECIAL_JUDGES, "cancellation_requested", side_effect=[False, True] ), mock.patch.object( SPECIAL_JUDGES.shutil, "rmtree", @@ -604,6 +655,68 @@ def terminate(self): result["pre_cleanup_result"]["execution_status"], "cancelled" ) + def test_interactive_custom_cancel_check_cleans_both_trees_and_runtime(self): + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + actor_script = root / "actor.py" + actor_script.write_text( + "import os, pathlib, subprocess, sys, time\n" + "parent_pid = pathlib.Path(sys.argv[1])\n" + "child_pid = pathlib.Path(sys.argv[2])\n" + "parent_pid.write_text(str(os.getpid()), encoding='utf-8')\n" + "child = (\"import os,pathlib,sys,time;\"\n" + " \"pathlib.Path(sys.argv[1]).write_text(str(os.getpid()), encoding='utf-8');\"\n" + " \"time.sleep(30)\")\n" + "subprocess.Popen([sys.executable, '-c', child, os.fspath(child_pid)])\n" + "time.sleep(30)\n", + encoding="utf-8", + ) + actor_pid_files = [ + root / "contestant.pid", + root / "contestant-child.pid", + root / "interactor.pid", + root / "interactor-child.pid", + ] + (root / "case.in").write_bytes(b"") + (root / "case.ans").write_bytes(b"") + runtime_dirs = set(root.glob(".probhub-interactive-*")) + + def cancel_check(): + return all(path.is_file() for path in actor_pid_files) + + with self.assertRaisesRegex( + JUDGE_MODULE.ProcessCancelled, "execution cancelled" + ): + SPECIAL_JUDGES.execute_interactive_session( + [ + sys.executable, + str(actor_script), + str(actor_pid_files[0]), + str(actor_pid_files[1]), + ], + [ + sys.executable, + str(actor_script), + str(actor_pid_files[2]), + str(actor_pid_files[3]), + ], + root / "case.in", + root / "case.ans", + work_dir=root, + time_limit=10, + idle_limit=10, + cancel_check=cancel_check, + ) + + self.assertEqual(set(root.glob(".probhub-interactive-*")), runtime_dirs) + self.assertTrue( + all(path.is_file() for path in actor_pid_files), + "interactive actor or descendant did not start", + ) + self.assert_processes_dead([ + int(path.read_text(encoding="utf-8")) for path in actor_pid_files + ]) + def test_interactive_preparation_failure_is_structured_start_error(self): with tempfile.TemporaryDirectory() as temp, mock.patch.object( SPECIAL_JUDGES.Path, @@ -667,6 +780,60 @@ def terminate(self): self.assertEqual(result["resources"]["contestant"]["memory"], 5) self.assertEqual(result["resources"]["interactor"]["memory"], 100) + def test_interactor_resource_failure_survives_same_round_exit(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 + + def __init__(self, memory, exit_during_sample=False): + self.peak_memory_mb = memory + self.exit_during_sample = exit_during_sample + self.proc = FakeProcess() + + def sample(self): + if self.exit_during_sample: + self.proc.returncode = 1 + return 1, self.peak_memory_mb + + def terminate(self): + if self.proc.returncode is None: + self.proc.returncode = -1 + + managed = iter([FakeManaged(5), FakeManaged(100, exit_during_sample=True)]) + + def fake_spawn(*_args, **_kwargs): + item = next(managed) + if item.exit_during_sample: + time.sleep(0.06) + return item + + 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, + memory_limit_mb=50, + ) + self.assertEqual(result["status"], "FAIL", result) + self.assertEqual(result["actor"], "interactor", result) + self.assertEqual(result["execution_status"], "memory_limit", result) + self.assertEqual(result["failure_kind"], "resource_limit", result) + self.assertEqual(result["termination_reason"], "memory_limit", result) + def test_interactive_deadlines_exclude_process_setup(self): class FakeProcess: def __init__(self): @@ -1031,6 +1198,51 @@ def test_interactive_fast_exit_is_submission_re_not_infrastructure_fail(self): self.assertEqual(case["failure_kind"], "runtime_error", case) self.assertEqual(case["termination_reason"], "completed", case) + def test_interactor_failure_wins_over_simultaneous_contestant_re(self): + with tempfile.TemporaryDirectory() as temp: + problem = self.write_problem( + Path(temp), + {"type": "interactive", "interactor": "code/interactor.cpp"}, + "int main(){ return 7; }", + "int main(){ return 7; }", + {"interactor.cpp": "int main(){ return 3; }"}, + ) + result, events = self.run_judge(problem) + self.assertNotEqual(result.returncode, 0) + case = next( + event for event in events + if event.get("type") == "case" and event.get("kind") == "std" + ) + self.assertEqual(case["status"], "FAIL", case) + self.assertEqual(case["actor"], "interactor", case) + self.assertEqual(case["failure_kind"], "judge_failure", case) + self.assertEqual(case["termination_reason"], "completed", case) + + def test_interactor_failure_wins_over_simultaneous_contestant_ole(self): + with tempfile.TemporaryDirectory() as temp: + problem = self.write_problem( + Path(temp), + {"type": "interactive", "interactor": "code/interactor.cpp"}, + """ + #include + int main(){ + for (int i = 0; i < 2000000; ++i) std::cout.put('x'); + } + """, + "int main(){ return 7; }", + {"interactor.cpp": "int main(){ return 3; }"}, + ) + result, events = self.run_judge(problem) + self.assertNotEqual(result.returncode, 0) + case = next( + event for event in events + if event.get("type") == "case" and event.get("kind") == "std" + ) + self.assertEqual(case["status"], "FAIL", case) + self.assertEqual(case["actor"], "interactor", case) + self.assertEqual(case["failure_kind"], "judge_failure", 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: problem = self.write_problem( From a95670572b0e0c097aa514766617a7c3b0dfcdb5 Mon Sep 17 00:00:00 2001 From: greenthree <1395214327@qq.com> Date: Wed, 5 Aug 2026 19:12:18 +0800 Subject: [PATCH 2/2] fix Windows Unicode judge paths --- probhub/special_judges.py | 24 +++++++++++---- tests/test_special_judges.py | 60 ++++++++++++++++++++++++++++++++++-- 2 files changed, 76 insertions(+), 8 deletions(-) diff --git a/probhub/special_judges.py b/probhub/special_judges.py index ba3ca8a..ccdd1b5 100644 --- a/probhub/special_judges.py +++ b/probhub/special_judges.py @@ -105,6 +105,18 @@ def _feedback_message(feedback_dir, fallback="", limit_bytes=MAX_CHECKER_DIAGNOS } +def _child_path_argument(path, cwd): + """Prefer a relative argument so narrow-path Judge runtimes avoid Unicode parents.""" + value = os.fspath(path) + if cwd is None or not os.path.isabs(value): + return value + try: + return os.path.relpath(value, start=os.fspath(cwd)) + except ValueError: + # Windows cannot form a relative path across drives. + return value + + 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)] @@ -199,9 +211,9 @@ def run_checker_to_files( execution = run_managed_to_files( [ *checker_command, - os.fspath(input_path), - os.fspath(answer_path), - os.fspath(feedback_dir), + _child_path_argument(input_path, cwd), + _child_path_argument(answer_path, cwd), + _child_path_argument(feedback_dir, cwd), ], input_path=contestant_output_path, stdout_path=stdout_path, @@ -703,9 +715,9 @@ def finish(outcome, *, timeout_kind=None, traffic_evidence=None): interactor_managed = spawn_managed( [ *_command_list(interactor_command), - os.fspath(input_path), - os.fspath(answer_path), - os.fspath(feedback_dir), + _child_path_argument(input_path, work_dir), + _child_path_argument(answer_path, work_dir), + _child_path_argument(feedback_dir, work_dir), ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, diff --git a/tests/test_special_judges.py b/tests/test_special_judges.py index 5165405..2ee28b7 100644 --- a/tests/test_special_judges.py +++ b/tests/test_special_judges.py @@ -89,6 +89,59 @@ 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_checker_uses_relative_file_arguments_under_unicode_cwd(self): + captured = {} + + def run_checker(command, **kwargs): + captured["command"] = command + captured["cwd"] = kwargs["cwd"] + feedback = Path(kwargs["cwd"], command[-1]) + (feedback / "judgemessage.txt").write_text("accepted", encoding="utf-8") + return { + "reason": "completed", + "returncode": 0, + "time": 0.01, + "memory": 1, + "memory_enforced": True, + "process_limit_enforced": True, + "output_bytes": 8, + "retained_output_bytes": 8, + "stdout_retained_bytes": 0, + "stderr_retained_bytes": 0, + "output_truncated": False, + } + + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) / "\u4e2d\u6587 workspace" + root.mkdir() + input_path = root / "case.in" + answer_path = root / "case.ans" + output_path = root / "contestant.out" + for path in (input_path, answer_path, output_path): + path.write_bytes(b"1\n") + with mock.patch.object( + SPECIAL_JUDGES, "run_managed_to_files", side_effect=run_checker + ): + result = SPECIAL_JUDGES.run_checker_to_files( + ["checker"], + input_path, + answer_path, + output_path, + timeout=1, + cwd=root, + ) + + self.assertEqual(result["verdict"], "AC") + self.assertEqual(captured["cwd"], root) + for argument, expected in zip( + captured["command"][-3:], + (input_path, answer_path, None), + ): + self.assertFalse(Path(argument).is_absolute(), argument) + if expected is not None: + self.assertEqual((root / argument).resolve(), expected.resolve()) + + def test_custom_contestant_failure_fields_are_normalized_before_checker(self): root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, root, True) @@ -381,10 +434,13 @@ def terminate(self): spawned = [] - def fake_spawn(command, **_kwargs): + def fake_spawn(command, **kwargs): managed = FakeManaged() if spawned: - Path(command[-1], "judgemessage.txt").mkdir() + feedback = Path(command[-1]) + if not feedback.is_absolute(): + feedback = Path(kwargs["cwd"]) / feedback + (feedback / "judgemessage.txt").mkdir() spawned.append(managed) return managed