reasoning_engine_adapter template: stream_query always fails with TypeError, surfacing as HTTP 200 with an empty body

[RomanZhulpoPaysera]

Summary

The agent_runtime scaffold's reasoning_engine_adapter.py iterates every
streaming class-method with async for. AdkApp.register_operations() reports
two distinct streaming modes, and only one of them yields an async iterator:

• "stream" → stream_query, a plain generator
• "async_stream" → async_stream_query, an async generator

The template merges both into one set and then assumes the async form, so every
stream_query call raises. Because the StreamingResponse is already committed
by the time the body iterator runs, the exception never reaches the caller: the
client receives HTTP 200 with a zero-byte body, and the deploy, the engine
resource and the CLI all report success.

stream_query is the contract the Vertex AI Console Playground and Gemini
Enterprise (ADK registration) both use, so the practical effect is an agent that
is mute on its two primary human-facing surfaces while every automated signal
says healthy.

Version

• google-agents-cli 1.3.1
• google-adk[gcp,otel-gcp,bigquery-analytics] 2.7.0
• google-cloud-aiplatform 1.164.0
• deployment target agent_runtime, region europe-west1

Present identically in the released package and on main today, at
src/google/agents/cli/scaffold/deployment_targets/agent_runtime/python/{{cookiecutter.agent_directory}}/app_utils/reasoning_engine_adapter.py.

Reproduce

1. agents-cli scaffold create <name> --deployment-target agent_runtime
2. agents-cli deploy
3. POST {engine}:streamQuery with
   {"class_method": "stream_query", "input": {"user_id": "u", "message": "hi"}}

Observed: HTTP 200, Content-Type: application/json, 0 bytes, ~0.3 s.
Container logs:

File "/code/app/app_utils/reasoning_engine_adapter.py", line 84, in generator
    async for event in method(**(body.get("input") or {})):
TypeError: 'async for' requires an object with __aiter__ method, got generator

Expected: the events yielded by stream_query.

Root cause

# line 60 — the two modes are merged, losing the distinction
streaming_methods = set(operations.get("stream", [])) | set(
    operations.get("async_stream", [])
)

# line 84 — the async form is then assumed for both
async def generator():
    async for event in method(**(body.get("input") or {})):
        yield json.dumps(event) + "\n"

The sibling non-streaming handler in the same file already makes exactly this
distinction (line 98, inspect.iscoroutinefunction), so the file is internally
inconsistent rather than uniformly wrong.

Suggested fix

Branch on the returned object rather than the registered mode — that stays
correct however operations are registered. The synchronous generator should not
be iterated inline, since that blocks the event loop for the duration of an LLM
round-trip on a container serving multiple concurrent requests:

from starlette.concurrency import iterate_in_threadpool

async def generator():
    stream = method(**(body.get("input") or {}))
    if hasattr(stream, "__aiter__"):
        async for event in stream:
            yield json.dumps(event) + "\n"
    else:
        async for event in iterate_in_threadpool(stream):
            yield json.dumps(event) + "\n"

Verified against a live Agent Runtime deployment: before the change,
stream_query returned 0 bytes; after, it returns the full event stream.

Note on severity

The failure mode is worse than a crash. Nothing in the deploy output, the
engine resource state, or the HTTP status distinguishes it from success — only
the container logs do. A team that does not read them can reasonably conclude
their agent is healthy while it answers nothing.