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.*

Bắt đầu tại vertexai/agent_engines/templates/adk.py ở async_stream_query và so sánh đường dẫn session_events của nó với streaming_agent_run_with_events quanh các dòng 1364 và 1393. Đọc set_up() và entry point google/adk/cli/fast_api.py đã được triển khai để hiểu những thuộc tính trong bộ nhớ nào có sẵn. Được xem là hoàn tất khi việc phát lại các event được cung cấp hoạt động trong tiến trình và trên một Agent Engine đã triển khai mà không có AttributeError, session mồ côi hoặc stream phản hồi trống.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.