diff --git a/nemo_gym/openai_utils.py b/nemo_gym/openai_utils.py index 07341b76a85..208d14ba062 100644 --- a/nemo_gym/openai_utils.py +++ b/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, diff --git a/resources_servers/equivalence_llm_judge/README.md b/resources_servers/equivalence_llm_judge/README.md index a7881a27a70..c13a74b1cfd 100644 --- a/resources_servers/equivalence_llm_judge/README.md +++ b/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). diff --git a/resources_servers/equivalence_llm_judge/app.py b/resources_servers/equivalence_llm_judge/app.py index 2b43e12aaae..d33a446ffa8 100644 --- a/resources_servers/equivalence_llm_judge/app.py +++ b/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, diff --git a/resources_servers/equivalence_llm_judge/tests/test_app.py b/resources_servers/equivalence_llm_judge/tests/test_app.py index 123f01b3f73..18434c3e160 100644 --- a/resources_servers/equivalence_llm_judge/tests/test_app.py +++ b/resources_servers/equivalence_llm_judge/tests/test_app.py @@ -12,6 +12,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json from copy import deepcopy from pathlib import Path from unittest.mock import AsyncMock, MagicMock @@ -20,6 +21,7 @@ from pytest import approx, fixture, mark from nemo_gym.config_types import AggregateMetricsRequest, ModelServerRef +from nemo_gym.judge import judge_failsafe from nemo_gym.openai_utils import ( NeMoGymResponse, NeMoGymResponseCreateParamsNonStreaming, @@ -90,6 +92,107 @@ def _msg(self, text: str) -> NeMoGymResponseOutputMessage: type="message", ) + @mark.parametrize("limit,answer_length,skip", [(100, 101, True), (100, 100, False), (None, 100001, False)]) + async def test_final_answer_limit_preserves_response(self, config, limit, answer_length, skip): + assert config.max_answer_chars is None + if limit is not None: + config.max_answer_chars = limit + client = MagicMock(spec=ServerClient) + reply = MagicMock(ok=True) + reply.read = AsyncMock(return_value=self._create_response("judge", self._msg("[[A=B]]"))) + client.post = AsyncMock(return_value=reply) + server = LLMJudgeResourcesServer(config=config, server_client=client) + response = NeMoGymResponse.model_validate_json(self._create_response("policy", self._msg("é" * answer_length))) + request = LLMJudgeVerifyRequest( + responses_create_params=NeMoGymResponseCreateParamsNonStreaming(input=[]), + response=response, + expected_answer="answer", + ) + before = request.model_dump() + result = await server.verify(request) + assert request.model_dump() == before + assert result.response.model_dump() == before["response"] + if skip: + client.post.assert_not_awaited() + assert result.reward == 0 + assert result.judge_evaluations == [] + assert result.failure_reason == "final_answer_too_long: 101 characters exceeds 100" + else: + client.post.assert_awaited_once() + assert result.reward == 1 + assert result.failure_reason is None + + async def test_answer_limit_excludes_thinking_and_previous_messages(self, config): + config.max_answer_chars = 100 + client = MagicMock(spec=ServerClient) + reply = MagicMock(ok=True) + reply.read = AsyncMock(return_value=self._create_response("judge", self._msg("[[A=B]]"))) + client.post = AsyncMock(return_value=reply) + server = LLMJudgeResourcesServer(config=config, server_client=client) + response_data = json.loads(self._create_response("policy", self._msg("short answer"))) + response_data["output"][:0] = [ + self._msg("previous answer " * 100).model_dump(), + { + "id": "thinking", + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "thinking " * 10000}], + }, + ] + request = LLMJudgeVerifyRequest( + responses_create_params=NeMoGymResponseCreateParamsNonStreaming(input=[]), + response=NeMoGymResponse.model_validate(response_data), + expected_answer="answer", + ) + result = await server.verify(request) + assert result.reward == 1 + client.post.assert_awaited_once() + prompt = result.judge_evaluations[0].responses_create_params.input[-1].content + assert "short answer" in prompt + assert "thinking" not in prompt + assert "previous answer" not in prompt + + async def test_answer_limit_precedes_regex_and_counts_all_text_blocks(self, config): + config.max_answer_chars = 100 + config.response_extract_regex = r"Answer: (.*)" + client = MagicMock(spec=ServerClient) + client.post = AsyncMock() + server = LLMJudgeResourcesServer(config=config, server_client=client) + message = self._msg("x" * 95) + message.content.append(NeMoGymResponseOutputText(annotations=[], text="Answer: x", type="output_text")) + request = LLMJudgeVerifyRequest( + responses_create_params=NeMoGymResponseCreateParamsNonStreaming(input=[]), + response=NeMoGymResponse.model_validate_json(self._create_response("policy", message)), + expected_answer="x", + ) + result = await server.verify(request) + assert result.reward == 0 + assert result.failure_reason == "final_answer_too_long: 105 characters exceeds 100" + client.post.assert_not_awaited() + + async def test_exhausted_judge_failure_preserves_response_for_recovery(self, config): + client = MagicMock(spec=ServerClient) + client.post = AsyncMock(side_effect=RuntimeError("upstream judge unavailable")) + server = LLMJudgeResourcesServer(config=config, server_client=client) + request = LLMJudgeVerifyRequest( + responses_create_params=NeMoGymResponseCreateParamsNonStreaming(input=[]), + response=NeMoGymResponse.model_validate_json(self._create_response("policy", self._msg("answer"))), + expected_answer="answer", + ) + result = await judge_failsafe(server.verify)(request) + failed = json.loads(result.body) + assert failed["_ng_failure_class"] == "judge_failed" + assert "upstream judge unavailable" in failed["_ng_failure_judge_error"] + assert failed["response"] == request.response.model_dump(mode="json") + assert failed["expected_answer"] == "answer" + client.post.assert_awaited_once() + + reply = MagicMock(ok=True) + reply.read = AsyncMock(return_value=self._create_response("judge", self._msg("[[A=B]]"))) + client.post = AsyncMock(return_value=reply) + recovered = await server.verify(request) + assert recovered.reward == 1 + assert recovered.response.model_dump() == request.response.model_dump() + async def test_verify_equal_then_confirm(self, config: LLMJudgeResourcesServerConfig) -> None: server_mock = MagicMock(spec=ServerClient) # Default: check_twice_swap = False @@ -202,38 +305,47 @@ async def test_unexpected_judge_output_defaults_to_not_equal(self, config: LLMJu res = await rs.verify(req) assert res.reward == approx(0.0) - async def test_missing_assistant_text_uses_configured_failure_message( - self, config: LLMJudgeResourcesServerConfig + @mark.parametrize("case", ["missing", "reasoning_only", "empty", "whitespace", "no_text", "last_empty"]) + async def test_empty_final_answer_skips_judge_and_preserves_response( + self, config: LLMJudgeResourcesServerConfig, case: str ) -> None: - server_mock = MagicMock(spec=ServerClient) - cfg = config.model_copy(deep=True) - cfg.msg_extraction_failure = "[CUSTOM EXTRACTION FAILURE]" - rs = LLMJudgeResourcesServer(config=cfg, server_client=server_mock) - - post_mock = MagicMock() - post_mock.read = AsyncMock(return_value=self._create_response("f", self._msg("[[A!=B]]"))) - server_mock.post = AsyncMock(return_value=post_mock) - - req = LLMJudgeVerifyRequest( + config.check_twice_swap = True + config.check_full_generation_on_fail = True + client = MagicMock(spec=ServerClient) + client.post = AsyncMock() + server = LLMJudgeResourcesServer(config=config, server_client=client) + outputs = { + "missing": [], + "reasoning_only": [ + { + "id": "thinking", + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "The answer is 42"}], + } + ], + "empty": [self._msg("")], + "whitespace": [self._msg(" \n\t ")], + "no_text": [self._msg("").model_copy(update={"content": []})], + "last_empty": [self._msg("Earlier answer"), self._msg("")], + } + response = json.loads(self._create_response("policy", self._msg(""))) + response["output"] = outputs[case] + request = LLMJudgeVerifyRequest( responses_create_params=NeMoGymResponseCreateParamsNonStreaming(input=[]), - response=NeMoGymResponse( - id="r", - created_at=0.0, - model="m", - object="response", - output=[], - parallel_tool_calls=False, - tool_choice="none", - tools=[], - ), - expected_answer="x", + response=NeMoGymResponse.model_validate(response), + expected_answer="42", + template_metadata={"output_regex": r"ANSWER: (.*)"}, ) + before = request.model_dump() - res = await rs.verify(req) + result = await server.verify(request) - judge_prompt = res.judge_evaluations[0].responses_create_params.input[-1].content - assert cfg.msg_extraction_failure in judge_prompt - assert res.reward == approx(0.0) + client.post.assert_not_awaited() + assert result.reward == 0 + assert result.failure_reason == "empty_final_answer" + assert result.judge_evaluations == [] + assert result.response.model_dump() == before["response"] + assert request.model_dump() == before async def test_swap_fails_uses_configured_reward(self, config: LLMJudgeResourcesServerConfig) -> None: server_mock = MagicMock(spec=ServerClient) @@ -311,7 +423,10 @@ async def test_per_record_regex_extraction(self, config: LLMJudgeResourcesServer # Verify the regex extraction worked by checking judge was called once assert server_mock.post.call_count == 1 - async def test_full_generation_rescue_on_extraction_failure(self, config: LLMJudgeResourcesServerConfig) -> None: + @mark.parametrize("pattern", [r"ANSWER:\s*(.+)", r"(?=The)"]) + async def test_full_generation_rescue_on_extraction_failure( + self, config: LLMJudgeResourcesServerConfig, pattern: str + ) -> None: """When regex-extracted answer fails, retry with full generation for partial credit.""" server_mock = MagicMock(spec=ServerClient) cfg = config.model_copy(deep=True) @@ -332,7 +447,7 @@ async def test_full_generation_rescue_on_extraction_failure(self, config: LLMJud model_create_params = NeMoGymResponseCreateParamsNonStreaming( input=[{"role": "user", "content": "What is 2+2?"}] ) - # Model output: correct answer is in full text but regex won't match properly + # Model output contains an answer even if extraction misses or matches empty text. model_response = NeMoGymResponse( id="resp", created_at=0.0, @@ -344,12 +459,12 @@ async def test_full_generation_rescue_on_extraction_failure(self, config: LLMJud tools=[], ) - # Regex pattern that won't match the model output + # Both a missing match and an empty match must retain full-generation rescue. req = LLMJudgeVerifyRequest( responses_create_params=deepcopy(model_create_params), response=model_response.model_copy(deep=True), expected_answer="4", - template_metadata={"output_regex": r"ANSWER:\s*(.+)"}, # Won't match + template_metadata={"output_regex": pattern}, ) res = await rs.verify(req) diff --git a/responses_api_models/openai_model/README.md b/responses_api_models/openai_model/README.md index 873d4f17537..b577ac69d70 100644 --- a/responses_api_models/openai_model/README.md +++ b/responses_api_models/openai_model/README.md @@ -1,5 +1,28 @@ # Description +OpenAI-compatible model server using Gym's shared HTTP client. + +## HTTP retries + +HTTP 404 and 408 are retryable by default, alongside the existing transient +server and rate-limit errors. This applies to every dataset and to policy and +judge clients. A permanent 404 still fails after the bounded attempt budget. +Retries resend the same request with the existing fixed 0.5-second delay +between attempts and preserve the terminal error body. + +`max_http_attempts` defaults to three total attempts. Set it per model server: + +```yaml +judge_model: + responses_api_models: + openai_model: + max_http_attempts: 5 +``` + +A value of one disables HTTP retries. Connection-error retries are controlled +separately. Internal Gym clients retain their existing unbounded extension for +rate-limit statuses; 404 and 408 never trigger that extension. + # Licensing information Code: Apache 2.0 diff --git a/responses_api_models/openai_model/app.py b/responses_api_models/openai_model/app.py index f72ac9eb549..ba0a1a0419b 100644 --- a/responses_api_models/openai_model/app.py +++ b/responses_api_models/openai_model/app.py @@ -24,6 +24,7 @@ SimpleResponsesAPIModel, ) from nemo_gym.openai_utils import ( + MAX_NUM_TRIES, NeMoGymAsyncOpenAI, NeMoGymChatCompletion, NeMoGymChatCompletionCreateParamsNonStreaming, @@ -39,6 +40,7 @@ class SimpleModelServerConfig(BaseResponsesAPIModelConfig): extra_body: Dict[str, Any] = Field(default_factory=dict) openai_default_headers: Dict[str, str] = Field(default_factory=dict) + max_http_attempts: int = Field(default=MAX_NUM_TRIES, ge=1) max_concurrent_requests: Optional[int] = Field( default=None, @@ -68,6 +70,7 @@ def model_post_init(self, context): base_url=self.config.openai_base_url, api_key=self.config.openai_api_key, default_headers=self.config.openai_default_headers, + max_http_attempts=self.config.max_http_attempts, ) self._semaphore = ( asyncio.Semaphore(self.config.max_concurrent_requests) diff --git a/responses_api_models/openai_model/tests/test_app.py b/responses_api_models/openai_model/tests/test_app.py index 1aaafc5ff93..c22a4a7aafd 100644 --- a/responses_api_models/openai_model/tests/test_app.py +++ b/responses_api_models/openai_model/tests/test_app.py @@ -63,7 +63,7 @@ def _response_data() -> dict: class TestApp: - def _setup_server(self, max_concurrent_requests=None, drop_input_reasoning_items=False): + def _setup_server(self, max_concurrent_requests=None, drop_input_reasoning_items=False, **kwargs): config = SimpleModelServerConfig( host="0.0.0.0", port=8081, @@ -74,12 +74,19 @@ def _setup_server(self, max_concurrent_requests=None, drop_input_reasoning_items name="test_model_server", max_concurrent_requests=max_concurrent_requests, drop_input_reasoning_items=drop_input_reasoning_items, + **kwargs, ) return SimpleModelServer(config=config, server_client=MagicMock(spec=ServerClient, global_config_dict={})) async def test_sanity(self) -> None: self._setup_server() + async def test_retry_configuration_is_scoped_to_model_server(self): + judge = self._setup_server(max_http_attempts=5) + policy = self._setup_server() + assert judge._client.max_http_attempts == 5 + assert policy._client.max_http_attempts == 3 + async def test_chat_completions(self, monkeypatch: MonkeyPatch, tmp_path) -> None: server = self._setup_server() server.server_client.global_config_dict = { diff --git a/tests/unit_tests/test_openai_utils.py b/tests/unit_tests/test_openai_utils.py index decbea8e2c2..c72b5885bc1 100644 --- a/tests/unit_tests/test_openai_utils.py +++ b/tests/unit_tests/test_openai_utils.py @@ -27,7 +27,7 @@ get_origin, get_type_hints, ) -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, call import openai import pytest @@ -126,6 +126,10 @@ def _response_with_output(output: list) -> dict: class TestOpenAIUtils: + def test_invalid_retry_configuration_rejected(self): + with pytest.raises(ValidationError): + NeMoGymAsyncOpenAI(api_key="abc", base_url="https://example.com/v1", max_http_attempts=0) + async def test_NeMoGymAsyncOpenAI(self) -> None: NeMoGymAsyncOpenAI(api_key="abc", base_url="https://api.openai.com/v1") @@ -142,6 +146,69 @@ async def test_external_endpoint_retries_are_bounded(self, monkeypatch: pytest.M assert request.await_count == MAX_NUM_TRIES + @pytest.mark.parametrize("status", [404, 408]) + async def test_retry_reuses_request_with_fixed_delay(self, monkeypatch, status): + failure = SimpleNamespace(status=status, content=SimpleNamespace(read=AsyncMock(return_value=b"temporary"))) + success = SimpleNamespace(status=200) + request = AsyncMock(side_effect=[failure, failure, success]) + sleep = AsyncMock() + monkeypatch.setattr("nemo_gym.openai_utils.request", request) + monkeypatch.setattr("nemo_gym.openai_utils.sleep", sleep) + client = NeMoGymAsyncOpenAI(api_key="abc", base_url="https://example.com/v1") + payload = {"model": "judge", "input": [{"role": "user", "content": "preserved answer"}]} + original = deepcopy(payload) + + result = await client._request_with_retry(method="POST", url="https://example.com/v1/responses", json=payload) + + assert result is success + assert request.await_count == 3 + assert all(c.kwargs["json"] == original for c in request.await_args_list) + assert payload == original + assert sleep.await_args_list == [call(0.5), call(0.5)] + + @pytest.mark.parametrize("status", [400, 401, 403]) + async def test_non_retryable_http_errors_are_returned_once(self, monkeypatch, status): + response = SimpleNamespace(status=status) + request = AsyncMock(return_value=response) + sleep = AsyncMock() + monkeypatch.setattr("nemo_gym.openai_utils.request", request) + monkeypatch.setattr("nemo_gym.openai_utils.sleep", sleep) + client = NeMoGymAsyncOpenAI(api_key="abc", base_url="https://example.com/v1") + + assert await client._request_with_retry() is response + request.assert_awaited_once() + sleep.assert_not_awaited() + + @pytest.mark.parametrize("status", [404, 408]) + @pytest.mark.parametrize("attempts", [1, 3, 5]) + @pytest.mark.parametrize("internal", [False, True]) + async def test_configured_attempt_limit_preserves_terminal_error(self, monkeypatch, status, attempts, internal): + replies = [ + SimpleNamespace(status=status, content=SimpleNamespace(read=AsyncMock(return_value=b"error body"))) + for _ in range(attempts) + ] + request = AsyncMock(side_effect=replies) + sleep = AsyncMock() + monkeypatch.setattr("nemo_gym.openai_utils.request", request) + monkeypatch.setattr("nemo_gym.openai_utils.sleep", sleep) + + async def raise_status(response): + assert response is replies[-1] + response.content.read.assert_not_awaited() + raise RuntimeError("terminal error") + + monkeypatch.setattr("nemo_gym.openai_utils.raise_for_status", raise_status) + client = NeMoGymAsyncOpenAI( + api_key="abc", + base_url="https://example.com/v1", + max_http_attempts=attempts, + internal=internal, + ) + with pytest.raises(RuntimeError, match="terminal error"): + await client._request_with_retry() + assert request.await_count == attempts + assert sleep.await_args_list == [call(0.5)] * (attempts - 1) + class TestNeMoGymResponseCreateParamsNonStreaming: def test_seed_rejected_at_top_level(self) -> None: