[Live/bidi_stream_query] Session state silently dropped — first queue message 'state' field never read, async_create_session called without state=

[smwitkowski]

🔴 Required Information

Describe the Bug:

bidi_stream_query() silently drops session state from the first queue message. The state field is never read from first_request, and when no session_id is provided, async_create_session(user_id=user_id) is called without the state= parameter — creating a stateless session. Tools that read ToolContext.state receive an empty dict.

The same state flows correctly through async_stream_query() when the session is pre-created with async_create_session(state=...).

This blocks any application that needs to inject client-side state (user profile, page context, preferences) into a live/bidi streaming session.

Steps to Reproduce:

1. Create app/agent.py:

import json
from google.adk.agents import Agent
from google.adk.tools import FunctionTool, ToolContext
from google.genai.types import GenerateContentConfig

def get_name(tool_context: ToolContext) -> str:
 """Retrieve the user's name from session state."""
 name = tool_context.state.get("name", "NO_NAME_IN_STATE")
 return json.dumps({"name": name})

root_agent = Agent(
 model="gemini-2.5-flash",
 name="state_test_agent",
 instruction=(
 "You are a test agent. When the user says hello, ALWAYS call get_name first. "
 "If it returns a real name, respond with 'Hello' followed by that name. "
 "If it returns 'NO_NAME_IN_STATE', respond 'Hello stranger!'"
 ),
 tools=[FunctionTool(get_name)],
 generate_content_config=GenerateContentConfig(temperature=0.0),
)

2. Create app/agent_engine_app.py:

from vertexai.preview.reasoning_engines.templates.adk import AdkApp
from app.agent import root_agent

class AgentEngineApp(AdkApp):
 def register_operations(self):
 operations = super().register_operations()
 operations["bidi_stream"] = ["bidi_stream_query"]
 operations["async_stream"] = ["async_stream_query"]
 return operations

agent_engine = AgentEngineApp(agent=root_agent)

3. Run the following test that instruments async_create_session to prove state is dropped:

import asyncio
from app.agent_engine_app import agent_engine

def extract_text(event):
 content = event.get("content", {})
 if isinstance(content, dict):
 for part in content.get("parts", []):
 if isinstance(part, dict) and part.get("text"):
 return part["text"]
 return ""

async def main():
 # TEST 1: async_stream_query — state flows correctly
 session = await agent_engine.async_create_session(
 user_id="async-user", state={"name": "Alice"}
 )
 response = ""
 async for event in agent_engine.async_stream_query(
 user_id="async-user",
 session_id=session.id,
 message={"role": "user", "parts": [{"text": "Hello"}]},
 ):
 response += extract_text(event)
 print(f"async_stream_query response: {response}")
 print(f" 'Alice' in response: {'alice' in response.lower()}") # True

 # TEST 2: bidi_stream_query — instrument to prove state is dropped
 captured_kwargs = []
 original = agent_engine.async_create_session

 async def instrumented(*args, **kwargs):
 captured_kwargs.append(kwargs)
 return await original(*args, **kwargs)

 agent_engine.async_create_session = instrumented

 queue = asyncio.Queue()
 await queue.put({
 "user_id": "bidi-user",
 "state": {"name": "Charlie"}, # THIS IS SILENTLY DROPPED
 })
 await queue.put({
 "content": {"role": "user", "parts": [{"text": "Hello"}]},
 })

 try:
 async with asyncio.timeout(10):
 async for event in agent_engine.bidi_stream_query(queue):
 pass
 except Exception:
 pass # Expected: Live API model error (irrelevant to the bug)
 finally:
 agent_engine.async_create_session = original

 print(f"\nbidi_stream_query called async_create_session with:")
 for kw in captured_kwargs:
 print(f" kwargs = {kw}")
 state_passed = any(kw.get("state") for kw in captured_kwargs)
 print(f" state= parameter present: {state_passed}") # False — BUG

asyncio.run(main())

4. Output:

async_stream_query response: Hello Alice!
 'Alice' in response: True

bidi_stream_query called async_create_session with:
 kwargs = {'user_id': 'bidi-user'}
 state= parameter present: False

Expected Behavior:

