AdkApp.async_stream_query(session_events=...) always raises AttributeError, and fails as a silent HTTP 200 on a deployed Agent Engine

## Summary

`AdkApp.async_stream_query(session_events=...)` always fails, because `async_create_session()` returns a serialized `dict` and that dict is passed straight to `BaseSessionService.append_event(session=...)`, which requires a `Session`.

On a **deployed** Agent Engine this fails silently: the engine answers **HTTP 200 with an empty response stream**. No error event, no error status, nothing on the wire. The traceback only appears in Cloud Logging. Every caller of `session_events` therefore sees what looks like a successful turn where the agent had nothing to say.

## Environment

- `google-cloud-aiplatform` 1.163.0 and 2.0.1 (current release; the bug is in both)
- `google-adk` 2.6.3
- Python 3.12, and Python 3.11 in the deployed container

## Cause

In `vertexai/agent_engines/templates/adk.py`:

```python
if not session_id:
session = await self.async_create_session(user_id=user_id) # returns self._serialize(session) -> dict
session_id = session["id"]
if session_events is not None:
...
for event in session_events:
if not isinstance(event, Event):
event = Event.model_validate(event)
await session_service.append_event(
session=session, # <-- dict, but append_event needs a Session
event=event,
)
```

- `adk.py:1182` in 1.163.0
- `adk.py:1229` in 2.0.1

`async_create_session` ends with `return self._serialize(session)` (1.163.0 `adk.py:1591`), so the value bound to `session` is a plain dict. That is correct for the method's own contract, since it has to be JSON-serializable for the reasoning-engine wire protocol. It is just not what `append_event` accepts.

Two different `AttributeError`s result, depending on the session service, which makes this look like two unrelated bugs:

- `InMemorySessionService` (in-process): `AttributeError: 'dict' object has no attribute 'app_name'` (`in_memory_session_service.py:326`)
- `VertexAiSessionService` (deployed engine): `AttributeError: 'dict' object has no attribute 'events'` (`vertex_ai_session_service.py:387` -> `base_session_service.py:164`)

Note that `session_events=[]` does **not** reproduce it: the loop body never runs. At least one event is needed.

## Reproduction

No credentials, no project and no deployment needed. This fails before any model or API call:

```python
import asyncio

from google.adk.agents import Agent
from vertexai.agent_engines.templates.adk import AdkApp

app = AdkApp(agent=Agent(model="gemini-2.5-flash", name="repro"))

PRIOR = [
{
"id": "e1",
"invocation_id": "i1",
"author": "user",
"timestamp": 1.0,
"content": {"role": "user", "parts": [{"text": "my name is Ada"}]},
}
]

async def main():
async for event in app.async_stream_query(
message="what is my name?", user_id="u", session_events=PRIOR
):
print(event)

asyncio.run(main())
```

```
File ".../vertexai/agent_engines/templates/adk.py", line 1182, in async_stream_query
await session_service.append_event(
File ".../google/adk/sessions/in_memory_session_service.py", line 326, in append_event
app_name = session.app_name
AttributeError: 'dict' object has no attribute 'app_name'
```

Deployed to Agent Engine, the same call over `:streamQuery` returns HTTP 200 with an empty body, and Cloud Logging shows the `'dict' object has no attribute 'events'` variant.

## Expected

`session_events` initializes the new session with the supplied events, and the query then runs against that history. That is what the docstring promises: "The session events to use for the query. This will be used to initialize the session if `session_id` is not provided."

## Suggested fix

*Updated 2026-08-31 — the original suggestion (re-fetch the `Session` before appending) fixes the crash only. Kept below because [my comment about #7119](https://github.com/googleapis/python-aiplatform/issues/7118#issuecomment-5477469427) refers to it.*

Mirror `streaming_agent_run_with_events`, which faces the same "no session id, initialize from supplied events" case and already handles it: `in_memory_session_service` and `in_memory_runner` rather than the managed services, with `delete_session` in a `finally` (`adk.py:1364`, `:1393`; both attrs populated by `set_up()` at `:1091`, `:1094`).

This fixes the crash, the orphaned session (#7119), and the per-event latency in one change. `InMemorySessionService.create_session` returns a real `Session`, so there is no dict to re-fetch; nothing is written to the managed store, so there is no session to leak; and appends against a process-local service are not round trips, so replay stops scaling with transcript length. Verified on a deployed engine in `europe-west1`.

Implementation constraint: `_tmpl_attrs["in_memory_*"]` cannot be read directly, because `adk deploy agent_engine` (google-adk 2.6.3) runs `adk api_server`, which builds its own `AdkApp` (`google/adk/cli/fast_api.py:760`) and sets `_tmpl_attrs["runner"]` itself, so `set_up()` never runs and those attrs are absent. The existing `if not self._tmpl_attrs.get("runner"): self.set_up()` guard does not fire either, since `runner` is set. Reading them would pass in-process and `AttributeError` on `None` in the deployed server — the same silent failure as this issue.

Original suggestion (fixes the crash only)

Re-fetch the `Session` object before appending, keeping `async_create_session`'s serialized return value as it is:

```python
created = await self.async_create_session(user_id=user_id)
session_id = created["id"]
if session_events is not None:
session = await session_service.get_session(
app_name=self._app_name(), user_id=user_id, session_id=session_id
)
for event in session_events:
if not isinstance(event, Event):
event = Event.model_validate(event)
await session_service.append_event(session=session, event=event)
```

I verified this in an `AdkApp` subclass, in-process and on a deployed engine in `europe-west1`. With it, a caller-held transcript replays correctly and the agent continues the conversation, including transcripts recorded in an earlier session.

Two adjacent things worth considering while this is open:

1. **The silent failure is the more serious half.** An exception raised inside the streaming generator becomes an empty 200 response with no error payload. Even after this bug is fixed, a caller has no way to distinguish "the agent produced no events" from "user code raised". Surfacing an error event, or a non-200, would prevent a whole class of undiagnosable failures.
2. **`append_event` per event is O(n) round trips** against `VertexAiSessionService`. Replaying an 11-event transcript to a deployed engine added roughly 4.5s to time-to-first-event over the `session_id` path, about 0.4s per event, which makes `session_events` impractical for long conversations even once it works. *Updated: no bulk-append path is needed — the fix above removes the round trips rather than batching them.*

Start in vertexai/agent_engines/templates/adk.py at async_stream_query and compare its session_events path with streaming_agent_run_with_events around lines 1364 and 1393. Read set_up() and the deployed google/adk/cli/fast_api.py entry point to understand which in-memory attributes are available. Done means replaying supplied events works in-process and on a deployed Agent Engine without an AttributeError, orphaned session, or empty response stream.

Scritto dal modello di indicizzazione a partire dal testo della issue.