fix: use loop.is_closed() for stale loop state validation in BigQuery plugin

The previous approach in PR google#4457 used `_queue._loop` to validate loop
identity, but `asyncio.Queue._loop` is `None` on Python 3.10+, causing
the check to always treat states as stale and break the plugin entirely.

This fix uses `loop.is_closed()` — a public, reliable API — to detect
and clean up stale loop states in `_batch_processor_prop`, `_get_loop_state`,
and `flush`.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: caohy1988

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -1643,11 +1643,23 @@ def __getattribute__(self, name: str) -> Any:
       return self._write_stream_prop
     return super().__getattribute__(name)
 
+  def _cleanup_stale_loop_states(self) -> None:
+    """Removes entries for event loops that have been closed."""
+    stale = [loop for loop in self._loop_state_by_loop if loop.is_closed()]
+    for loop in stale:
+      logger.warning(
+          "Cleaning up stale loop state for closed loop %s (id=%s).",
+          loop,
+          id(loop),
+      )
+      del self._loop_state_by_loop[loop]
+
   @property
   def _batch_processor_prop(self) -> Optional["BatchProcessor"]:
     """The batch processor for the current loop (backward compatibility)."""
     try:
       loop = asyncio.get_running_loop()
+      self._cleanup_stale_loop_states()
       if loop in self._loop_state_by_loop:
         return self._loop_state_by_loop[loop].batch_processor
     except RuntimeError:
@@ -1700,6 +1712,7 @@ async def _get_loop_state(self) -> _LoopState:
         The loop-specific state object containing clients and processors.
     """
     loop = asyncio.get_running_loop()
+    self._cleanup_stale_loop_states()
     if loop in self._loop_state_by_loop:
       return self._loop_state_by_loop[loop]
 
@@ -1771,6 +1784,7 @@ async def flush(self) -> None:
     """
     try:
       loop = asyncio.get_running_loop()
+      self._cleanup_stale_loop_states()
       if loop in self._loop_state_by_loop:
         await self._loop_state_by_loop[loop].batch_processor.flush()
     except RuntimeError:

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -2246,3 +2246,110 @@ async def test_generation_config_logging(
 
     if "labels" in gen_config_kwargs:
       assert attributes.get("labels") == gen_config_kwargs["labels"]
+
+
+class TestLoopStateValidation:
+  """Tests for loop state validation and stale loop cleanup."""
+
+  def _make_plugin(self):
+    """Creates a plugin instance without starting it."""
+    return bigquery_agent_analytics_plugin.BigQueryAgentAnalyticsPlugin(
+        project_id=PROJECT_ID,
+        dataset_id=DATASET_ID,
+        table_id=TABLE_ID,
+    )
+
+  def _make_loop_state(self):
+    """Creates a mock _LoopState with batch_processor and write_client."""
+    state = mock.MagicMock()
+    state.batch_processor = mock.MagicMock(
+        spec=bigquery_agent_analytics_plugin.BatchProcessor
+    )
+    state.batch_processor.flush = mock.AsyncMock()
+    state.write_client = mock.MagicMock()
+    return state
+
+  def test_cleanup_stale_loop_states_removes_closed_loops(self):
+    """Closed loops should be removed from _loop_state_by_loop."""
+    plugin = self._make_plugin()
+
+    closed_loop = mock.MagicMock(spec=asyncio.AbstractEventLoop)
+    closed_loop.is_closed.return_value = True
+
+    plugin._loop_state_by_loop[closed_loop] = self._make_loop_state()
+
+    plugin._cleanup_stale_loop_states()
+
+    assert closed_loop not in plugin._loop_state_by_loop
+
+  def test_cleanup_stale_loop_states_keeps_open_loops(self):
+    """Open loops should not be removed from _loop_state_by_loop."""
+    plugin = self._make_plugin()
+
+    open_loop = mock.MagicMock(spec=asyncio.AbstractEventLoop)
+    open_loop.is_closed.return_value = False
+
+    plugin._loop_state_by_loop[open_loop] = self._make_loop_state()
+
+    plugin._cleanup_stale_loop_states()
+
+    assert open_loop in plugin._loop_state_by_loop
+
+  def test_cleanup_removes_only_closed_loops(self):
+    """Only closed loops should be removed; open ones stay."""
+    plugin = self._make_plugin()
+
+    open_loop = mock.MagicMock(spec=asyncio.AbstractEventLoop)
+    open_loop.is_closed.return_value = False
+    closed_loop = mock.MagicMock(spec=asyncio.AbstractEventLoop)
+    closed_loop.is_closed.return_value = True
+
+    plugin._loop_state_by_loop[open_loop] = self._make_loop_state()
+    plugin._loop_state_by_loop[closed_loop] = self._make_loop_state()
+
+    plugin._cleanup_stale_loop_states()
+
+    assert open_loop in plugin._loop_state_by_loop
+    assert closed_loop not in plugin._loop_state_by_loop
+
+  @pytest.mark.asyncio
+  async def test_batch_processor_prop_returns_processor_for_open_loop(
+      self,
+  ):
+    """_batch_processor_prop returns processor for the current loop."""
+    plugin = self._make_plugin()
+
+    loop = asyncio.get_running_loop()
+    state = self._make_loop_state()
+    plugin._loop_state_by_loop[loop] = state
+
+    assert plugin._batch_processor_prop is state.batch_processor
+
+    # Clean up
+    del plugin._loop_state_by_loop[loop]
+
+  @pytest.mark.asyncio
+  async def test_batch_processor_prop_cleans_closed_loop_entry(self):
+    """Accessing _batch_processor_prop cleans up closed loop entries."""
+    plugin = self._make_plugin()
+
+    closed_loop = mock.MagicMock(spec=asyncio.AbstractEventLoop)
+    closed_loop.is_closed.return_value = True
+    plugin._loop_state_by_loop[closed_loop] = self._make_loop_state()
+
+    # Accessing the prop should clean up the closed loop entry
+    _ = plugin._batch_processor_prop
+    assert closed_loop not in plugin._loop_state_by_loop
+
+  @pytest.mark.asyncio
+  async def test_flush_cleans_stale_states(self):
+    """flush() should clean up stale loop states before flushing."""
+    plugin = self._make_plugin()
+
+    closed_loop = mock.MagicMock(spec=asyncio.AbstractEventLoop)
+    closed_loop.is_closed.return_value = True
+    plugin._loop_state_by_loop[closed_loop] = self._make_loop_state()
+
+    await plugin.flush()
+
+    assert closed_loop not in plugin._loop_state_by_loop