bidi_stream_query() should read state from the first queue message and pass it to async_create_session(). The get_name tool should return "Charlie" via ToolContext.state, and the agent should respond with "Hello Charlie!".

Observed Behavior:

bidi_stream_query() ignores the state field in the first queue message. async_create_session() is called with only user_id= — no state= parameter. The session is created stateless. ToolContext.state is empty. The agent responds with "Hello stranger!" instead of "Hello Charlie!".

The state field is visible in the queue message dict but is never read at adk.py:1170-1176.

Environment Details:

• ADK Library Version: 1.27.1 (also verified in source: v1.28.0 live_request_queue.py and aiplatform v1.145.0 adk.py — both unfixed)
• Desktop OS: macOS 15.4
• Python Version: 3.13.0

Model Information:

• Are you using LiteLLM: No
• Which model: gemini-2.5-flash (for async path), gemini-live-2.5-flash-native-audio (for bidi/live path)

---

🟡 Optional Information

Regression:

This has likely never worked — bidi_stream_query() has never read state from the first queue message. The state parameter was added to async_create_session() but the bidi path was not updated to use it.

How often has this issue occurred?:

Always (100%) — deterministic. State is never read from the first bidi message.

Minimal Reproduction Code:

See Steps to Reproduce above. The key proof is instrumenting async_create_session:

# What bidi_stream_query passes:
async_create_session(user_id='bidi-user')
# No state= parameter

# What it should pass:
async_create_session(user_id='bidi-user', state={'name': 'Charlie'})

Root Cause:

Two issues in the ADK/vertexai stack:

1. bidi_stream_query() never reads state from first queue message

vertexai/preview/reasoning_engines/templates/adk.py (lines ~1169-1182 in v1.141.0):

first_request = await request_queue.get()
user_id = first_request.get("user_id") # read
session_id = first_request.get("session_id") # read
run_config = first_request.get("run_config") # read
first_live_request = first_request.get("live_request") # read
# state = first_request.get("state") # NEVER READ

if not session_id:
 session = await self.async_create_session(user_id=user_id) # no state=
 session_id = session.id

async_create_session() at line ~1350 accepts state: Optional[Dict[str, Any]] = None, but bidi_stream_query never passes it.

2. LiveRequest has no state field for subsequent messages

google/adk/agents/live_request_queue.py (lines 26-57):

class LiveRequest(BaseModel):
 content: Optional[types.Content] = None
 blob: Optional[types.Blob] = None
 activity_start: Optional[types.ActivityStart] = None
 activity_end: Optional[types.ActivityEnd] = None
 close: bool = False
 # NO state field — Pydantic silently drops unknown fields

_forward_requests() at line ~1194 calls LiveRequest.model_validate(request), which drops any state or state_delta key via Pydantic validation.

Suggested Fix:

3-line change in bidi_stream_query():

first_request = await request_queue.get()
user_id = first_request.get("user_id")
session_id = first_request.get("session_id")
state = first_request.get("state") # ← ADD: read state from first message
run_config = first_request.get("run_config")
first_live_request = first_request.get("live_request")

if not session_id:
 session = await self.async_create_session(
 user_id=user_id,
 state=state, # ← ADD: pass state to session creation
 )
 session_id = session.id

For the case where session_id IS provided but state needs to be updated, a state_delta mechanism on LiveRequest would also be needed — but the above fix addresses the most common pattern (new session with initial state).

Related Issues:

• [Live] run_live(): Fire-and-forget tool calls cause duplicate model responses via orphaned function response reinjection #4902 — [Live] Fire-and-forget tool calls cause duplicate responses (our previous report, same project)
• Restoring a saved session from the database for a Bidi streaming Agent doesn't work #1317 — Restoring saved session for bidi streaming agent (closed as fixed, but only addressed session_id — not state)
• BIDI Input and Output Transcriptions Added as NULL Events to DataBaseSessionService #3172 — BIDI transcriptions added as NULL events (related state persistence issue)
• [Live] Multiple responses after agent transfer and repeat response on session resumption #3395 — Multiple responses after agent transfer + session resumption

[wuliang229]

It's fixed in googleapis/python-aiplatform@37b5a0f.