fix(agents): await cancelled tasks in _merge_agent_run_pre_3_11 to prevent aclose() RuntimeError

Merge #5416

### Link to Issue or Description of Change

Fixes #5297

### Description

On Python 3.10, `ParallelAgent` uses `_merge_agent_run_pre_3_11` instead of
`asyncio.TaskGroup`. When a sub-agent raises an exception, the `finally` block
cancelled all internal tasks with `task.cancel()` but **did not await them**.
This left the `process_an_agent` coroutines still executing their own `finally`
blocks — which hold references to the sub-agent async generators — when
`_run_async_impl` subsequently called `aclose()` on those generators, raising:

```
RuntimeError: aclose(): asynchronous generator is already running
```

This secondary error masked the original sub-agent exception (e.g., a
Pydantic validation failure from a structured-output agent).

**Fix:** add `await asyncio.gather(*tasks, return_exceptions=True)` after
cancellation so all tasks — including their generator cleanup — complete fully
before the caller can invoke `aclose()` on the generators.

### Changes

- `src/google/adk/agents/parallel_agent.py`: one line added to `_merge_agent_run_pre_3_11` finally block
- `tests/unittests/agents/test_parallel_agent.py`: regression test that directly exercises `_merge_agent_run_pre_3_11` with a slow generator + failing generator, then calls `aclose()` — without the fix this raises `RuntimeError`

### Testing Plan

- New test `test_merge_agent_run_pre_3_11_no_aclose_error_on_failure` added to `tests/unittests/agents/test_parallel_agent.py`
- All 9 existing parallel agent tests continue to pass

```
pytest tests/unittests/agents/test_parallel_agent.py -v
======================== 9 passed in 5.46s ===========================
```

Co-authored-by: Bo Yang <ybo@google.com>
COPYBARA_INTEGRATE_REVIEW=#5416 from Koushik-Salammagari:fix/parallel-agent-pre-311-aclose-error 1e11fd7
PiperOrigin-RevId: 931247148

Author: 2 people

src/google/adk/agents/parallel_agent.py

@@ -146,6 +146,7 @@ async def process_an_agent(events_for_one_agent):
   finally:
     for task in tasks:
       task.cancel()
+    await asyncio.gather(*tasks, return_exceptions=True)
 
 
 @deprecated(

tests/unittests/agents/test_parallel_agent.py

@@ -20,6 +20,7 @@
 from google.adk.agents.base_agent import BaseAgent
 from google.adk.agents.base_agent import BaseAgentState
 from google.adk.agents.invocation_context import InvocationContext
+from google.adk.agents.parallel_agent import _merge_agent_run_pre_3_11
 from google.adk.agents.parallel_agent import ParallelAgent
 from google.adk.agents.sequential_agent import SequentialAgent
 from google.adk.agents.sequential_agent import SequentialAgentState
@@ -373,3 +374,38 @@ async def test_stop_agent_if_sub_agent_fails(
     async for _ in agen:
       # The infinite agent could iterate a few times depending on scheduling.
       pass
+
+
+async def _slow_agent_with_cleanup_delay():
+  """Async generator that sleeps in its finally block to simulate cleanup."""
+  try:
+    await asyncio.sleep(10)
+    yield 'slow-event'
+  finally:
+    await asyncio.sleep(0.05)
+
+
+async def _failing_agent():
+  """Async generator that raises after a short delay."""
+  await asyncio.sleep(0.01)
+  raise ValueError('simulated sub-agent failure')
+  yield  # pragma: no cover
+
+
+@pytest.mark.asyncio
+async def test_merge_agent_run_pre_3_11_no_aclose_error_on_failure():
+  """Regression test for Python 3.10 RuntimeError: aclose() already running.
+
+  _merge_agent_run_pre_3_11 must await all cancelled tasks before returning so
+  that generators are fully released before the caller invokes aclose() on them.
+  """
+  agent_runs = [_slow_agent_with_cleanup_delay(), _failing_agent()]
+
+  with pytest.raises(ValueError, match='simulated sub-agent failure'):
+    async for _ in _merge_agent_run_pre_3_11(agent_runs):
+      pass
+
+  # If tasks were not properly awaited, aclose() on a still-running generator
+  # would raise RuntimeError here.
+  for agen in agent_runs:
+    await agen.aclose()