fix: clear both _credentials and _user_credentials on pickle

Addresses review feedback:

- Non-picklable user-provided credentials (e.g., compute_engine with
  requests.Session) broke pickle.dumps() because _user_credentials
  was preserved in __getstate__.  Now both _credentials and
  _user_credentials are cleared; credentials re-resolve via ADC
  after unpickle.

- Fork safety: _reset_runtime_state documents that user-provided
  credentials are kept as-is (we cannot re-create them), while
  ADC-resolved credentials are cleared for re-resolution.

- GCS client: always pass credentials explicitly when available;
  GCSOffloader.__init__ always creates a client eagerly (the "lazy
  restoration" claim in the previous commit was incorrect — the
  pre-credentials code had the same eager behavior).

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

Author: caohy1988

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -2201,17 +2201,18 @@ async def _lazy_setup(self, **kwargs) -> None:
 
       self.offloader = None
       if self.config.gcs_bucket_name:
-        gcs_client = None
+        # GCSOffloader always creates a storage.Client eagerly
+        # (line 1329: storage_client or storage.Client(...)).
+        # Pass credentials so it uses the same auth as the other
+        # clients; omit when None to let it use ADC.
+        gcs_kwargs = {"project": self.project_id}
         if self._credentials is not None:
-          gcs_client = storage.Client(
-              project=self.project_id,
-              credentials=self._credentials,
-          )
+          gcs_kwargs["credentials"] = self._credentials
         self.offloader = GCSOffloader(
             self.project_id,
             self.config.gcs_bucket_name,
             self._executor,
-            storage_client=gcs_client,
+            storage_client=storage.Client(**gcs_kwargs),
         )
 
       self.parser = HybridContentParser(
@@ -2547,8 +2548,10 @@ def __getstate__(self):
     state["_init_pid"] = 0
     # Credential objects may hold non-picklable transport state
     # (e.g., requests.Session in compute_engine.Credentials).
-    # Clear and re-resolve from _user_credentials on unpickle.
+    # Clear both so pickle succeeds regardless of credential type.
+    # After unpickle, credentials are re-resolved via ADC.
     state["_credentials"] = None
+    state["_user_credentials"] = None
     return state
 
   def __setstate__(self, state):
@@ -2557,11 +2560,10 @@ def __setstate__(self, state):
     # code versions so _ensure_started does not raise AttributeError.
     state.setdefault("_init_pid", 0)
     state.setdefault("_user_credentials", None)
-    # Restore _credentials from _user_credentials (if user provided
-    # them); ADC-resolved credentials will be re-resolved on next
+    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.
-    if state.get("_credentials") is None:
-      state["_credentials"] = state.get("_user_credentials")
     self.__dict__.update(state)
 
   def _reset_runtime_state(self) -> None:
@@ -2616,8 +2618,10 @@ def _reset_runtime_state(self) -> None:
     self._startup_error = None
     self._is_shutting_down = False
     self._init_pid = os.getpid()
-    # Credentials may hold stale HTTP transport state after fork.
-    # Re-resolve from _user_credentials on next _create_loop_state.
+    # For ADC-resolved credentials, clear so they are re-resolved
+    # in the child process.  For user-provided credentials, keep
+    # the original object — we cannot re-create it.  The user is
+    # responsible for providing fork-safe credentials if needed.
     self._credentials = self._user_credentials
 
   async def __aenter__(self) -> BigQueryAgentAnalyticsPlugin: