fix(evals): stop the judge grading mixed runners or skipping a grown condition set

`group_responses()` keyed groups on (case_id, trial) alone and assigned
`groups[key][condition] = response`. run_evals.py keys its own resume on
(case_id, trial, condition, runner), so one responses file legitimately holds
several runners; the later runner's rows overwrote the earlier runner's, and
every condition was then graded from whichever runner happened to be written
last, with no warning and exit 0. Raise on the collision instead.

The judge's own resume key had the same shape: `judged` was a set of
(case_id, trial), so a group judged in an earlier pass under a narrower
--conditions was skipped forever. Judging baseline+candidate and appending
comparator responses afterwards is the documented flow, because run_evals.py
writes one condition per invocation; the rerun printed "skip judged" for every
group, wrote no comparator rows, and exited 0, after which `run_evals.py score`
gated on the two conditions that happened to be present. Record the conditions
covered by each written group instead. Re-judging a partially covered group is
refused with the missing conditions named, because the scorer rejects duplicate
rows: re-judging would fail there anyway, and skipping is what lost the
condition silently.

Author: Matthew-Selvam

scripts/judge.py

@@ -143,10 +143,26 @@ def parse_judge_scores(
 
 
 def group_responses(rows: list[dict[str, Any]]) -> dict[tuple, dict[str, str]]:
-    """Collect responses into {(case_id, trial): {condition: response}} groups."""
+    """Collect responses into {(case_id, trial): {condition: response}} groups.
+
+    A condition may appear at most once per group. One (case, trial) can carry
+    rows from several runner invocations -- `run_evals.py` keys its own resume on
+    `(case_id, trial, condition, runner)`, so a file with two runners is a shape
+    the harness produces. Collapsing those rows into one dict silently grades one
+    runner's answers under the other runner's name and reports success, so the
+    collision is raised instead of overwritten.
+    """
     groups: dict[tuple, dict[str, str]] = defaultdict(dict)
     for row in rows:
-        groups[(row["case_id"], row["trial"])][row["condition"]] = row["response"]
+        key = (row["case_id"], row["trial"])
+        condition = row["condition"]
+        if condition in groups[key]:
+            raise ValueError(
+                f"{row['case_id']}/trial {row['trial']}: two responses for the "
+                f"{condition} condition. Judge one runner's responses per file "
+                f"(run_evals.py keys completed rows by runner as well)."
+            )
+        groups[key][condition] = row["response"]
     return dict(groups)
 
 
@@ -280,9 +296,34 @@ def main(argv: Optional[list[str]] = None) -> int:
             file=sys.stderr,
         )
 
-    judged: set[tuple] = set()
+    # Which conditions each already-written group covers. Keying only on
+    # (case_id, trial) would skip a group judged in an earlier pass under a
+    # narrower --conditions, leaving a requested condition ungraded while the
+    # run still exits 0. A group is either complete for this run (skipped) or
+    # absent; a partially covered one cannot be re-judged without writing
+    # duplicate rows for the conditions that are already there, and the scorer
+    # rejects duplicates, so it stops the run instead.
+    written: dict[tuple, set[str]] = defaultdict(set)
     if args.output.exists():
-        judged = {(row["case_id"], row["trial"]) for row in run_evals.read_jsonl(args.output)}
+        for row in run_evals.read_jsonl(args.output):
+            written[(row["case_id"], row["trial"])].add(row["condition"])
+    partial = sorted(
+        key for key in complete if written[key] and not required.issubset(written[key])
+    )
+    if partial:
+        missing = {
+            key: sorted(required - written[key]) for key in partial
+        }
+        raise ValueError(
+            f"{args.output} already holds scores for "
+            + ", ".join(
+                f"{case_id}/trial {trial} without {', '.join(missing[(case_id, trial)])}"
+                for case_id, trial in partial
+            )
+            + ". Judge into a fresh --output file, or remove those rows first: "
+            "re-judging would write duplicate rows for the conditions already present."
+        )
+    judged = {key for key in complete if required.issubset(written[key])}
 
     config = json.loads(args.runner_config.read_text(encoding="utf-8"))
     runner = config[args.runner]
@@ -295,7 +336,8 @@ def main(argv: Optional[list[str]] = None) -> int:
     with args.output.open("a", encoding="utf-8") as destination:
         for key in sorted(complete):
             case_id, trial = key
-            if key in judged:
+            pending = required - written[key]
+            if not pending:
                 print(f"skip judged {case_id}/trial {trial}")
                 continue
             if case_id not in cases:

tests/test_judge.py

