Live tool function_response lacks live_session_id, causing orphan response ValueError after reconnect/history rebuild

[smwitkowski]

🔴 Required Information

Describe the Bug:

In ADK Live mode, a model-generated function_call event can be persisted with live_session_id, while the matching ADK-generated function_response event is persisted without live_session_id.

That mixed metadata causes ADK history reconstruction to fail if a later Live run/reconnect starts from a session where the latest relevant event is that function response.

The failing shape is:

function_call event:
  author=<agent>
  role=model
  function_call.id=<id>
  live_session_id=<live session id>

function_response event:
  author=<same agent>
  role=user
  function_response.id=<same id>
  live_session_id=None

During _get_contents(), the Live-scoped model function_call is treated as context and converted into text. The matching function_response remains structured. _rearrange_events_for_latest_function_response() then sees a structured response but cannot find the structured matching call, and raises:

ValueError: No function call event found for function responses ids: {...}

This appears to be caused by the Live tool response path in google/adk/flows/llm_flows/functions.py: handle_function_calls_live() receives the original function_call_event with live_session_id, but __build_response_event() creates the matching function_response event without preserving that Live session metadata.

Steps to Reproduce:

1. Install ADK:

   pip install google-adk==1.32.0

2. Save the script below as repro_adk_live_organic_tool_response_orphan.py.

3. Run it:

   python repro_adk_live_organic_tool_response_orphan.py

4. Observe that ADK organically creates:

   • a function_call event with live_session_id
   • a matching function_response event without live_session_id

5. Observe that a subsequent _get_contents() history rebuild raises the orphan function response ValueError.

Expected Behavior:

A Live tool call and its matching tool response should remain pairable during history reconstruction.

Either:

• the Live-generated function_call and ADK-generated function_response should carry consistent Live metadata, or
• the history builder should avoid converting away the structured function_call while leaving its matching structured function_response behind.

In practice, reconnecting/resuming a Live session after a tool response should not permanently poison the session history.

Observed Behavior:

ADK raises:

ValueError: No function call event found for function responses ids: {'b00a1bcc-42b5-4dc4-9ba2-11c15816b8b1'}

The function call is not absent from the session. It exists, but was converted out of structured form during history rebuild because it had live_session_id. The matching function response remains structured because it did not have
live_session_id.

Environment Details:

• ADK Library Version: google-adk 1.32.0
• Desktop OS: macOS
• Python Version: Python 3.13

Model Information:

• Are you using LiteLLM: No
• Which model is being used: gemini-live-2.5-flash-native-audio

---

🟡 Optional Information

Regression:

N/A. We have reproduced this on google-adk 1.32.0.

Logs:

Expected output from the repro:

[generated events]
[event] {
  "author": "root_agent",
  "role": "model",
  "live_session_id": "live-session-1",
  "function_calls": [
    {
      "id": "b00a1bcc-42b5-4dc4-9ba2-11c15816b8b1",
      "name": "choice_group",
      "args": {
        "question": "Which option should we show?"
      }
    }
  ],
  "function_responses": []
}
[event] {
  "author": "root_agent",
  "role": "user",
  "live_session_id": null,
  "function_calls": [],
  "function_responses": [
    {
      "id": "b00a1bcc-42b5-4dc4-9ba2-11c15816b8b1",
      "name": "choice_group",
      "response": {
        "result": {
          "rich_content": {
            "type": "choice_group",
            "question": "Which option should we show?",
            "choices": [
              {
                "text": "Tell me about Scarlet Lady"
              },
              {
                "text": "What is included in the fare?"
              },
              {
                "text": "How do dining reservations work?"
              }
            ]
          }
        }
      }
    }
  ]
}

[history rebuild]
RAISED ValueError: No function call event found for function responses ids: {'b00a1bcc-42b5-4dc4-9ba2-11c15816b8b1'}

Additional Context:

We first observed this as a follow-on failure after Agent Engine/Live reconnect/resumption. The client reconnects, the same persisted ADK session history is used, and bidi_stream_query fails repeatedly because every new stream trips the same history reconstruction error.

This is separate from transport-level Live TTL/session-resumption behavior. The transport can reconnect successfully; the failure happens when ADK rebuilds model input from persisted events whose latest relevant boundary is a Live tool response.

The relevant source boundaries appear to be:

• google/adk/flows/llm_flows/functions.py
  • handle_function_calls_live()
  • _execute_single_function_call_live()
  • __build_response_event()
  • merge_parallel_function_response_events()
• google/adk/flows/llm_flows/contents.py
  • _is_other_agent_reply()
  • _rearrange_events_for_latest_function_response()

Suggested fix direction:

1. In the Live function-call path, preserve the originating function_call_event.live_session_id on matching function response events.
2. If parallel function response events are merged, preserve the base response event's live_session_id on the merged event.
3. Add a regression test where:
   • a Live model function_call has live_session_id
   • ADK builds the matching function_response
   • history reconstruction via _get_contents(... preserve_function_call_ids=True) does not raise

