fix: Fix transcript finish

Merge #3324

**Problem:**
ADK seems to not pass output/input transcription `finished` flag from the Gemini.

**Solution:**
Relaxation of checking conditions of valid llm_response for input/output transcription.

### Testing Plan

Unit test `test_receive_transcript_finished` checks if input/output transcription message with no text but `finished` flag is received.

**Unit Tests:**

- [x] I have added or updated unit tests for my change.
- [x] All unit tests pass locally.

<img width="785" height="373" alt="image" src="https://github.com/user-attachments/assets/6d870e9f-1372-4808-91a9-38578c1b3729" />

**Manual End-to-End (E2E) Tests:**

Configure Gemini Agent to produce input & output transcriptions. Observe incoming transcript messages - to see if the finished flag appears at the end agents & users statement.

### Checklist

- [x] I have read the [CONTRIBUTING.md](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) document.
- [x] I have performed a self-review of my own code.
- [x] I have commented my code, particularly in hard-to-understand areas.
- [x] I have added tests that prove my fix is effective or that my feature works.
- [x] New and existing unit tests pass locally with my changes.
- [x] I have manually tested my changes end-to-end.
- [x] Any dependent changes have been merged and published in downstream modules.

### Additional context

The mentioned `finished` flag can be obtained in native Gemini APIs like via Websocket but not via ADK.

Co-authored-by: Hangfei Lin <[email protected]>
COPYBARA_INTEGRATE_REVIEW=#3324 from ChrisQlasty:fix/transcript_finish e7b8e5e
PiperOrigin-RevId: 828987996

Author: 2 people

src/google/adk/models/gemini_llm_connection.py

@@ -165,18 +165,12 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
               yield self.__build_full_text_response(text)
               text = ''
             yield llm_response
-          if (
-              message.server_content.input_transcription
-              and message.server_content.input_transcription.text
-          ):
+          if message.server_content.input_transcription:
             llm_response = LlmResponse(
                 input_transcription=message.server_content.input_transcription,
             )
             yield llm_response
-          if (
-              message.server_content.output_transcription
-              and message.server_content.output_transcription.text
-          ):
+          if message.server_content.output_transcription:
             llm_response = LlmResponse(
                 output_transcription=message.server_content.output_transcription
             )

tests/unittests/models/test_gemini_llm_connection.py

@@ -112,6 +112,46 @@ async def test_close(gemini_connection, mock_gemini_session):
 
 
 @pytest.mark.asyncio
+@pytest.mark.parametrize('tx_direction', ['input', 'output'])
+async def test_receive_transcript_finished(
+    gemini_connection, mock_gemini_session, tx_direction
+):
+  """Test receive_transcript_finished for input and output transcription."""
+
+  finished_tx = types.Transcription(finished=True)
+
+  msg = mock.Mock()
+  msg.tool_call = None
+  msg.usage_metadata = None
+  msg.session_resumption_update = None
+  msg.server_content.model_turn = None
+  msg.server_content.interrupted = False
+  msg.server_content.turn_complete = False
+  msg.server_content.input_transcription = (
+      finished_tx if tx_direction == 'input' else None
+  )
+  msg.server_content.output_transcription = (
+      finished_tx if tx_direction == 'output' else None
+  )
+
+  async def gen():
+    yield msg
+
+  mock_gemini_session.receive = mock.Mock(return_value=gen())
+
+  responses = []
+  async for r in gemini_connection.receive():
+    responses.append(r)
+
+  attr_name = f'{tx_direction}_transcription'
+  tx_resps = [r for r in responses if getattr(r, attr_name)]
+  assert tx_resps, f'Expected {tx_direction} transcription response'
+
+  transcription = getattr(tx_resps[0], attr_name)
+  assert transcription.finished is True
+  assert not transcription.text
+
+
 async def test_receive_usage_metadata_and_server_content(
     gemini_connection, mock_gemini_session
 ):