@@ -58,6 +58,43 @@ def test_groups_missing_a_condition_are_partitioned_out_not_dropped(self):
         self.assertEqual([("direct-answer", 1)], sorted(complete))
         self.assertEqual([("casual-message", 1)], sorted(incomplete))
 
+    def test_two_responses_for_one_condition_are_rejected_not_overwritten(self):
+        # run_evals.py keys its own resume on (case, trial, condition, runner),
+        # so one responses file legitimately holds several runners. Collapsing
+        # them into one dict would grade the last runner's answers under the
+        # condition name and report success.
+        rows = [
+            {
+                "case_id": "direct-answer",
+                "trial": 1,
+                "condition": condition,
+                "runner": runner,
+                "response": f"{condition}-from-{runner}",
+            }
+            for runner in ("claude", "codex")
+            for condition in ("baseline", "candidate")
+        ]
+
+        with self.assertRaisesRegex(ValueError, "two responses for the baseline condition"):
+            judge.group_responses(rows)
+
+    def test_the_same_condition_from_one_runner_still_groups(self):
+        rows = [
+            {
+                "case_id": "direct-answer",
+                "trial": 1,
+                "condition": condition,
+                "runner": "claude",
+                "response": condition,
+            }
+            for condition in ("baseline", "candidate")
+        ]
+
+        self.assertEqual(
+            {"baseline": "baseline", "candidate": "candidate"},
+            judge.group_responses(rows)[("direct-answer", 1)],
+        )
+
 
 class ParseJudgeScoresTest(unittest.TestCase):
     @staticmethod
@@ -411,6 +448,99 @@ def test_runner_failure_skips_its_group_and_continues(self):
             self.assertEqual(2, len(rows))
             self.assertNotEqual(0, exit_code, "skipped groups must not report success")
 
+    def test_a_group_judged_under_narrower_conditions_is_not_silently_skipped(self):
+        # run_evals.py writes one condition per invocation, so judging
+        # baseline+candidate and appending comparator responses afterwards is
+        # the documented flow. Skipping on (case_id, trial) alone would drop the
+        # comparator with exit 0, and `run_evals.py score` would then gate on
+        # the two conditions that happen to be present.
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            responses = tmp_path / "responses.jsonl"
+            responses.write_text(
+                "".join(
+                    json.dumps(
+                        {
+                            "case_id": "direct-answer",
+                            "trial": 1,
+                            "condition": condition,
+                            "runner": "stub",
+                            "response": f"{condition} answer",
+                        }
+                    )
+                    + "\n"
+                    for condition in ("baseline", "candidate")
+                )
+            )
+            verdict = tmp_path / "verdict.json"
+            verdict.write_text(
+                json.dumps({"A": self.VERDICT, "B": self.VERDICT, "C": self.VERDICT})
+            )
+            runner_config = tmp_path / "runners.json"
+            runner_config.write_text(
+                json.dumps(
+                    {
+                        "stub": {
+                            "command": ["sh", "-c", f"cat >/dev/null; cat {verdict}"],
+                            "response_format": "text",
+                        }
+                    }
+                )
+            )
+            output = tmp_path / "scores.jsonl"
+            common = [
+                "--responses",
+                str(responses),
+                "--cases",
+                str(ROOT / "evals" / "cases.jsonl"),
+                "--rubric",
+                str(ROOT / "evals" / "rubric.md"),
+                "--runner-config",
+                str(runner_config),
+                "--runner",
+                "stub",
+                "--output",
+                str(output),
+            ]
+
+            self.assertEqual(0, judge.main(common))
+
+            with responses.open("a") as handle:
+                handle.write(
+                    json.dumps(
+                        {
+                            "case_id": "direct-answer",
+                            "trial": 1,
+                            "condition": "comparator",
+                            "runner": "stub",
+                            "response": "comparator answer",
+                        }
+                    )
+                    + "\n"
+                )
+
+            with self.assertRaisesRegex(ValueError, "without comparator"):
+                judge.main(
+                    [
+                        "--responses",
+                        str(responses),
+                        "--cases",
+                        str(ROOT / "evals" / "cases.jsonl"),
+                        "--rubric",
+                        str(ROOT / "evals" / "rubric.md"),
+                        "--runner-config",
+                        str(runner_config),
+                        "--runner",
+                        "stub",
+                        "--conditions",
+                        "baseline",
+                        "candidate",
+                        "comparator",
+                        "--output",
+                        str(output),
+                    ]
+                )
+
 
 if __name__ == "__main__":
     unittest.main()