fix: rollback of path segment validation in InMemoryArtifactService and GcsArtifactService

Co-authored-by: Jason Zhang <jasoncz@google.com>
PiperOrigin-RevId: 944123180

Author: jazhang00

src/google/adk/artifacts/artifact_util.py

@@ -134,32 +134,3 @@ def validate_artifact_reference_scope(
         "Session-scoped artifact references must stay within the same"
         " session scope."
     )
-
-
-def validate_path_segment(value: str, field_name: str) -> None:
-  """Rejects values that could alter the constructed path.
-
-  Args:
-    value: The caller-supplied identifier (e.g. user_id or session_id).
-    field_name: Human-readable name used in the error message.
-
-  Raises:
-    InputValidationError: If the value contains path separators, traversal
-      segments, or null bytes.
-  """
-  if not value:
-    raise input_validation_error.InputValidationError(
-        f"{field_name} must not be empty."
-    )
-  if "\x00" in value:
-    raise input_validation_error.InputValidationError(
-        f"{field_name} must not contain null bytes."
-    )
-  if "/" in value or "\\" in value:
-    raise input_validation_error.InputValidationError(
-        f"{field_name} {value!r} must not contain path separators."
-    )
-  if value in (".", "..") or ".." in value.split("/"):
-    raise input_validation_error.InputValidationError(
-        f"{field_name} {value!r} must not contain traversal segments."
-    )

src/google/adk/artifacts/file_artifact_service.py

@@ -34,7 +34,6 @@
 from pydantic import ValidationError
 from typing_extensions import override
 
-from . import artifact_util
 from ..errors.input_validation_error import InputValidationError
 from .base_artifact_service import ArtifactVersion
 from .base_artifact_service import BaseArtifactService
@@ -143,14 +142,39 @@ def _is_user_scoped(session_id: Optional[str], filename: str) -> bool:
   return session_id is None or _file_has_user_namespace(filename)
 
 
+def _validate_path_segment(value: str, field_name: str) -> None:
+  """Rejects values that could alter the constructed filesystem path.
+
+  Args:
+    value: The caller-supplied identifier (e.g. user_id or session_id).
+    field_name: Human-readable name used in the error message.
+
+  Raises:
+    InputValidationError: If the value contains path separators, traversal
+      segments, or null bytes.
+  """
+  if not value:
+    raise InputValidationError(f"{field_name} must not be empty.")
+  if "\x00" in value:
+    raise InputValidationError(f"{field_name} must not contain null bytes.")
+  if "/" in value or "\\" in value:
+    raise InputValidationError(
+        f"{field_name} {value!r} must not contain path separators."
+    )
+  if value in (".", "..") or ".." in value.split("/"):
+    raise InputValidationError(
+        f"{field_name} {value!r} must not contain traversal segments."
+    )
+
+
 def _user_artifacts_dir(base_root: Path) -> Path:
   """Returns the path that stores user-scoped artifacts."""
   return base_root / "artifacts"
 
 
 def _session_artifacts_dir(base_root: Path, session_id: str) -> Path:
   """Returns the path that stores session-scoped artifacts."""
-  artifact_util.validate_path_segment(session_id, "session_id")
+  _validate_path_segment(session_id, "session_id")
   return base_root / "sessions" / session_id / "artifacts"
 
 
@@ -232,7 +256,7 @@ def __init__(self, root_dir: Path | str):
 
   def _base_root(self, user_id: str, /) -> Path:
     """Returns the artifacts root directory for a user."""
-    artifact_util.validate_path_segment(user_id, "user_id")
+    _validate_path_segment(user_id, "user_id")
     return self.root_dir / "users" / user_id
 
   def _scope_root(
@@ -245,7 +269,7 @@ def _scope_root(
     base = self._base_root(user_id)
     if _is_user_scoped(session_id, filename):
       return _user_artifacts_dir(base)
-    if session_id is None:
+    if not session_id:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
@@ -519,7 +543,7 @@ def _list_artifact_keys_sync(
 
     base_root = self._base_root(user_id)
 
-    if session_id is not None:
+    if session_id:
       session_root = _session_artifacts_dir(base_root, session_id)
       for artifact_dir in _iter_artifact_dirs(session_root):
         metadata = self._latest_metadata(artifact_dir)

src/google/adk/artifacts/gcs_artifact_service.py

@@ -167,16 +167,13 @@ def _get_blob_prefix(
       session_id: Optional[str] = None,
   ) -> str:
     """Constructs the blob name prefix in GCS for a given artifact."""
-    artifact_util.validate_path_segment(app_name, "app_name")
-    artifact_util.validate_path_segment(user_id, "user_id")
     if self._file_has_user_namespace(filename):
       return f"{app_name}/{user_id}/user/{filename}"
 
     if session_id is None:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
-    artifact_util.validate_path_segment(session_id, "session_id")
     return f"{app_name}/{user_id}/{session_id}/{filename}"
 
   def _get_blob_name(
@@ -371,10 +368,6 @@ def _load_artifact(
   def _list_artifact_keys(
       self, app_name: str, user_id: str, session_id: Optional[str]
   ) -> list[str]:
-    artifact_util.validate_path_segment(app_name, "app_name")
-    artifact_util.validate_path_segment(user_id, "user_id")
-    if session_id is not None:
-      artifact_util.validate_path_segment(session_id, "session_id")
     filenames = set()
 
     if session_id:

src/google/adk/artifacts/in_memory_artifact_service.py

@@ -85,16 +85,13 @@ def _artifact_path(
     Returns:
         The constructed artifact path.
     """
-    artifact_util.validate_path_segment(app_name, "app_name")
-    artifact_util.validate_path_segment(user_id, "user_id")
     if self._file_has_user_namespace(filename):
       return f"{app_name}/{user_id}/user/{filename}"
 
     if session_id is None:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
-    artifact_util.validate_path_segment(session_id, "session_id")
     return f"{app_name}/{user_id}/{session_id}/{filename}"
 
   @override
@@ -218,10 +215,6 @@ async def load_artifact(
   async def list_artifact_keys(
       self, *, app_name: str, user_id: str, session_id: Optional[str] = None
   ) -> list[str]:
-    artifact_util.validate_path_segment(app_name, "app_name")
-    artifact_util.validate_path_segment(user_id, "user_id")
-    if session_id is not None:
-      artifact_util.validate_path_segment(session_id, "session_id")
     usernamespace_prefix = f"{app_name}/{user_id}/user/"
     session_prefix = (
         f"{app_name}/{user_id}/{session_id}/" if session_id else None