[Unverified tag check](1) llm-judge runs a read-only investigate job

engine/hooks/llm-judge/README.md

@@ -94,6 +94,25 @@ most 300 characters, taken from the end of stderr or the error text.
 Tests use it to plug in small fake runners. If it is set but not that shape,
 `ask` raises `ValueError` instead of quietly falling back to the real runners.
 
+## Investigate mode
+
+A job opts in with `"mode": "investigate"`. It uses a read-only runner set:
+
+1. **codex**: `codex exec --skip-git-repo-check --sandbox read-only -c notify=[] PROMPT`
+2. **claude**: `claude -p --model haiku --settings '{"disableAllHooks": true}' --allowedTools Read Grep Glob --disallowedTools Write Edit NotebookEdit Bash -- PROMPT`
+
+`cursor-agent` is not used because it has no read-only switch.
+
+An investigate job may carry `timeout_seconds`. The judge caps it at 600
+seconds. If it is missing or not a number, the runner gets 60 seconds.
+
+An investigate job may carry `cwd`. The judge uses it only when it is an
+absolute path to a folder that exists. Otherwise the runner uses a fresh temp
+folder, and `judge.log` gets a line saying the cwd was refused.
+
+`CATSTACK_LLM_JUDGE_RUNNERS` still replaces the selected runner set. A job that
+gets no answer from any runner still comes back `unchecked`.
+
 ## Three outcomes
 
 `verdict(job, result)` turns an `ask` result into one of:

engine/hooks/llm-judge/judge.py

@@ -14,6 +14,7 @@
 import uuid
 
 TIMEOUT_SECONDS = 60
+INVESTIGATE_TIMEOUT_CAP = 600
 KILL_GRACE_SECONDS = 5
 REASON_LIMIT = 300
 PROMPT_SLOT = "{prompt}"
@@ -25,6 +26,10 @@
     ("claude", ["claude", "-p", "--model", "haiku", "--settings", '{"disableAllHooks": true}', PROMPT_SLOT]),
     ("cursor", ["cursor-agent", "-p", "--output-format", "text", PROMPT_SLOT]),
 )
+INVESTIGATE_RUNNERS = (
+    ("codex", ["codex", "exec", "--skip-git-repo-check", "--sandbox", "read-only", "-c", "notify=[]", PROMPT_SLOT]),
+    ("claude", ["claude", "-p", "--model", "haiku", "--settings", '{"disableAllHooks": true}', "--allowedTools", "Read", "Grep", "Glob", "--disallowedTools", "Write", "Edit", "NotebookEdit", "Bash", "--", PROMPT_SLOT]),
+)
 
 
 def state_root() -> str:
@@ -49,10 +54,11 @@ def valid_runner(entry: object) -> bool:
     )
 
 
-def runners() -> list[tuple[str, list[str]]]:
+def runners(mode: object = None) -> list[tuple[str, list[str]]]:
+    default = INVESTIGATE_RUNNERS if mode == "investigate" else DEFAULT_RUNNERS
     raw = os.environ.get(RUNNERS_ENV)
     if not raw:
-        return [(name, list(argv)) for name, argv in DEFAULT_RUNNERS]
+        return [(name, list(argv)) for name, argv in default]
     try:
         parsed = json.loads(raw)
     except ValueError as exc:
@@ -96,30 +102,43 @@ def stop_group(proc: subprocess.Popen) -> str:
     return stderr or ""
 
 
-def run_runner(name: str, argv: list[str], prompt: str) -> tuple[dict, dict | None]:
+def bounded_timeout(timeout_seconds: object) -> int | float:
+    if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)):
+        timeout_seconds = TIMEOUT_SECONDS
+    return min(timeout_seconds, INVESTIGATE_TIMEOUT_CAP)
+
+
+def run_runner(name: str, argv: list[str], prompt: str, timeout_seconds: object = TIMEOUT_SECONDS, cwd: object = None) -> tuple[dict, dict | None]:
     if shutil.which(argv[0]) is None:
         return failed(name, "not installed"), None
     command = [prompt if item == PROMPT_SLOT else item for item in argv]
     env = dict(os.environ)
     env[CHILD_ENV] = "1"
-    with tempfile.TemporaryDirectory(prefix="llm-judge-") as cwd:
+    timeout = bounded_timeout(timeout_seconds)
+    with tempfile.TemporaryDirectory(prefix="llm-judge-") as temp_cwd:
+        runner_cwd = temp_cwd
+        if cwd is not None:
+            if isinstance(cwd, str) and os.path.isabs(cwd) and os.path.isdir(cwd):
+                runner_cwd = cwd
+            else:
+                log(f"runner {name}: refused cwd {cwd!r}")
         try:
             proc = subprocess.Popen(
                 command,
                 stdin=subprocess.DEVNULL,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 text=True,
-                cwd=cwd,
+                cwd=runner_cwd,
                 env=env,
                 start_new_session=True,
             )
         except OSError as exc:
             return failed(name, clip(type(exc).__name__, str(exc))), None
         try:
-            stdout, stderr = proc.communicate(timeout=TIMEOUT_SECONDS)
+            stdout, stderr = proc.communicate(timeout=timeout)
         except subprocess.TimeoutExpired:
-            return failed(name, clip(f"timed out after {TIMEOUT_SECONDS}s", stop_group(proc))), None
+            return failed(name, clip(f"timed out after {timeout}s", stop_group(proc))), None
     if proc.returncode != 0:
         return failed(name, clip(f"exit {proc.returncode}", stderr)), None
     answer = last_json_object(stdout)