Conceptually:

# functions.py, Live tool response path
function_response_event = __build_response_event(
    tool,
    function_response,
    tool_context,
    invocation_context,
)
function_response_event.live_session_id = function_call_event.live_session_id

and for merged parallel tool responses:

merged_event = Event(
    invocation_id=base_event.invocation_id,
    author=base_event.author,
    branch=base_event.branch,
    content=types.Content(role='user', parts=merged_parts),
    actions=merged_actions,
    live_session_id=base_event.live_session_id,
)

Minimal Reproduction Code:

This repro does not call Gemini or Agent Engine. It simulates only the inbound Live model response that Gemini/ADK's Live connection would provide: an LlmResponse containing a function_call and a live_session_id.

Everything after that is ADK's normal Live path:

1. BaseLlmFlow._postprocess_live() finalizes the model function_call event.
2. ADK's Live tool handler runs a real FunctionTool.
3. ADK builds the matching function_response event.
4. The generated events are appended to an ADK session.
5. A later history rebuild raises the orphan response ValueError.

from __future__ import annotations

import asyncio
import json
from typing import Any

from google.adk.agents.invocation_context import InvocationContext
from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.run_config import RunConfig
from google.adk.events.event import Event
from google.adk.flows.llm_flows.base_llm_flow import BaseLlmFlow
from google.adk.flows.llm_flows.contents import _get_contents
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.tools.function_tool import FunctionTool
from google.genai import types


APP_NAME = "organic-live-history-repro"
USER_ID = "user"
SESSION_ID = "session"
AGENT_NAME = "root_agent"
INVOCATION_ID = "invocation-1"
CALL_ID = "b00a1bcc-42b5-4dc4-9ba2-11c15816b8b1"
LIVE_SESSION_ID = "live-session-1"


def choice_group(question: str) -> dict[str, Any]:
    return {
        "result": {
            "rich_content": {
                "type": "choice_group",
                "question": question,
                "choices": [
                    {"text": "Tell me about Scarlet Lady"},
                    {"text": "What is included in the fare?"},
                    {"text": "How do dining reservations work?"},
                ],
            }
        }
    }


def event_summary(event: Event) -> dict[str, Any]:
    parts = event.content.parts if event.content and event.content.parts else []
    return {
        "author": event.author,
        "role": event.content.role if event.content else None,
        "live_session_id": event.live_session_id,
        "function_calls": [
            part.function_call.model_dump(exclude_none=True, mode="json")
            for part in parts
            if part.function_call
        ],
        "function_responses": [
            part.function_response.model_dump(exclude_none=True, mode="json")
            for part in parts
            if part.function_response
        ],
    }


async def main() -> None:
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name=APP_NAME,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )

    tool = FunctionTool(choice_group)
    agent = LlmAgent(
        name=AGENT_NAME,
        model="gemini-live-2.5-flash-native-audio",
        tools=[tool],
    )

    invocation_context = InvocationContext(
        session_service=session_service,
        invocation_id=INVOCATION_ID,
        agent=agent,
        session=session,
        run_config=RunConfig(),
    )

    llm_request = LlmRequest()
    llm_request.tools_dict[tool.name] = tool

    # This is the only simulated input: the shape produced when Gemini Live
    # asks to call a tool. ADK's Live connection stamps model responses with
    # live_session_id before BaseLlmFlow sees them.
    live_model_tool_call = LlmResponse(
        model_version="gemini-live-2.5-flash-native-audio",
        live_session_id=LIVE_SESSION_ID,
        content=types.Content(
            role="model",
            parts=[
                types.Part(
                    function_call=types.FunctionCall(
                        id=CALL_ID,
                        name="choice_group",
                        args={"question": "Which option should we show?"},
                    )
                )
            ],
        ),
    )

    flow = BaseLlmFlow()
    base_model_event = Event(
        invocation_id=INVOCATION_ID,
        author=AGENT_NAME,
    )

    generated_events: list[Event] = []
    async for event in flow._postprocess_live(
        invocation_context,
        llm_request,
        live_model_tool_call,
        base_model_event,
    ):
        generated_events.append(event)
        await session_service.append_event(session=session, event=event)

    print("[generated events]")
    for event in generated_events:
        print(json.dumps(event_summary(event), indent=2))

    print("\n[history rebuild]")
    try:
        contents = _get_contents(
            current_branch=None,
            events=session.events,
            agent_name=AGENT_NAME,
            preserve_function_call_ids=True,
        )
    except Exception as exc:
        print(f"RAISED {type(exc).__name__}: {exc}")
        return

    print(f"OK contents={len(contents)}")
    for content in contents:
        print(json.dumps(content.model_dump(exclude_none=True, mode="json")))


if __name__ == "__main__":
    asyncio.run(main())

How often has this issue occurred?:

• Always (100%) for the deterministic local repro above.
• In deployed Live/Agent Engine usage, it is timing-dependent: it appears when a reconnect/resume or new stream starts from history where the latest relevant persisted boundary is the ADK-created tool response.