fix(cli): stream_reasoning_engine raises StopIteration RuntimeError on sync generators

[surajit-1306]

Link to Issue or Description of Change

Closes : #6093
Problem:
On Agent Engine deployments served by the ADK API server, every call to the
/api/stream_reasoning_engine route with a synchronous streaming class_method
(e.g. stream_query) ends with RuntimeError: coroutine raised StopIteration
after the last chunk is streamed.

The cause is the sync-to-async adapter _aiter_from_iter in
src/google/adk/cli/fast_api.py (lines 916–922 in v2.2.0):

  async def _aiter_from_iter(iterator):
    while True:   
      try:
        chunk = await run_in_threadpool(next, iterator)
        yield chunk
      except StopIteration:
        break

The except StopIteration is unreachable. When the iterator is exhausted,
next() raises StopIteration inside the worker thread, anyio sets it on a
future, and it propagates out of the run_in_threadpool coroutine frame.
Python (PEP 479) forbids StopIteration escaping a coroutine and converts it
to RuntimeError("coroutine raised StopIteration") before the except clause
ever sees it.

Affected versions: Regression introduced in v2.2.0 — the route and the
buggy adapter were added in the same commit. Not present in the v1.x line
(verified absent at v1.35.0).

Solution:
Stop relying on StopIteration crossing the await boundary; use a sentinel
default so iterator exhaustion never raises across it:

  _SENTINEL = object()

  async def _aiter_from_iter(iterator):
    while True:
      chunk = await run_in_threadpool(next, iterator, _SENTINEL)
      if chunk is _SENTINEL:
        break
      yield chunk

This is the minimal, idiomatic fix; the stream now terminates cleanly when the
sync generator is exhausted.

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Added test_gemini_stream_reasoning_engine_sync_generator plus a
test_app_with_gemini_enterprise_sync_stream fixture in
tests/unittests/cli/test_fast_api.py. The pre-existing stream test used an
async generator (the isasyncgenfunction branch) and never exercised the
buggy sync-generator path. The new test fails on the unpatched code with
RuntimeError and passes with the fix.

pytest summary:

  $ pytest tests/unittests/cli/test_fast_api.py -k stream_reasoning_engine -q
  3 passed, 79 deselected   

  $ pytest tests/unittests/cli/test_fast_api.py -q
  82 passed

Manual End-to-End (E2E) Tests:

The failure and the fix reproduce standalone in ~15 lines, independent of any
model or deployment:

  import asyncio
  from starlette.concurrency import run_in_threadpool

  async def _aiter_from_iter(iterator):  # old, buggy version
      while True:
          try:
              chunk = await run_in_threadpool(next, iterator)
              yield chunk
          except StopIteration:
              break

  async def main():
      def gen():
          yield 1
          yield 2
      async for c in _aiter_from_iter(gen()):
          print("chunk:", c)

  asyncio.run(main())
  # chunk: 1
  # chunk: 2
  # RuntimeError: coroutine raised StopIteration   <-- before the fix

With the sentinel version above, the same script prints the two chunks and
exits cleanly with no exception. Originally observed on a live Vertex AI Agent
Engine deployment (google-adk==2.2.0, Python 3.11) where every stream_query
call logged the RuntimeError after the final chunk.

Checklist

• I have read the CONTRIBUTING.md document.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end.
• Any dependent changes have been merged and published in downstream modules.

Additional context

Original server traceback:

  ERROR:    Exception in ASGI application
  Traceback (most recent call last):
    File ".../starlette/responses.py", line 250, in stream_response
      async for chunk in self.body_iterator:
    File ".../google/adk/cli/fast_api.py", line 797, in json_generator
      async for chunk in output:
    File ".../google/adk/cli/fast_api.py", line 919, in _aiter_from_iter
      chunk = await run_in_threadpool(next, iterator)
    File ".../starlette/concurrency.py", line 32, in run_in_threadpool
      return await anyio.to_thread.run_sync(func)
    File ".../anyio/to_thread.py", line 63, in run_sync
      return await get_async_backend().run_sync_in_worker_thread(
    File ".../anyio/_backends/_asyncio.py", line 2518, in run_sync_in_worker_thread
      return await future
  RuntimeError: coroutine raised StopIteration

Occurs 100% of the time on every sync streaming request once the generator is
exhausted. The bug is model-agnostic (purely in the FastAPI streaming adapter).

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.