fix: Windows Path Handling and Normalize Cross-Platform Path Resolution in AgentLoader

Merge https://github.com/google/adk-python/pull/3609

Co-authored-by: George Weale <gweale@google.com>
COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/3609 from p-kris10:fix/windows-cmd 8cb0310bd4450097a0a7714eaac6521b6a447442
PiperOrigin-RevId: 836395714
This commit is contained in:
Kristen Pereira
2025-11-24 16:06:24 -08:00
committed by Copybara-Service
parent cf21ca3584
commit a1c09b724b
4 changed files with 143 additions and 8 deletions
+1 -1
View File
@@ -132,7 +132,7 @@ def resolve_agent_reference(
else:
return from_config(
os.path.join(
referencing_agent_config_abs_path.rsplit("/", 1)[0],
os.path.dirname(referencing_agent_config_abs_path),
ref_config.config_path,
)
)
+8 -7
View File
@@ -58,7 +58,7 @@ class AgentLoader(BaseAgentLoader):
"""
def __init__(self, agents_dir: str):
self.agents_dir = agents_dir.rstrip("/")
self.agents_dir = str(Path(agents_dir))
self._original_sys_path = None
self._agent_cache: dict[str, Union[BaseAgent, App]] = {}
@@ -272,12 +272,13 @@ class AgentLoader(BaseAgentLoader):
f"No root_agent found for '{agent_name}'. Searched in"
f" '{actual_agent_name}.agent.root_agent',"
f" '{actual_agent_name}.root_agent' and"
f" '{actual_agent_name}/root_agent.yaml'.\n\nExpected directory"
f" structure:\n <agents_dir>/\n {actual_agent_name}/\n "
" agent.py (with root_agent) OR\n root_agent.yaml\n\nThen run:"
f" adk web <agents_dir>\n\nEnsure '{agents_dir}/{actual_agent_name}' is"
" structured correctly, an .env file can be loaded if present, and a"
f" root_agent is exposed.{hint}"
f" '{actual_agent_name}{os.sep}root_agent.yaml'.\n\nExpected directory"
f" structure:\n <agents_dir>{os.sep}\n "
f" {actual_agent_name}{os.sep}\n agent.py (with root_agent) OR\n "
" root_agent.yaml\n\nThen run: adk web <agents_dir>\n\nEnsure"
f" '{os.path.join(agents_dir, actual_agent_name)}' is structured"
" correctly, an .env file can be loaded if present, and a root_agent"
f" is exposed.{hint}"
)
def _record_origin_metadata(
@@ -12,14 +12,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import ntpath
import os
from pathlib import Path
from textwrap import dedent
from typing import Literal
from typing import Type
from unittest import mock
from google.adk.agents import config_agent_utils
from google.adk.agents.agent_config import AgentConfig
from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.base_agent_config import BaseAgentConfig
from google.adk.agents.common_configs import AgentRefConfig
from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.loop_agent import LoopAgent
from google.adk.agents.parallel_agent import ParallelAgent
@@ -280,3 +285,91 @@ other_field: other value
config.root.model_dump()
)
assert my_custom_config.other_field == "other value"
@pytest.mark.parametrize(
("config_rel_path", "child_rel_path", "child_name", "instruction"),
[
(
Path("main.yaml"),
Path("sub_agents/child.yaml"),
"child_agent",
"I am a child agent",
),
(
Path("level1/level2/nested_main.yaml"),
Path("sub/nested_child.yaml"),
"nested_child",
"I am nested",
),
],
)
def test_resolve_agent_reference_resolves_relative_paths(
config_rel_path: Path,
child_rel_path: Path,
child_name: str,
instruction: str,
tmp_path: Path,
):
"""Verify resolve_agent_reference resolves relative sub-agent paths."""
config_file = tmp_path / config_rel_path
config_file.parent.mkdir(parents=True, exist_ok=True)
child_config_path = config_file.parent / child_rel_path
child_config_path.parent.mkdir(parents=True, exist_ok=True)
child_config_path.write_text(dedent(f"""
agent_class: LlmAgent
name: {child_name}
model: gemini-2.0-flash
instruction: {instruction}
""").lstrip())
config_file.write_text(dedent(f"""
agent_class: LlmAgent
name: main_agent
model: gemini-2.0-flash
instruction: I am the main agent
sub_agents:
- config_path: {child_rel_path.as_posix()}
""").lstrip())
ref_config = AgentRefConfig(config_path=child_rel_path.as_posix())
agent = config_agent_utils.resolve_agent_reference(
ref_config, str(config_file)
)
assert agent.name == child_name
config_dir = os.path.dirname(str(config_file.resolve()))
assert config_dir == str(config_file.parent.resolve())
expected_child_path = os.path.join(config_dir, *child_rel_path.parts)
assert os.path.exists(expected_child_path)
def test_resolve_agent_reference_uses_windows_dirname():
"""Ensure Windows-style config references resolve via os.path.dirname."""
ref_config = AgentRefConfig(config_path="sub\\child.yaml")
recorded: dict[str, str] = {}
def fake_from_config(path: str):
recorded["path"] = path
return "sentinel"
with (
mock.patch.object(
config_agent_utils,
"from_config",
autospec=True,
side_effect=fake_from_config,
),
mock.patch.object(config_agent_utils.os, "path", ntpath),
):
referencing = r"C:\workspace\agents\main.yaml"
result = config_agent_utils.resolve_agent_reference(ref_config, referencing)
expected_path = ntpath.join(
ntpath.dirname(referencing), ref_config.config_path
)
assert result == "sentinel"
assert recorded["path"] == expected_path
@@ -12,12 +12,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import ntpath
import os
from pathlib import Path
from pathlib import PureWindowsPath
import re
import sys
import tempfile
from textwrap import dedent
from google.adk.cli.utils import agent_loader as agent_loader_module
from google.adk.cli.utils.agent_loader import AgentLoader
from pydantic import ValidationError
import pytest
@@ -280,6 +284,43 @@ class TestAgentLoader:
assert agent2 is not agent3
assert agent1.agent_id != agent2.agent_id != agent3.agent_id
def test_error_messages_use_os_sep_consistently(self):
"""Verify error messages use os.sep instead of hardcoded '/'."""
del self
with tempfile.TemporaryDirectory() as temp_dir:
loader = AgentLoader(temp_dir)
agent_name = "missing_agent"
expected_path = os.path.join(temp_dir, agent_name)
with pytest.raises(ValueError) as exc_info:
loader.load_agent(agent_name)
exc_info.match(re.escape(expected_path))
exc_info.match(re.escape(f"{agent_name}{os.sep}root_agent.yaml"))
exc_info.match(re.escape(f"<agents_dir>{os.sep}"))
def test_agent_loader_with_mocked_windows_path(self, monkeypatch):
"""Mock Path() to simulate Windows behavior and catch regressions.
REGRESSION TEST: Fails with rstrip('/'), passes with str(Path()).
"""
del self
windows_path = "C:\\Users\\dev\\agents\\"
with monkeypatch.context() as m:
m.setattr(
agent_loader_module,
"Path",
lambda path_str: PureWindowsPath(path_str),
)
loader = AgentLoader(windows_path)
expected = str(PureWindowsPath(windows_path))
assert loader.agents_dir == expected
assert not loader.agents_dir.endswith("\\")
assert not loader.agents_dir.endswith("/")
def test_agent_not_found_error(self):
"""Test that appropriate error is raised when agent is not found."""
with tempfile.TemporaryDirectory() as temp_dir: