refactor(scripts): extract shared Gemini quota-error predicate (9 sites)

patterns/eval-change.md

@@ -3,7 +3,7 @@ governs:
   - evolution/
   - eval/
   - scripts/memory_indexer.py
-last_verified: "2026-07-11" # auto-bump @1783777889
+last_verified: "2026-07-14" # auto-bump @1784021189
 test_tasks:
   - "Add a new DeepEval metric under eval/ for the core_qa test suite"
   - "Add a new judge backend to evolution/judge/ using the provider registry"

scripts/_gemini_quota.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+"""Shared Gemini quota-error predicate.
+
+Extracted from duplicated `"429" in str(e) or "RESOURCE_EXHAUSTED" in str(e)`
+checks scattered across scripts/drift_check.py, scripts/trec_atom_benchmark.py,
+scripts/memory_indexer.py, scripts/bench/rule_following_judge.py, and
+scripts/bench/attention_dilution_probe.py. Intentionally scoped to ONLY this
+predicate — the surrounding retry/fallback bodies differ meaningfully per
+file (temperatures, token limits, exhaustion-tracking, retry cadence) and are
+NOT consolidated here.
+"""
+
+
+def is_quota_error(exc: Exception) -> bool:
+    """Return True if `exc` looks like a Gemini per-minute/per-day quota error."""
+    msg = str(exc)
+    return "429" in msg or "RESOURCE_EXHAUSTED" in msg

scripts/bench/attention_dilution_probe.py

@@ -35,6 +35,7 @@
 import argparse
 import datetime
 import hashlib
+import importlib.util
 import itertools
 import json
 import os
@@ -54,6 +55,7 @@
 _SCRIPTS_DIR = _BENCH_DIR.parent
 _REPO_ROOT = _SCRIPTS_DIR.parent
 _SP_PATH = _SCRIPTS_DIR / "standards_pack.py"
+_GQ_PATH = _SCRIPTS_DIR / "_gemini_quota.py"
 _FIXTURES_DIR = _BENCH_DIR / "fixtures"
 _DEFAULT_OUTPUT_DIR = _BENCH_DIR / "results"
 _PAIRWISE_CACHE_PATH = _BENCH_DIR / "attention_dilution_pairwise_cache.json"
