fix(qa): stop a retried trial inheriting the failed attempt's classification

A trial that fails an attempt, gets classified HARNESS_ERROR, then passes on
a later attempt kept the HARNESS_ERROR forever -- describing artifacts the
row no longer held. Three defects compound to produce that:

- `_prepare_trial_run` cleared every result field before an attempt (reward,
  result, tokens, trial_s3_key, ...) but left `analysis`/`analysis_status`.
  QA is terminal-sticky, so the stale verdict could never be revisited.
- The task itself was never reopened. `maybe_start_qa_stage` only fires from
  PENDING/RUNNING, so a task that reached COMPLETED while an attempt was
  failing stayed closed through the attempt that passed, and no QA pass was
  ever enqueued again.
- The classifier picked the first `task-*` job dir from an unsorted
  `iterdir()`. Every attempt re-uploads into the same S3 prefix without
  clearing it, so a retried trial's tree holds one job dir per attempt and
  the classifier analysed an arbitrary one.

`_prepare_trial_run` now clears the analysis fields alongside the results
they describe, and reopens a COMPLETED/FAILED task the way `append_trials`
already does. Reopening is limited to the terminal statuses because reaching
them is what proves no QA job is in flight, so it cannot race one.

The classifier now resolves the current attempt from the job-level
`result.json`, whose `reward_stats`/`exception_stats` name that attempt's
directory, and falls back to sorted order so the choice is deterministic
rather than readdir-dependent.

Verified against four FVSmith runs wedged at QA_PROBE showing 3/3
HARNESS_ERROR on sweeps whose pass@1 was 1.0.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: charlesyhuang

oddish/src/oddish/analyze/classifier.py

@@ -228,6 +228,62 @@ def classify_trial(
     )
 
 
+def _attempt_dirs_named_by(root_data: dict) -> set[str]:
+    """Job directories the job-level ``result.json`` attributes to this attempt.
+
+    Harbor mints a fresh ``task-<slug>__<id>`` directory per attempt inside a
+    per-trial wrapper, and every attempt re-uploads that wrapper to the same S3
+    prefix without clearing it -- so a retried trial's downloaded tree holds one
+    such directory per attempt, side by side. Only the *latest* attempt
+    overwrites the job-level ``result.json``, and its ``reward_stats`` /
+    ``exception_stats`` name that attempt's directory. That is the only signal
+    in the tree that tells the current attempt from the stale ones.
+    """
+    names: set[str] = set()
+    stats = root_data.get("stats")
+    evals = stats.get("evals") if isinstance(stats, dict) else None
+    if not isinstance(evals, dict):
+        return names
+    for eval_data in evals.values():
+        if not isinstance(eval_data, dict):
+            continue
+        for key in ("reward_stats", "exception_stats"):
+            grouped = eval_data.get(key)
+            if not isinstance(grouped, dict):
+                continue
+            for entry in grouped.values():
+                # reward_stats nests one level deeper than exception_stats:
+                # {metric: {value: [names]}} vs {exception: [names]}.
+                buckets = entry.values() if isinstance(entry, dict) else [entry]
+                for bucket in buckets:
+                    if isinstance(bucket, list):
+                        names.update(n for n in bucket if isinstance(n, str))
+    return names
+
+
+def _current_attempt_result(trial_dir: Path, root_data: dict) -> Path | None:
+    """Pick the nested per-attempt ``result.json`` this trial's row describes.
+
+    Prefer the directory the job-level result names; fall back to the first in
+    sorted order so the choice is at least deterministic when the job result
+    names nothing we can see (older layouts, partial uploads).
+    """
+    candidates = sorted(
+        subdir
+        for subdir in trial_dir.iterdir()
+        if subdir.is_dir()
+        and subdir.name.startswith("task-")
+        and (subdir / "result.json").exists()
+    )
+    if not candidates:
+        return None
+    named = _attempt_dirs_named_by(root_data)
+    for subdir in candidates:
+        if subdir.name in named:
+            return subdir / "result.json"
+    return candidates[0] / "result.json"
+
+
 class TrialClassifier:
     """Classifies trial outcomes using Claude Code to identify task quality issues."""
 
