[Python][Live] Background tool output can reach the parent model during the transfer delay

[ehtesham-zahoor]

🔴 Required Information

Describe the Bug:

In Live mode, when an agent transfers to a sub-agent, run_live sleeps for DEFAULT_TRANSFER_AGENT_DELAY (1.0s) before it cancels the send task, closes the parent connection, and stops the parent's background tools:

# 2.8.0: src/google/adk/flows/llm_flows/base_llm_flow.py:784-798
transfer_to_agent = event.actions.transfer_to_agent
if transfer_to_agent:
  await asyncio.sleep(DEFAULT_TRANSFER_AGENT_DELAY)
  # cancel the tasks that belongs to the closed connection.
  send_task.cancel()
  await llm_connection.close()
  # The sub agent takes over the live request queue, so this agent's
  # background tools have to stop here rather than when this run_live
  # eventually returns: ... until then a tool of this agent would keep
  # feeding function responses to a model that never made those calls.
  await self._stop_background_tool_tasks(invocation_context)

For the whole of that second the connection is open and send_task is still draining the live request queue. A non-blocking tool belonging to the parent that completes inside the window posts its function response onto the shared queue (functions.py; streaming tools do the same), send_task picks it up, and it is forwarded to the parent model — which is exactly what the comment above, the _stop_background_tool_tasks docstring, and test_streaming_tool_stops_when_its_agent_hands_off all describe as the thing to prevent.

The guard is correct and clearly deliberate. It is simply placed after the delay instead of before it, so it closes the door once the window has already elapsed.

Same code on main after the live-flow extraction, at src/google/adk/flows/llm_flows/_live_llm_flow.py:814-830.

Steps to Reproduce:

1. Install: pip install google-adk==2.8.0
2. Save the script under Minimal Reproduction Code as repro.py
3. Run: python repro.py
4. The tool's function response reaches the connection at ~0.26s while closed=False; the connection closes only at ~1.01s

Expected Behavior:

Once the transfer event has been yielded, no function response from the parent agent's background tools should reach the parent model's connection. The project states this intent in three places:

• base_llm_flow.py, on the cleanup call itself: "a tool of this agent would keep feeding function responses to a model that never made those calls."
• base_llm_flow.py, _stop_background_tool_tasks docstring: "a tool kept running after its agent was done, feeding function responses into a live request queue that by then belonged to another agent, or to nobody at all."
• tests/unittests/streaming/test_live_tool_shutdown.py, test_streaming_tool_stops_when_its_agent_hands_off: "a tool still running for the previous agent would push function responses at a model that never called it."

Observed Behavior:

The response is delivered to the still-open parent connection:

google-adk layout: base_llm_flow.BaseLlmFlow.run_live
DEFAULT_TRANSFER_AGENT_DELAY = 1.0   tool completes at 0.25s
  [ 0.01s] transfer event YIELDED to caller
  [ 0.25s] background tool COMPLETES, enqueues response
  [ 0.26s] connection RECEIVED content  closed=False     <-- reaches the parent model
  [ 1.01s] connection CLOSED

late output reached the parent connection: True
REPRODUCED

Environment Details:

• ADK Library Version (pip show google-adk): 2.8.0 (also reproduced on main at commit c506ddf3)
• Desktop OS: macOS 15.1.1 (arm64) — also reproduced in python:3.12-slim-bookworm
• Python Version (python -V): 3.14.7 (also reproduced on 3.12.14)

Model Information:

• Are you using LiteLLM: N/A
• Which model is being used: N/A — the reproduction substitutes a fake BaseLlmConnection. The flow, the live request queue, the send task and the background tool task are all real; only the transport is faked, so no model traffic is involved.

---

🟡 Optional Information

Regression:

No — this has not worked in a previous version. The cleanup call was introduced already positioned after the delay, in 0088abbe "fix(live): stop background tool tasks when a live agent run ends" (2026-08-14), which is contained in v2.8.0. The later 65b382d5 "refactor: extract live flow logic from base_llm_flow into a dedicated module" only moved the block into _live_llm_flow.py without changing the order.

Logs:

Control — the same script with the tool completing at 2.0s, outside the 1.0s window. The guard fires and cancels the tool, so nothing is delivered. This confirms the harness detects the difference rather than always printing REPRODUCED:

DEFAULT_TRANSFER_AGENT_DELAY = 1.0   tool completes at 2.0s
  [ 0.01s] transfer event YIELDED to caller
  [ 1.01s] connection CLOSED
  [ 1.02s] background tool CANCELLED by the guard

late output reached the parent connection: False
not reproduced

