From 44ba620c8b43db329e4abc056ff7e339e9f34232 Mon Sep 17 00:00:00 2001 From: Jeff Hamons Date: Sun, 12 Jul 2026 19:06:57 -0500 Subject: [PATCH] Detect terminal engine-down errors and fail remaining tasks fast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Grok Build 402 "usage balance exhausted" incident caused every task on that engine to burn both retry attempts (~5s each) before failing, and all of it landed in the scoreboard as ordinary model failures. - detect_engine_down_reason() classifies worker output (tail-scoped, ~2000 chars) for billing/auth terminal patterns: HTTP 402, "payment required", "balance exhausted", insufficient credits/quota, expired auth/token/ session/api-key, invalid api key. - RingerRunner tracks which engines are down for the run; once one task reveals it, every other task on that engine (running or queued) fails fast with status "engine-down" / verdict ENGINE_DOWN instead of burning a second attempt or a fresh taskdir. - aggregate_model_log_rows / aggregate_model_scoreboard_rows now exclude ENGINE_DOWN-verdict tasks entirely, so a billing outage no longer drags down a model's pass_rate/first_try_pass_rate. - engines/mock_worker.py gains a MOCK_ENGINE_DOWN directive for testing. Deviation from the "verification executes the artifact" invariant: the engine-down path skips running the check entirely (nothing to verify from a billing/auth failure, and PASS is never claimed). Stdin-closed, explicit sandbox mode, and "logs carry raw worker output" are unaffected — the new [ringer.py]-prefixed log lines follow the same convention as existing attempt-lifecycle lines. 🤖 Generated by JeffOS --- engines/mock_worker.py | 7 + ringer.py | 137 +++++++++++++++++- tests/test_engine_down.py | 288 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 431 insertions(+), 1 deletion(-) create mode 100644 tests/test_engine_down.py diff --git a/engines/mock_worker.py b/engines/mock_worker.py index e290baf..fe27f53 100755 --- a/engines/mock_worker.py +++ b/engines/mock_worker.py @@ -13,6 +13,10 @@ def has_fail_directive(spec: str) -> bool: return any(line.strip() == "MOCK_FAIL" for line in spec.splitlines()) +def has_engine_down_directive(spec: str) -> bool: + return any(line.strip() == "MOCK_ENGINE_DOWN" for line in spec.splitlines()) + + def parse_blocks(spec: str) -> list[tuple[str, str]]: lines = spec.splitlines() blocks: list[tuple[str, str]] = [] @@ -73,6 +77,9 @@ def main(argv: list[str]) -> int: return 2 spec = argv[-1] + if has_engine_down_directive(spec): + print("mock-worker: 402 Payment Required: usage balance exhausted", file=sys.stderr) + return 1 if has_fail_directive(spec): print("mock-worker: simulated failure") return 1 diff --git a/ringer.py b/ringer.py index aa75f84..80e06b9 100755 --- a/ringer.py +++ b/ringer.py @@ -1069,6 +1069,7 @@ class WorkerResult: tokens: int | None error: str | None = None reported_model: str | None = None + engine_down_reason: str | None = None @dataclass(frozen=True) @@ -1268,14 +1269,16 @@ def snapshot(self) -> dict[str, Any]: tasks.append(task_state) pass_count = sum(1 for item in tasks if item["status"] == "pass") fail_count = sum(1 for item in tasks if item["status"] == "fail") + engine_down_count = sum(1 for item in tasks if item["status"] == "engine-down") running_count = sum( 1 for item in tasks if item["status"] in {"running", "verifying", "retrying"} ) totals = { "running": running_count, - "done": pass_count + fail_count, + "done": pass_count + fail_count + engine_down_count, "pass": pass_count, "fail": fail_count, + "engine_down": engine_down_count, "tokens": sum(int(item["tokens"] or 0) for item in tasks), } return { @@ -1307,6 +1310,9 @@ def build_summary(self) -> dict[str, int]: return { "pass": sum(1 for runtime in self.runtimes if runtime.status == "pass"), "fail": sum(1 for runtime in self.runtimes if runtime.status == "fail"), + "engine_down": sum( + 1 for runtime in self.runtimes if runtime.status == "engine-down" + ), "tokens": sum(int(runtime.tokens or 0) for runtime in self.runtimes), } @@ -2388,6 +2394,7 @@ def scan_run_states(state_dir: Path) -> list[dict[str, Any]]: "fail": "var(--fail)", "error": "var(--fail)", "timeout": "var(--fail)", + "engine-down": "var(--waiting)", "running": "var(--running)", "retrying": "var(--running)", "verifying": "var(--running)", @@ -4813,6 +4820,13 @@ def model_log_row_is_reserved_fixture(row: dict[str, Any]) -> bool: return model_log_text(row.get("model")) in RESERVED_FIXTURE_MODELS +def model_log_row_is_engine_down(row: dict[str, Any]) -> bool: + # A terminal engine-level billing/auth failure isn't the model failing + # the task — it never got a fair attempt — so it must not drag down + # pass_rate/first_try_pass_rate the way a genuine FAIL would. + return model_log_text(row.get("verdict")).upper() == "ENGINE_DOWN" + + def model_reasoning_effort_keys(rows: list[dict[str, Any]]) -> set[tuple[str, str, bool]]: keys: set[tuple[str, str, bool]] = set() for row in rows: @@ -4948,6 +4962,8 @@ def aggregate_model_log_rows( final = ordered[-1] if model_log_row_is_reserved_fixture(final): continue + if model_log_row_is_engine_down(final): + continue group_engine = model_log_row_engine(final) group_model = model_log_row_model(final) group_task_type = model_log_row_task_type(final) @@ -6191,6 +6207,8 @@ def aggregate_model_scoreboard_rows( final = ordered[-1] if model_log_row_is_reserved_fixture(final): continue + if model_log_row_is_engine_down(final): + continue group_engine = model_log_row_engine(final) group_model = model_log_row_model(final) group_task_type = model_log_row_task_type(final) @@ -7397,6 +7415,7 @@ def __init__( self.verifier = Verifier() self.semaphore = asyncio.Semaphore(manifest.max_parallel) self.active_processes: dict[int, asyncio.subprocess.Process] = {} + self.engine_down: dict[str, str] = {} async def run(self) -> int: self.manifest.workdir.mkdir(parents=True, exist_ok=True) @@ -7447,10 +7466,54 @@ async def kill_all_workers(self) -> None: if proc.returncode is None: kill_process_group(proc) + def _engine_down_reason(self, engine_name: str) -> str | None: + with self.lock: + return self.engine_down.get(engine_name) + + def _mark_engine_down(self, engine_name: str, reason: str) -> str: + with self.lock: + existing = self.engine_down.get(engine_name) + if existing is not None: + return existing + self.engine_down[engine_name] = reason + return reason + + async def _finalize_engine_down( + self, + runtime: TaskRuntime, + reason: str, + *, + worker: WorkerResult | None, + attempt: int, + duration_ms: int, + ) -> None: + with self.lock: + runtime.attempts = attempt + runtime.status = "engine-down" + runtime.final_verdict = "ENGINE_DOWN" + runtime.ended_at_monotonic = time.monotonic() + effective_worker = worker or WorkerResult(returncode=None, timed_out=False, tokens=None) + verify = VerifyResult( + ok=False, + check_returncode=None, + check_timed_out=False, + raw_output_excerpt=( + f"[ringer.py] engine '{runtime.task.engine}' unavailable ({reason}); " + "task not attempted/verified — skipped, not a model failure." + ), + ) + self._log_attempt( + runtime, runtime.task.spec, attempt > 1, effective_worker, verify, "ENGINE_DOWN", duration_ms + ) + async def _run_task(self, runtime: TaskRuntime) -> None: async with self.semaphore: with self.lock: runtime.started_at_monotonic = time.monotonic() + down_reason = self._engine_down_reason(runtime.task.engine) + if down_reason is not None: + await self._finalize_engine_down(runtime, down_reason, worker=None, attempt=1, duration_ms=0) + return prepared, prepare_error = await self._prepare_taskdir(runtime) if not prepared: await self._record_prepare_error(runtime, prepare_error or "taskdir preparation failed") @@ -7458,6 +7521,12 @@ async def _run_task(self, runtime: TaskRuntime) -> None: current_spec = runtime.task.spec max_attempts = 2 for attempt in range(1, max_attempts + 1): + down_reason = self._engine_down_reason(runtime.task.engine) + if down_reason is not None: + await self._finalize_engine_down( + runtime, down_reason, worker=None, attempt=attempt, duration_ms=0 + ) + return retrying = attempt > 1 with self.lock: runtime.attempts = attempt @@ -7469,6 +7538,13 @@ async def _run_task(self, runtime: TaskRuntime) -> None: runtime.status = "verifying" if worker.tokens is not None: runtime.tokens = (runtime.tokens or 0) + worker.tokens + if worker.engine_down_reason is not None: + reason = self._mark_engine_down(runtime.task.engine, worker.engine_down_reason) + duration_ms = int((time.monotonic() - attempt_started) * 1000) + await self._finalize_engine_down( + runtime, reason, worker=worker, attempt=attempt, duration_ms=duration_ms + ) + return verify = await self.verifier.verify(runtime.task, runtime.taskdir) verdict = verdict_for(worker, verify) with self.lock: @@ -7771,14 +7847,30 @@ async def _run_worker(self, runtime: TaskRuntime, spec: str, attempt: int) -> Wo output_tail = capture.text() tokens = parse_token_count(output_tail, engine.token_regex) reported_model = parse_reported_model(output_tail, engine.model_report_regex) + engine_down_reason = ( + None + if timed_out or proc.returncode == 0 + # Scan only the last stretch of output: a terminal engine error is + # the last thing the process prints before it dies. Scanning the + # full ~1MB capture would also catch a coding task's own mid-run + # narration about payment/auth handling it's implementing. + else detect_engine_down_reason(output_tail[-2000:]) + ) if timed_out: append_text(log_path, f"\n[ringer.py] worker timed out after {runtime.task.timeout_s}s\n") append_text(log_path, f"[ringer.py] attempt {attempt} exited rc={proc.returncode}\n") + if engine_down_reason is not None: + append_text( + log_path, + f"[ringer.py] engine-down detected ({engine_down_reason}); " + f"marking engine '{runtime.task.engine}' down for this run\n", + ) return WorkerResult( returncode=proc.returncode, timed_out=timed_out, tokens=tokens, reported_model=reported_model, + engine_down_reason=engine_down_reason, ) async def _tee_stream( @@ -8002,6 +8094,49 @@ def verdict_for(worker: WorkerResult, verify: VerifyResult) -> str: return "FAIL" +# Terminal engine-level failures (billing exhausted, auth expired) are not the +# model's fault and can't be fixed by retrying the same task — unlike a normal +# FAIL, they mean every other queued task on this engine will fail the same +# way. Detected from the worker's own stdout/stderr; each tag is a rough +# classification for the raw log, not something callers branch on. +ENGINE_DOWN_PATTERNS: tuple[tuple[str, "re.Pattern[str]"], ...] = ( + ("http_402", re.compile(r'"status"\s*:\s*402\b')), + ("http_402", re.compile(r'"code"\s*:\s*402\b')), + ("payment_required", re.compile(r"payment required", re.IGNORECASE)), + ("balance_exhausted", re.compile(r"balance exhausted", re.IGNORECASE)), + ( + "insufficient_balance", + re.compile(r"insufficient (?:balance|credits?|quota|funds)", re.IGNORECASE), + ), + ( + "auth_expired", + re.compile( + r"(?:auth(?:entication|orization)?|token|api[\s_-]?key|session)" + r"[^\n]{0,20}\bexpired\b", + re.IGNORECASE, + ), + ), + ( + "auth_expired", + re.compile( + r"\bexpired\b[^\n]{0,20}" + r"(?:auth(?:entication|orization)?|token|api[\s_-]?key|session)", + re.IGNORECASE, + ), + ), + ("invalid_api_key", re.compile(r"invalid[\s_-]*api[\s_-]?key", re.IGNORECASE)), + ("unauthorized", re.compile(r"\b401\b[^\n]{0,40}unauthorized", re.IGNORECASE)), + ("unauthorized", re.compile(r"unauthorized[^\n]{0,40}\b401\b", re.IGNORECASE)), +) + + +def detect_engine_down_reason(output: str) -> str | None: + for reason, pattern in ENGINE_DOWN_PATTERNS: + if pattern.search(output): + return reason + return None + + def build_run_id(run_name: str) -> str: stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") safe_name = re.sub(r"[^A-Za-z0-9_.-]+", "-", run_name.strip()).strip("-") diff --git a/tests/test_engine_down.py b/tests/test_engine_down.py new file mode 100644 index 0000000..a2f3c65 --- /dev/null +++ b/tests/test_engine_down.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +"""Terminal engine-level failures (billing exhausted, auth expired) must not +burn task attempts or count against a model's pass rate. + +Regression for the 2026-07-12 incident: a Grok Build 402 "usage balance +exhausted" error caused every task on that engine to burn both attempts +(~5s each) before failing, and all of it landed in the scoreboard as +ordinary model failures. +""" +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from ringer import ( # noqa: E402 + aggregate_model_log_rows, + aggregate_model_scoreboard_rows, + detect_engine_down_reason, +) + + +def toml_string(value: object) -> str: + return json.dumps(str(value)) + + +class DetectEngineDownReasonTests(unittest.TestCase): + def test_matches_the_incident_message(self) -> None: + self.assertIsNotNone( + detect_engine_down_reason("402 Payment Required: usage balance exhausted") + ) + + def test_matches_json_style_status_codes(self) -> None: + self.assertIsNotNone(detect_engine_down_reason('{"error": {"status": 402}}')) + self.assertIsNotNone(detect_engine_down_reason('{"error": {"code": 402}}')) + + def test_matches_auth_expired_variants(self) -> None: + self.assertIsNotNone(detect_engine_down_reason("API key has expired")) + self.assertIsNotNone(detect_engine_down_reason("your session expired, please log in")) + self.assertIsNotNone(detect_engine_down_reason("token expired at 2026-07-12")) + self.assertIsNotNone(detect_engine_down_reason("invalid_api_key: invalid API key")) + + def test_matches_insufficient_credit_phrasing(self) -> None: + self.assertIsNotNone(detect_engine_down_reason("insufficient credits remaining")) + self.assertIsNotNone(detect_engine_down_reason("Insufficient quota for this request")) + + def test_ignores_ordinary_worker_output(self) -> None: + self.assertIsNone(detect_engine_down_reason("mock-worker: wrote 1 file(s): hello.txt")) + self.assertIsNone( + detect_engine_down_reason("diff shows 402 lines changed across the repo") + ) + self.assertIsNone(detect_engine_down_reason("")) + + +class AggregationExcludesEngineDownTests(unittest.TestCase): + """The scoreboard and model log must not treat ENGINE_DOWN like FAIL.""" + + def rows(self) -> list[dict[str, object]]: + return [ + { + "run_id": "run1", + "task_key": "a", + "worker_engine": "grok-build", + "model": "grok-build", + "task_type": "code-feature", + "verdict": "PASS", + "duration_ms": 100, + "worker_tokens": 10, + "retry": False, + "logged_at": "2026-07-12T10:00:00+00:00", + }, + { + "run_id": "run1", + "task_key": "b", + "worker_engine": "grok-build", + "model": "grok-build", + "task_type": "code-feature", + "verdict": "ENGINE_DOWN", + "duration_ms": 50, + "worker_tokens": 0, + "retry": False, + "logged_at": "2026-07-12T10:00:05+00:00", + }, + { + "run_id": "run1", + "task_key": "c", + "worker_engine": "grok-build", + "model": "grok-build", + "task_type": "code-feature", + "verdict": "ENGINE_DOWN", + "duration_ms": 0, + "worker_tokens": None, + "retry": False, + "logged_at": "2026-07-12T10:00:06+00:00", + }, + ] + + def test_model_log_aggregation_excludes_engine_down_tasks(self) -> None: + groups = aggregate_model_log_rows(self.rows()) + self.assertEqual(1, len(groups)) + group = groups[0] + # Only the genuine PASS counts; the two ENGINE_DOWN tasks are excluded + # entirely rather than counted as failures. + self.assertEqual(1, group["tasks"]) + self.assertEqual(1, group["passed"]) + self.assertEqual(0, group["failed"]) + self.assertEqual(1.0, group["pass_rate"]) + self.assertEqual(1.0, group["first_try_pass_rate"]) + + def test_model_scoreboard_aggregation_excludes_engine_down_tasks(self) -> None: + entries = aggregate_model_scoreboard_rows(self.rows()) + self.assertEqual(1, len(entries)) + entry = entries[0] + self.assertEqual(1, entry["tasks"]) + self.assertEqual(1, entry["passed"]) + self.assertEqual(0, entry["failed"]) + self.assertEqual(1.0, entry["pass_rate"]) + + +class EngineDownFastFailEndToEndTests(unittest.TestCase): + def test_engine_down_short_circuits_remaining_tasks_without_burning_attempts(self) -> None: + with tempfile.TemporaryDirectory() as temp_root: + root = Path(temp_root) + home = root / "home" + ringer_home = root / "ringer-home" + state_dir = root / "state" + workdir = root / "work" + config_path = root / "config.toml" + manifest_path = root / "manifest.json" + + home.mkdir() + ringer_home.mkdir() + + config_path.write_text( + "\n".join( + [ + f"state_dir = {toml_string(state_dir)}", + "", + "[eval]", + 'backend = "jsonl"', + f"jsonl_path = {toml_string(root / 'runs.jsonl')}", + "", + "[artifact]", + "enabled = false", + "", + "[engines.mock]", + f"bin = {toml_string(sys.executable)}", + "args_template = [", + f" {toml_string(ROOT / 'engines' / 'mock_worker.py')},", + ' "{spec}",', + "]", + "sandbox_args = []", + "full_access_args = []", + "", + ] + ), + encoding="utf-8", + ) + + # max_parallel=1 makes the ordering deterministic: task-one burns + # its one real (and only) attempt discovering the engine is down; + # task-two and task-three must never even spawn a worker. + manifest_path.write_text( + json.dumps( + { + "run_name": "engine-down-test", + "workdir": str(workdir), + "max_parallel": 1, + "worktrees": False, + "tasks": [ + { + "key": "task-one", + "engine": "mock", + "spec": "You are the deterministic mock worker.\nMOCK_ENGINE_DOWN", + "check": "test -f impossible.txt || { echo FAIL: never runs; exit 1; }", + }, + { + "key": "task-two", + "engine": "mock", + "spec": ( + "You are the deterministic mock worker. Write only the file " + "described in this MOCK_FILE block.\n" + "MOCK_FILE: hello.txt\n" + "hello from mock\n" + "MOCK_END" + ), + "check": "grep -q hello hello.txt || { echo FAIL: missing; exit 1; }", + "expect_files": ["hello.txt"], + }, + { + "key": "task-three", + "engine": "mock", + "spec": ( + "You are the deterministic mock worker. Write only the file " + "described in this MOCK_FILE block.\n" + "MOCK_FILE: hello.txt\n" + "hello from mock\n" + "MOCK_END" + ), + "check": "grep -q hello hello.txt || { echo FAIL: missing; exit 1; }", + "expect_files": ["hello.txt"], + }, + ], + }, + indent=2, + ), + encoding="utf-8", + ) + + env = os.environ.copy() + env["HOME"] = str(home) + env["RINGER_HOME"] = str(ringer_home) + env["XDG_CONFIG_HOME"] = str(root / "xdg-config") + + proc = subprocess.run( + [ + sys.executable, + "ringer.py", + "run", + str(manifest_path), + "--config", + str(config_path), + "--no-dashboard", + "--identity", + "engine-down-test", + ], + cwd=ROOT, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + timeout=30, + ) + + combined_output = proc.stdout + proc.stderr + self.assertEqual(1, proc.returncode, combined_output) + + # All three tasks end in the distinct engine-down status, none of + # them marked "fail" (which would count against the model). + for key in ("task-one", "task-two", "task-three"): + self.assertRegex( + combined_output, + re.compile( + rf"^{re.escape(key)}\s+engine-down\s+ENGINE_DOWN\s+1\s+", + re.MULTILINE, + ), + combined_output, + ) + + # task-one actually ran the mock worker and hit the 402. + task_one_log = workdir / "task-one" / "worker.log" + self.assertTrue(task_one_log.exists(), combined_output) + self.assertIn("402 Payment Required", task_one_log.read_text(encoding="utf-8")) + attempt_starts = re.findall( + r"^\[ringer\.py\] attempt (\d) started \d{4}-", + task_one_log.read_text(encoding="utf-8"), + flags=re.MULTILINE, + ) + self.assertEqual(["1"], attempt_starts, "task-one must not retry a dead engine") + + # task-two and task-three must never spawn a worker at all: no + # taskdir, no log, no wasted attempt. + self.assertFalse((workdir / "task-two").exists(), combined_output) + self.assertFalse((workdir / "task-three").exists(), combined_output) + + # The model log must carry ENGINE_DOWN verdicts, not FAIL, so the + # scoreboard doesn't count this as the model failing three tasks. + log_rows = [ + json.loads(line) + for line in (root / "runs.jsonl").read_text(encoding="utf-8").splitlines() + if line.strip() + ] + self.assertEqual(3, len(log_rows)) + for row in log_rows: + self.assertEqual("ENGINE_DOWN", row["verdict"]) + + +if __name__ == "__main__": + unittest.main(verbosity=2)