fix: Prevent .env files from overriding existing environment variables

This change modifies `load_dotenv_for_agent` to first capture the environment variables already set in the process. After loading the `.env` file with `override=True`, it restores the values of these initially set variables, ensuring that explicitly set environment variables are not overwritten by the `.env` file. A new environment variable, `ADK_DISABLE_LOAD_DOTENV`, is also introduced to completely skip the `.env` loading process

Close #4020
Close $4018

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 852981654

src/google/adk/cli/utils/envs.py

@@ -12,12 +12,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
+import functools
 import logging
 import os
 
 from dotenv import load_dotenv
 
-logger = logging.getLogger(__file__)
+from ...utils.env_utils import is_env_enabled
+
+logger = logging.getLogger('google_adk.' + __name__)
+
+_ADK_DISABLE_LOAD_DOTENV_ENV_VAR = 'ADK_DISABLE_LOAD_DOTENV'
+
+
+@functools.lru_cache(maxsize=1)
+def _get_explicit_env_keys() -> frozenset[str]:
+  """Returns env var keys set before ADK loads any `.env` files.
+
+  This snapshot is used to preserve user-provided environment variables while
+  still allowing later `.env` files to override earlier ones via
+  `override=True`.
+  """
+  return frozenset(os.environ)
 
 
 def _walk_to_root_until_found(folder, filename) -> str:
@@ -35,7 +53,19 @@ def _walk_to_root_until_found(folder, filename) -> str:
 def load_dotenv_for_agent(
     agent_name: str, agent_parent_folder: str, filename: str = '.env'
 ):
-  """Loads the .env file for the agent module."""
+  """Loads the `.env` file for the agent module.
+
+  Explicit environment variables (present before the first `.env` load) are
+  preserved, while values loaded from `.env` may be overridden by later `.env`
+  loads.
+  """
+  if is_env_enabled(_ADK_DISABLE_LOAD_DOTENV_ENV_VAR):
+    logger.info(
+        'Skipping %s loading because %s is enabled.',
+        filename,
+        _ADK_DISABLE_LOAD_DOTENV_ENV_VAR,
+    )
+    return
 
   # Gets the folder of agent_module as starting_folder
   starting_folder = os.path.abspath(
@@ -43,7 +73,13 @@ def load_dotenv_for_agent(
   )
   dotenv_file_path = _walk_to_root_until_found(starting_folder, filename)
   if dotenv_file_path:
+    explicit_env_keys = _get_explicit_env_keys()
+    explicit_env = {
+        key: os.environ[key] for key in explicit_env_keys if key in os.environ
+    }
+
     load_dotenv(dotenv_file_path, override=True, verbose=True)
+    os.environ.update(explicit_env)
     logger.info(
         'Loaded %s file for %s at %s',
         filename,

tests/unittests/cli/utils/test_envs.py

@@ -0,0 +1,92 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Unit tests for dotenv loading utilities."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import google.adk.cli.utils.envs as envs
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _clear_explicit_env_cache() -> None:
+  envs._get_explicit_env_keys.cache_clear()
+
+
+def test_load_dotenv_for_agent_preserves_explicit_env(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+  agents_dir = tmp_path / "agents"
+  agent_dir = agents_dir / "agent1"
+  agent_dir.mkdir(parents=True)
+
+  explicit_key = "ADK_TEST_EXPLICIT_ENV"
+  from_dotenv_key = "ADK_TEST_FROM_DOTENV"
+
+  monkeypatch.setenv(explicit_key, "explicit")
+  monkeypatch.delenv(from_dotenv_key, raising=False)
+  envs._get_explicit_env_keys.cache_clear()
+
+  (agent_dir / ".env").write_text(
+      f"{explicit_key}=from_dotenv\n{from_dotenv_key}=from_dotenv\n"
+  )
+
+  envs.load_dotenv_for_agent("agent1", str(agents_dir))
+
+  assert os.environ[explicit_key] == "explicit"
+  assert os.environ[from_dotenv_key] == "from_dotenv"
+
+
+def test_load_dotenv_for_agent_overrides_previous_dotenv(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+  agents_dir = tmp_path / "agents"
+  agent1_dir = agents_dir / "agent1"
+  agent2_dir = agents_dir / "agent2"
+  agent1_dir.mkdir(parents=True)
+  agent2_dir.mkdir(parents=True)
+
+  key = "ADK_TEST_DOTENV_OVERRIDE"
+  monkeypatch.delenv(key, raising=False)
+
+  (agent1_dir / ".env").write_text(f"{key}=one\n")
+  envs.load_dotenv_for_agent("agent1", str(agents_dir))
+  assert os.environ[key] == "one"
+
+  (agent2_dir / ".env").write_text(f"{key}=two\n")
+  envs.load_dotenv_for_agent("agent2", str(agents_dir))
+  assert os.environ[key] == "two"
+
+
+def test_load_dotenv_for_agent_respects_disable_flag(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+  agents_dir = tmp_path / "agents"
+  agent_dir = agents_dir / "agent1"
+  agent_dir.mkdir(parents=True)
+
+  key = "ADK_TEST_DISABLE_DOTENV"
+  monkeypatch.delenv(key, raising=False)
+  monkeypatch.setenv("ADK_DISABLE_LOAD_DOTENV", "1")
+  envs._get_explicit_env_keys.cache_clear()
+
+  (agent_dir / ".env").write_text(f"{key}=from_dotenv\n")
+
+  envs.load_dotenv_for_agent("agent1", str(agents_dir))
+
+  assert key not in os.environ