@@ -109,6 +111,28 @@
 """
 
 
+# ---------------------------------------------------------------------------
+# Module loaders
+# ---------------------------------------------------------------------------
+
+
+def _load_gq():
+    """Load _gemini_quota as a module. scripts/bench/ is not on sys.path (this
+    file invokes standards_pack.py as a subprocess rather than in-process, so
+    unlike rule_following_judge.py it had no prior importlib plumbing) — same
+    spec_from_file_location pattern as rule_following_judge.py's `_load_gq`.
+    """
+    if "_gemini_quota" in sys.modules:
+        return sys.modules["_gemini_quota"]
+    spec = importlib.util.spec_from_file_location("_gemini_quota", _GQ_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Cannot load _gemini_quota from {_GQ_PATH}")
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["_gemini_quota"] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
 # ---------------------------------------------------------------------------
 # Standards-pack subprocess invocation
 # ---------------------------------------------------------------------------
@@ -433,7 +457,7 @@ def _judge_recall(
         except Exception as e:
             msg = str(e)
             last_err = msg
-            if "429" in msg or "RESOURCE_EXHAUSTED" in msg:
+            if _load_gq().is_quota_error(e):
                 if "PerDay" in msg:
                     exhausted.add(model)
                 else:
@@ -655,7 +679,7 @@ def _judge_pair(
         except Exception as e:
             msg = str(e)
             last_err = msg
-            if "429" in msg or "RESOURCE_EXHAUSTED" in msg:
+            if _load_gq().is_quota_error(e):
                 if "PerDay" in msg:
                     exhausted.add(model)
                 else:

scripts/bench/rule_following_judge.py

@@ -56,6 +56,7 @@
 _SCRIPTS_DIR = _BENCH_DIR.parent
 _REPO_ROOT = _SCRIPTS_DIR.parent
 _SP_PATH = _SCRIPTS_DIR / "standards_pack.py"
+_GQ_PATH = _SCRIPTS_DIR / "_gemini_quota.py"
 _DEFAULT_PROBES = _SCRIPTS_DIR / "tests" / "fixtures" / "rule_following_probes.jsonl"
 _DEFAULT_OUTPUT_DIR = _BENCH_DIR / "results"
 _JUDGE_CACHE_PATH = _BENCH_DIR / "rule_following_judge_cache.json"
@@ -131,6 +132,22 @@ def _load_sp():
     return mod
 
 
+def _load_gq():
+    """Load _gemini_quota as a module. Same importlib pattern as `_load_sp` —
+    scripts/bench/ is not on sys.path, so a plain `from _gemini_quota import
+    is_quota_error` would break direct script execution.
+    """
+    if "_gemini_quota" in sys.modules:
+        return sys.modules["_gemini_quota"]
+    spec = importlib.util.spec_from_file_location("_gemini_quota", _GQ_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Cannot load _gemini_quota from {_GQ_PATH}")
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["_gemini_quota"] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
 # ---------------------------------------------------------------------------
 # Fixture I/O
 # ---------------------------------------------------------------------------
@@ -372,7 +389,7 @@ def _judge_one(
         except Exception as e:  # noqa: BLE001 — broad on purpose for fallback
             msg = str(e)
             last_err = msg[:200]
-            if "429" in msg or "RESOURCE_EXHAUSTED" in msg:
+            if _load_gq().is_quota_error(e):
                 if "PerDay" in msg:
                     exhausted.add(model)
                     print(f"  {model} daily quota exhausted, skipping", file=sys.stderr)

scripts/drift_check.py

@@ -37,6 +37,8 @@
 import sys
 from pathlib import Path
 
+from _gemini_quota import is_quota_error
+
 PROJECT_ROOT = Path(__file__).parent.parent
 
 
@@ -841,8 +843,7 @@ def call_gemini(prompt: str) -> str | None:
                 )
                 return (response.text or "").strip()
             except Exception as e:
-                err = str(e)
-                if "429" in err or "RESOURCE_EXHAUSTED" in err:
+                if is_quota_error(e):
                     continue
                 print(f"  WARN: Gemini error ({model}): {e}", file=sys.stderr)
                 return None
@@ -1047,8 +1048,7 @@ def call_gemini(prompt: str) -> str | None:
                 )
                 return (response.text or "").strip()
             except Exception as e:
-                err = str(e)
-                if "429" in err or "RESOURCE_EXHAUSTED" in err:
+                if is_quota_error(e):
                     continue
                 print(f"  WARN: Gemini error ({model}): {e}", file=sys.stderr)
                 return None
@@ -1195,8 +1195,7 @@ def check_contradictions(project_root: Path, pattern_filter: str | None = None)
             response_text = (response.text or "").strip()
             break
         except Exception as e:
-            err = str(e)
-            if "429" in err or "RESOURCE_EXHAUSTED" in err:
+            if is_quota_error(e):
                 continue
             print(f"WARN: Gemini error ({model}): {e}", file=sys.stderr)
             return 0

scripts/memory_indexer.py

@@ -34,6 +34,7 @@
 from _time import local_now, utc_now  # noqa: E402
 from _exit_codes import SUCCESS, ABSTAIN, USAGE_ERROR, NOT_FOUND, AUTH_ERROR, INTERNAL_ERROR
 from _agent_io import is_agent_context, compact_json, select_fields
+from _gemini_quota import is_quota_error
 
 import sqlite_vec
 from google import genai
@@ -1853,12 +1854,6 @@ def _extract_content_for_llm(content: str, max_chars: int = 6000) -> str:
     return trimmed[:max_chars]
 
 
-def _is_quota_error(exc: Exception) -> bool:
-    """Return True if `exc` looks like a Gemini per-minute / per-day quota error."""
-    msg = str(exc)
-    return "429" in msg or "RESOURCE_EXHAUSTED" in msg
-
-
 def _generate_with_fallback(
     prompt,
     *,
@@ -1892,7 +1887,7 @@ def _generate_with_fallback(
             try:
                 return client.models.generate_content(model=model, **kwargs)
             except Exception as exc:
-                if _is_quota_error(exc):
+                if is_quota_error(exc):
                     if attempt == 1:
                         # brief within-model backoff — helps per-minute RPM caps
                         time.sleep(1)

scripts/tests/test_gemini_quota.py

@@ -0,0 +1,40 @@
+"""Tests for scripts/_gemini_quota.py — the shared Gemini quota-error predicate.
+
+Extracted from duplicated `"429" in str(e) or "RESOURCE_EXHAUSTED" in str(e)`
+checks across drift_check.py, trec_atom_benchmark.py, memory_indexer.py, and
+the two bench judge scripts. This test only covers the predicate itself.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_SCRIPTS_DIR = str(Path(__file__).resolve().parent.parent)
+if _SCRIPTS_DIR not in sys.path:
+    sys.path.insert(0, _SCRIPTS_DIR)
+
+from _gemini_quota import is_quota_error
+
+
+def test_matches_429():
+    assert is_quota_error(Exception("429 Too Many Requests")) is True
+
+
+def test_matches_resource_exhausted():
+    assert is_quota_error(Exception("RESOURCE_EXHAUSTED: quota exceeded")) is True
+
+
+def test_matches_both_markers_present():
+    assert is_quota_error(Exception("429 RESOURCE_EXHAUSTED PerDay limit")) is True
+
+
+def test_does_not_match_unrelated_error():
+    assert is_quota_error(Exception("500 Internal Server Error")) is False
+
+
+def test_does_not_match_empty_message():
+    assert is_quota_error(Exception("")) is False
+
+
+def test_does_not_match_connection_error():
+    assert is_quota_error(ConnectionError("connection reset by peer")) is False

scripts/trec_atom_benchmark.py

@@ -42,6 +42,8 @@
 if _SCRIPTS_DIR not in sys.path:
     sys.path.insert(0, _SCRIPTS_DIR)
 
+from _gemini_quota import is_quota_error  # noqa: E402
+
 BENCH_DIR = Path(os.path.expanduser("~/.deus/bench"))
 LOG_PATH = Path(os.path.expanduser("~/.deus/memory_tree_queries.jsonl"))
 DB_PATH = Path(os.path.expanduser("~/.deus/memory.db"))
@@ -412,7 +414,7 @@ def stage_judge(judge: str = "gemini", judge_model: str = "gemma4:e4b", fresh: b
                         time.sleep(4)
                         break
                     except Exception as e:
-                        if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
+                        if is_quota_error(e):
                             if "PerDay" in str(e):
                                 exhausted_models.add(model)
                                 print(f"  {model} daily quota exhausted, skipping", file=sys.stderr)
@@ -463,7 +465,7 @@ def stage_judge(judge: str = "gemini", judge_model: str = "gemma4:e4b", fresh: b
                                 time.sleep(4)
                                 break
                             except Exception as e:
-                                if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
+                                if is_quota_error(e):
                                     if "PerDay" in str(e):
                                         exhausted_models.add(model)
                                     else: