test: Add test for interrupted signal preservation with pending text

Add test_receive_interrupted_with_pending_text_preserves_flag to verify:
- interrupted flag is preserved when flushing pending text
- grounding_metadata is carried through to the flushed response
- accumulated text is properly merged before interruption

Addresses MEDIUM priority review comment about missing test coverage for edge cases.

Author: VedantMadane

tests/unittests/models/test_gemini_llm_connection.py

@@ -991,3 +991,92 @@ async def mock_receive_generator():
   assert tool_call_response is not None
   # The grounding_metadata should be carried over to tool_call
   assert tool_call_response.grounding_metadata == mock_grounding_metadata
+
+
+@pytest.mark.asyncio
+async def test_receive_interrupted_with_pending_text_preserves_flag(
+    gemini_connection, mock_gemini_session
+):
+  """Test that interrupted flag is preserved when flushing pending text."""
+  mock_grounding_metadata = types.GroundingMetadata(
+      retrieval_queries=['test query'],
+  )
+
+  # First message with text content and grounding
+  mock_content1 = types.Content(
+      role='model', parts=[types.Part.from_text(text='partial')]
+  )
+  mock_server_content1 = mock.Mock()
+  mock_server_content1.model_turn = mock_content1
+  mock_server_content1.interrupted = False
+  mock_server_content1.input_transcription = None
+  mock_server_content1.output_transcription = None
+  mock_server_content1.turn_complete = False
+  mock_server_content1.generation_complete = False
+  mock_server_content1.grounding_metadata = mock_grounding_metadata
+
+  message1 = mock.Mock()
+  message1.usage_metadata = None
+  message1.server_content = mock_server_content1
+  message1.tool_call = None
+  message1.session_resumption_update = None
+
+  # Second message with more text
+  mock_content2 = types.Content(
+      role='model', parts=[types.Part.from_text(text=' text')]
+  )
+  mock_server_content2 = mock.Mock()
+  mock_server_content2.model_turn = mock_content2
+  mock_server_content2.interrupted = False
+  mock_server_content2.input_transcription = None
+  mock_server_content2.output_transcription = None
+  mock_server_content2.turn_complete = False
+  mock_server_content2.generation_complete = False
+  mock_server_content2.grounding_metadata = None
+
+  message2 = mock.Mock()
+  message2.usage_metadata = None
+  message2.server_content = mock_server_content2
+  message2.tool_call = None
+  message2.session_resumption_update = None
+
+  # Third message with interrupted signal
+  mock_server_content3 = mock.Mock()
+  mock_server_content3.model_turn = None
+  mock_server_content3.interrupted = True
+  mock_server_content3.input_transcription = None
+  mock_server_content3.output_transcription = None
+  mock_server_content3.turn_complete = False
+  mock_server_content3.generation_complete = False
+  mock_server_content3.grounding_metadata = None
+
+  message3 = mock.Mock()
+  message3.usage_metadata = None
+  message3.server_content = mock_server_content3
+  message3.tool_call = None
+  message3.session_resumption_update = None
+
+  async def mock_receive_generator():
+    yield message1
+    yield message2
+    yield message3
+
+  receive_mock = mock.Mock(return_value=mock_receive_generator())
+  mock_gemini_session.receive = receive_mock
+
+  responses = [resp async for resp in gemini_connection.receive()]
+
+  # Find the full text response that should have been flushed with interrupted=True
+  full_text_responses = [
+      r for r in responses if r.content and not r.partial and r.interrupted
+  ]
+  assert (
+      len(full_text_responses) > 0
+  ), 'Should have interrupted full text response'
+
+  # The full text response should have the accumulated text
+  assert full_text_responses[0].content.parts[0].text == 'partial text'
+  # And should carry the grounding_metadata
+  assert full_text_responses[0].grounding_metadata == mock_grounding_metadata
+  # And should have interrupted=True
+  assert full_text_responses[0].interrupted is True