fix(eval): close judge lifecycle warnings

Author: neutronstar238

examples/optimization/eval_optimize_loop/README.md

@@ -111,13 +111,15 @@ a harmless explanation rewrite does not zero out an otherwise correct route.
 
 `environment_snapshot` records the git commit, dirty flag, Python version, SDK
 version when installed, model name, redacted base URL host, seed, command, and
-optimizer config path. It never records API keys. Provider/runtime warnings are
-not globally suppressed: an expected DeepSeek response-schema downgrade remains
-observable as a provider compatibility warning.
-The pipeline awaits `Runner.close()` on its own execution paths. Even so,
-upstream OpenAI/httpx may still emit Python 3.13 stream-shutdown diagnostics.
-Those warnings remain observable rather than being suppressed or automatically
-attributed to this example.
+optimizer config path. It never records API keys. When a judge provider does
+not support native JSON schemas, the judge uses JSON-object mode and validates
+the response locally instead of sending an ignored schema.
+Provider and runtime warnings are not globally suppressed. Judge calls run
+non-streaming and explicitly close their agent run so normal online evaluation
+does not leave OpenAI/httpx stream-shutdown diagnostics behind.
+The pipeline awaits `Runner.close()` for its agent execution paths.
+If an upstream OpenAI/httpx diagnostic still occurs outside this lifecycle,
+warnings remain observable rather than being filtered.
 
 Report error text redacts provider URLs, configured provider credentials,
 standalone provider-key shapes, and semantic credential markers while retaining

tests/agents/core/test_llm_processor.py

@@ -12,6 +12,7 @@
 from unittest.mock import Mock
 
 import pytest
+from unittest.mock import patch
 
 from trpc_agent_sdk.agents._base_agent import BaseAgent
 from trpc_agent_sdk.agents.core._llm_processor import LlmProcessor
@@ -23,6 +24,8 @@
 
 
 class _StubAgent(BaseAgent):
+    instruction: str = ""
+
     async def _run_async_impl(self, ctx):
         yield
 
@@ -159,6 +162,34 @@ def test_event_without_content_skips_planning(self, model, invocation_context):
 
 
 class TestCallLlmAsync:
+
+    @pytest.mark.asyncio
+    async def test_non_streaming_drains_model_before_yielding_event(self, invocation_context):
+        """A non-streaming event is exposed only after the model generator is closed."""
+
+        class ClosingModel(MockLLMModel):
+
+            def __init__(self):
+                super().__init__(model_name="test-llmproc-model")
+                self.finished = False
+
+            async def _generate_async_impl(self, request, stream=False, ctx=None):
+                try:
+                    yield LlmResponse(content=Content(parts=[Part(text="complete")]))
+                finally:
+                    self.finished = True
+
+        model = ClosingModel()
+        processor = LlmProcessor(model)
+        event_stream = processor.call_llm_async(LlmRequest(), invocation_context, stream=False)
+
+        with patch("trpc_agent_sdk.agents.core._llm_processor.trace_call_llm"):
+            event = await anext(event_stream)
+
+        assert event.content.parts[0].text == "complete"
+        assert model.finished is True
+        await event_stream.aclose()
+
     def test_yields_events_for_responses(self, model, invocation_context):
         proc = LlmProcessor(model)
         request = LlmRequest()

tests/agents/test_llm_agent_ext.py

@@ -409,5 +409,37 @@ async def run():
         assert events[0].error_code == "build_error"
 
 
+class TestRunAsyncImplStreaming:
+
+    def test_disables_model_stream_when_run_config_disables_streaming(self, invocation_context):
+        """A non-streaming run must not create a provider SSE response."""
+
+        class RecordingModel(MockLLMModel):
+
+            def __init__(self):
+                super().__init__(model_name="test-llm-ext-model")
+                self.stream_args: list[bool] = []
+
+            async def _generate_async_impl(self, request, stream=False, ctx=None):
+                self.stream_args.append(stream)
+                yield LlmResponse(content=Content(parts=[Part(text="recorded")]))
+
+        model = RecordingModel()
+        agent = _agent(model=model)
+        invocation_context.agent = agent
+        invocation_context.run_config = RunConfig(streaming=False)
+        invocation_context.override_messages = [
+            Content(role="user", parts=[Part(text="evaluate this response")]),
+        ]
+
+        async def run():
+            async for _ in agent._run_async_impl(invocation_context):
+                pass
+
+        asyncio.run(run())
+
+        assert model.stream_args == [False]
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])