@@ -128,10 +147,12 @@ def run_runner(name: str, argv: list[str], prompt: str) -> tuple[dict, dict | No
     return {"runner": name, "ok": True, "reason": "answered"}, answer
 
 
-def ask(prompt: str) -> dict:
+def ask(prompt: str, mode: object = None, timeout_seconds: object = None, cwd: object = None) -> dict:
+    if timeout_seconds is None:
+        timeout_seconds = TIMEOUT_SECONDS
     attempts = []
-    for name, argv in runners():
-        attempt, answer = run_runner(name, argv, prompt)
+    for name, argv in runners(mode):
+        attempt, answer = run_runner(name, argv, prompt, timeout_seconds=timeout_seconds, cwd=cwd)
         attempts.append(attempt)
         if answer is not None:
             return {"outcome": "answered", "runner": name, "answer": answer, "attempts": attempts}
@@ -216,7 +237,7 @@ def run_job(path: str) -> dict:
             raise ValueError(f"job file holds a JSON {type(loaded).__name__}, not an object")
         job = dict(loaded)
         job.setdefault("id", stem)
-        result = verdict(job, ask(str(job["prompt"])))
+        result = verdict(job, ask(str(job["prompt"]), mode=job.get("mode"), timeout_seconds=job.get("timeout_seconds", TIMEOUT_SECONDS), cwd=job.get("cwd")))
     except Exception as exc:
         print(f"catstack-hook-error llm-judge: {type(exc).__name__}: {exc}", file=sys.stderr)
         log(f"job {job.get('id')} failed: {type(exc).__name__}: {exc}\n{traceback.format_exc()}")

engine/hooks/llm-judge/tests/test_judge.py

@@ -124,6 +124,90 @@ def test_malformed_runners_env_refuses_instead_of_running_defaults(self):
     def test_default_runner_order_is_codex_then_claude_then_cursor(self):
         self.assertEqual([name for name, _ in judge.runners()], ["codex", "claude", "cursor"])
 
+    def test_investigate_runner_argv_is_read_only_and_excludes_cursor(self):
+        self.assertEqual(
+            judge.runners("investigate"),
+            [
+                (
+                    "codex",
+                    [
+                        "codex",
+                        "exec",
+                        "--skip-git-repo-check",
+                        "--sandbox",
+                        "read-only",
+                        "-c",
+                        "notify=[]",
+                        judge.PROMPT_SLOT,
+                    ],
+                ),
+                (
+                    "claude",
+                    [
+                        "claude",
+                        "-p",
+                        "--model",
+                        "haiku",
+                        "--settings",
+                        '{"disableAllHooks": true}',
+                        "--allowedTools",
+                        "Read",
+                        "Grep",
+                        "Glob",
+                        "--disallowedTools",
+                        "Write",
+                        "Edit",
+                        "NotebookEdit",
+                        "Bash",
+                        "--",
+                        judge.PROMPT_SLOT,
+                    ],
+                ),
+            ],
+        )
+
+    def test_investigate_runners_env_replaces_investigate_defaults(self):
+        custom = ["probe", [PY, "-c", "print('{}')", "{prompt}"]]
+        self.use_runners(custom)
+        self.assertEqual(judge.runners("investigate"), [("probe", custom[1])])
+
+    def test_investigate_job_threads_timeout_and_cwd_to_runner(self):
+        path = os.path.join(self.state.name, "jobs", "investigate-job.json")
+        with tempfile.TemporaryDirectory() as cwd:
+            judge.write_json_atomic(path, self.job(id="investigate-job", mode="investigate", timeout_seconds=123, cwd=cwd))
+            calls = []
+
+            def capture(name, argv, prompt, timeout_seconds=judge.TIMEOUT_SECONDS, cwd=None):
+                calls.append((name, timeout_seconds, cwd))
+                return {"runner": name, "ok": True, "reason": "answered"}, {"match": True}
+
+            with patch.object(judge, "run_runner", side_effect=capture):
+                result = judge.run_job(path)
+
+        self.assertEqual(result["outcome"], "hit")
+        self.assertEqual(calls, [("codex", 123, cwd)])
+
+    def test_investigate_timeout_is_capped_at_600_seconds(self):
+        self.assertEqual(judge.bounded_timeout(999), 600)
+
+    def test_non_investigate_job_still_gets_default_timeout_and_empty_temp_cwd(self):
+        self.use_runners(runner("env", "import json, os; print(json.dumps({'cwd': os.getcwd(), 'entries': os.listdir('.')}))"))
+        path = os.path.join(self.state.name, "jobs", "default-job.json")
+        judge.write_json_atomic(path, self.job(id="default-job"))
+        calls = []
+        original = judge.run_runner
+
+        def capture(name, argv, prompt, timeout_seconds=judge.TIMEOUT_SECONDS, cwd=None):
+            calls.append((timeout_seconds, cwd))
+            return original(name, argv, prompt, timeout_seconds=timeout_seconds, cwd=cwd)
+
+        with patch.object(judge, "run_runner", side_effect=capture):
+            result = judge.run_job(path)
+
+        self.assertEqual(calls, [(judge.TIMEOUT_SECONDS, None)])
+        self.assertEqual(result["answer"]["entries"], [])
+        self.assertFalse(os.path.exists(result["answer"]["cwd"]))
+
 
 class TestVerdict(JudgeTestCase):
     def test_hit_when_every_hit_key_is_true(self):