@@ -280,12 +336,9 @@ async def classify_trial(
             try:
                 root_data = json.loads(result_path.read_text())
                 if "n_total_trials" in root_data or "stats" in root_data:
-                    for subdir in trial_dir.iterdir():
-                        if subdir.is_dir() and subdir.name.startswith("task-"):
-                            nested_result = subdir / "result.json"
-                            if nested_result.exists():
-                                result_path = nested_result
-                                break
+                    nested = _current_attempt_result(trial_dir, root_data)
+                    if nested is not None:
+                        result_path = nested
             except Exception:
                 pass
 

oddish/src/oddish/workers/queue/trial_handler.py

@@ -503,6 +503,18 @@ async def _prepare_trial_run(
         # reservation mid-run. Settlement overwrites it with the actual key.
         trial.phase_timing = None
         trial.has_trajectory = False
+        # A re-attempt invalidates the previous attempt's QA verdict: the reward,
+        # result and artifacts it was derived from were just cleared above. QA is
+        # terminal-sticky (``_trial_needs_classification`` skips SUCCESS/FAILED),
+        # so leaving it would pin the old attempt's classification -- e.g. a
+        # HARNESS_ERROR from an infra-killed attempt riding on a trial that goes
+        # on to pass -- and no later QA pass would ever revisit it.
+        trial.analysis = None
+        trial.analysis_status = None
+        trial.analysis_error = None
+        trial.analysis_started_at = None
+        trial.analysis_finished_at = None
+        trial.analysis_log = None
         trial.attempts += 1
 
         if not trial.idempotency_key:
@@ -512,6 +524,27 @@ async def _prepare_trial_run(
         if task and task.status == TaskStatus.PENDING:
             task.status = TaskStatus.RUNNING
             task.started_at = utcnow()
+        elif task and task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED):
+            # The task already closed out -- QA ran and synthesized a verdict --
+            # yet this trial still had attempts left and is running again. Its
+            # old result (and the verdict built on it) describe a run that no
+            # longer exists, and ``maybe_start_qa_stage`` only fires from
+            # PENDING/RUNNING, so leaving the task closed would strand the new
+            # attempt permanently unclassified. Reopen it the way
+            # ``append_trials`` reopens a finished task when live trials appear.
+            #
+            # Only from the terminal statuses: reaching them is precisely what
+            # proves no QA job is in flight, so this can never race one. A task
+            # still in VERDICT_PENDING/ANALYZING has a live QA job that will
+            # pick the trial up on its own now that its analysis is cleared.
+            task.status = TaskStatus.RUNNING
+            task.finished_at = None
+            if task.run_analysis:
+                task.verdict = None
+                task.verdict_status = None
+                task.verdict_error = None
+                task.verdict_started_at = None
+                task.verdict_finished_at = None
 
         task_id = task.id if task else trial.task_id
         task_name = task.name if task else trial.task_id

oddish/tests/analyze/test_current_attempt_result.py

@@ -0,0 +1,139 @@
+"""A retried trial's downloaded tree holds one job dir per attempt.
+
+Harbor mints a fresh ``task-<slug>__<id>`` directory per attempt, and every
+attempt re-uploads into the same S3 prefix without clearing it, so the
+classifier sees every attempt side by side. Only the last attempt overwrites
+the job-level ``result.json``; picking any other dir classifies a stale
+attempt and can label a passing trial HARNESS_ERROR.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src"))
+
+from oddish.analyze.classifier import (  # noqa: E402
+    _attempt_dirs_named_by,
+    _current_attempt_result,
+)
+
+
+def _job_dir(trial_dir: Path, name: str, payload: dict) -> Path:
+    job = trial_dir / name
+    job.mkdir(parents=True)
+    (job / "result.json").write_text(json.dumps(payload))
+    return job
+
+
+def _root_reward(dir_name: str, reward: str = "1.0") -> dict:
+    return {
+        "n_total_trials": 1,
+        "stats": {
+            "evals": {
+                "grok-build__m__adhoc": {
+                    "reward_stats": {"reward": {reward: [dir_name]}},
+                    "exception_stats": {},
+                }
+            }
+        },
+    }
+
+
+def _root_exception(dir_name: str, exc: str = "EnvironmentStartTimeoutError") -> dict:
+    return {
+        "n_total_trials": 1,
+        "stats": {
+            "evals": {
+                "grok-build__m__adhoc": {
+                    "reward_stats": {},
+                    "exception_stats": {exc: [dir_name]},
+                }
+            }
+        },
+    }
+
+
+def test_picks_the_dir_the_job_result_names_not_the_first(monkeypatch, tmp_path):
+    """The regression, pinned against an adversarial directory order.
+
+    ``iterdir`` yields in filesystem order, so the old "first ``task-*`` dir
+    wins" scan picked an arbitrary attempt. Force the stale attempt to come
+    first: only a scan that consults the job-level result gets this right.
+    """
+    _job_dir(tmp_path, "task-x__AAAA", {"verifier_result": None})
+    _job_dir(
+        tmp_path, "task-x__ZZZZ", {"verifier_result": {"rewards": {"reward": 1.0}}}
+    )
+
+    real_iterdir = Path.iterdir
+    monkeypatch.setattr(Path, "iterdir", lambda self: iter(sorted(real_iterdir(self))))
+
+    picked = _current_attempt_result(tmp_path, _root_reward("task-x__ZZZZ"))
+
+    assert picked == tmp_path / "task-x__ZZZZ" / "result.json"
+
+
+def test_named_dir_wins_for_an_errored_final_attempt(tmp_path):
+    _job_dir(
+        tmp_path, "task-x__AAAA", {"verifier_result": {"rewards": {"reward": 1.0}}}
+    )
+    _job_dir(tmp_path, "task-x__BBBB", {"exception_info": "boom"})
+
+    picked = _current_attempt_result(tmp_path, _root_exception("task-x__BBBB"))
+
+    assert picked == tmp_path / "task-x__BBBB" / "result.json"
+
+
+def test_single_attempt_is_unchanged(tmp_path):
+    _job_dir(tmp_path, "task-x__ONLY", {"verifier_result": None})
+
+    picked = _current_attempt_result(tmp_path, _root_reward("task-x__ONLY"))
+
+    assert picked == tmp_path / "task-x__ONLY" / "result.json"
+
+
+def test_falls_back_deterministically_when_the_job_result_names_nothing(tmp_path):
+    """Older layouts name no dirs; the pick must still be stable, not readdir order."""
+    _job_dir(tmp_path, "task-x__BBBB", {})
+    _job_dir(tmp_path, "task-x__AAAA", {})
+
+    picked = _current_attempt_result(tmp_path, {"n_total_trials": 1, "stats": {}})
+
+    assert picked == tmp_path / "task-x__AAAA" / "result.json"
+
+
+def test_ignores_dirs_without_a_result(tmp_path):
+    (tmp_path / "task-x__AAAA").mkdir()
+    _job_dir(tmp_path, "task-x__BBBB", {})
+
+    picked = _current_attempt_result(tmp_path, {"n_total_trials": 1, "stats": {}})
+
+    assert picked == tmp_path / "task-x__BBBB" / "result.json"
+
+
+def test_no_job_dirs_yields_none(tmp_path):
+    assert _current_attempt_result(tmp_path, _root_reward("task-x__AAAA")) is None
+
+
+def test_named_dirs_survive_both_stat_shapes():
+    payload = {
+        "stats": {
+            "evals": {
+                "e1": {
+                    "reward_stats": {"reward": {"1.0": ["a"], "0.0": ["b"]}},
+                    "exception_stats": {"Boom": ["c"]},
+                },
+                "e2": {"reward_stats": {}, "exception_stats": {}},
+            }
+        }
+    }
+
+    assert _attempt_dirs_named_by(payload) == {"a", "b", "c"}
+
+
+def test_malformed_job_result_names_nothing():
+    for payload in ({}, {"stats": None}, {"stats": {"evals": []}}):
+        assert _attempt_dirs_named_by(payload) == set()