Require the checker to complete in run_pass_at_1 (block early-exit reward hack)

src/trinity/orchestration/reward.py

@@ -45,6 +45,7 @@
 import json
 import os
 import re
+import secrets
 import subprocess
 import sys
 import tempfile
@@ -1189,6 +1190,36 @@ def _functional_harness(fn_name: str, stdin_data: str, expected_output: str) ->
     return "\n".join(lines) + "\n"
 
 
+def _run_checked_script(candidate_code: str, checker: str, *, timeout_s: int) -> bool:
+    """Run ``candidate_code`` then ``checker``, requiring the checker to run to
+    completion — not merely a zero exit status.
+
+    The assert- and call-based flavors append verification code *after* the
+    untrusted candidate. Judging a pass by exit code alone let a candidate that
+    terminates the process early — ``sys.exit(0)``, ``exit()``/``quit()``,
+    ``raise SystemExit(0)``, or even a hard ``os._exit(0)`` — short-circuit the
+    appended checks and score correct regardless of correctness.
+
+    We append a write of a per-run random sentinel *after* the checker. A pass
+    requires exit code 0 AND the sentinel present in stdout, so any early
+    termination before the checker's asserts run fails: the sentinel is never
+    emitted. The token is unguessable per run, so a candidate cannot forge it by
+    printing a fixed string.
+    """
+    sentinel = "__TRINITY_PASS_" + secrets.token_hex(16) + "__"
+    script = (
+        candidate_code
+        + "\n\n"
+        + checker
+        + "\n"
+        + "import sys as _trinity_sys\n"
+        + f"_trinity_sys.stdout.write({sentinel!r})\n"
+        + "_trinity_sys.stdout.flush()\n"
+    )
+    ok, stdout = _exec_script_capture(script, stdin_data="", timeout_s=timeout_s)
+    return ok and sentinel in stdout
+
+
 def _run_one_test(
     code: str, test: object, timeout_s: int, fn_name: str | None = None
 ) -> bool:
@@ -1211,8 +1242,8 @@ def _run_one_test(
         if ttype == "functional" and has_io:
             stdin_data = str(test.get("stdin", test.get("input", "")))
             expected = str(test.get("expected_stdout", test.get("output", "")))
-            script = code + "\n\n" + _functional_harness(fn_name or "", stdin_data, expected)
-            return _exec_script(script, stdin_data="", timeout_s=timeout_s)
+            checker = _functional_harness(fn_name or "", stdin_data, expected)
+            return _run_checked_script(code, checker, timeout_s=timeout_s)
 
     stdin_data: str | None = None
     expected_stdout: str | None = None
@@ -1245,8 +1276,7 @@ def _run_one_test(
         return False
 
     if assert_block is not None:
-        script = code + "\n\n" + assert_block + "\n"
-        return _exec_script(script, stdin_data="", timeout_s=timeout_s)
+        return _run_checked_script(code, assert_block, timeout_s=timeout_s)
 
     # stdin/stdout test.
     ok, stdout = _exec_script_capture(
@@ -1312,12 +1342,6 @@ def _apply() -> None:  # pragma: no cover - runs only in the child process.
     return _apply
 
 
-def _exec_script(script: str, *, stdin_data: str, timeout_s: int) -> bool:
-    """Run a script; pass iff it exits 0 within the timeout. No output check."""
-    ok, _ = _exec_script_capture(script, stdin_data=stdin_data, timeout_s=timeout_s)
-    return ok
-
-
 def _exec_script_capture(
     script: str, *, stdin_data: str, timeout_s: int
 ) -> tuple[bool, str]:

tests/test_reward_sandbox.py

@@ -69,3 +69,59 @@ def test_rlimits_do_not_break_ordinary_grading():
     code = "import sys\nprint(int(sys.stdin.read()) + 1)"
     tests = [{"input": "41\n", "output": "42"}]
     assert R.run_pass_at_1(code, tests, timeout_s=5) is True
+
+
+# ---------------------------------------------------------------------------
+# Early-exit reward hack: a candidate must not short-circuit the appended
+# checker by terminating the process with a zero exit status before the
+# asserts run. Pass now requires the checker to run to completion.
+# ---------------------------------------------------------------------------
+_WRONG_ADD = "def add(a, b):\n    return 999\n"
+_ADD_ASSERT = ["assert add(2, 3) == 5"]
+
+
+def test_assert_rejects_sys_exit_before_checker():
+    hack = _WRONG_ADD + "import sys\nsys.exit(0)\n"
+    assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False
+
+
+def test_assert_rejects_exit_builtin_before_checker():
+    hack = _WRONG_ADD + "exit()\n"
+    assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False
+
+
+def test_assert_rejects_os_exit_before_checker():
+    # os._exit is a hard exit that bypasses SystemExit handling; the missing
+    # success sentinel still fails it.
+    hack = _WRONG_ADD + "import os\nos._exit(0)\n"
+    assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False
+
+
+def test_assert_rejects_raise_systemexit_before_checker():
+    hack = _WRONG_ADD + "raise SystemExit(0)\n"
+    assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False
+
+
+def test_assert_correct_code_still_passes():
+    good = "def add(a, b):\n    return a + b\n"
+    assert R.run_pass_at_1(good, _ADD_ASSERT, timeout_s=5) is True
+
+
+def test_assert_rejects_forged_success_string():
+    # A candidate cannot fake a pass by printing a plausible marker; the real
+    # sentinel is random per run.
+    hack = _WRONG_ADD + "print('__TRINITY_PASS__')\nimport sys\nsys.exit(0)\n"
+    assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False
+
+
+def test_functional_rejects_sys_exit_before_checker():
+    # LiveCodeBench call-based path: wrong Solution that exits before the harness.
+    bad = (
+        "class Solution:\n"
+        "    def twoSum(self, nums, target):\n"
+        "        return [9, 9]\n"
+        "import sys\n"
+        "sys.exit(0)\n"
+    )
+    spec = {"tests": [{"input": "[2,7,11,15]\n9", "output": "[0,1]", "testtype": "functional"}], "fn_name": "twoSum"}
+    assert R.score_text("livecodebench", "```python\n" + bad + "```", spec) == 0.0