Parallel calls to the same streaming tool leave an orphan task after live teardown

[CorgiBoyG]

Required Information

Describe the Bug:

When one live model response contains multiple calls to the same async-generator tool, ADK starts each call concurrently but stores only one task under InvocationContext.active_streaming_tools[tool.name]. Each registration overwrites the previous task reference. As a result, run teardown and stop_streaming(function_name) can only cancel the last registered call; earlier calls continue writing to the live request queue after their agent run has ended.

The violated lifecycle invariant is that every background tool task started by a live agent run must remain tracked until it completes or the run tears it down.

Steps to Reproduce:

1. Create an async-generator FunctionTool that yields once and then waits.
2. Build one live function-call event containing two calls to that tool with different call IDs/arguments.
3. Pass the event through handle_function_calls_live().
4. Wait until both tool tasks have started.
5. Call SingleFlow()._stop_background_tool_tasks(invocation_context).
6. Inspect both tasks.

Expected Behavior:

Both calls are tracked and cancelled during live-run teardown.

Observed Behavior:

Only the task currently stored in active_streaming_tools[tool.name].task is cancelled. The other task remains pending. This reproduces consistently on current main (b0180620f4c2f4f4467a89c37a30f75bf849700b).

Environment Details:

• ADK Library Version: current main at b0180620
• Desktop OS: macOS
• Python Version: 3.10.20

Model Information:

• Are you using LiteLLM: No
• Which model is being used: N/A (unit-level live function-call event)

Additional Context

The overwrite occurs in _process_function_live_helper() when it assigns a newly created task to active_streaming_tools[tool.name].task. stop_background_tool_tasks() later enumerates one task per registry entry, so it has no reference to the earlier call. Non-blocking live tools avoid the same collision by including the function-call ID in their task key.

I intend to keep all concurrent tasks for a streaming tool tracked while preserving the existing name-based stop_streaming behavior, and add a regression test through the real live function-call path. I can send the focused fix and test within 24 hours.

Regression:

Not confirmed.

Logs:

assert all(task.done() for task in tasks)
E assert False

Minimal Reproduction Code:

await handle_function_calls_live(
    invocation_context,
    event_with_two_calls_to_monitor,
    {monitor_tool.name: monitor_tool},
)
await both_started.wait()
await SingleFlow()._stop_background_tool_tasks(invocation_context)
assert all(task.done() for task in started_tasks)  # fails: one is still pending

How often has this issue occurred?:

• Always (100%)

[llalitkumarrr]

Hello @CorgiBoyG,

Apologies for the delayed response. Could you please share an end to end code sample which include agent and calls? This will help us reproduce the issue and perform a detailed analysis.

[CorgiBoyG]

Thanks for taking a look. Here is a deterministic end-to-end reproduction through Agent and Runner.run_live. It uses the repository's MockModel only to force two calls in one model response without requiring credentials; tool dispatch and teardown use the normal live runtime path.

import asyncio
from contextlib import aclosing

from google.adk.agents.llm_agent import Agent
from google.adk.agents.run_config import RunConfig
from google.adk.live import LiveRequestQueue
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types

from tests.unittests import testing_utils


async def main() -> None:
  started_tasks: list[asyncio.Task] = []
  both_started = asyncio.Event()

  async def monitor(value: str):
    started_tasks.append(asyncio.current_task())
    if len(started_tasks) == 2:
      both_started.set()
    yield {'value': value}
    await asyncio.Event().wait()

  duplicate_calls = LlmResponse(
      content=types.Content(
          role='model',
          parts=[
              types.Part.from_function_call(
                  name='monitor', args={'value': 'first'}
              ),
              types.Part.from_function_call(
                  name='monitor', args={'value': 'second'}
              ),
          ],
      ),
      turn_complete=False,
  )
  agent = Agent(
      name='root_agent',
      model=testing_utils.MockModel.create([duplicate_calls]),
      tools=[monitor],
  )
  session_service = InMemorySessionService()
  session = await session_service.create_session(
      app_name='repro', user_id='user'
  )
  runner = Runner(
      app_name='repro', agent=agent, session_service=session_service
  )
  live_request_queue = LiveRequestQueue()
  live_request_queue.send_realtime(
      types.Blob(data=b'question', mime_type='audio/pcm')
  )

  try:
    async with aclosing(
        runner.run_live(
            user_id='user',
            session_id=session.id,
            live_request_queue=live_request_queue,
            run_config=RunConfig(response_modalities=['TEXT']),
        )
    ) as events:
      async for _ in events:
        if both_started.is_set():
          break

    for _ in range(10):
      await asyncio.sleep(0)
    print('started task count:', len(started_tasks))
    print('done after run teardown:', [task.done() for task in started_tasks])
  finally:
    for task in started_tasks:
      task.cancel()
    await asyncio.gather(*started_tasks, return_exceptions=True)


asyncio.run(main())

Run from a development checkout with:

PYTHONPATH=src:. .venv/bin/python repro.py

On current main at b0180620:

started task count: 2
done after run teardown: [False, True]

On PR #7059 at 33602a56:

started task count: 2
done after run teardown: [True, True]

The first result shows that both calls started, but only the task left in active_streaming_tools['monitor'].task was reachable during teardown. The earlier task remained pending until the explicit cleanup in the sample.