Stop a retried trial inheriting the failed attempt's classification

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,10 @@ async def _prepare_trial_run(
         # reservation mid-run. Settlement overwrites it with the actual key.
         trial.phase_timing = None
         trial.has_trajectory = False
+        # A stale analysis from an earlier attempt is dropped in
+        # ``_run_post_trial_hooks``, not here: clearing it at attempt start would
+        # leave the trial unclassified for the whole attempt, racing an in-flight
+        # QA job that snapshots its work list up front.
         trial.attempts += 1
 
         if not trial.idempotency_key:
@@ -966,6 +970,50 @@ async def _run_post_trial_hooks(trial_id: str) -> None:
                     f"{type(exc).__name__}: {exc}[/red]"
                 )
 
+        # This trial re-ran, so any classification on it describes an earlier
+        # attempt -- one whose reward, result and artifacts
+        # ``_prepare_trial_run`` cleared. QA is terminal-sticky
+        # (``_trial_needs_classification`` skips SUCCESS/FAILED), so leaving it
+        # would pin that attempt's verdict forever: a HARNESS_ERROR from an
+        # infra-killed attempt riding on a trial that went on to pass.
+        #
+        # Cleared here rather than at attempt start, under the task + trial
+        # locks this function already holds, so it is atomic with the
+        # ``maybe_start_qa_stage`` re-enqueue below and the trial is terminal
+        # (its artifacts complete). Clearing at attempt start would instead
+        # leave the trial unclassified for the length of an attempt, racing an
+        # in-flight QA job that snapshots its work list up front.
+        #
+        # No timestamp comparison, and none would be safe: a label such a job
+        # wrote mid-attempt -- off the row ``_prepare_trial_run`` had already
+        # wiped -- carries a NEWER stamp than this attempt's own start, yet is
+        # exactly the kind that must go. Probes are the one exception; the probe
+        # path writes their classification during settlement, just above.
+        if trial.attempts > 1 and not trial.is_probe and trial.analysis_status:
+            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
+            # ``maybe_start_qa_stage`` only fires from PENDING/RUNNING. If QA
+            # already closed this task out while the attempt was still running,
+            # nothing would re-enqueue it and the trial would strand
+            # unclassified -- so reopen the task the way ``append_trials``
+            # reopens a finished task when live trials appear. COMPLETED is the
+            # only status that reaches here: this function returns above on a
+            # FAILED task, and VERDICT_PENDING/ANALYZING mean a QA job is live
+            # and will classify the trial itself now that its analysis is gone.
+            if task.status == TaskStatus.COMPLETED:
+                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
+
         await maybe_gate_llm_trials(session, trial_id)
         if await maybe_start_qa_stage(session, trial_id):
             console.print(

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

oddish/tests/test_harbor_runner.py

@@ -1721,6 +1721,12 @@ async def test_post_trial_hooks_run_for_completed_trial(monkeypatch):
         harbor_stage="completed",
         # An LLM agent: baselines take the skip path (test_qa_skips_baselines).
         agent="claude-code",
+        # First attempt, never classified: the stale-analysis clear
+        # (test_retry_clears_stale_analysis) reads these and must no-op here.
+        attempts=1,
+        started_at=None,
+        analysis_started_at=None,
+        analysis_finished_at=None,
     )
     calls = []