fix(artifacts): prevent filenames from colliding with version storage

src/google/adk/artifacts/file_artifact_service.py

@@ -473,7 +473,9 @@ async def save_artifact(
     (``"images/photo.png"``), or explicitly user-scoped
     (``"user:shared/diagram.png"``). All values are interpreted relative to the
     computed scope root; absolute paths or inputs that traverse outside that
-    root (for example ``"../../secret.txt"``) raise ``ValueError``.
+    root (for example ``"../../secret.txt"``) raise ``ValueError``. The final
+    name ``metadata.json`` and any ``versions`` path component are reserved for
+    the service's storage layout and are rejected in any casing.
     """
     return await asyncio.to_thread(
         self._save_artifact_sync,
@@ -512,6 +514,17 @@ def _save_artifact_sync(
           " is stored under the artifact's own name and would overwrite the"
           " metadata document."
       )
+    if any(
+        part.casefold() == "versions"
+        for part in _to_posix_path(
+            _strip_user_namespace(filename).strip()
+        ).parts
+    ):
+      raise InputValidationError(
+          f"Artifact filename {filename!r} is reserved: an artifact path may"
+          " not contain a 'versions' component (in any casing) because that"
+          " directory stores artifact versions."
+      )
     artifact_dir.mkdir(parents=True, exist_ok=True)
 
     next_version, staging_dir, version_dir = _reserve_version_dir(artifact_dir)

tests/unittests/artifacts/test_artifact_service.py

@@ -3247,6 +3247,105 @@ async def test_save_artifact_rejects_reserved_metadata_filename(
     )
 
 
+@pytest.mark.parametrize(
+    ("filename", "session_id"),
+    [
+        ("versions/report.txt", "session"),
+        ("nested/VeRsIoNs/report.txt", "session"),
+        ("nested/report.txt/versions", "session"),
+        (r"nested\versions\report.txt", "session"),
+        ("user:shared/versions/report.txt", "session"),
+        ("shared/versions/report.txt", None),
+    ],
+)
+@pytest.mark.asyncio
+async def test_file_save_rejects_reserved_versions_path_without_writing(
+    tmp_path, filename, session_id
+):
+  """A reserved versions component is rejected before disk mutation."""
+  root = tmp_path / "artifacts"
+  service = FileArtifactService(root_dir=root)
+  before = list(root.rglob("*"))
+
+  with pytest.raises(InputValidationError, match="versions"):
+    await service.save_artifact(
+        app_name="app",
+        user_id="user",
+        session_id=session_id,
+        filename=filename,
+        artifact=types.Part(text="payload"),
+    )
+
+  assert list(root.rglob("*")) == before
+
+
+@pytest.mark.parametrize(
+    "filename",
+    ["nested/releases/report.txt", "nested/versions.txt", "reversions/file"],
+)
+@pytest.mark.asyncio
+async def test_file_save_allows_nonreserved_nested_paths(tmp_path, filename):
+  """Nested names without an exact versions component remain valid."""
+  service = FileArtifactService(root_dir=tmp_path / "versions")
+
+  version = await service.save_artifact(
+      app_name="versions",
+      user_id="versions",
+      session_id="versions",
+      filename=filename,
+      artifact=types.Part(text="payload"),
+  )
+
+  assert version == 0
+  assert await service.load_artifact(
+      app_name="versions",
+      user_id="versions",
+      session_id="versions",
+      filename=filename,
+  ) == types.Part(text="payload")
+
+
+@pytest.mark.asyncio
+async def test_reserved_versions_path_stays_readable_and_deletable(tmp_path):
+  """Legacy artifacts using a versions component remain accessible."""
+  service = FileArtifactService(root_dir=tmp_path)
+  filename = "project/versions/readme.txt"
+  artifact_dir = service._artifact_dir(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename=filename,
+  )
+  version_dir = artifact_dir / "versions" / "0"
+  version_dir.mkdir(parents=True)
+  (version_dir / "readme.txt").write_text("legacy", encoding="utf-8")
+  file_artifact_service._write_metadata(
+      version_dir / "metadata.json",
+      filename=filename,
+      mime_type=None,
+      version=0,
+      canonical_uri=(version_dir / "readme.txt").as_uri(),
+      custom_metadata=None,
+      display_name=None,
+  )
+
+  loaded = await service.load_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename=filename,
+  )
+  await service.delete_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename=filename,
+  )
+
+  assert loaded == types.Part(text="legacy")
+  assert not artifact_dir.exists()
+
+
 @pytest.mark.asyncio
 async def test_reserved_metadata_filename_stays_deletable(tmp_path):
   """A name rejected on write must still be removable.
@@ -3425,15 +3524,18 @@ async def test_list_artifact_keys_survives_metadata_path_shadowed_by_dir(
 ):
   """A directory where a metadata document is expected must not raise."""
   service = FileArtifactService(root_dir=tmp_path)
-  # Creates `<user scope>/a/versions/0/metadata.json` as a *directory*, which
-  # made every subsequent listing for this user fail with IsADirectoryError.
-  await service.save_artifact(
-      app_name="app",
-      user_id="user",
-      session_id="session",
-      filename="user:a/versions/0/metadata.json/payload.txt",
-      artifact=types.Part(text="x"),
+  version_dir = (
+      tmp_path
+      / "apps"
+      / "app"
+      / "users"
+      / "user"
+      / "artifacts"
+      / "a"
+      / "versions"
+      / "0"
   )
+  (version_dir / "metadata.json").mkdir(parents=True)
 
   keys = await service.list_artifact_keys(
       app_name="app", user_id="user", session_id="session"

tests/unittests/cli/test_fast_api.py

@@ -33,6 +33,7 @@
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.agents.run_config import RunConfig
 from google.adk.artifacts.base_artifact_service import ArtifactVersion
+from google.adk.artifacts.file_artifact_service import FileArtifactService
 from google.adk.cli import fast_api as fast_api_module
 from google.adk.cli.fast_api import get_fast_api_app
 from google.adk.errors.input_validation_error import InputValidationError
@@ -2400,6 +2401,51 @@ def test_save_artifact_returns_400_on_validation_error(
   assert response.json()["detail"] == "invalid artifact"
 
 
+def test_file_artifact_save_rejects_reserved_versions_path(
+    tmp_path,
+    test_session_info,
+    mock_session_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  """The HTTP API surfaces file storage path collisions as HTTP 400."""
+  service = FileArtifactService(root_dir=tmp_path / "artifacts")
+  client = _create_test_client(
+      mock_session_service,
+      service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+  )
+  info = test_session_info
+  url = (
+      f"/apps/{info['app_name']}/users/{info['user_id']}/sessions/"
+      f"{info['session_id']}/artifacts"
+  )
+
+  rejected = client.post(
+      url,
+      json={
+          "filename": "project/versions/report.txt",
+          "artifact": {"text": "x"},
+      },
+  )
+  accepted = client.post(
+      url,
+      json={
+          "filename": "project/releases/report.txt",
+          "artifact": {"text": "x"},
+      },
+  )
+
+  assert rejected.status_code == 400
+  assert "versions" in rejected.json()["detail"]
+  assert accepted.status_code == 200
+
+
 def test_save_artifact_returns_500_on_unexpected_error(
     test_app, create_test_session, mock_artifact_service
 ):