fix: preserve picklable credentials, drop only non-picklable ones

The previous commit cleared both _credentials and _user_credentials
unconditionally in __getstate__, silently dropping picklable user
credentials (service-account, AnonymousCredentials) after unpickle.
This caused the plugin to fall back to ADC instead of the user's
configured identity.

Now __getstate__ tests whether _user_credentials is picklable:
- Picklable: preserved — survives pickle and restored on unpickle
- Non-picklable: dropped gracefully — falls back to ADC

Adds two tests covering both paths.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: caohy1988

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -2546,12 +2546,18 @@ def __getstate__(self):
     state["_startup_error"] = None
     state["_is_shutting_down"] = False
     state["_init_pid"] = 0
-    # Credential objects may hold non-picklable transport state
-    # (e.g., requests.Session in compute_engine.Credentials).
-    # Clear both so pickle succeeds regardless of credential type.
-    # After unpickle, credentials are re-resolved via ADC.
+    # _credentials is always runtime-resolved; clear unconditionally.
     state["_credentials"] = None
-    state["_user_credentials"] = None
+    # Preserve _user_credentials if they are picklable (e.g.,
+    # service-account, AnonymousCredentials).  Drop only when
+    # pickle would fail (e.g., compute_engine.Credentials holding
+    # a requests.Session).
+    import pickle as _pickle
+
+    try:
+      _pickle.dumps(state.get("_user_credentials"))
+    except Exception:
+      state["_user_credentials"] = None
     return state
 
   def __setstate__(self, state):
@@ -2561,9 +2567,11 @@ def __setstate__(self, state):
     state.setdefault("_init_pid", 0)
     state.setdefault("_user_credentials", None)
     state.setdefault("_credentials", None)
-    # Both _credentials and _user_credentials are cleared during
-    # pickle.  Credentials will be re-resolved via ADC on the next
-    # _create_loop_state call.
+    # Restore _credentials from _user_credentials if available so
+    # _create_loop_state uses the user's identity.  When both are
+    # None (non-picklable credentials were dropped), ADC is used.
+    if state["_credentials"] is None and state["_user_credentials"]:
+      state["_credentials"] = state["_user_credentials"]
     self.__dict__.update(state)
 
   def _reset_runtime_state(self) -> None:

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -2093,6 +2093,56 @@ async def test_pickle_safety(self, mock_auth_default, mock_bq_client):
     finally:
       await plugin.shutdown()
 
+  @pytest.mark.asyncio
+  async def test_pickle_preserves_picklable_credentials(
+      self, mock_auth_default, mock_bq_client
+  ):
+    """Picklable user credentials survive pickle/unpickle."""
+    import pickle
+
+    picklable_creds = FakeCredentials()
+    plugin = bigquery_agent_analytics_plugin.BigQueryAgentAnalyticsPlugin(
+        PROJECT_ID,
+        DATASET_ID,
+        table_id=TABLE_ID,
+        credentials=picklable_creds,
+    )
+    pickled = pickle.dumps(plugin)
+    unpickled = pickle.loads(pickled)
+    # User-provided picklable credentials are preserved.
+    assert unpickled._user_credentials is not None
+    assert unpickled._credentials is not None
+    await plugin.shutdown()
+
+  @pytest.mark.asyncio
+  async def test_pickle_drops_non_picklable_credentials(
+      self, mock_auth_default, mock_bq_client
+  ):
+    """Non-picklable user credentials are dropped gracefully."""
+    import pickle
+
+    class NonPicklableCreds(google.auth.credentials.Credentials):
+
+      def refresh(self, request):
+        pass
+
+      def __getstate__(self):
+        raise TypeError("cannot pickle")
+
+    plugin = bigquery_agent_analytics_plugin.BigQueryAgentAnalyticsPlugin(
+        PROJECT_ID,
+        DATASET_ID,
+        table_id=TABLE_ID,
+        credentials=NonPicklableCreds(),
+    )
+    # Should not raise — non-picklable credentials are dropped.
+    pickled = pickle.dumps(plugin)
+    unpickled = pickle.loads(pickled)
+    # Credentials fall back to None (ADC on next use).
+    assert unpickled._user_credentials is None
+    assert unpickled._credentials is None
+    await plugin.shutdown()
+
   @pytest.mark.asyncio
   async def test_span_hierarchy_llm_call(
       self,