With the proposed one-line move applied, the same 0.25s tool is cancelled immediately, nothing reaches the connection, and the connection still closes at 1.01s — the delay's own purpose is unaffected:

DEFAULT_TRANSFER_AGENT_DELAY = 1.0   tool completes at 0.25s
  [ 0.01s] transfer event YIELDED to caller
  [ 0.01s] background tool CANCELLED by the guard
  [ 1.01s] connection CLOSED

late output reached the parent connection: False
not reproduced

Additional Context:

Related to #6541, which produced the surrounding transfer-gating code. The review scope there explicitly kept "the connection-close/delay and resumption-reset logic in that block untouched", so this ordering question was out of scope — which may be why it has not surfaced before.

The change suggested below stays inside that same boundary: it moves only _stop_background_tool_tasks, and leaves the delay, send_task.cancel() and llm_connection.close() exactly where they are. The delay does matter — just above the transfer branch the transfer's own function response is queued with live_request_queue.send_content(...), and the sleep is what gives send_task the chance to deliver it before cancellation. Fencing the parent's tools rather than the queue keeps that behaviour intact, which the third log above confirms.

Proposed fix (against main; happy to open a PR on confirmation):

   transfer_to_agent = event.actions.transfer_to_agent
   if transfer_to_agent:
+    await flow._stop_background_tool_tasks(invocation_context)
     await asyncio.sleep(base_llm_flow.DEFAULT_TRANSFER_AGENT_DELAY)
     # cancel the tasks that belongs to the closed connection.
     send_task.cancel()
     await llm_connection.close()
-    await flow._stop_background_tool_tasks(invocation_context)

For regression coverage I would extend test_streaming_tool_stops_when_its_agent_hands_off to assert that nothing reaches the parent connection between the transfer event and the close. As written, that test asserts the tool's task is done and its tick count has stopped — both true once the transfer has completed — so it passes with the defect present.

Minimal Reproduction Code:

Self-contained; no API key and no model access needed. The version shims at the top let the same file run on released 2.8.0 and on main after the extraction refactor.

"""Repro: a live background tool's function response reaches the parent model
during the agent-transfer delay.

  pip install google-adk==2.8.0 && python repro.py

Expected (intended behaviour): nothing reaches the parent connection.
Observed: the response arrives at ~0.26s while the connection is still open;
the connection closes at ~1.02s.

Works on released 2.8.0 (branch inline in base_llm_flow) and on main after the
_live_llm_flow extraction.
"""

from __future__ import annotations

import asyncio
from contextlib import aclosing

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.events.event_actions import EventActions
from google.adk.flows.llm_flows import base_llm_flow
try:  # main
    from google.adk.live.live_request_queue import LiveRequestQueue
except ImportError:  # released 2.8.0
    from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.sessions.session import Session
from google.genai import types

try:  # main, after the live-flow extraction
    from google.adk.flows.llm_flows import _live_llm_flow

    LAYOUT = "_live_llm_flow.run_live_flow"

    def run_live(flow, ctx):
        return _live_llm_flow.run_live_flow(flow, ctx)

except ImportError:  # released 2.8.0, branch inline in base_llm_flow
    LAYOUT = "base_llm_flow.BaseLlmFlow.run_live"

    def run_live(flow, ctx):
        return flow.run_live(ctx)


TOOL_COMPLETES_AT = 0.25  # anything < DEFAULT_TRANSFER_AGENT_DELAY hits the window


def _t(t0):
    return asyncio.get_running_loop().time() - t0


class FakeConnection:
    """Stands in for the parent model's live connection."""

    def __init__(self):
        self.sent, self.closed, self.t0 = [], False, 0.0

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self.close()

    async def send_history(self, contents):
        pass

    async def send_content(self, content):
        await self._send_content(content)

    async def send_realtime(self, blob):
        pass

    async def _send_content(self, content, partial=False):
        self.sent.append(content)
        print(f"  [{_t(self.t0):5.2f}s] connection RECEIVED content  closed={self.closed}")

    async def close(self):
        if not self.closed:
            self.closed = True
            print(f"  [{_t(self.t0):5.2f}s] connection CLOSED")


class FakeLlm:
    def __init__(self, conn):
        self.conn = conn
        self.model = "fake"

    def connect(self, llm_request):
        return self.conn


class ChildAgent:
    name = "child_agent"

    async def run_live(self, ctx):
        if False:
            yield


