feat: pause batch on high rate limit window

src/lilith_agent/runner.py

@@ -164,6 +164,8 @@ def run_agent_on_questions(graph: Any, questions: list[dict], checkpoint_dir: st
         BatchAbortRateLimitError,
         QuestionRateLimitStreakError,
         RateLimitCooldownError,
+        batch_rate_limit_pause_seconds,
+        clear_batch_rate_limit_window,
         get_cheap_model,
         rate_limit_question_scope,
     )
@@ -179,6 +181,14 @@ def run_agent_on_questions(graph: Any, questions: list[dict], checkpoint_dir: st
             with ephemeral_memory():
                 return graph.invoke(task_state, {"configurable": {"thread_id": task_id}})
 
+    def _maybe_pause_for_batch_rate_limit() -> None:
+        pause_seconds = batch_rate_limit_pause_seconds()
+        if pause_seconds is None:
+            return
+        log_runner.warning("[runner] pausing batch for %ss due to rate limit window", pause_seconds)
+        time.sleep(pause_seconds)
+        clear_batch_rate_limit_window()
+
     for idx, question in enumerate(questions, start=1):
         reset_vision_state()
         task_id = question.get("task_id")
@@ -236,10 +246,12 @@ def run_agent_on_questions(graph: Any, questions: list[dict], checkpoint_dir: st
         except RateLimitCooldownError as exc:
             log_runner.warning("[runner] task=%s rate limited after retry: %s", task_id, exc)
             answers.append({"task_id": task_id, "submitted_answer": "AGENT ERROR: RATE LIMITED"})
+            _maybe_pause_for_batch_rate_limit()
             continue
         except QuestionRateLimitStreakError as exc:
             log_runner.warning("[runner] task=%s rate limit streak: %s", task_id, exc)
             answers.append({"task_id": task_id, "submitted_answer": "AGENT ERROR: RATE LIMITED"})
+            _maybe_pause_for_batch_rate_limit()
             continue
         except BatchAbortRateLimitError as exc:
             log_runner.warning("[runner] task=%s batch abort rate limit: %s", task_id, exc)
@@ -261,6 +273,7 @@ def run_agent_on_questions(graph: Any, questions: list[dict], checkpoint_dir: st
                     "submitted_answer": f"AGENT ERROR: {exc}",
                 }
             )
+            _maybe_pause_for_batch_rate_limit()
             continue
 
         last_m = result["messages"][-1]
@@ -297,6 +310,7 @@ def run_agent_on_questions(graph: Any, questions: list[dict], checkpoint_dir: st
             (submitted_answer[:160] + "…") if len(submitted_answer) > 160 else submitted_answer,
         )
         answers.append({"task_id": task_id, "submitted_answer": submitted_answer.strip()})
+        _maybe_pause_for_batch_rate_limit()
 
     return answers
 

tests/test_runner.py

@@ -227,3 +227,33 @@ def test_runner_stops_batch_and_writes_abort_marker_on_daily_quota(tmp_path: Pat
     assert data["reason"] == "daily quota exhausted"
     assert not (tmp_path / "task-abort.json").exists()
     assert not (tmp_path / "task-never.json").exists()
+
+
+class _GraphAlwaysSucceeds:
+    def __init__(self):
+        self.calls = 0
+
+    def invoke(self, state, config):
+        self.calls += 1
+        return {"messages": [AIMessage(content=f"answer-{self.calls}")]}
+
+
+def test_runner_pauses_batch_when_window_trips(monkeypatch, tmp_path: Path):
+    pauses = [300, None]
+    sleeps = []
+    monkeypatch.setattr("lilith_agent.models.batch_rate_limit_pause_seconds", lambda: pauses.pop(0))
+    monkeypatch.setattr("lilith_agent.models.clear_batch_rate_limit_window", lambda: None)
+    monkeypatch.setattr("lilith_agent.runner.time.sleep", sleeps.append)
+    monkeypatch.setattr("lilith_agent.runner._final_formatting_cleanup", lambda model, question, raw, llm_formatter_enabled=True: raw)
+
+    answers = run_agent_on_questions(
+        _GraphAlwaysSucceeds(),
+        [
+            {"task_id": "task-a", "question": "a"},
+            {"task_id": "task-b", "question": "b"},
+        ],
+        tmp_path,
+    )
+
+    assert sleeps == [300]
+    assert [answer["task_id"] for answer in answers] == ["task-a", "task-b"]