fix: reject resumes across incompatible dataset and model identities

.github/workflows/devdex-eval.yml

@@ -59,7 +59,7 @@ jobs:
       - name: Install
         run: |
           python -m pip install --upgrade pip
-          pip install pytest
+          pip install pytest numpy
       - name: Dataset + scorer tests
         # the DIRECTORY, not a hand-listed set -- a new test file must not need a CI edit to run
         run: python -m pytest devdex/tests -q

benchmark/register_arm.py

@@ -67,5 +67,8 @@ def run_agent_mode(engine, track="all", limit=None, n_runs=1, workers=3, dataset
         if limit:
             cmd += ["--limit", str(limit)]
         print(f"\n=== {t}: {' '.join(cmd[1:])}", flush=True)
-        rc |= subprocess.call(cmd, env=env, cwd=str(ROOT))
+        status = subprocess.call(cmd, env=env, cwd=str(ROOT))
+        if status == 78:
+            return status  # rejected resume: do not run further tracks or report old records
+        rc |= status
     return rc

benchmark/run_benchmark.py

@@ -202,6 +202,8 @@ def main():
         rc = register_arm.run_agent_mode(
             engine=label, track=a.track, limit=a.limit,
             n_runs=a.n_runs, workers=a.workers, datasets=DATA)
+        if rc == 78:
+            return rc  # saved records were rejected; they must not become a new score
         if rc:
             print(f"\none or more tracks exited non-zero (rc={rc}); scoring what landed", flush=True)
 

devdex/harness/runner_sdk.py

@@ -776,7 +776,7 @@ def _is_dead(row):
     return bool(err) and any(d in err for d in _DEAD_ON_RESUME)
 
 
-def _resume(pass_tag, track, arm, items):
+def _resume(pass_tag, track, arm, items, *, gt_sha=None, model=None):
     """Load a partial cell if one exists and return (done_rows, remaining_items)."""
     path = _out(f"{pass_tag}_{track}_{arm}.json")
     if not os.path.exists(path):
@@ -790,6 +790,19 @@ def _resume(pass_tag, track, arm, items):
     else:
         keep = [r for r in done if not _is_dead(r)]
         redo = {r.get("qid") for r in done if _is_dead(r)}
+    # Qids can remain stable while their queries/golds or driver change. Reusing those
+    # measurements would mix experiments; _one_pass later stamps the current config on all
+    # rows, concealing that mismatch. Unknown legacy identity is not evidence of a match.
+    for row in keep:
+        for key, expected in (("gt_sha", gt_sha), ("model", model)):
+            if expected is not None and row.get(key) != expected:
+                print(
+                    f"cannot resume {path}: qid={row.get('qid')!r} has {key}={row.get(key)!r}, "
+                    f"expected {expected!r}. Start a fresh run with a new --label and without "
+                    "--auto-resume, or use a fresh --output-dir; saved records are unchanged.",
+                    file=sys.stderr)
+                # EX_CONFIG: the suite must not score/relabel the incompatible saved cell.
+                raise SystemExit(78)
     seen = {r.get("qid") for r in keep}
     todo = [x for x in items if x["id"] not in seen]
     if done:
@@ -830,7 +843,7 @@ def _one_pass(items, pass_tag, pass_no):
            "skill": SKILL_NAME, "skill_sha": SKILL_SHA, "skill_provenance": SKILL_PROV}
     _CFG_SNAPSHOT.update(cfg)
 
-    rows, items = _resume(pass_tag, TRACK, ARM_NAME, items)
+    rows, items = _resume(pass_tag, TRACK, ARM_NAME, items, gt_sha=GT_SHA, model=MODEL)
     if not items:
         print("nothing to do — cell already complete", flush=True)
         return rows

devdex/run_eval.py