class Flow(base_llm_flow.BaseLlmFlow):
    def __init__(self, conn):
        super().__init__()
        self.conn = conn

    async def _preprocess_async(self, ctx, req):
        if False:
            yield

    # main calls _get_llm; released 2.8.0 calls the name-mangled __get_llm
    def _get_llm(self, ctx):
        return FakeLlm(self.conn)

    def _BaseLlmFlow__get_llm(self, ctx):
        return FakeLlm(self.conn)

    # released 2.8.0 passes an extra event_id; main does not
    async def _receive_from_model(self, llm_connection, *args):
        ctx = args[-2]
        yield Event(
            invocation_id=ctx.invocation_id,
            author="root_agent",
            actions=EventActions(transfer_to_agent="child_agent"),
        )

    def _get_agent_to_run(self, ctx, name):
        return ChildAgent()


async def main():
    queue = LiveRequestQueue()
    agent = LlmAgent(
        name="root_agent",
        model="gemini-2.0-flash",
        sub_agents=[LlmAgent(name="child_agent", model="gemini-2.0-flash")],
    )
    ctx = InvocationContext(
        invocation_id="invocation",
        agent=agent,
        session=Session(id="s", app_name="app", user_id="u"),
        session_service=InMemorySessionService(),
        live_request_queue=queue,
        run_config=RunConfig(),
        active_non_blocking_tool_tasks={},
    )
    conn = FakeConnection()
    flow = Flow(conn)

    async def late_tool():
        # Mirrors functions.py — a completed non-blocking tool posts its
        # function response onto the shared live request queue.
        await asyncio.sleep(TOOL_COMPLETES_AT)
        print(f"  [{_t(conn.t0):5.2f}s] background tool COMPLETES, enqueues response")
        queue.send_content(
            types.Content(
                role="user",
                parts=[types.Part.from_function_response(name="late_tool", response={"r": 1})],
            )
        )

    print(f"google-adk layout: {LAYOUT}")
    print(
        f"DEFAULT_TRANSFER_AGENT_DELAY = {base_llm_flow.DEFAULT_TRANSFER_AGENT_DELAY}"
        f"   tool completes at {TOOL_COMPLETES_AT}s"
    )
    conn.t0 = asyncio.get_running_loop().time()
    task = asyncio.create_task(late_tool())
    ctx.active_non_blocking_tool_tasks = {"late_tool": task}

    # The transfer branch runs AFTER `yield event`, so the generator has to be
    # drained; pulling a single event leaves the flow suspended at the yield.
    async def drain():
        async with aclosing(run_live(flow, ctx)) as events:
            async for ev in events:
                if ev.actions and ev.actions.transfer_to_agent:
                    print(f"  [{_t(conn.t0):5.2f}s] transfer event YIELDED to caller")

    drainer = asyncio.create_task(drain())
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout=5.0)
    except asyncio.CancelledError:
        print(f"  [{_t(conn.t0):5.2f}s] background tool CANCELLED by the guard")
    try:
        await asyncio.wait_for(asyncio.shield(drainer), timeout=4.0)
    except asyncio.TimeoutError:
        drainer.cancel()
    await asyncio.sleep(0.05)

    print(f"\nlate output reached the parent connection: {bool(conn.sent)}")
    print("REPRODUCED" if conn.sent else "not reproduced")


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

How often has this issue occurred?:

• Intermittently (<50%) — it requires a background tool of the transferring agent to complete inside the 1.0s window.

---

Disclosure: I used an AI assistant for repository navigation, reproduction support, and drafting this report. I reviewed the code path, ran the reproduction and the control, and verified the proposed fix myself.

[ehtesham-zahoor]

Picking this one up now — opening a PR shortly. Flagging it here so nobody duplicates the work; if someone is already on it, say so and I will drop mine.

Thanks for the interest! I have the one-line patch and the verification already done; the control and fixed-ordering runs in the issue are from my working tree, so I'd like to take this one.

Per the contributing guide I'm waiting on a maintainer to confirm the approach before opening the PR. @surajksharma07 @wuliang229, happy to send it whenever you're comfortable with the ordering change.

[surajksharma07]

@ehtesham-zahoor Moving _stop_background_tool_tasks above the sleep/cancel/close looks like the right workaround and matches what you already verified. Run it against the existing live-shutdown tests plus a couple more transfer timings to be safe and once you're confident it holds go ahead and open the PR.

[ehtesham-zahoor]

@ehtesham-zahoor Moving _stop_background_tool_tasks above the sleep/cancel/close looks like the right workaround and matches what you already verified. Run it against the existing live-shutdown tests plus a couple more transfer timings to be safe and once you're confident it holds go ahead and open the PR.

Opened #7011 — five delay/completion timings plus a streaming-tool case; four of the six fail on unmodified main, and each asserts the tool actually started.

Note: the existing test_streaming_tool_stops_when_its_agent_hands_off passes with the defect present — it observes state after the handoff completes, so the window was never covered.