AdkApp.async_stream_query(session_events=...) creates a managed session per call and never deletes it

[brandhaug]

Summary

Every async_stream_query(session_events=...) call against a deployed Agent Engine creates a persisted Vertex session, appends the caller's supplied transcript to it, and abandons it. The session id is never returned to the caller, so it cannot be deleted, and it persists until TTL — 365 days by default.

This is independent of #7118. That issue is about session_events raising an AttributeError; this is about the session created before the raise. The fix suggested in #7118 does not address it, so applying that fix alone changes the behaviour from "leaks a session and fails" to "leaks a session and succeeds".

Environment

• google-cloud-aiplatform 1.163.0 and 2.0.1 (read from 1.162.0 in the wheel cache; AdkApp.async_stream_query(session_events=...) always raises AttributeError, and fails as a silent HTTP 200 on a deployed Agent Engine #7118 confirms the same code at adk.py:1182 in 1.163.0 and adk.py:1229 in 2.0.1)
• google-adk 2.6.3
• Deployed Agent Engine, europe-west1

Cause

In vertexai/agent_engines/templates/adk.py, async_stream_query:

if not session_id:
    session = await self.async_create_session(user_id=user_id)   # persisted session
    session_id = session["id"]
    if session_events is not None:
        session_service = self._tmpl_attrs.get("session_service")   # VertexAiSessionService when deployed
        for event in session_events:
            await session_service.append_event(session=session, event=event)

The generator's finally calls only _force_flush_otel. There is no delete_session.

Three things make it unrecoverable for the caller:

1. The id is never surfaced. It is bound to a local and not yielded. google/adk/events/event.py defines no session_id field, so nothing in the response stream carries it either.
2. It accumulates per turn, not per conversation. session_events exists for callers that hold their own history and replay it, so an N-turn conversation leaves N sessions, each holding a longer prefix of the same transcript.
3. No retention control reaches it. ttl and expire_time cannot be passed through async_stream_query, so every one of them gets the 365-day default.

Reproduction

Against a deployed engine:

async for event in remote_app.async_stream_query(
    message="hello", user_id="u", session_events=[]
):
    print(event)

session_events=[] is deliberate: the loop body never runs, so there is no exception and none of #7118 is involved. Then list sessions on the engine — one new session per call, none of them reachable from the code that made them.

Impact

The main use for session_events is a caller that owns the transcript and does not want a server-side copy of user-supplied conversation content. As implemented, a server-side copy is created anyway, in a resource the caller cannot address, delete, or set a retention policy on. For anything handling personal data, that turns an opt-out into an undeletable copy with the longest available lifetime.

Suggested fix

The sibling _StreamRunRequest path already does the right thing: when no session_id is supplied it uses in_memory_session_service and deletes the session in a finally (around adk.py:1364). Routing the session_events branch the same way fits — the session is throwaway by construction, so it does not need to be managed. It also removes the per-event round trips noted in #7118's second follow-up point, since in-memory appends are local.

Related observation

The async_create_session call sits above the session_events check, so any async_stream_query call without a session_id creates a managed session that is never deleted and whose id is never returned. That broader case may be intended — the docstring says a new session will be created — but the id not being returned makes it unusable as a resumable session either way. Filing this scoped to session_events, where there is a clear expectation of no server-side copy.