fix: surface error for empty STOP model turn in non-streaming mode

An empty model turn (finish_reason=STOP with no content parts) was silently
becoming an empty final response. The non-streaming flow now surfaces it as a
MODEL_RETURNED_NO_CONTENT error event. Detection lives in the flow and only in
non-streaming mode, not in LlmResponse.create, so streaming consumers that
batch parts across chunks (where a terminal finish-only chunk legitimately
follows earlier content) are unaffected.

Close #5631

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 939915020

Author: GWeale

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -71,6 +71,11 @@ class _ReconnectSentinel(Event):
 
 _ADK_AGENT_NAME_LABEL_KEY = 'adk_agent_name'
 
+_NO_CONTENT_ERROR_CODE = 'MODEL_RETURNED_NO_CONTENT'
+_NO_CONTENT_ERROR_MESSAGE = (
+    'The model returned no content (finish_reason=STOP with empty parts).'
+)
+
 # Timing configuration
 DEFAULT_TRANSFER_AGENT_DELAY = 1.0
 DEFAULT_TASK_COMPLETION_DELAY = 1.0
@@ -1066,6 +1071,23 @@ async def _postprocess_async(
       async for event in agen:
         yield event
 
+    # A non-streaming turn that finishes with STOP but has no content parts would
+    # otherwise be skipped below and become a silent empty final response;
+    # surface it as an actionable error instead. Streaming is excluded
+    # because a terminal finish-only chunk legitimately follows content already
+    # streamed in earlier chunks.
+    if (
+        not llm_response.partial
+        and llm_response.error_code is None
+        and llm_response.finish_reason == types.FinishReason.STOP
+        and (not llm_response.content or not llm_response.content.parts)
+        and invocation_context.run_config.streaming_mode != StreamingMode.SSE
+    ):
+      llm_response.error_code = _NO_CONTENT_ERROR_CODE
+      llm_response.error_message = (
+          llm_response.error_message or _NO_CONTENT_ERROR_MESSAGE
+      )
+
     # Skip the model response event if there is no content and no error code.
     # This is needed for the code executor to trigger another loop.
     if (

tests/unittests/flows/llm_flows/test_base_llm_flow.py

@@ -1644,6 +1644,58 @@ def _make_agent_tree():
   return root, child1, child2
 
 
+@pytest.mark.asyncio
+async def test_empty_stop_after_tool_call_surfaces_error_event():
+  """Regression test for empty Gemini turn after a successful tool call (#5631).
+
+  Turn 1 returns a function_call which executes successfully, then turn 2
+  returns Content(role='model', parts=[]) with finish_reason=STOP and no error.
+  In non-streaming mode the flow must surface that empty turn as an error event
+  instead of a silent empty final response.
+  """
+  function_call_part = types.Part.from_function_call(
+      name='increase_by_one', args={'x': 1}
+  )
+
+  turn_1 = LlmResponse(
+      content=types.Content(role='model', parts=[function_call_part]),
+      finish_reason=types.FinishReason.STOP,
+  )
+  # An empty Gemini turn: STOP with no content parts and no error from the model.
+  turn_2 = LlmResponse(
+      content=types.Content(role='model', parts=[]),
+      finish_reason=types.FinishReason.STOP,
+  )
+
+  function_called = 0
+
+  def increase_by_one(x: int) -> int:
+    nonlocal function_called
+    function_called += 1
+    return x + 1
+
+  mock_model = testing_utils.MockModel.create(responses=[turn_1, turn_2])
+  agent = Agent(name='root_agent', model=mock_model, tools=[increase_by_one])
+  runner = testing_utils.InMemoryRunner(agent)
+  events = runner.run('test')
+
+  assert function_called == 1, 'Tool should still execute on turn 1'
+
+  function_call_events = [e for e in events if e.get_function_calls()]
+  function_response_events = [e for e in events if e.get_function_responses()]
+  assert len(function_call_events) == 1
+  assert len(function_response_events) == 1
+
+  # The empty turn 2 must surface as an error event, not an empty final.
+  error_events = [e for e in events if e.error_code]
+  assert len(error_events) == 1
+  err = error_events[0]
+  assert err.error_code == 'MODEL_RETURNED_NO_CONTENT'
+  assert err.error_message
+  # And it must be the run's final event (no silent empty event after it).
+  assert events[-1] is err
+
+
 @pytest.mark.asyncio
 async def test_transfer_to_sibling_disallowed_raises_value_error():
   """Transfer to sibling raises ValueError when disallow_transfer_to_peers is True."""

tests/unittests/models/test_llm_response.py

@@ -345,7 +345,12 @@ def test_llm_response_create_error_case_with_citation_metadata():
 
 
 def test_llm_response_create_empty_content_with_stop_reason():
-  """Test LlmResponse.create() with empty content and stop finish reason."""
+  """Empty content + STOP stays a successful response at the model layer.
+
+  Surfacing the empty turn as an error is the flow's job (non-streaming only);
+  the model/streaming layer must not classify a terminal finish-only chunk as
+  an error or it breaks streaming consumers that batch parts across chunks.
+  """
   generate_content_response = types.GenerateContentResponse(
       candidates=[
           types.Candidate(
@@ -359,6 +364,26 @@ def test_llm_response_create_empty_content_with_stop_reason():
 
   assert response.error_code is None
   assert response.content is not None
+  assert response.finish_reason == types.FinishReason.STOP
+
+
+def test_llm_response_create_non_empty_parts_with_stop_is_success():
+  """Regression guard: real text + STOP must remain a successful response."""
+  generate_content_response = types.GenerateContentResponse(
+      candidates=[
+          types.Candidate(
+              content=types.Content(
+                  role='model', parts=[types.Part(text='ok')]
+              ),
+              finish_reason=types.FinishReason.STOP,
+          )
+      ]
+  )
+
+  response = LlmResponse.create(generate_content_response)
+
+  assert response.error_code is None
+  assert response.content is not None
 
 
 def test_llm_response_create_includes_model_version():

tests/unittests/utils/test_streaming_utils.py

@@ -188,7 +188,12 @@ async def test_close_with_error(self):
   async def test_empty_content_produces_empty_final_frame(
       self, use_progressive_sse
   ):
-    """A candidate with an empty parts list produces an empty final frame."""
+    """A candidate with empty parts + STOP passes through without an error.
+
+    A terminal empty STOP chunk must not be classified as an error at the
+    streaming layer; consumers that batch parts across chunks rely on it
+    passing through cleanly.
+    """
     with temporary_feature_override(
         FeatureName.PROGRESSIVE_SSE_STREAMING, use_progressive_sse
     ):
@@ -208,6 +213,7 @@ async def test_empty_content_produces_empty_final_frame(
 
       assert len(results) == 1
       assert results[0].content is not None
+      assert results[0].error_code is None
       assert closed_response is not None
       assert closed_response.partial is False
       assert closed_response.content is None