fix(eval): recover judge errors and skip empty or oversized answers

nemo_gym/openai_utils.py

@@ -1221,11 +1221,13 @@ class NeMoGymChatCompletionCreateParamsNonStreaming(BaseModel):
 ########################################
 
 # See https://platform.openai.com/docs/guides/error-codes/api-errors
+# 404 can be a transient model-routing failure; retries remain bounded.
+# 408 is a request timeout.
 # 500 is internal server error, which may sporadically occur
 # 502 is Bad gateway (when the endpoint is overloaded)
 # 504 is Gateway timeout (when the endpoint config has too low of a gateway timeout setting for the model to finish generating)
 RATE_LIMIT_ERROR_CODES = [429, 502, 503, 504, 520]
-RETRY_ERROR_CODES = RATE_LIMIT_ERROR_CODES + [500]
+RETRY_ERROR_CODES = RATE_LIMIT_ERROR_CODES + [404, 408, 500]
 
 
 class NeMoGymAsyncOpenAI(BaseModel):  # pragma: no cover
@@ -1247,6 +1249,8 @@ class NeMoGymAsyncOpenAI(BaseModel):  # pragma: no cover
         ),
     )
 
+    max_http_attempts: int = Field(default=MAX_NUM_TRIES, ge=1)
+
     default_headers: Dict[str, str] = Field(
         default_factory=dict,
         description="Extra headers to include in every request.",
@@ -1266,7 +1270,7 @@ async def _request(self, **request_kwargs: Dict) -> ClientResponse:
         return await self._request_with_retry(**request_kwargs)
 
     async def _request_with_retry(self, **request_kwargs: Dict) -> ClientResponse:
-        max_num_tries = MAX_NUM_TRIES
+        max_num_tries = self.max_http_attempts
         tries = 0
         while tries < max_num_tries:
             tries += 1
@@ -1277,8 +1281,12 @@ async def _request_with_retry(self, **request_kwargs: Dict) -> ClientResponse:
                 if response.status in RATE_LIMIT_ERROR_CODES and self.internal:
                     max_num_tries += 1
 
-                content = (await response.content.read()).decode()
-                kind = "rate_limit" if response.status in RATE_LIMIT_ERROR_CODES else "server_error"
+                # Preserve the final error body for raise_for_status and avoid sleeping
+                # after the last attempt. Reading intermediate bodies releases sockets.
+                if tries >= max_num_tries:
+                    break
+                content = (await response.content.read()).decode(errors="replace")
+                kind = "rate_limit" if response.status in RATE_LIMIT_ERROR_CODES else "http_error"
                 print(
                     f"[model_retry url={request_kwargs.get('url')} status={response.status} kind={kind} try={tries} max_tries={max_num_tries} error_msg={content[:200]}]",
                     flush=True,

resources_servers/equivalence_llm_judge/README.md

@@ -19,6 +19,39 @@ https://huggingface.co/datasets/nvidia/Nemotron-RL-knowledge-openqa
 - `reward_if_full_generation_succeeds` (float, default 0.5): reward when full generation check succeeds after extraction failure. Set to 1.0 for full credit.
 - `extraction_length_threshold` (int, default 120): skip regex extraction when expected answer exceeds this length. Use full generation instead. Only applies when per-record regex is present. Set to null to disable.
 
+### Empty answers, answer-length guard, and judge recovery
+
+Missing assistant messages and empty or whitespace-only final assistant text
+score zero without a judge call, with `failure_reason: empty_final_answer`.
+The full response is retained, including any reasoning. This check applies before
+regex extraction, so nonempty answers still use the existing extraction and
+full-generation fallback logic even when a regex produces an empty match.
+
+`max_answer_chars` is optional and defaults to `null` (no limit) for every dataset.
+To protect a judge endpoint from oversized answers, set it on the resources server:
+
+```yaml
+equivalence_llm_judge:
+  resources_servers:
+    equivalence_llm_judge:
+      max_answer_chars: 100000
+```
+
+When enabled, longer final assistant answers score zero without a judge call.
+The full response is retained, with `failure_reason` starting with
+`final_answer_too_long`. The check counts characters in the last assistant message
+before regex extraction, excludes reasoning and earlier messages, and does not
+stop generation early. Enabling the limit changes reward assignment, so choose it
+explicitly in the configuration for your workload (evaluation, synthetic data
+generation, or RL).
+
+If judge calls exhaust their retries, the shared judge failsafe saves the policy
+response in the failures sidecar as `judge_failed`, excluded from scored aggregates.
+Recover these rows with `gym eval reverify --judge-failed-only` and the same
+verifier/judge configuration. Successful grades are preserved. This applies to
+new sidecar-tagged failures; older in-band `JUDGE_ERROR` rows require migration
+before using it. A run with pending grades is incomplete.
+
 ### Input schema
 Accepts the same outer request structure as other resources servers:
 - `responses_create_params`: the original model query (used here to extract a question/context string from user messages for the judge prompt).

resources_servers/equivalence_llm_judge/app.py

@@ -38,7 +38,7 @@
     SimpleResourcesServer,
 )
 from nemo_gym.config_types import ModelServerRef
-from nemo_gym.judge import JudgeError, call_judge
+from nemo_gym.judge import call_judge
 from nemo_gym.openai_utils import (
     NeMoGymEasyInputMessage,
     NeMoGymResponse,
@@ -137,6 +137,11 @@ class LLMJudgeResourcesServerConfig(BaseResourcesServerConfig):
     # returned; otherwise, the entire last match is used.
     response_extract_regex: Optional[str] = None
     msg_extraction_failure: str = "[NO VALID ANSWER EXTRACTED]"
+    max_answer_chars: Optional[int] = Field(
+        default=None,
+        gt=0,
+        description="Score longer final assistant answers zero without calling the judge. Thinking is excluded.",
+    )
 
     # Swap check: Run second judge pass with swapped expected/generated to detect positional bias
     check_twice_swap: bool = False
@@ -189,16 +194,14 @@ class LLMJudgeVerifyRequest(LLMJudgeRunRequest, BaseVerifyRequest):
     pass
 
 
-# Marks a rollout whose verdict is absent because the judge service failed,
-# as distinct from a judge that ran and returned no parseable verdict.
+# Legacy in-band service failures remain readable when aggregating old runs.
+# New failures go through the shared judge_failed sidecar instead.
 JUDGE_ERROR_LABEL = "JUDGE_ERROR"
 
 
 class JudgeEvaluation(BaseModel):
     responses_create_params: NeMoGymResponseCreateParamsNonStreaming
-    # None when the judge could not be reached or returned an unusable payload.
-    # The rollout is still recorded so the failure is visible in the artifacts
-    # rather than taking down the run that produced it.
+    # None on legacy rows whose judge call failed.
     response: Optional[NeMoGymResponse] = None
     # Extracted verdict token from judge output, e.g., "[[A=B]]" or "[[A!=B]]",
     # or JUDGE_ERROR_LABEL when the judge itself failed.
@@ -490,15 +493,26 @@ async def verify(self, body: LLMJudgeVerifyRequest) -> LLMJudgeVerifyResponse:
         """Verify model response by comparing with expected answer using LLM judge.
 
         Flow:
-        1. Extract question and expected answer
+        1. Extract expected answer; reject empty or over-limit final text; extract question
         2. Determine extraction regex (per-record override, length threshold)
         3. Extract answer to judge (could be regex-extracted OR full generation)
         4. Run first judge evaluation on extracted answer
         5. Handle failure → rescue with full generation or immediate fail
         6. Handle success → swap check or immediate success
         """
-        # Step 1: Extract question and expected answer
+        # Step 1: Check the raw final answer before applying extraction or calling the judge.
         expected = _extract_expected_answer(body) or ""
+        answer = _extract_last_assistant_text(body, extract_regex=None)
+        if not answer:
+            result = self._make_response(body, expected, reward=0.0, evaluations=[])
+            result.failure_reason = "empty_final_answer"
+            return result
+        if self.config.max_answer_chars is not None and len(answer) > self.config.max_answer_chars:
+            result = self._make_response(body, expected, reward=0.0, evaluations=[])
+            result.failure_reason = (
+                f"final_answer_too_long: {len(answer)} characters exceeds {self.config.max_answer_chars}"
+            )
+            return result
         question = _extract_question_text(body.responses_create_params, self.config.question_extract_regex)
 
         # Step 2: Determine extraction regex (None if long answer triggers threshold)
@@ -544,31 +558,15 @@ async def _generate_judge_evaluation(
         responses_create_params.input = msgs
 
         async with self._judge_endpoint_max_concurrency:
-            try:
-                judge_response = await call_judge(
-                    self.server_client,
-                    server_name=cfg.judge_model_server.name,
-                    url_path="/v1/responses",
-                    json=responses_create_params,
-                    response_model=NeMoGymResponse,
-                )
-            except JudgeError as e:
-                print(
-                    f"DEBUG: LLMJudgeResourcesServer: judge model server HTTP POST error: {e}",
-                    flush=True,
-                )
-                # Do not re-raise. The judge is a separate service, and a
-                # transient failure from it is not a failure of the rollout:
-                # propagating here aborts the whole evaluation and discards every
-                # rollout already generated, which can be many hours of work.
-                # Record the failure on the rollout and score it not-equal, so
-                # the run completes and a downstream check can decide whether the
-                # judge-error rate makes the score untrustworthy.
-                return False, JudgeEvaluation(
-                    responses_create_params=responses_create_params,
-                    response=None,
-                    verdict_label=JUDGE_ERROR_LABEL,
-                )
+            # Let the shared verify-endpoint failsafe preserve exhausted judge
+            # failures for `gym eval reverify --judge-failed-only`.
+            judge_response = await call_judge(
+                self.server_client,
+                server_name=cfg.judge_model_server.name,
+                url_path="/v1/responses",
+                json=responses_create_params,
+                response_model=NeMoGymResponse,
+            )
 
         eval_record = JudgeEvaluation(
             responses_create_params=responses_create_params,