Skip to content

Wrong-check-reflect tests use the shared judge test base #528

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mergify merged 1 commit into from
Sep 14, 2026
+13 −15
Merged

Wrong-check-reflect tests use the shared judge test base #528

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 13 additions & 15 deletions engine/hooks/wrong-check-reflect/tests/test_hooks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,8 @@

sys.path.append(os.path.dirname(detect.LLM_JUDGE_PATH))
import inbox as judge_inbox # noqa: E402
import judge # noqa: E402
import phrases # noqa: E402
from judge_test_base import JudgeTestCase # noqa: E402


PY = sys.executable
Expand Down Expand Up @@ -74,18 +74,16 @@ def transcript_line(role: str, text: str) -> str:
return json.dumps({"type": role, "message": {"role": role, "content": [{"type": "text", "text": text}]}})


class TestWrongCheckReflect(unittest.TestCase):
class TestWrongCheckReflect(JudgeTestCase):
def setUp(self):
super().setUp()
self.reflect_state = tempfile.TemporaryDirectory()
self.judge_state = tempfile.TemporaryDirectory()
self.env = patch.dict(os.environ, {
self.reflect_env = patch.dict(os.environ, {
flags.REFLECT_ENFORCEMENT: "1",
"WRONG_CHECK_REFLECT_STATE_DIR": self.reflect_state.name,
judge.STATE_ENV: self.judge_state.name,
judge.RUNNERS_ENV: json.dumps([ANSWERS_HIT]),
})
self.env.start()
os.environ.pop(judge.CHILD_ENV, None)
self.reflect_env.start()
self.use_runners(ANSWERS_HIT)
detect.STATE_DIR = self.reflect_state.name
detect._judge.cache_clear()
detect._phrases.cache_clear()
Expand All @@ -98,14 +96,14 @@ def tearDown(self):
deadline = time.monotonic() + 15
while self.jobs() and time.monotonic() < deadline:
time.sleep(0.1)
self.env.stop()
self.judge_state.cleanup()
self.reflect_env.stop()
self.reflect_state.cleanup()
detect._judge.cache_clear()
detect._phrases.cache_clear()
super().tearDown()

def jobs(self) -> list[str]:
folder = os.path.join(self.judge_state.name, "jobs")
folder = os.path.join(self.state.name, "jobs")
return os.listdir(folder) if os.path.isdir(folder) else []

def write_transcript(self, *lines: tuple[str, str], name: str = "session.jsonl") -> str:
Expand Down Expand Up @@ -158,14 +156,14 @@ def test_enqueue_is_off_unless_the_flag_is_on(self):
self.assertEqual(self.jobs(), [])

def test_claude_stop_queues_job_for_normal_reply(self):
os.environ[judge.RUNNERS_ENV] = json.dumps([SLOW_CLEAN])
self.use_runners(SLOW_CLEAN)
path = self.write_transcript(("assistant", HIT_TEXT))
blocked, err = run_claude({"transcript_path": path})
self.assertFalse(blocked)
self.assertEqual(err, "")
jobs = self.wait_for_jobs(1)
self.assertEqual(len(jobs), 1)
with open(os.path.join(self.judge_state.name, "jobs", jobs[0]), encoding="utf-8") as handle:
with open(os.path.join(self.state.name, "jobs", jobs[0]), encoding="utf-8") as handle:
job = json.load(handle)
self.assertEqual(job["hook"], "wrong-check-reflect")
self.assertEqual(job["transcript"], path)
Expand All @@ -177,7 +175,7 @@ def test_hit_verdict_reaches_agent_as_dictionary_on_hit(self):
self.assertEqual(self.wait_for_messages(path), [detect.FOLLOWUP])

def test_clean_verdict_says_nothing(self):
os.environ[judge.RUNNERS_ENV] = json.dumps([ANSWERS_CLEAN])
self.use_runners(ANSWERS_CLEAN)
path = self.write_transcript(("assistant", OPTION_TEXT))
self.assertIsNotNone(detect.enqueue_judge({"transcript_path": path}))
deadline = time.monotonic() + 15
Expand All @@ -186,7 +184,7 @@ def test_clean_verdict_says_nothing(self):
self.assertEqual(judge_inbox.messages(path), [])

def test_unchecked_verdict_says_could_not_judge(self):
os.environ[judge.RUNNERS_ENV] = json.dumps([MISSING])
self.use_runners(MISSING)
path = self.write_transcript(("assistant", COUNT_TEXT))
self.assertIsNotNone(detect.enqueue_judge({"transcript_path": path}))
messages = self.wait_for_messages(path)
Expand Down
Loading