fix: avoid false-positive retries on sub-agent and no-tool-call responses

The empty response retry was too aggressive — it triggered on:
1. Sub-agents (AgentTool, ParallelAgent) that legitimately return no content
2. First LLM calls with no prior tool execution

Fixes:
- Add null guard for last_event in is_final_response check (NoneType crash)
- Only retry after at least one tool call in the invocation, since the
  bug only manifests when models return empty after processing tool results
- Remove append_event for resume message (caused session state corruption
  in pause/resume flows and leaked to UI)
- Silent retry instead (proven 100% recovery rate in production tests)
- Update scenario tests to include tool call before empty response

Author: gurjot-05

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

@@ -69,13 +69,6 @@
 # tool execution. Prevents infinite loops if the model keeps returning empty.
 _MAX_EMPTY_RESPONSE_RETRIES = 2
 
-# Message injected into the conversation when the model returns an empty
-# response, nudging it to resume execution on the next attempt.
-_EMPTY_RESPONSE_RESUME_MESSAGE = (
-    'Your previous response was empty. '
-    'Please resume execution from where you left off.'
-)
-
 
 def _has_meaningful_content(event: Event) -> bool:
   """Returns whether the event has content worth showing to the user.
@@ -782,27 +775,35 @@ async def run_async(
   ) -> AsyncGenerator[Event, None]:
     """Runs the flow."""
     empty_response_count = 0
+    has_prior_tool_call = False
     while True:
       last_event = None
       async with Aclosing(self._run_one_step_async(invocation_context)) as agen:
         async for event in agen:
           last_event = event
           yield event
+          # Track if any tool calls have been executed in this invocation.
+          # Empty responses are only retried after tool execution because
+          # that is where models intermittently return 0 output tokens.
+          if event.get_function_calls():
+            has_prior_tool_call = True
 
       # Determine if the model returned an empty / useless response that
       # should be retried.  Three cases:
-      #   1. No event at all (model/adapter yielded nothing)
-      #   2. Last event is partial with no meaningful content (streaming +
+      #   1. Last event is partial with no meaningful content (streaming +
       #      thinking: only thought chunks arrived, no final response)
-      #   3. Last event is a final response with no meaningful content
+      #   2. Last event is a final response with no meaningful content
       #      (non-streaming empty response, or streaming empty aggregated)
       is_empty_response = False
-      if not last_event:
-        is_empty_response = True
-      elif last_event.partial and not _has_meaningful_content(last_event):
+      if (
+          last_event
+          and last_event.partial
+          and not _has_meaningful_content(last_event)
+      ):
         is_empty_response = True
       elif (
-          last_event.is_final_response()
+          last_event
+          and last_event.is_final_response()
           and not _has_meaningful_content(last_event)
           and last_event.author == invocation_context.agent.name
       ):
@@ -815,35 +816,15 @@ async def run_async(
 
       if (
           is_empty_response
+          and has_prior_tool_call
           and empty_response_count < _MAX_EMPTY_RESPONSE_RETRIES
       ):
         empty_response_count += 1
         logger.warning(
-            'Model returned an empty response (attempt %d/%d),'
-            ' injecting resume message and re-prompting.',
+            'Model returned an empty response (attempt %d/%d), re-prompting.',
             empty_response_count,
             _MAX_EMPTY_RESPONSE_RETRIES,
         )
-        # Inject a resume nudge into the session so the next LLM call
-        # sees it in its context and is more likely to continue.
-        # We append directly to the session (not yield) so that the
-        # message reaches the model but is NOT sent to the UI/SSE stream.
-        resume_event = Event(
-            id=Event.new_id(),
-            invocation_id=invocation_context.invocation_id,
-            author='user',
-            branch=invocation_context.branch,
-            content=types.Content(
-                role='user',
-                parts=[
-                    types.Part.from_text(text=_EMPTY_RESPONSE_RESUME_MESSAGE)
-                ],
-            ),
-        )
-        await invocation_context.session_service.append_event(
-            session=invocation_context.session,
-            event=resume_event,
-        )
         continue
 
       if (

tests/unittests/flows/llm_flows/test_base_llm_flow_partial_handling.py

@@ -91,18 +91,14 @@ async def test_run_async_breaks_on_final_response():
 
 
 @pytest.mark.asyncio
-async def test_run_async_retries_then_breaks_on_no_last_event():
-  """Test that run_async retries when there is no last event, then breaks."""
-  # Create a mock model that returns empty responses (no content).
-  # Need enough responses to cover initial call + max retries.
-  from google.adk.flows.llm_flows.base_llm_flow import _MAX_EMPTY_RESPONSE_RETRIES
+async def test_run_async_breaks_on_no_last_event():
+  """Test that run_async breaks when there is no last event."""
+  # content=None is filtered by _postprocess_async, so no events are
+  # yielded and last_event stays None.  The loop should break immediately
+  # (no retry) because this is legitimate for sub-agents.
+  empty_response = LlmResponse(content=None, partial=False)
 
-  empty_responses = [
-      LlmResponse(content=None, partial=False)
-      for _ in range(_MAX_EMPTY_RESPONSE_RETRIES + 1)
-  ]
-
-  mock_model = testing_utils.MockModel.create(responses=empty_responses)
+  mock_model = testing_utils.MockModel.create(responses=[empty_response])
 
   agent = Agent(name='test_agent', model=mock_model)
   invocation_context = await testing_utils.create_invocation_context(
@@ -116,12 +112,8 @@ async def test_run_async_retries_then_breaks_on_no_last_event():
   async for event in flow.run_async(invocation_context):
     events.append(event)
 
-  # Resume events are appended to session (not yielded), so no user
-  # events should appear in the output stream.  Verify retries happened
-  # by checking how many responses were consumed.
-  assert mock_model.response_index == _MAX_EMPTY_RESPONSE_RETRIES
-  leaked = [e for e in events if e.author == 'user']
-  assert len(leaked) == 0, 'Resume messages must not leak to output'
+  # Should have no events because empty responses are filtered out
+  assert len(events) == 0
 
 
 @pytest.mark.asyncio