tests/test_agent_qa_rerun.py
Rama main · réplica de lectura del repositorio original.
"""The 2026-10-18 re-run's harness: version-2 scoring, the usage gate, the
API-failure rule, the server spec, the win rule and the v2 drift check.
No model, no broker."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "benchmarks" / "agent_qa"))
import build_hard # noqa: E402
import run # noqa: E402
from analyze import win_rule # noqa: E402
from score import oracle_reply, score, wrong_reply # noqa: E402
KNOWN = {"laptop": "laptop", "desktop": "desktop", "desktop-wsl": "desktop", "gt76": "gt76"}
V2 = [
{"id": "a", "category": "window_count", "truth": {"count": 13}},
{"id": "b", "category": "event_count", "truth": {"count": 7}},
{"id": "c", "category": "window_ids", "truth": {"job_ids": [4424, 4433, 4508]}},
{"id": "d", "category": "busiest_host",
"truth": {"tally": {"desktop": 11, "laptop": 9}, "names": ["desktop", "desktop-wsl"],
"known_names": KNOWN}},
{"id": "e", "category": "busiest_project",
"truth": {"tally": {"agrigen": 10, "asset-gen": 5}, "project": "agrigen"}},
{"id": "f", "category": "total_runtime",
"truth": {"seconds": 5103.342, "tolerance_s": 5.0, "job_ids": [1, 2, 3, 4, 5]}},
{"id": "g", "category": "queue_wait", "truth": {"seconds": 2280.27}},
{"id": "h", "category": "log_join",
"truth": {"lines": {"6603": "x", "6676": "y"}, "job_id": 6676}},
] # fmt: skip
def test_v2_oracle_replies_score_right_and_near_misses_wrong() -> None:
for q in V2:
assert score(q, oracle_reply(q))["verdict"] == "right", q["id"]
assert score(q, wrong_reply(q))["verdict"] == "wrong", q["id"]
@pytest.mark.parametrize(
("qid", "reply", "verdict"),
[
("d", "desktop-wsl", "right"), # an alias of the top host
("d", "desktop, laptop", "wrong"), # two hosts is not one answer
("e", "`Agrigen`", "right"),
("e", "agrigen-trellis", "wrong"),
("f", "5108", "right"), # 4.66 s off, inside 1 s per job
("f", "5109", "wrong"),
("f", "1:25:03", "right"),
("g", "2281", "right"),
("g", "2282", "wrong"),
("h", "6676", "right"),
("h", "6603, 6676", "wrong"),
("c", "4508, 4424, 4433", "right"),
("c", "4424, 4433", "wrong"),
("a", "13 jobs", "right"),
],
)
def test_v2_matching_rules(qid: str, reply: str, verdict: str) -> None:
q = next(q for q in V2 if q["id"] == qid)
assert score(q, f"ANSWER: {reply}")["verdict"] == verdict
class Clock:
def __init__(self) -> None:
self.t = 1_000_000.0
self.slept: list[float] = []
def now(self) -> float:
return self.t
def sleep(self, s: float) -> None:
self.slept.append(s)
self.t += s
def _rl(five: float, seven: float = 0.2, status: str = "allowed") -> dict:
return {"status": status, "five_hour": five, "five_hour_resets": 1_003_600,
"seven_day": seven, "seven_day_resets": 2_000_000} # fmt: skip
def test_gate_pauses_at_085_until_the_reset_and_records_it(tmp_path: Path) -> None:
c = Clock()
g = run.Gate(tmp_path / "pauses.jsonl", sleep=c.sleep, now=c.now)
g.update(_rl(0.84), "alt h01 rep1") # positive control: below the line, no pause
g.wait()
assert c.slept == [] and not (tmp_path / "pauses.jsonl").exists()
g.update(_rl(0.85), "mcp h01 rep1")
g.wait()
assert c.t == 1_003_660 # resetsAt + 60 s
log = [json.loads(ln) for ln in (tmp_path / "pauses.jsonl").read_text().splitlines()]
assert len(log) == 1 and log[0]["after"] == "mcp h01 rep1" and log[0]["five_hour"] == 0.85
def test_gate_pauses_on_a_rejection_and_stops_on_seven_day(tmp_path: Path) -> None:
c = Clock()
g = run.Gate(tmp_path / "p.jsonl", sleep=c.sleep, now=c.now)
g.update(_rl(0.40, status="rejected"), "x")
assert g.until == 1_003_660
g.update(_rl(0.40, seven=0.79), "y") # below the stop line
with pytest.raises(run.UsageStop):
g.update(_rl(0.40, seven=0.80), "z")
def _events(*, rejected: bool, result: dict | None) -> list[dict]:
ev: list[dict] = [{"type": "system", "subtype": "init"}]
if rejected:
ev.append({"type": "rate_limit_event", "rate_limit_info": {"status": "rejected"}})
if result is not None:
ev.append({"type": "result", **result})
return ev
def test_api_failure_is_a_rejection_before_an_answer_and_nothing_else() -> None:
ok = {"subtype": "success", "is_error": False, "result": "ANSWER: 3"}
err = {"subtype": "success", "is_error": True, "result": "API Error: rate limit"}
assert run.api_failure(_events(rejected=True, result=err), 0) == "rate_limited"
assert run.api_failure(_events(rejected=True, result=None), 1) == "rate_limited"
# A rejection the run recovered from, and a plain answer, are not failures.
assert run.api_failure(_events(rejected=True, result=ok), 0) is None
assert run.api_failure(_events(rejected=False, result=ok), 0) is None
# A timeout stays in as an invalid run.
assert run.api_failure(_events(rejected=False, result=None), "timeout") is None
assert (
run.api_failure(_events(rejected=False, result={"terminal_reason": "api_error"}), 0)
== "api_error"
)
def test_drop_api_failures_moves_each_triple_once(tmp_path: Path) -> None:
res = tmp_path / "results.jsonl"
rows = [
{"arm": "alt", "qid": "h01", "rep": 1, "api_failure": "rate_limited"},
{"arm": "mcp", "qid": "h01", "rep": 1, "api_failure": None},
]
res.write_text("".join(json.dumps(r) + "\n" for r in rows))
run.drop_api_failures(res)
assert [r["arm"] for r in map(json.loads, res.read_text().splitlines())] == ["mcp"]
# The re-run failed the same way: it stays this time.
with res.open("a") as fh:
fh.write(json.dumps(rows[0]) + "\n")
run.drop_api_failures(res)
assert len(res.read_text().splitlines()) == 2
assert len(res.with_suffix(".api_failures.jsonl").read_text().splitlines()) == 1
def test_server_spec_takes_a_full_commit_only() -> None:
sha = "d2817e4" + "0" * 33
assert run.check_server_spec(f"git+https://github.com/musharna/jobd@{sha}") == sha
assert run.check_server_spec(run.SERVER) is None
for bad in ("git+https://github.com/musharna/jobd@d2817e4", "jobd[mcp]==0.5.50",
"git+https://github.com/someone/jobd@" + sha): # fmt: skip
with pytest.raises(SystemExit):
run.check_server_spec(bad)
cmd = run.server_cmd(f"git+https://github.com/musharna/jobd@{sha}", "2026-10-18")
assert cmd == ["uvx", "--exclude-newer", "2026-10-18", "--from",
f"jobd[mcp] @ git+https://github.com/musharna/jobd@{sha}", "jobd-mcp"] # fmt: skip
assert run.server_cmd(run.SERVER, None) == run.MCP_SERVER
def _rows(right: dict[str, list[str]], qids: list[str]) -> list[dict]:
return [
{"arm": a, "qid": q, "rep": k, "valid": True,
"verdict": "right" if q in right[a] else "wrong"}
for a in ("alt", "mcp") for q in qids for k in (1, 2, 3)
] # fmt: skip
def test_win_rule_win_tie_loss() -> None:
qids = [f"h{i:02d}" for i in range(38)]
base = qids[:20]
win = win_rule(_rows({"alt": base, "mcp": qids[:32]}, qids), qids)
assert win["verdict"] == "WIN" and win["only_mcp"] == 12 and win["mcnemar_p"] < 0.05
assert win_rule(_rows({"alt": base, "mcp": qids[:23]}, qids), qids)["verdict"] == "TIE"
# Non-inferiority: 3 fewer is a loss even though p > 0.05.
loss = win_rule(_rows({"alt": base, "mcp": qids[:17]}, qids), qids)
assert loss["verdict"] == "LOSS" and loss["mcnemar_p"] > 0.05
assert win_rule(_rows({"alt": base, "mcp": qids[:18]}, qids), qids)["verdict"] == "TIE"
def test_win_rule_counts_an_invalid_run_as_not_right() -> None:
qids = ["h01"]
rows = _rows({"alt": qids, "mcp": qids}, qids)
rows[-1]["valid"] = False
out = win_rule(rows, qids)
assert out["mcp_right_all"] == 0 and out["alt_right_all"] == 1
def _job(i, project, state, fin, worker="laptop", start=None):
return {"id": i, "project": project, "state": state, "worker": worker,
"finished_at": fin, "started_at": start or fin, "submitted_at": fin} # fmt: skip
def test_v2_truths_are_half_open_and_drift_sees_a_changed_answer() -> None:
jobs = [
_job(1, "p", "failed", "2026-10-01T00:00:00+00:00"), # at the start: in
_job(2, "p", "failed", "2026-10-01T06:00:00+00:00"), # at the end: out
_job(3, "q", "failed", "2026-10-01T03:00:00+00:00", worker="desktop"),
_job(
4, "p", "completed", "2026-10-01T02:00:00+00:00", start="2026-10-01T01:59:00+00:00"
), # fmt: skip
]
w = {"after": "2026-10-01T00:00:00Z", "before": "2026-10-01T06:00:00Z"}
assert build_hard.t_window_ids(w, jobs, None) == {"job_ids": [1, 3]}
assert build_hard.t_window_count({**w, "project": "p", "state": "failed"}, jobs, None) == {
"count": 1
}
assert build_hard.t_busiest_project({**w, "state": "failed"}, jobs, None)["tally"] == {
"p": 1, "q": 1,
} # fmt: skip
assert build_hard.unique_top({"p": 1, "q": 1}) is None
assert build_hard.t_total_runtime({**w, "project": "p"}, jobs, None)["seconds"] == 60.0
key = {"questions": [{"id": "h01", "category": "window_ids", "params": w,
"truth": {"job_ids": [1, 3]}}]} # fmt: skip
class FakeBroker:
def __init__(self, rows):
self.rows = rows
def get_with_total(self, path, **params):
return self.rows, len(self.rows)
assert build_hard.drift(FakeBroker(jobs), key) == [] # positive control
moved = [*jobs[:2], {**jobs[2], "finished_at": "2026-10-01T07:00:00+00:00"}, jobs[3]]
assert build_hard.drift(FakeBroker(moved), key) == ["h01 (window_ids): answer changed"]