@@ -63,7 +63,8 @@ def main():
     ap.add_argument("--resume-dir", default=None,
                     help="continue into an existing runs/<timestamp>-<label> directory instead "
                          "of creating a new one; finished qids are skipped. Use for vendors with "
-                         "a daily request cap (Mintlify: 1,000/day).")
+                         "a daily request cap (Mintlify: 1,000/day). Retained rows must record "
+                         "the same dataset hash and model; legacy unstamped rows need a fresh run.")
     ap.add_argument("--auto-resume", action="store_true",
                     help="reuse an existing runs/ dir with the same --label instead of minting a "
                          "new timestamp, so a killed run continues from its last flush. Makes a "

devdex/scorer/suite.py

@@ -393,6 +393,12 @@ def run_cell(track, arm, driver, n_runs, concurrency, out_root, label=None, limi
             argv.append(model)
         print(f"\n=== pass {i}/{n_runs}  {label}  -> {pdir}", flush=True)
         rc = subprocess.call(argv, cwd=str(HARNESS), env=env)
+        if rc == 78:
+            # runner_sdk reserves EX_CONFIG for rejected resume identity. Scoring the old
+            # file here would still publish it under this invocation's dataset/model.
+            print("cannot resume incompatible saved measurements; start a fresh run",
+                  file=sys.stderr)
+            raise SystemExit(78)
         if rc != 0:
             print(f"    pass {i} exited {rc}", flush=True)
 
@@ -458,4 +464,3 @@ def compare_to_baseline(results, baseline_path):
                        "claim. Re-run the baseline with --n-runs 3 to compare distributions.")
     return out
 
-

devdex/tests/_resume_shim.py

@@ -5,6 +5,7 @@
 """
 import ast
 import os
+import sys
 from pathlib import Path
 
 RUNNER = Path(__file__).resolve().parents[1] / "harness" / "runner_sdk.py"
@@ -19,7 +20,7 @@ def make_ns(out_dir, monkeypatch):
             if (isinstance(n, (ast.FunctionDef,)) and n.name in WANT)
             or (isinstance(n, ast.Assign)
                 and any(getattr(t, "id", None) in WANT for t in n.targets))]
-    ns = {"os": os, "json": __import__("json"),
+    ns = {"os": os, "sys": sys, "json": __import__("json"),
           "_out": lambda name: os.path.join(os.environ["OUT_DIR"], "tasks", name)}
     exec(compile(ast.Module(body=keep, type_ignores=[]), str(RUNNER), "exec"), ns)
     return ns

devdex/tests/test_resume_identity.py

@@ -0,0 +1,125 @@
+"""Resume cannot relabel measurements from a different or unknown experiment."""
+import ast
+import inspect
+import json
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from devdex.tests._resume_shim import RUNNER, make_resume
+
+
+def saved(tmp_path, rows):
+    path = tmp_path / "tasks/p1_repo_external.json"
+    path.parent.mkdir(parents=True)
+    path.write_text(json.dumps(rows))
+    return path
+
+
+def resume(tmp_path, monkeypatch):
+    callback = make_resume(tmp_path, monkeypatch)
+    # Also run against the original four-argument function, so baseline failures show
+    # incompatible rows being accepted rather than just a changed function signature.
+    identity = ({"gt_sha": "current-gt", "model": "current-model"}
+                if "gt_sha" in inspect.signature(callback).parameters else {})
+    return callback("p1", "repo", "external", [{"id": "a"}], **identity)
+
+
+@pytest.mark.parametrize("old", [
+    {"gt_sha": "old-gt", "model": "current-model"},
+    {"gt_sha": "current-gt", "model": "old-model"},
+    {"model": "current-model"},
+    {"gt_sha": "current-gt"},
+])
+def test_incompatible_or_unstamped_measurement_is_rejected_without_overwrite(old, tmp_path,
+                                                                          monkeypatch, capsys):
+    path = saved(tmp_path, [{"qid": "a", **old}])
+    original = path.read_bytes()
+    with pytest.raises(SystemExit) as error:
+        resume(tmp_path, monkeypatch)
+    assert error.value.code == 78
+    assert "cannot resume" in capsys.readouterr().err
+    assert path.read_bytes() == original
+
+
+def test_identical_dataset_and_model_keep_finished_items(tmp_path, monkeypatch):
+    saved(tmp_path, [{"qid": "a", "gt_sha": "current-gt", "model": "current-model"}])
+    keep, todo = resume(tmp_path, monkeypatch)
+    assert [r["qid"] for r in keep] == ["a"]
+    assert todo == []
+
+
+def test_dead_rows_are_retried_under_current_identity(tmp_path, monkeypatch):
+    saved(tmp_path, [{"qid": "a", "error": "timeout", "gt_sha": "old-gt"}])
+    keep, todo = resume(tmp_path, monkeypatch)
+    assert keep == []
+    assert todo == [{"id": "a"}]
+
+
+def test_keep_dead_does_not_bypass_identity_check(tmp_path, monkeypatch):
+    monkeypatch.setenv("RESUME_KEEP_DEAD", "1")
+    saved(tmp_path, [{"qid": "a", "error": "timeout", "gt_sha": "old-gt"}])
+    with pytest.raises(SystemExit) as error:
+        resume(tmp_path, monkeypatch)
+    assert error.value.code == 78
+
+
+def test_fresh_run_needs_no_saved_identity(tmp_path, monkeypatch):
+    assert resume(tmp_path, monkeypatch) == ([], [{"id": "a"}])
+
+
+def test_real_runner_supplies_current_dataset_and_model_to_resume():
+    tree = ast.parse(RUNNER.read_text())
+    calls = [node for node in ast.walk(tree) if isinstance(node, ast.Call)
+             and isinstance(node.func, ast.Name) and node.func.id == "_resume"]
+    assert len(calls) == 1
+    arguments = {keyword.arg: ast.unparse(keyword.value) for keyword in calls[0].keywords}
+    assert arguments.get("gt_sha") == "GT_SHA"
+    assert arguments.get("model") == "MODEL"
+
+
+def test_suite_does_not_rescore_or_relabel_rejected_saved_cell(tmp_path, monkeypatch):
+    from devdex.scorer import suite
+
+    run = tmp_path / "saved-run"
+    cell = run / "run1/tasks/p1_repo_external.json"
+    cell.parent.mkdir(parents=True)
+    cell.write_text(json.dumps([{"qid": "a", "gt_sha": "old", "model": "old"}]))
+    result = run / "results.json"
+    result.write_text('{"model": "old", "dataset": "old.jsonl"}')
+    before = (cell.read_bytes(), result.read_bytes())
+    monkeypatch.setattr(suite.subprocess, "call", lambda *args, **kwargs: 78)
+    with pytest.raises(SystemExit) as error:
+        suite.run_cell(track="repo", arm="external", driver="opus", n_runs=1, concurrency=1,
+                       out_root=str(tmp_path), resume_dir=str(run), model_override="new")
+    assert error.value.code == 78
+    assert (cell.read_bytes(), result.read_bytes()) == before
+
+
+def test_register_arm_stops_all_tracks_and_preserves_resume_error(monkeypatch):
+    import subprocess
+    from benchmark import register_arm
+
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "synthetic-test-key")
+    monkeypatch.setenv("DEVDEX_EXT_MCP_URL", "https://example.test/mcp")
+    calls = []
+    def call(*args, **kwargs):
+        calls.append(args)
+        return 78
+    monkeypatch.setattr(subprocess, "call", call)
+    assert register_arm.run_agent_mode("example", track="all") == 78
+    assert len(calls) == 1
+
+
+def test_public_cli_never_reports_rejected_saved_records(monkeypatch):
+    from benchmark import run_benchmark as public
+
+    monkeypatch.setitem(sys.modules, "register_arm", SimpleNamespace(run_agent_mode=lambda **kw: 78))
+    monkeypatch.setattr(sys, "argv", ["run_benchmark.py", "--name", "example", "--mcp-url",
+                                     "https://example.test/mcp"])
+    called = []
+    monkeypatch.setattr(public, "report", lambda *args: called.append(args))
+    monkeypatch.setenv("DEVDEX_EXT_MCP_URL", "https://example.test/mcp")
+    assert public.main() == 78
+    assert called == []