Fix generation quota pacing and withholding failures

README.md

@@ -121,6 +121,8 @@ Set `GROQ_API_KEY` in the root `.env`. Existing installations can use the `GROK_
 
 The model namespace starts with `openai/`, but requests go to **Groq's API**. The model is configured centrally in `rag_app/generation.py`.
 
+Generation uses remaining-token/reset headers to pace calls and keeps quota waits within request deadlines. The token estimate is approximate and applies to one provider instance, not a shared multi-worker quota. Offline case collection saves private writer/repair/verifier results and content-free HTTP statuses for diagnosis; these artifacts stay outside Git and are never returned as released answers.
+
 ### Prepare or restore artifacts
 
 This is a code-only checkout, not a bundled dataset or one-command model download. Restore the matching prepared artifacts before serving:

rag_app/api.py

@@ -166,6 +166,7 @@ def worker():
                 provider = runtime.get('provider')
                 if provider:
                     provider.usage.clear()
+                    provider.attempts.clear()
                     provider.deadline = started + 180
                 context = sessions.context(session.id)
                 question = body.question
@@ -191,7 +192,7 @@ def worker():
                             'pipeline': pipeline_version(),
                             'index': retrieve.index_identity() if hasattr(retrieve, 'index_identity') else 'injected_test_index',
                             'model': MODEL,
-                            'prompt': 'strict_v1', 'generation': {'provider': 'groq', 'reasoning_effort': 'low', 'max_output_tokens': 1500}}
+                            'prompt': 'strict_v2', 'generation': {'provider': 'groq', 'reasoning_effort': 'low', 'max_output_tokens': 1500}}
                 cached = None if regen or result else answers.get(session.id, identity, session_expires_at=session.expires_at)
                 if cached:
                     result = dict(cached)
@@ -246,6 +247,7 @@ def worker():
                     if not completed:
                         provider_calls = list(provider.usage)
                     provider.usage.clear()
+                    provider.attempts.clear()
                 if not completed:
                     write_metrics({'request_id': rid, 'status': 'cancelled_or_timed_out',
                                    'metrics': {'provider_calls': provider_calls}})

rag_app/generation.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import math
 import os
 import random
 import re
@@ -44,6 +45,20 @@ def __init__(self, api_key: str | None = None, *, model: str = MODEL,
             raise GenerationError("Groq API key is not configured")
         self.model, self.timeout, self.post = model, min(timeout, 75), post
         self.usage: list[dict[str, Any]] = []
+        self.attempts: list[dict[str, Any]] = []
+        self._remaining_tokens, self._quota_ready_at = float("inf"), 0.0
+
+    def _remember_quota(self, headers):
+        try:
+            remaining = float(headers["x-ratelimit-remaining-tokens"])
+            reset = re.fullmatch(r"(?:(\d+(?:\.\d+)?)m)?(?:(\d+(?:\.\d+)?)s)?", headers["x-ratelimit-reset-tokens"])
+            if reset and any(reset.groups()) and math.isfinite(remaining) and remaining >= 0:
+                seconds = float(reset[1] or 0) * 60 + float(reset[2] or 0)
+                if math.isfinite(seconds):
+                    self._remaining_tokens = remaining
+                    self._quota_ready_at = time.monotonic() + seconds + 2
+        except (KeyError, TypeError, ValueError):
+            pass
 
     def json_call(self, prompt: str, schema: dict[str, Any]) -> dict[str, Any]:
         url = "https://api.groq.com/openai/v1/chat/completions"
@@ -57,30 +72,40 @@ def json_call(self, prompt: str, schema: dict[str, Any]) -> dict[str, Any]:
         started = time.monotonic()
         deadline = min(started + self.timeout, getattr(self, "deadline", float("inf")))
         payload = None
-        for attempt in range(2):
+        for attempt in range(3):
+            # shortcut: estimate English prompt tokens; quota is per client, not shared across processes.
+            wait = max(0, self._quota_ready_at - time.monotonic()) if self._remaining_tokens < len(prompt) // 4 + 1600 else 0
+            if wait:
+                if wait >= deadline - time.monotonic():
+                    raise GenerationError("Quota reset exceeds request deadline", category="quota_deadline")
+                time.sleep(wait)
             remaining = deadline - time.monotonic()
             if remaining <= 0:
-                raise GenerationError("Groq request failed") from None
+                raise GenerationError("Groq request deadline exceeded", category="request_deadline") from None
             try:
                 response = self.post(url, headers=headers, json=body, timeout=remaining)
+                self.attempts.append({"status": response.status_code})
+                self._remember_quota(response.headers)
                 response.raise_for_status()
                 payload = response.json()
                 break
             except httpx.HTTPStatusError as exc:
                 status = exc.response.status_code
-                retry = attempt == 0 and (status == 429 or 500 <= status <= 599)
+                retry = attempt < 2 and (status == 429 or 500 <= status <= 599)
                 retry_after = 0.0
                 if retry:
                     try:
-                        retry_after = max(0.0, float(exc.response.headers.get("Retry-After", 0)))
+                        value = float(exc.response.headers.get("Retry-After", 0))
+                        retry_after = max(0.0, value) if math.isfinite(value) else 0
                     except (TypeError, ValueError):
                         pass
-                delay = retry_after + random.uniform(0.05, 0.2)
+                delay = max(5 if status == 429 else 0, retry_after + 2) + random.uniform(0.05, 0.2)
                 if retry and delay < deadline - time.monotonic():
                     time.sleep(delay)
                     continue
                 raise GenerationError(f"Groq HTTP {status}", category=f"http_{status}") from None
             except httpx.TransportError as exc:
+                self.attempts.append({"status": None, "error_type": type(exc).__name__})
                 raise GenerationError("Groq transport " + type(exc).__name__, category="transport_" + type(exc).__name__) from None
             except (ValueError, TypeError) as exc:
                 raise GenerationError("Groq returned an invalid HTTP response", category="invalid_response") from None
@@ -184,6 +209,20 @@ class AnswerGenerator:
 
     def __init__(self, json_call: Callable[[str, dict[str, Any]], dict[str, Any]]):
         self.json_call = json_call
+        self.trace: list[dict[str, Any]] = []
+
+    def _call(self, stage, prompt, schema):
+        # Private diagnostics only; never included in the released answer.
+        event = {"stage": stage}
+        self.trace.append(event)
+        try:
+            result = self.json_call(prompt, schema)
+        except GenerationError as exc:
+            event["error_type"] = exc.category
+            raise
+        event["result"] = deepcopy(result)
+        return result
+
 
     @staticmethod
     def _json(value: Any) -> str:
@@ -201,12 +240,13 @@ def _write(self, question: str, evidence: list[dict[str, Any]], repair: dict[str
         schema = deepcopy(ANSWER_SCHEMA)
         schema["properties"]["citations"]["items"]["enum"] = list(dict.fromkeys(
             str(item["id"]) for item in evidence if "id" in item))
-        result = self.json_call(prompt, schema)
+        result = self._call("repair" if repair else "write", prompt, schema)
         if (result.get("status") not in {"answered", "needs_clarification", "insufficient_evidence"} or
-                not isinstance(result.get("answer"), str) or not result["answer"].strip() or
+                not isinstance(result.get("answer"), str) or
+                (result.get("status") == "answered" and not result["answer"].strip()) or
                 not isinstance(result.get("citations"), list) or
                 not all(isinstance(c, str) for c in result["citations"])):
-            raise GenerationError("Groq returned an invalid answer shape")
+            raise GenerationError("Groq returned an invalid answer shape", category="invalid_answer_shape")
         return result
 
     @staticmethod
@@ -230,17 +270,22 @@ def _non_answer(status: str) -> dict[str, Any]:
         return {"status": "insufficient_evidence", "answer": "The available sources do not support a complete answer.", "citations": []}
 
     def _verify(self, question: str, evidence: list[dict[str, Any]], draft: dict[str, Any]) -> dict[str, Any]:
-        result = self.json_call("Verify support and completeness using only evidence. Check missing required steps and version conflicts. "
+        result = self._call("verify", "Verify support and completeness using only evidence. "
+            "The supported field means the answer's claims are backed by evidence, not that a product supports a feature. "
+            "An evidence-backed answer saying a feature is NOT supported must have supported=true. "
+            "Judge completeness against the question asked; do not require unrequested setup or remediation steps. "
+            "Check missing required steps and version conflicts for the same product, component and conditions. "
+            "A complete verdict requires supported=true and complete=true; otherwise use evidence_gap or clarify. "
             "If the draft contains approved_answer, also check that the presented answer preserves its meaning and qualifications. "
             f"Question: {question}\nEvidence: {self._json(evidence)}\nDraft: {self._json(draft)}", VERIFY_SCHEMA)
         if (result.get("verdict") not in {"complete", "evidence_gap", "clarify"} or
                 not isinstance(result.get("supported"), bool) or not isinstance(result.get("complete"), bool) or
                 not isinstance(result.get("missing_steps"), list) or not isinstance(result.get("version_conflicts"), list) or
                 not isinstance(result.get("findings"), list)):
-            raise GenerationError("Groq returned an invalid verification shape")
+            raise GenerationError("Groq returned an invalid verification shape", category="invalid_verification_shape")
         return result
-
     def generate(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, Any]:
+        self.trace.clear()
         source_ids = {str(item["id"]) for item in evidence if "id" in item}
         if not evidence or not source_ids:
             return self._non_answer("insufficient_evidence")
@@ -256,7 +301,7 @@ def generate(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, A
             if draft["status"] != "answered":
                 return self._non_answer(draft["status"])
             if self._citation_error(draft, source_ids):
-                raise GenerationError("Answer contains invalid or missing citations")
+                raise GenerationError("Answer contains invalid or missing citations", category="invalid_citations")
         verdict = self._verify(question, evidence, draft)
         if (verdict["verdict"] == "complete" and verdict["supported"] and verdict["complete"]
                 and not verdict["missing_steps"] and not verdict["version_conflicts"]):
@@ -275,7 +320,7 @@ def generate(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, A
         if repaired["status"] != "answered":
             return self._non_answer(repaired["status"])
         if self._citation_error(repaired, source_ids):
-            raise GenerationError("Answer contains invalid or missing citations")
+            raise GenerationError("Answer contains invalid or missing citations", category="invalid_citations")
         final_verdict = self._verify(question, evidence, repaired)
         if (final_verdict["verdict"] == "complete" and final_verdict["supported"] and final_verdict["complete"]
                 and not final_verdict["missing_steps"] and not final_verdict["version_conflicts"]):

scripts/eval_ragas.py

@@ -189,14 +189,16 @@ def collect(args):
             started = time.perf_counter()
             retrieved = retriever.retrieve(question)
             provider.usage.clear()
+            provider.attempts.clear()
             provider.deadline = time.monotonic() + 180
             try:
                 answer = generator.generate(question, retrieved['sources'])
             except GenerationError as exc:
                 answer = {'status': 'generation_error', 'answer': '', 'error_type': exc.category}
             case = make_case(q, original_docs[q['DOCUMENT']]['text'], retrieved, answer)
             case['generation'] = {'seconds': time.perf_counter() - started,
-                                  'provider_calls': list(provider.usage), 'error_type': answer.get('error_type')}
+                                  'provider_calls': list(provider.usage), 'error_type': answer.get('error_type'),
+                                  'http_attempts': list(provider.attempts), 'trace': list(generator.trace)}
             save(path, case)
             print(f"  saved {case['status']}; {len(case['sources'])} passages", flush=True)
     finally:

tests/test_app_generation.py

@@ -108,6 +108,59 @@ def test_missing_key_is_bounded_error(self):
         with self.assertRaisesRegex(GenerationError, "not configured"):
             GroqREST(api_key="")
 
+    def test_empty_nonanswers_are_safe_but_empty_answers_are_errors(self):
+        for status in ("insufficient_evidence", "needs_clarification"):
+            result, calls = self.run_with([{"status": status, "answer": "", "citations": []}])
+            self.assertEqual(result["status"], status)
+            self.assertTrue(result["answer"])
+            self.assertEqual(len(calls), 1)
+        with self.assertRaises(GenerationError) as error:
+            self.run_with([{"status": "answered", "answer": "", "citations": []}])
+        self.assertEqual(error.exception.category, "invalid_answer_shape")
+
+    def test_private_trace_records_verifier_rejection_without_changing_output(self):
+        replies = [self.answer.copy(), {**self.complete, "verdict": "evidence_gap", "complete": False}]
+        generator = AnswerGenerator(lambda *args: replies.pop(0))
+        result = generator.generate("How?", self.evidence)
+        self.assertEqual(result["status"], "insufficient_evidence")
+        self.assertEqual([x["stage"] for x in generator.trace], ["write", "verify"])
+        self.assertEqual(generator.trace[-1]["result"]["verdict"], "evidence_gap")
+        self.assertNotIn("trace", result)
+
+    def test_quota_headers_pace_next_call_and_respect_deadline(self):
+        clock = [0.0]
+        calls = []
+        def sleep(seconds): clock[0] += seconds
+        def post(url, **kwargs):
+            calls.append(clock[0])
+            return httpx.Response(200, request=httpx.Request("POST", url),
+                headers={"x-ratelimit-remaining-tokens": "10", "x-ratelimit-reset-tokens": "1m2s"},
+                json={"choices": [{"message": {"content": "{}"}, "finish_reason": "stop"}]})
+        with patch("rag_app.generation.time.monotonic", lambda: clock[0]), patch("rag_app.generation.time.sleep", sleep):
+            client = GroqREST("secret", post=post)
+            client.json_call("test", {})
+            client.json_call("test", {})
+            self.assertGreaterEqual(calls[1]-calls[0], 64)
+            client.deadline = clock[0] + 5
+            with self.assertRaises(GenerationError) as error: client.json_call("test", {})
+            self.assertEqual(error.exception.category, "quota_deadline")
+            self.assertEqual(len(calls), 2)
+
+    def test_two_rate_limits_get_buffered_bounded_retries(self):
+        clock, calls = [0.0], []
+        def post(url, **kwargs):
+            calls.append(clock[0])
+            return httpx.Response(429 if len(calls) < 3 else 200,
+                request=httpx.Request("POST", url), headers={"retry-after": "nan"},
+                json={"choices": [{"message": {"content": "{}"}, "finish_reason": "stop"}]})
+        def sleep(seconds): clock[0] += seconds
+        with patch("rag_app.generation.time.monotonic", lambda: clock[0]), patch("rag_app.generation.time.sleep", sleep):
+            client = GroqREST("secret", post=post)
+            self.assertEqual(client.json_call("test", {}), {})
+        self.assertEqual(len(calls), 3)
+        self.assertGreaterEqual(calls[1]-calls[0], 5)
+        self.assertGreaterEqual(calls[2]-calls[1], 5)
+
     def test_unreferenced_citation_array_does_not_pass(self):
         uncited = {**self.answer, "answer": "Use X without an inline citation."}
         calls = []