fix(artifacts): keep in-memory scopes separate when identifiers contain /

src/google/adk/artifacts/in_memory_artifact_service.py

@@ -20,6 +20,8 @@
 from typing import cast
 from typing import Optional
 from typing import Union
+from urllib.parse import quote
+from urllib.parse import unquote
 
 from google.genai import types
 from pydantic import BaseModel
@@ -34,6 +36,21 @@
 
 logger = logging.getLogger("google_adk." + __name__)
 
+_SESSION_NAMESPACE = "s"
+_USER_NAMESPACE = "u"
+
+
+def _encode_segment(value: str) -> str:
+  """Encodes one storage-key segment so embedded ``/`` cannot shift scopes.
+
+  Identifiers may contain ``/`` by design (the shared validator allows it), so
+  joining raw identifiers would let distinct ``(app_name, user_id)`` pairs
+  address the same stored entry. Percent-encoding removes ``/`` from every
+  segment; the leading namespace tag keeps a session id such as ``user`` from
+  colliding with the user namespace.
+  """
+  return quote(value, safe="")
+
 
 @dataclasses.dataclass
 class _ArtifactEntry:
@@ -90,14 +107,25 @@ def _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}"
+      return "/".join((
+          _USER_NAMESPACE,
+          _encode_segment(app_name),
+          _encode_segment(user_id),
+          _encode_segment(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}"
+    return "/".join((
+        _SESSION_NAMESPACE,
+        _encode_segment(app_name),
+        _encode_segment(user_id),
+        _encode_segment(session_id),
+        _encode_segment(filename),
+    ))
 
   @override
   async def save_artifact(
@@ -241,18 +269,31 @@ async def list_artifact_keys(
     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/"
+    # Encoded segments contain no '/', so a trailing-slash prefix matches
+    # exactly this scope and no other.
+    usernamespace_prefix = "/".join((
+        _USER_NAMESPACE,
+        _encode_segment(app_name),
+        _encode_segment(user_id),
+        "",
+    ))
     session_prefix = (
-        f"{app_name}/{user_id}/{session_id}/" if session_id else None
+        "/".join((
+            _SESSION_NAMESPACE,
+            _encode_segment(app_name),
+            _encode_segment(user_id),
+            _encode_segment(session_id),
+            "",
+        ))
+        if session_id
+        else None
     )
     filenames = []
     for path in self.artifacts:
       if session_prefix and path.startswith(session_prefix):
-        filename = path.removeprefix(session_prefix)
-        filenames.append(filename)
+        filenames.append(unquote(path.removeprefix(session_prefix)))
       elif path.startswith(usernamespace_prefix):
-        filename = path.removeprefix(usernamespace_prefix)
-        filenames.append(filename)
+        filenames.append(unquote(path.removeprefix(usernamespace_prefix)))
     return sorted(filenames)
 
   @override

tests/unittests/artifacts/test_artifact_service.py

@@ -4167,3 +4167,175 @@ async def test_load_artifact_survives_malformed_metadata_on_binary_payload(
   )
 
   assert loaded is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("first", "second"),
+    [
+        (("team/project", "alice"), ("team", "project/alice")),
+        (("team", "project/alice"), ("team/project", "alice")),
+    ],
+)
+async def test_in_memory_slash_in_identifiers_keeps_scopes_separate(
+    first, second
+):
+  """Distinct (app_name, user_id) pairs with embedded '/' stay isolated.
+
+  The shared validator allows '/' inside app and user identifiers, so joining
+  raw identifiers into one storage key would let either scope read, version
+  and delete the other's artifacts.
+  """
+  artifact_service = InMemoryArtifactService()
+  saved = types.Part.from_text(text="first scope value")
+  assert (
+      await artifact_service.save_artifact(
+          app_name=first[0],
+          user_id=first[1],
+          session_id="s1",
+          filename="notes.txt",
+          artifact=saved,
+      )
+      == 0
+  )
+
+  # The second scope has its own empty history.
+  assert (
+      await artifact_service.load_artifact(
+          app_name=second[0],
+          user_id=second[1],
+          session_id="s1",
+          filename="notes.txt",
+      )
+      is None
+  )
+  assert (
+      await artifact_service.list_versions(
+          app_name=second[0],
+          user_id=second[1],
+          session_id="s1",
+          filename="notes.txt",
+      )
+      == []
+  )
+  assert (
+      await artifact_service.list_artifact_keys(
+          app_name=second[0], user_id=second[1], session_id="s1"
+      )
+      == []
+  )
+
+  # Saving through the second scope allocates its own version 0.
+  assert (
+      await artifact_service.save_artifact(
+          app_name=second[0],
+          user_id=second[1],
+          session_id="s1",
+          filename="notes.txt",
+          artifact=types.Part.from_text(text="second scope value"),
+      )
+      == 0
+  )
+  loaded_first = await artifact_service.load_artifact(
+      app_name=first[0],
+      user_id=first[1],
+      session_id="s1",
+      filename="notes.txt",
+  )
+  assert loaded_first is not None
+  assert loaded_first.text == "first scope value"
+  assert await artifact_service.list_versions(
+      app_name=first[0],
+      user_id=first[1],
+      session_id="s1",
+      filename="notes.txt",
+  ) == [0]
+
+  # Deleting through the second scope leaves the first scope intact.
+  await artifact_service.delete_artifact(
+      app_name=second[0],
+      user_id=second[1],
+      session_id="s1",
+      filename="notes.txt",
+  )
+  reloaded_first = await artifact_service.load_artifact(
+      app_name=first[0],
+      user_id=first[1],
+      session_id="s1",
+      filename="notes.txt",
+  )
+  assert reloaded_first is not None
+  assert reloaded_first.text == "first scope value"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "filename", ["sub/notes.txt", "user:preferences/sub/style.json"]
+)
+async def test_in_memory_filename_with_slash_round_trips_in_listing(
+    filename,
+):
+  """A filename containing '/' keeps its exact form through save and list.
+
+  The storage key percent-encodes every segment, so a '/' inside the filename
+  cannot be mistaken for a scope boundary, and list_artifact_keys decodes the
+  filename back to the caller's original spelling.
+  """
+  artifact_service = InMemoryArtifactService()
+  assert (
+      await artifact_service.save_artifact(
+          app_name="app0",
+          user_id="user0",
+          session_id="sess0" if not filename.startswith("user:") else None,
+          filename=filename,
+          artifact=types.Part.from_text(text="value"),
+      )
+      == 0
+  )
+
+  assert await artifact_service.list_artifact_keys(
+      app_name="app0",
+      user_id="user0",
+      session_id="sess0" if not filename.startswith("user:") else None,
+  ) == [filename]
+  loaded = await artifact_service.load_artifact(
+      app_name="app0",
+      user_id="user0",
+      session_id="sess0" if not filename.startswith("user:") else None,
+      filename=filename,
+  )
+  assert loaded is not None
+  assert loaded.text == "value"
+
+
+@pytest.mark.asyncio
+async def test_in_memory_hierarchical_app_name_lists_only_its_own_files():
+  """A hierarchical app_name lists its own artifacts, scope-exact.
+
+  Regression guard for the aliasing fix: the encoded key contains no raw '/',
+  so a trailing-slash prefix for app0's scope matches exactly that scope even
+  when app0 itself has an embedded '/'.
+  """
+  artifact_service = InMemoryArtifactService()
+  for app_name, user_id, filename in [
+      ("team/project", "alice", "shared.txt"),
+      ("team", "project/alice", "other.txt"),
+      ("team/project", "bob", "bob.txt"),
+  ]:
+    await artifact_service.save_artifact(
+        app_name=app_name,
+        user_id=user_id,
+        session_id="s1",
+        filename=filename,
+        artifact=types.Part.from_text(text="value"),
+    )
+
+  assert await artifact_service.list_artifact_keys(
+      app_name="team/project", user_id="alice", session_id="s1"
+  ) == ["shared.txt"]
+  assert await artifact_service.list_artifact_keys(
+      app_name="team", user_id="project/alice", session_id="s1"
+  ) == ["other.txt"]
+  assert await artifact_service.list_artifact_keys(
+      app_name="team/project", user_id="bob", session_id="s1"
+  ) == ["bob.txt"]