tests/evaluation/test_llm_judge_multi_model.py

@@ -290,6 +290,62 @@ async def test_legacy_single_judge_model(self):
         assert r.overall_eval_status == EvalStatus.PASSED
 
 
+class TestJudgeModelResponseFormatCapabilities:
+
+    @staticmethod
+    def _capture_judge_agent_construction(metric):
+        captured = []
+
+        class CapturingJudgeAgent:
+
+            def __init__(self, model, config, system_prompt, output_schema=None, tools=None, planner=None):
+                captured.append({
+                    "model": model,
+                    "config": config,
+                    "output_schema": output_schema,
+                })
+
+        with patch("trpc_agent_sdk.evaluation._llm_judge._JudgeAgent", CapturingJudgeAgent):
+            LLMJudge(metric)
+        return captured
+
+    @staticmethod
+    def _rubric_metric(model_name):
+        return EvalMetric(
+            metric_name="llm_rubric_response",
+            threshold=0.5,
+            criterion={
+                "llm_judge": {
+                    "judge_model": {
+                        "model_name": model_name,
+                    },
+                },
+            },
+        )
+
+    def test_deepseek_judge_uses_json_mode_without_native_schema(self):
+        from trpc_agent_sdk.models.openai_adapter import _deepseek
+
+        captured = self._capture_judge_agent_construction(self._rubric_metric("deepseek-v4-flash"))
+
+        assert len(captured) == 1
+        judge = captured[0]
+        assert judge["output_schema"] is None
+        assert judge["config"].response_mime_type == "application/json"
+        with patch.object(_deepseek.logger, "warning") as warning:
+            response_format = judge["model"]._build_response_format(judge["config"])
+        assert response_format == {"type": "json_object"}
+        warning.assert_not_called()
+
+    def test_schema_capable_judge_keeps_native_schema(self):
+        captured = self._capture_judge_agent_construction(self._rubric_metric("gpt-4o"))
+
+        assert len(captured) == 1
+        judge = captured[0]
+        assert judge["output_schema"] is not None
+        assert judge["config"].response_mime_type is None
+
+
 class TestUnknownAggregatorRaisesAtConstruction:
 
     def test_unknown_aggregator_raises(self):

tests/evaluation/test_llm_judge_think.py

@@ -143,6 +143,77 @@ def test_generation_config_thinking_config_used_when_think_is_none(self):
 
 class TestJudgeAgentPlanner:
 
+    @pytest.mark.asyncio
+    async def test_judge_agent_closes_its_agent_run(self):
+        captured = {"closed": False}
+
+        class _FakeAgentRun:
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                raise StopAsyncIteration
+
+            async def aclose(self):
+                captured["closed"] = True
+
+        agent_run = _FakeAgentRun()
+
+        class _FakeLlmAgent:
+
+            def __init__(self, **kwargs):
+                pass
+
+            def run_async(self, ctx):
+                return agent_run
+
+        class _FakeInvocationContext:
+
+            def __init__(self, **kwargs):
+                pass
+
+        with patch("trpc_agent_sdk.evaluation._llm_judge.LlmAgent", _FakeLlmAgent), patch(
+                "trpc_agent_sdk.evaluation._llm_judge.InvocationContext", _FakeInvocationContext):
+            judge = _JudgeAgent(
+                model=object(),
+                config=None,
+                system_prompt="sp",
+            )
+            await judge.get_response("evaluate this response")
+
+        assert captured["closed"] is True
+
+    @pytest.mark.asyncio
+    async def test_judge_agent_disables_model_streaming(self):
+        captured: dict[str, Any] = {}
+
+        class _FakeLlmAgent:
+
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+
+            async def run_async(self, ctx):
+                captured["run_config"] = ctx.run_config
+                if False:  # pragma: no cover - makes this an async generator
+                    yield None
+
+        class _FakeInvocationContext:
+
+            def __init__(self, **kwargs):
+                self.run_config = kwargs["run_config"]
+
+        with patch("trpc_agent_sdk.evaluation._llm_judge.LlmAgent", _FakeLlmAgent), patch(
+                "trpc_agent_sdk.evaluation._llm_judge.InvocationContext", _FakeInvocationContext):
+            judge = _JudgeAgent(
+                model=object(),
+                config=None,
+                system_prompt="sp",
+            )
+            await judge.get_response("evaluate this response")
+
+        assert captured["run_config"].streaming is False
+
     def test_judge_agent_accepts_planner_and_forwards_to_llm_agent(self):
         captured: dict[str, Any] = {}
 

