llm-judge: find Codex's transcript by thread-id, and deliver subagent verdicts to the parent

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")))

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

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()

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:

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")):