diff --git a/engine/hooks/_sdk/transcripts.py b/engine/hooks/_sdk/transcripts.py new file mode 100644 index 000000000..2629a2ba3 --- /dev/null +++ b/engine/hooks/_sdk/transcripts.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import glob +import os +import re + +CODEX_SESSIONS_ENV = "CATSTACK_CODEX_SESSIONS_DIR" +CODEX_HOME_ENV = "CODEX_HOME" +THREAD_ID = re.compile(r"^[0-9A-Za-z-]{8,64}$") + + +def codex_sessions_root() -> str: + override = os.environ.get(CODEX_SESSIONS_ENV) + if override: + return override + home = os.environ.get(CODEX_HOME_ENV) or os.path.join(os.path.expanduser("~"), ".codex") + return os.path.join(home, "sessions") + + +def codex_rollout(payload: dict) -> str: + thread = payload.get("thread-id") or payload.get("thread_id") + if not isinstance(thread, str) or not THREAD_ID.match(thread): + return "" + pattern = os.path.join(codex_sessions_root(), "*", "*", "*", f"rollout-*-{thread}.jsonl") + matches = sorted(glob.glob(pattern)) + return matches[-1] if matches else "" + + +def subagent_transcripts(transcript: str) -> list[str]: + if not transcript.endswith(".jsonl"): + return [] + folder = os.path.join(transcript[: -len(".jsonl")], "subagents") + return sorted(glob.glob(os.path.join(folder, "*.jsonl"))) diff --git a/engine/hooks/llm-judge/inbox.py b/engine/hooks/llm-judge/inbox.py index 669ed910d..262739830 100644 --- a/engine/hooks/llm-judge/inbox.py +++ b/engine/hooks/llm-judge/inbox.py @@ -5,6 +5,7 @@ import os import judge +from transcripts import codex_rollout, subagent_transcripts NO_TRANSCRIPT = "llm-judge: {harness} payload has no transcript path, so finished verdicts were not checked" REPORT_LIMIT = 600 @@ -31,7 +32,7 @@ def resolve_transcript(payload: dict) -> str: candidate = os.path.join(root, project, "agent-transcripts", conv, f"{conv}.jsonl") if os.path.isfile(candidate): return candidate - return "" + return codex_rollout(payload) def _turn_text(data: dict) -> str: @@ -167,7 +168,10 @@ def unchecked_message(item: dict) -> str: def messages(transcript: str) -> list[str]: out = [] - for item in judge.drain(transcript): + drained = [] + for path in [transcript, *subagent_transcripts(transcript)]: + drained.extend(judge.drain(path)) + for item in drained: outcome = item.get("outcome") if outcome == "clean": continue diff --git a/engine/hooks/llm-judge/tests/test_inbox.py b/engine/hooks/llm-judge/tests/test_inbox.py index a6ad26a91..8787882f7 100644 --- a/engine/hooks/llm-judge/tests/test_inbox.py +++ b/engine/hooks/llm-judge/tests/test_inbox.py @@ -355,5 +355,78 @@ def test_no_transcript_says_unchecked_on_stderr(self): self.assertIn("no transcript path", self.run_codex([json.dumps({"type": "agent-turn-complete", "thread-id": "t1"})])) + +REAL_CODEX_NOTIFY_KEYS = ["client", "cwd", "input-messages", "last-assistant-message", "thread-id", "turn-id", "type"] +THREAD = "01a0ce9a-301b-7d42-bb5c-b9a9a4cfe8c6" + + +class TestTranscriptResolution(InboxTestCase): + def setUp(self): + super().setUp() + sessions = os.path.join(self.work.name, "codex-sessions") + day = os.path.join(sessions, "2026", "09", "23") + os.makedirs(day) + self.rollout = os.path.join(day, f"rollout-2026-09-23T22-10-06-{THREAD}.jsonl") + with open(self.rollout, "w", encoding="utf-8") as handle: + handle.write("{}\n") + self.sessions_env = patch.dict(os.environ, {"CATSTACK_CODEX_SESSIONS_DIR": sessions}) + self.sessions_env.start() + + def tearDown(self): + self.sessions_env.stop() + super().tearDown() + + def real_codex_payload(self): + payload = { + "type": "agent-turn-complete", + "thread-id": THREAD, + "turn-id": "01a0ce9a-30dc-73f1-bfe6-ccce644446e1", + "cwd": self.work.name, + "client": "codex_exec", + "input-messages": ["Reply with the single word ok."], + "last-assistant-message": "ok", + } + self.assertEqual(sorted(payload), REAL_CODEX_NOTIFY_KEYS) + return payload + + def test_real_codex_notify_payload_resolves_to_its_rollout_file(self): + self.assertEqual(inbox.resolve_transcript(self.real_codex_payload()), self.rollout) + + def test_hit_for_a_codex_rollout_reaches_the_next_codex_notify(self): + self.transcript = self.rollout + self.seed(ANSWERS_TRUE) + self.assertEqual(self.run_codex([json.dumps(self.real_codex_payload())]), ON_HIT + "\n") + + def test_a_codex_home_override_is_where_the_rollout_is_looked_up(self): + home = os.path.join(self.work.name, "codex-home") + day = os.path.join(home, "sessions", "2026", "09", "23") + os.makedirs(day) + rollout = os.path.join(day, f"rollout-2026-09-23T22-40-00-{THREAD}.jsonl") + with open(rollout, "w", encoding="utf-8") as handle: + handle.write("{}\n") + self.sessions_env.stop() + try: + with patch.dict(os.environ, {"CODEX_HOME": home}, clear=False): + os.environ.pop("CATSTACK_CODEX_SESSIONS_DIR", None) + self.assertEqual(inbox.resolve_transcript(self.real_codex_payload()), rollout) + finally: + self.sessions_env.start() + + def test_unknown_or_unsafe_thread_id_resolves_to_nothing(self): + for thread in ("0000000-0000-not-there", "../../etc", "*"): + with self.subTest(thread=thread): + self.assertEqual(inbox.resolve_transcript({"thread-id": thread}), "") + + def test_a_subagent_verdict_is_delivered_to_the_parent_session(self): + subagent = os.path.join(self.transcript[: -len(".jsonl")], "subagents", "agent-a1.jsonl") + os.makedirs(os.path.dirname(subagent)) + with open(subagent, "w", encoding="utf-8") as handle: + handle.write("{}\n") + parent = self.transcript + self.transcript = subagent + self.seed(ANSWERS_TRUE) + self.assertEqual(inbox.messages(parent), [ON_HIT]) + self.assertEqual(inbox.messages(parent), []) + if __name__ == "__main__": unittest.main() diff --git a/engine/hooks/wrong-check-reflect/detect.py b/engine/hooks/wrong-check-reflect/detect.py index 0bb9339b3..e1af76c89 100644 --- a/engine/hooks/wrong-check-reflect/detect.py +++ b/engine/hooks/wrong-check-reflect/detect.py @@ -15,6 +15,7 @@ os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) from events import write_stage_event # noqa: E402 +from transcripts import codex_rollout # noqa: E402 from flags import enforcement_gate # noqa: E402 HOOKS_DIR = os.path.dirname(os.path.abspath(__file__)) @@ -258,7 +259,7 @@ def resolve_transcript(payload: dict) -> str: return candidate except OSError: pass - return "" + return codex_rollout(payload) def _is_assistant_line(data: dict) -> bool: diff --git a/engine/hooks/wrong-check-reflect/tests/test_hooks.py b/engine/hooks/wrong-check-reflect/tests/test_hooks.py index 27a62d0c6..c647eebd9 100644 --- a/engine/hooks/wrong-check-reflect/tests/test_hooks.py +++ b/engine/hooks/wrong-check-reflect/tests/test_hooks.py @@ -545,6 +545,25 @@ def test_claude_stop_records_the_queued_job_under_the_claude_harness(self): [("wrong-check-reflect", "claude", "transcript")]) self.assertEqual(queued[0]["finding_id"], self.wait_for_jobs(1)[0][: -len(".json")]) + def test_real_codex_notify_queues_against_its_rollout_file(self): + self.use_runners(SLOW_CLEAN) + thread = "01a0ce9a-301b-7d42-bb5c-b9a9a4cfe8c6" + day = os.path.join(self.reflect_state.name, "codex-sessions", "2026", "09", "23") + os.makedirs(day) + rollout = os.path.join(day, f"rollout-2026-09-23T22-10-06-{thread}.jsonl") + with open(rollout, "w", encoding="utf-8") as handle: + handle.write("{}\n") + payload = {"type": "agent-turn-complete", "thread-id": thread, "turn-id": "t", "cwd": "/", + "client": "codex_exec", "input-messages": ["hi"], "last-assistant-message": HIT_TEXT} + with patch.dict(os.environ, {"CATSTACK_CODEX_SESSIONS_DIR": os.path.dirname(os.path.dirname(os.path.dirname(day)))}): + err = run_codex_notify([json.dumps(payload)]) + self.assertEqual(err, "") + [job] = self.wait_for_jobs(1) + with open(os.path.join(self.state.name, "jobs", job), encoding="utf-8") as handle: + self.assertEqual(json.load(handle)["transcript"], rollout) + queued = [r for r in self.stage_rows() if r["action"] == "judge_queued"] + self.assertEqual([(r["harness"], r["reason"]) for r in queued], [("codex", "transcript")]) + def test_claude_malformed_stdin_fail_open(self): err = io.StringIO() with patch.object(sys, "stdin", io.StringIO("not-json")):