tests/models/test_openai_model_ext.py

@@ -1154,6 +1154,49 @@ def test_no_warning_for_supported(self):
 class TestGenerateAsyncEdgeCases:
     """Edge case tests for _generate_async_impl via generate_async."""
 
+    @pytest.mark.asyncio
+    async def test_streaming_response_is_closed_before_client(self):
+        """The OpenAI stream itself is closed before its owning client."""
+        model = _model()
+        content = Content(parts=[Part.from_text(text="hi")], role="user")
+        request = _request([content])
+        cleanup_order: list[str] = []
+
+        class ClosableStream:
+
+            def __init__(self):
+                self._yielded = False
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                if self._yielded:
+                    raise StopAsyncIteration
+                self._yielded = True
+                chunk = Mock()
+                chunk.model_dump.return_value = {"choices": [], "usage": None}
+                return chunk
+
+            async def close(self):
+                cleanup_order.append("stream")
+
+        stream = ClosableStream()
+
+        async def close_client():
+            cleanup_order.append("client")
+
+        with patch.object(model, "_create_async_client") as mock_factory:
+            mock_client = AsyncMock()
+            mock_client.chat.completions.create = AsyncMock(return_value=stream)
+            mock_client.close = AsyncMock(side_effect=close_client)
+            mock_factory.return_value = mock_client
+
+            async for _ in model.generate_async(request, stream=True):
+                pass
+
+        assert cleanup_order == ["stream", "client"]
+
     @pytest.mark.asyncio
     async def test_streaming_error_yields_error_response(self):
         """Streaming errors yield an error LlmResponse."""

trpc_agent_sdk/agents/_llm_agent.py

@@ -499,7 +499,8 @@ def accumulate_content(event: Event) -> None:
                 logger.debug("Starting LLM call for agent: %s", self.name)
 
                 # Use LlmProcessor to get unified events
-                async for event in llm_processor.call_llm_async(request, ctx, stream=True):
+                model_streaming = ctx.run_config.streaming if ctx.run_config is not None else True
+                async for event in llm_processor.call_llm_async(request, ctx, stream=model_streaming):
                     # Handle different event types by checking content and error status
                     if event.is_error():
                         # Error event - yield and stop

trpc_agent_sdk/agents/core/_llm_processor.py

@@ -67,6 +67,7 @@ async def call_llm_async(self,
         """
         author = context.agent.name
         logger.debug("Starting LLM call for agent: %s", author)
+        buffered_events: list[Event] = []
 
         try:
             # Step 1: Validate the request
@@ -131,7 +132,10 @@ def _append_function_calls(target: list[dict], calls: list) -> None:
                         if not llm_response.partial:
                             final_llm_response = llm_response
 
-                        yield event
+                        if stream:
+                            yield event
+                        else:
+                            buffered_events.append(event)
                 except Exception as ex:
                     metrics_error_type = type(ex).__name__
                     raise
@@ -164,6 +168,10 @@ def _append_function_calls(target: list[dict], calls: list) -> None:
         except Exception as ex:  # pylint: disable=broad-except
             logger.error("LLM call failed for agent %s: %s", author, ex)
             yield self._create_error_event(context, "llm_call_error", f"LLM call failed: {str(ex)}")
+        else:
+            if not stream:
+                for event in buffered_events:
+                    yield event
 
         logger.debug("LLM call completed for agent: %s", author)