fix: Port agent loader and file path resolution fixes to v1 (#6788)

Co-authored-by: GWeale <GWeale@users.noreply.github.com>

src/google/adk/cli/adk_web_server.py

@@ -827,6 +827,8 @@ class AdkWebServer:
       runner_dict: A dict of instantiated runners for each app.
   """
 
+  _allow_special_agents: bool = False
+
   def __init__(
       self,
       *,
@@ -866,6 +868,9 @@ class AdkWebServer:
 
   async def get_runner_async(self, app_name: str) -> Runner:
     """Returns the cached runner for the given app."""
+    # Rejected before any cleanup, cache lookup or .env walk, so a refused
+    # name cannot reach those paths.
+    self._reject_special_agent(app_name)
     # Handle cleanup
     if app_name in self.runners_to_clean:
       self.runners_to_clean.remove(app_name)
@@ -878,7 +883,7 @@ class AdkWebServer:
 
     # Create new runner
     envs.load_dotenv_for_agent(os.path.basename(app_name), self.agents_dir)
-    agent_or_app = self.agent_loader.load_agent(app_name)
+    agent_or_app = self._load_agent_or_app(app_name)
 
     # Instantiate extra plugins if configured
     extra_plugins_instances = self._instantiate_extra_plugins()
@@ -934,6 +939,22 @@ class AdkWebServer:
     self.runner_dict[app_name] = runner
     return runner
 
+  def _reject_special_agent(self, app_name: str) -> None:
+    """Refuses internal special agents unless they are explicitly enabled."""
+    if app_name.startswith("__") and not self._allow_special_agents:
+      raise HTTPException(
+          status_code=403,
+          detail=(
+              "Access to internal special agents is disabled in API server"
+              " mode."
+          ),
+      )
+
+  def _load_agent_or_app(self, app_name: str) -> BaseAgent | App:
+    """Loads an agent, refusing internal special agents unless enabled."""
+    self._reject_special_agent(app_name)
+    return self.agent_loader.load_agent(app_name)
+
   def _get_root_agent(self, agent_or_app: BaseAgent | App) -> BaseAgent:
     """Extract root agent from either a BaseAgent or App object."""
     if isinstance(agent_or_app, App):
@@ -1200,7 +1221,7 @@ class AdkWebServer:
     @app.get("/apps/{app_name}/app-info", response_model_exclude_none=True)
     async def get_adk_app_info(app_name: str) -> AppInfo:
       """Returns the detailed info for a given ADK app."""
-      agent_or_app = self.agent_loader.load_agent(app_name)
+      agent_or_app = self._load_agent_or_app(app_name)
       root_agent = self._get_root_agent(agent_or_app)
       if isinstance(root_agent, LlmAgent):
         return AppInfo(
@@ -1612,7 +1633,7 @@ class AdkWebServer:
       invocations = evals.convert_session_to_eval_invocations(session)
 
       # Populate the session with initial session state.
-      agent_or_app = self.agent_loader.load_agent(app_name)
+      agent_or_app = self._load_agent_or_app(app_name)
       root_agent = self._get_root_agent(agent_or_app)
       initial_session_state = create_empty_state(root_agent)
 
@@ -1759,7 +1780,7 @@ class AdkWebServer:
               status_code=400, detail=f"Eval set `{eval_set_id}` not found."
           )
 
-        agent_or_app = self.agent_loader.load_agent(app_name)
+        agent_or_app = self._load_agent_or_app(app_name)
         root_agent = self._get_root_agent(agent_or_app)
 
         eval_case_results = []
@@ -2195,7 +2216,7 @@ class AdkWebServer:
         app_name: The name of the agent/app
         dark_mode: Whether to use dark theme background color
       """
-      agent_or_app = self.agent_loader.load_agent(app_name)
+      agent_or_app = self._load_agent_or_app(app_name)
       root_agent = self._get_root_agent(agent_or_app)
 
       # Get graph with NO highlights (empty list) and specified theme
@@ -2227,7 +2248,7 @@ class AdkWebServer:
 
       function_calls = event.get_function_calls()
       function_responses = event.get_function_responses()
-      agent_or_app = self.agent_loader.load_agent(app_name)
+      agent_or_app = self._load_agent_or_app(app_name)
       root_agent = self._get_root_agent(agent_or_app)
       dot_graph = None
       if function_calls:

src/google/adk/cli/built_in_agents/utils/resolve_root_directory.py

@@ -42,33 +42,43 @@ def resolve_file_path(
     working_directory: Working directory to use as base (defaults to cwd)
 
   Returns:
-    Resolved absolute Path object
+    Resolved absolute Path object, guaranteed to be within the root directory.
+
+  Raises:
+    ValueError: If ``file_path`` resolves outside the root directory, e.g. via
+      ``..`` traversal or an absolute path pointing outside the root.
   """
   normalized_path = sanitize_generated_file_path(file_path)
   file_path_obj = Path(normalized_path)
 
-  # If already absolute, use as-is
-  if file_path_obj.is_absolute():
-    return file_path_obj
-
   # Get root directory from session state, default to "./"
   root_directory = "./"
   if session_state and "root_directory" in session_state:
     root_directory = session_state["root_directory"]
 
-  # Use the same resolution logic as the main function
   root_path_obj = Path(root_directory)
-
   if root_path_obj.is_absolute():
     resolved_root = root_path_obj
+  elif working_directory:
+    resolved_root = Path(working_directory) / root_directory
   else:
-    if working_directory:
-      resolved_root = Path(working_directory) / root_directory
-    else:
-      resolved_root = Path(os.getcwd()) / root_directory
+    resolved_root = Path(os.getcwd()) / root_directory
+  resolved_root = resolved_root.resolve()
 
-  # Resolve file path relative to root directory
-  return resolved_root / file_path_obj
+  if file_path_obj.is_absolute():
+    candidate = file_path_obj.resolve()
+  else:
+    candidate = (resolved_root / file_path_obj).resolve()
+
+  # Keep the resolved path within the root to block path-traversal escapes.
+  try:
+    candidate.relative_to(resolved_root)
+  except ValueError as exc:
+    raise ValueError(
+        f"File path {file_path!r} resolves outside the root directory"
+        f" {resolved_root}."
+    ) from exc
+  return candidate
 
 
 def resolve_file_paths(

src/google/adk/cli/fast_api.py

@@ -177,6 +177,9 @@ def get_fast_api_app(
   # initialize Agent Loader if not passed as argument
   if agent_loader is None:
     agent_loader = AgentLoader(agents_dir)
+  # Special internal agents back the dev UI only, so they stay unloadable
+  # unless the UI is being served.
+  agent_loader._allow_special_agents = web
 
   # Load services.py from agents_dir for custom service registration.
   load_services_module(agents_dir)
@@ -228,6 +231,9 @@ def get_fast_api_app(
       auto_create_session=auto_create_session,
       trigger_sources=trigger_sources,
   )
+  # The loader flag stops the import; this one turns the rejection into a 403
+  # rather than an uncaught error, and also covers a custom agent_loader.
+  adk_web_server._allow_special_agents = web
 
   # Callbacks & other optional args for when constructing the FastAPI instance
   extra_fast_api_args = {}

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

@@ -194,6 +194,11 @@ class AgentLoader(BaseAgentLoader):
     """Validate agent name to prevent arbitrary module imports."""
     # Strip the special agent prefix for validation
     if agent_name.startswith("__"):
+      if not self._allow_special_agents:
+        raise PermissionError(
+            f"Loading special internal agent {agent_name!r} is disabled in this"
+            " loader configuration."
+        )
       name_to_check = agent_name[2:]
       check_dir = os.path.abspath(SPECIAL_AGENTS_DIR)
     else:

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

@@ -28,6 +28,8 @@ from ...apps.app import App
 class BaseAgentLoader(ABC):
   """Abstract base class for agent loaders."""
 
+  _allow_special_agents: bool = False
+
   @abstractmethod
   def load_agent(self, agent_name: str) -> Union[BaseAgent, App]:
     """Loads an instance of an agent with the given name."""

tests/unittests/cli/test_adk_web_server_import_isolation.py

@@ -0,0 +1,50 @@
+# Copyright 2026 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.
+
+"""Import-isolation guard for adk_web_server.
+
+Importing ``adk_web_server`` must not eagerly pull in the Agent Builder agent
+stack. Doing so reaches ``google.adk.agents`` at import time and breaks
+downstream consumers that import ``adk_web_server`` while ``google.adk.agents``
+is still initializing.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+
+
+def test_importing_adk_web_server_does_not_import_agent_builder():
+  # Run in a fresh interpreter so the check is not polluted by modules that
+  # other tests already imported into sys.modules.
+  code = (
+      "import google.adk.cli.adk_web_server\n"
+      "import sys\n"
+      "forbidden = [\n"
+      "    'google.adk.cli.built_in_agents.agent',\n"
+      "    'google.adk.cli.built_in_agents.adk_agent_builder_assistant',\n"
+      "]\n"
+      "loaded = [name for name in forbidden if name in sys.modules]\n"
+      "assert not loaded, loaded\n"
+  )
+
+  result = subprocess.run(
+      [sys.executable, "-c", code],
+      capture_output=True,
+      text=True,
+      check=False,
+  )
+
+  assert result.returncode == 0, result.stderr

tests/unittests/cli/test_fast_api.py

@@ -653,6 +653,333 @@ bigquery_agent_analytics:
     assert getattr(runner.app, "_is_visual_builder_app", False) is True
 
 
+def _create_adk_web_server(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  """Helper to build an AdkWebServer backed by the mock service fixtures."""
+  from google.adk.cli.adk_web_server import AdkWebServer
+
+  return AdkWebServer(
+      agent_loader=mock_agent_loader,
+      session_service=mock_session_service,
+      memory_service=mock_memory_service,
+      artifact_service=mock_artifact_service,
+      credential_service=MagicMock(),
+      eval_sets_manager=mock_eval_sets_manager,
+      eval_set_results_manager=mock_eval_set_results_manager,
+      agents_dir=str(tmp_path),
+  )
+
+
+def test_get_runner_async_rejects_internal_special_agent_name(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  adk_web_server = _create_adk_web_server(
+      tmp_path,
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+  )
+
+  from fastapi import HTTPException
+
+  with pytest.raises(HTTPException) as exc_info:
+    asyncio.run(
+        adk_web_server.get_runner_async("__adk_agent_builder_assistant")
+    )
+
+  assert exc_info.value.status_code == 403
+  assert (
+      "Access to internal special agents is disabled in API server mode"
+      in exc_info.value.detail
+  )
+
+
+def test_get_runner_async_rejects_special_agent_already_in_the_cache(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  """A cached runner must not let a refused name bypass the 403."""
+  adk_web_server = _create_adk_web_server(
+      tmp_path,
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+  )
+  adk_web_server.runner_dict["__adk_agent_builder_assistant"] = MagicMock()
+
+  from fastapi import HTTPException
+
+  with pytest.raises(HTTPException) as exc_info:
+    asyncio.run(
+        adk_web_server.get_runner_async("__adk_agent_builder_assistant")
+    )
+
+  assert exc_info.value.status_code == 403
+
+
+def test_get_runner_async_accepts_internal_special_agent_name_when_enabled(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  adk_web_server = _create_adk_web_server(
+      tmp_path,
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+  )
+  adk_web_server._allow_special_agents = True
+
+  runner = asyncio.run(
+      adk_web_server.get_runner_async("__adk_agent_builder_assistant")
+  )
+
+  assert runner.app.name == "__adk_agent_builder_assistant"
+
+
+def test_app_info_rejects_internal_special_agent_name_without_web(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  client = _create_test_client(
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+      agents_dir=str(tmp_path),
+      web=False,
+  )
+
+  response = client.get("/apps/__adk_agent_builder_assistant/app-info")
+
+  assert response.status_code == 403
+  assert (
+      "Access to internal special agents is disabled in API server mode"
+      in response.json()["detail"]
+  )
+
+
+def test_app_info_allows_internal_special_agent_name_with_web(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  special_agent = LlmAgent(name="agent_builder_assistant")
+  mock_agent_loader.load_agent = lambda app_name: special_agent
+  client = _create_test_client(
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+      agents_dir=str(tmp_path),
+      web=True,
+  )
+
+  response = client.get("/apps/__adk_agent_builder_assistant/app-info")
+
+  assert response.status_code == 200
+  assert response.json()["rootAgentName"] == "agent_builder_assistant"
+
+
+def test_agent_loader_allows_special_agents_only_when_web_is_enabled(
+    tmp_path,
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  def build(web: bool) -> None:
+    _create_test_client(
+        mock_session_service,
+        mock_artifact_service,
+        mock_memory_service,
+        mock_agent_loader,
+        mock_eval_sets_manager,
+        mock_eval_set_results_manager,
+        agents_dir=str(tmp_path),
+        web=web,
+    )
+
+  build(web=False)
+  assert mock_agent_loader._allow_special_agents is False
+
+  build(web=True)
+  assert mock_agent_loader._allow_special_agents is True
+
+
+_SPECIAL_APP_NAME = "__adk_agent_builder_assistant"
+
+
+class _FlagIgnoringLoader:
+  """A caller-supplied loader that does not honour _allow_special_agents."""
+
+  def __init__(self):
+    self.requested = []
+
+  def load_agent(self, app_name):
+    self.requested.append(app_name)
+    return DummyAgent(name="agent_builder_assistant")
+
+  def list_agents(self):
+    return []
+
+
+def _create_api_server_client(loader, **overrides):
+  """Builds a TestClient over an AdkWebServer left in API server mode."""
+  from google.adk.cli.adk_web_server import AdkWebServer
+
+  kwargs = dict(
+      agent_loader=loader,
+      session_service=InMemorySessionService(),
+      memory_service=MagicMock(),
+      artifact_service=MagicMock(),
+      credential_service=MagicMock(),
+      eval_sets_manager=InMemoryEvalSetsManager(),
+      eval_set_results_manager=MagicMock(),
+      agents_dir=".",
+  )
+  kwargs.update(overrides)
+  adk_web_server = AdkWebServer(**kwargs)
+  fast_api_app = adk_web_server.get_fast_api_app(
+      setup_observer=lambda _observer, _server: None,
+      tear_down_observer=lambda _observer, _server: None,
+  )
+  return TestClient(fast_api_app)
+
+
+def test_dev_graph_rejects_internal_special_agent_name():
+  loader = _FlagIgnoringLoader()
+  client = _create_api_server_client(loader)
+
+  response = client.get(f"/dev/{_SPECIAL_APP_NAME}/graph")
+
+  assert response.status_code == 403
+  assert loader.requested == []
+
+
+def test_run_eval_rejects_internal_special_agent_name():
+  loader = _FlagIgnoringLoader()
+  eval_sets_manager = InMemoryEvalSetsManager()
+  eval_sets_manager.create_eval_set(_SPECIAL_APP_NAME, "eval_set_id")
+  client = _create_api_server_client(
+      loader, eval_sets_manager=eval_sets_manager
+  )
+
+  response = client.post(
+      f"/apps/{_SPECIAL_APP_NAME}/eval-sets/eval_set_id/run",
+      json={"evalMetrics": []},
+  )
+
+  assert response.status_code == 403
+  assert loader.requested == []
+
+
+def test_add_session_to_eval_set_rejects_internal_special_agent_name():
+  loader = _FlagIgnoringLoader()
+  session_service = InMemorySessionService()
+  asyncio.run(
+      session_service.create_session(
+          app_name=_SPECIAL_APP_NAME, user_id="user", session_id="session_id"
+      )
+  )
+  eval_sets_manager = InMemoryEvalSetsManager()
+  eval_sets_manager.create_eval_set(_SPECIAL_APP_NAME, "eval_set_id")
+  client = _create_api_server_client(
+      loader,
+      session_service=session_service,
+      eval_sets_manager=eval_sets_manager,
+  )
+
+  response = client.post(
+      f"/apps/{_SPECIAL_APP_NAME}/eval_sets/eval_set_id/add_session",
+      json={"evalId": "eval_id", "sessionId": "session_id", "userId": "user"},
+  )
+
+  assert response.status_code == 403
+  assert loader.requested == []
+
+
+def test_event_graph_rejects_internal_special_agent_name():
+  loader = _FlagIgnoringLoader()
+  session_service = AsyncMock()
+  session = Session(
+      id="session_id",
+      app_name=_SPECIAL_APP_NAME,
+      user_id="user",
+      state={},
+      events=[Event(author="dummy_agent")],
+  )
+  session_service.get_session.return_value = session
+  client = _create_api_server_client(loader, session_service=session_service)
+
+  response = client.get(
+      f"/apps/{_SPECIAL_APP_NAME}/users/user/sessions/session_id/events/"
+      f"{session.events[0].id}/graph"
+  )
+
+  assert response.status_code == 403
+  assert loader.requested == []
+
+
+def test_dev_graph_rejects_special_agent_before_the_loader_raises(tmp_path):
+  """The default loader's PermissionError must never reach the client."""
+  from google.adk.cli.utils.agent_loader import AgentLoader
+
+  client = _create_api_server_client(
+      AgentLoader(str(tmp_path)), agents_dir=str(tmp_path)
+  )
+
+  response = client.get(f"/dev/{_SPECIAL_APP_NAME}/graph")
+
+  assert response.status_code == 403
+
+
 @pytest.fixture
 def test_app(
     mock_session_service,

tests/unittests/cli/test_resolve_root_directory.py

@@ -0,0 +1,162 @@
+# Copyright 2026 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.
+
+"""Path-traversal containment tests for Agent Builder file tools."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest import mock
+
+from google.adk.cli.built_in_agents.tools.delete_files import delete_files
+from google.adk.cli.built_in_agents.tools.read_files import read_files
+from google.adk.cli.built_in_agents.tools.write_files import write_files
+from google.adk.cli.built_in_agents.utils.resolve_root_directory import resolve_file_path
+from google.adk.cli.built_in_agents.utils.resolve_root_directory import resolve_file_paths
+import pytest
+
+
+def _tool_context(root: Path) -> mock.MagicMock:
+  tool_context = mock.MagicMock()
+  tool_context._invocation_context.session.state = {"root_directory": str(root)}
+  return tool_context
+
+
+def test_resolve_file_path_allows_path_within_root(tmp_path):
+  resolved = resolve_file_path(
+      "sub/dir/file.txt", {"root_directory": str(tmp_path)}
+  )
+  assert resolved == (tmp_path / "sub" / "dir" / "file.txt").resolve()
+
+
+def test_resolve_file_path_allows_dot(tmp_path):
+  resolved = resolve_file_path(".", {"root_directory": str(tmp_path)})
+  assert resolved == tmp_path.resolve()
+
+
+def test_resolve_file_path_allows_interior_dotdot_within_root(tmp_path):
+  resolved = resolve_file_path(
+      "sub/../file.txt", {"root_directory": str(tmp_path)}
+  )
+  assert resolved == (tmp_path / "file.txt").resolve()
+
+
+def test_resolve_file_path_allows_absolute_within_root(tmp_path):
+  target = tmp_path / "nested" / "ok.txt"
+  resolved = resolve_file_path(str(target), {"root_directory": str(tmp_path)})
+  assert resolved == target.resolve()
+
+
+def test_resolve_file_path_rejects_relative_traversal(tmp_path):
+  with pytest.raises(ValueError):
+    resolve_file_path("../../escape.txt", {"root_directory": str(tmp_path)})
+
+
+def test_resolve_file_path_rejects_absolute_outside_root(tmp_path):
+  with pytest.raises(ValueError):
+    resolve_file_path("/etc/passwd", {"root_directory": str(tmp_path)})
+
+
+def test_resolve_file_paths_preserves_input_order(tmp_path):
+  state = {"root_directory": str(tmp_path)}
+
+  resolved = resolve_file_paths(["b.txt", "a.txt", "sub/c.txt"], state)
+
+  assert resolved == [
+      (tmp_path / "b.txt").resolve(),
+      (tmp_path / "a.txt").resolve(),
+      (tmp_path / "sub" / "c.txt").resolve(),
+  ]
+
+
+def test_resolve_file_paths_rejects_the_whole_batch_on_one_escape(tmp_path):
+  """One traversal attempt must fail the batch, not be silently dropped."""
+  with pytest.raises(ValueError):
+    resolve_file_paths(
+        ["ok.txt", "../escape.txt", "also_ok.txt"],
+        {"root_directory": str(tmp_path)},
+    )
+
+
+async def test_write_files_blocks_relative_traversal(
+    tmp_path, tmp_path_factory
+):
+  outside = tmp_path_factory.mktemp("outside")
+  payload = os.path.relpath(outside / "pwned.txt", tmp_path)
+
+  result = await write_files(
+      files={payload: "PWNED"}, tool_context=_tool_context(tmp_path)
+  )
+
+  assert not result["success"]
+  assert not (outside / "pwned.txt").exists()
+
+
+async def test_write_files_blocks_absolute_outside_root(
+    tmp_path, tmp_path_factory
+):
+  outside = tmp_path_factory.mktemp("outside")
+  target = outside / "abs.txt"
+
+  result = await write_files(
+      files={str(target): "PWNED"}, tool_context=_tool_context(tmp_path)
+  )
+
+  assert not result["success"]
+  assert not target.exists()
+
+
+async def test_write_files_allows_path_within_root(tmp_path):
+  result = await write_files(
+      files={"sub/ok.txt": "hello"}, tool_context=_tool_context(tmp_path)
+  )
+
+  assert result["success"]
+  assert (tmp_path / "sub" / "ok.txt").read_text() == "hello"
+
+
+async def test_read_files_blocks_relative_traversal(tmp_path, tmp_path_factory):
+  outside = tmp_path_factory.mktemp("outside")
+  secret = outside / "secret.txt"
+  secret.write_text("TOKEN=abc")
+  payload = os.path.relpath(secret, tmp_path)
+
+  result = await read_files(
+      file_paths=[payload], tool_context=_tool_context(tmp_path)
+  )
+
+  assert not result["success"]
+  assert all(
+      "TOKEN=abc" not in info.get("content", "")
+      for info in result["files"].values()
+  )
+
+
+async def test_delete_files_blocks_relative_traversal(
+    tmp_path, tmp_path_factory
+):
+  outside = tmp_path_factory.mktemp("outside")
+  victim = outside / "victim.txt"
+  victim.write_text("bye")
+  payload = os.path.relpath(victim, tmp_path)
+
+  result = await delete_files(
+      file_paths=[payload],
+      tool_context=_tool_context(tmp_path),
+      confirm_deletion=True,
+  )
+
+  assert not result["success"]
+  assert victim.exists()

tests/unittests/cli/utils/test_agent_loader.py

@@ -679,6 +679,7 @@ class TestAgentLoader:
 
         # Load the special agent
         loader = AgentLoader(str(regular_agents_dir))
+        loader._allow_special_agents = True
         agent = loader.load_agent("__helper")
 
         # Assert agent was loaded correctly
@@ -718,6 +719,7 @@ class TestAgentLoader:
 
         # Load the special agent twice
         loader = AgentLoader(str(regular_agents_dir))
+        loader._allow_special_agents = True
         agent1 = loader.load_agent("__cached_helper")
         agent2 = loader.load_agent("__cached_helper")
 
@@ -752,6 +754,7 @@ class TestAgentLoader:
         agent_loader.SPECIAL_AGENTS_DIR = str(special_agents_dir)
 
         loader = AgentLoader(str(regular_agents_dir))
+        loader._allow_special_agents = True
 
         # Try to load nonexistent special agent
         with pytest.raises(ValueError) as exc_info:
@@ -817,6 +820,7 @@ class TestAgentLoader:
 
         # Load the special agent
         loader = AgentLoader(str(regular_agents_dir))
+        loader._allow_special_agents = True
         agent = loader.load_agent("__yaml_helper")
 
         # Assert agent was loaded correctly
@@ -1006,3 +1010,20 @@ class TestAgentLoader:
       # 'subprocess' is a valid identifier but shouldn't be importable as an agent
       with pytest.raises(ValueError, match="Agent not found"):
         loader.load_agent("subprocess")
+
+  def test_validate_agent_name_rejects_special_agents_by_default(self):
+    """Special agents starting with __ are rejected by default (_allow_special_agents=False)."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+      loader = AgentLoader(temp_dir)
+      with pytest.raises(
+          PermissionError, match="Loading special internal agent"
+      ):
+        loader._validate_agent_name("__adk_agent_builder_assistant")
+
+  def test_validate_agent_name_allows_special_agents_when_enabled(self):
+    """Special agents starting with __ are allowed when _allow_special_agents=True."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+      loader = AgentLoader(temp_dir)
+      loader._allow_special_agents = True
+      # Should not raise any exception
+      loader._validate_agent_name("__adk_agent_builder_assistant")