diff --git a/src/google/adk/artifacts/file_artifact_service.py b/src/google/adk/artifacts/file_artifact_service.py
index e4a07be7e5..948a356d14 100644
--- a/src/google/adk/artifacts/file_artifact_service.py
+++ b/src/google/adk/artifacts/file_artifact_service.py
@@ -20,11 +20,10 @@
 from pathlib import PurePosixPath
 from pathlib import PureWindowsPath
 import shutil
+import tempfile
 from typing import Any
 from typing import Optional
 from typing import Union
-from urllib.parse import unquote
-from urllib.parse import urlparse
 
 from google.genai import types
 from pydantic import alias_generators
@@ -55,16 +54,91 @@ def _iter_artifact_dirs(root: Path) -> list[Path]:
   return artifact_dirs
 
 
-def _file_uri_to_path(uri: str) -> Optional[Path]:
-  """Converts a file:// URI to a filesystem path."""
-  parsed = urlparse(uri)
-  if parsed.scheme != "file":
+def _read_bytes_if_present(path: Path) -> Optional[bytes]:
+  """Reads a binary payload from disk.
+
+  The read is attempted directly instead of being guarded by an `exists()`
+  check so that a concurrent delete cannot be observed as a distinguishable
+  state between the check and the read.
+
+  Args:
+    path: Location of the payload.
+
+  Returns:
+    The file contents, or None if it is not a readable file.
+  """
+  try:
+    return path.read_bytes()
+  except FileNotFoundError:
+    return None
+  except OSError as exc:
+    logger.warning("Unreadable artifact payload at %s: %s", path, exc)
+    return None
+
+
+def _read_text_if_present(path: Path) -> Optional[str]:
+  """Reads a UTF-8 text payload from disk.
+
+  Args:
+    path: Location of the payload.
+
+  Returns:
+    The decoded file contents, or None if it is not a readable file.
+  """
+  try:
+    return path.read_text(encoding="utf-8")
+  except FileNotFoundError:
     return None
-  return Path(unquote(parsed.path))
+  except OSError as exc:
+    logger.warning("Unreadable artifact payload at %s: %s", path, exc)
+    return None
+
+
+def _umask_derived_file_mode() -> int:
+  """Returns the mode a normally created file would get from the umask.
 
+  Sampled once at import: reading the umask requires temporarily setting it,
+  which is process-global and would race against concurrent writers if done
+  per-write.
+
+  Returns:
+    The permission bits `open()` would produce for a new file.
+  """
+  umask = os.umask(0)
+  os.umask(umask)
+  return 0o666 & ~umask
+
+
+# Payloads are written through `open()`, which applies the umask, but the
+# metadata document is written through `tempfile.mkstemp`, which hardcodes
+# 0600. Without this the two files in a version directory end up readable by
+# different sets of principals.
+_DEFAULT_FILE_MODE = _umask_derived_file_mode()
 
 _USER_NAMESPACE_PREFIX = "user:"
 
+# Name of the per-version metadata document. A payload is stored alongside it
+# under the artifact directory's own name, so an artifact whose directory is
+# named `metadata.json` would have its payload written over the metadata
+# document. Callers may not use the name for that reason.
+_METADATA_FILENAME = "metadata.json"
+
+
+def _is_reserved_artifact_name(name: str) -> bool:
+  """Checks whether an artifact directory name collides with the metadata doc.
+
+  Compared caselessly because the collision is decided by the filesystem, and
+  the case-insensitive ones ADK supports (APFS, NTFS) resolve `Metadata.json`
+  and `metadata.json` to the same file.
+
+  Args:
+    name: The final path segment of the artifact directory.
+
+  Returns:
+    True if the name is reserved for internal use.
+  """
+  return name.casefold() == _METADATA_FILENAME.casefold()
+
 
 def _file_has_user_namespace(filename: str) -> bool:
   """Checks whether the file is scoped to the user namespace."""
@@ -210,7 +284,13 @@ def _versions_dir(artifact_dir: Path) -> Path:
 
 def _metadata_path(artifact_dir: Path, version: int) -> Path:
   """Returns the path to the metadata file for a specific version."""
-  return _versions_dir(artifact_dir) / str(version) / "metadata.json"
+  return _versions_dir(artifact_dir) / str(version) / _METADATA_FILENAME
+
+
+def _canonical_uri(artifact_dir: Path, version: int) -> str:
+  """Builds the canonical file:// URI for an artifact payload."""
+  payload_path = _versions_dir(artifact_dir) / str(version) / artifact_dir.name
+  return payload_path.resolve().as_uri()
 
 
 def _list_versions_on_disk(artifact_dir: Path) -> list[int]:
@@ -246,18 +326,25 @@ class FileArtifactService(BaseArtifactService):
 
   # Storage layout matches the cloud and in-memory services:
   # root/
-  # └── users/
-  #     └── {user_id}/
-  #         ├── sessions/
-  #         │   └── {session_id}/
-  #         │       └── artifacts/
-  #         │           └── {artifact_path}/  # derived from filename
-  #         │               └── versions/
-  #         │                   └── {version}/
-  #         │                       ├── {original_filename}
-  #         │                       └── metadata.json
-  #         └── artifacts/
-  #             └── {artifact_path}/...
+  # └── apps/
+  #     └── {app_name}/
+  #         └── users/
+  #             └── {user_id}/
+  #                 ├── sessions/
+  #                 │   └── {session_id}/
+  #                 │       └── artifacts/
+  #                 │           └── {artifact_path}/  # from filename
+  #                 │               └── versions/
+  #                 │                   └── {version}/
+  #                 │                       ├── {original_filename}
+  #                 │                       └── metadata.json
+  #                 └── artifacts/
+  #                     └── {artifact_path}/...
+  #
+  # Releases that predate the `apps/{app_name}` level wrote the same tree
+  # directly under `root/users`, which records no app name. A root can be
+  # shared by several apps, so that tree cannot be attributed to one of them
+  # and is never read from, written to or deleted.
   #
   # Artifact paths are derived from the provided filenames: separators create
   # nested directories, and path traversal is rejected to keep the layout
@@ -273,62 +360,60 @@ def __init__(self, root_dir: Path | str):
     self.root_dir = Path(root_dir).expanduser().resolve()
     self.root_dir.mkdir(parents=True, exist_ok=True)
 
-  def _base_root(self, user_id: str, /) -> Path:
-    """Returns the artifacts root directory for a user."""
+  def _base_root(self, app_name: str, user_id: str) -> Path:
+    """Returns the app-scoped root holding a user's artifacts."""
+    # An app name is allowed to nest. Vertex-managed sessions address an app by
+    # its full `projects/{p}/locations/{l}/reasoningEngines/{id}` resource name,
+    # so rejecting interior separators here would make every artifact call from
+    # those deployments fail. The shared validator still rejects a leading
+    # separator, traversal segments, null bytes and drive-qualified values, so
+    # the name cannot escape the root. `user_id` keeps the stricter rule below,
+    # which admits no separators at all.
+    artifact_util.validate_path_segment(app_name, "app_name")
     _validate_path_segment(user_id, "user_id")
-    return self.root_dir / "users" / user_id
+    return self.root_dir / "apps" / app_name / "users" / user_id
 
   def _scope_root(
       self,
-      user_id: str,
+      base_root: Path,
       session_id: Optional[str],
       filename: str,
   ) -> Path:
     """Returns the directory that represents the artifact scope."""
-    base = self._base_root(user_id)
     if _is_user_scoped(session_id, filename):
-      return _user_artifacts_dir(base)
+      return _user_artifacts_dir(base_root)
     if session_id is None:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
-    return _session_artifacts_dir(base, session_id)
+    return _session_artifacts_dir(base_root, session_id)
 
   def _artifact_dir(
       self,
+      app_name: str,
       user_id: str,
       session_id: Optional[str],
       filename: str,
   ) -> Path:
-    """Builds the directory path for an artifact."""
-    scope_root = self._scope_root(
-        user_id=user_id,
-        session_id=session_id,
-        filename=filename,
-    )
-    artifact_dir, _ = _resolve_scoped_artifact_path(scope_root, filename)
-    return artifact_dir
+    """Builds the directory that stores an artifact for an app."""
+    base_root = self._base_root(app_name, user_id)
+    return _resolve_scoped_artifact_path(
+        self._scope_root(base_root, session_id, filename), filename
+    )[0]
 
   def _build_artifact_version(
       self,
       *,
-      user_id: str,
-      session_id: Optional[str],
-      filename: str,
+      artifact_dir: Path,
       version: int,
       metadata: Optional[FileArtifactVersion],
   ) -> ArtifactVersion:
     """Creates an ArtifactVersion payload using on-disk metadata."""
-    canonical_uri = (
-        metadata.canonical_uri
-        if metadata and metadata.canonical_uri
-        else self._canonical_uri(
-            user_id=user_id,
-            session_id=session_id,
-            filename=filename,
-            version=version,
-        )
-    )
+    # Always recomputed from the storage layout rather than read back from the
+    # metadata document. For this service the two are equivalent for data this
+    # service wrote, and recomputing means a tampered document cannot dictate
+    # the URI handed to callers.
+    canonical_uri = _canonical_uri(artifact_dir, version)
     custom_metadata_val = metadata.custom_metadata if metadata else {}
     mime_type = metadata.mime_type if metadata else None
     return ArtifactVersion(
@@ -338,24 +423,6 @@ def _build_artifact_version(
         mime_type=mime_type,
     )
 
-  def _canonical_uri(
-      self,
-      *,
-      user_id: str,
-      session_id: Optional[str],
-      filename: str,
-      version: int,
-  ) -> str:
-    """Builds the canonical file:// URI for an artifact payload."""
-    artifact_dir = self._artifact_dir(
-        user_id=user_id,
-        session_id=session_id,
-        filename=filename,
-    )
-    stored_filename = artifact_dir.name
-    payload_path = _versions_dir(artifact_dir) / str(version) / stored_filename
-    return payload_path.resolve().as_uri()
-
   def _latest_metadata(
       self, artifact_dir: Path
   ) -> Optional[FileArtifactVersion]:
@@ -386,6 +453,7 @@ async def save_artifact(
     """
     return await asyncio.to_thread(
         self._save_artifact_sync,
+        app_name,
         user_id,
         filename,
         artifact,
@@ -395,6 +463,7 @@ async def save_artifact(
 
   def _save_artifact_sync(
       self,
+      app_name: str,
       user_id: str,
       filename: str,
       artifact: Union[types.Part, dict[str, Any]],
@@ -404,10 +473,21 @@ def _save_artifact_sync(
     """Saves an artifact to disk and returns its version."""
     artifact = ensure_part(artifact)
     artifact_dir = self._artifact_dir(
+        app_name=app_name,
         user_id=user_id,
         session_id=session_id,
         filename=filename,
     )
+    # Enforced here rather than in `_artifact_dir`, which reads and deletes
+    # share: an artifact stored under this name before the name was rejected
+    # must stay readable and, above all, deletable.
+    if _is_reserved_artifact_name(artifact_dir.name):
+      raise InputValidationError(
+          f"Artifact filename {filename!r} is reserved: an artifact may not be"
+          f" named {_METADATA_FILENAME!r} (in any casing) because its payload"
+          " is stored under the artifact's own name and would overwrite the"
+          " metadata document."
+      )
     artifact_dir.mkdir(parents=True, exist_ok=True)
 
     versions = _list_versions_on_disk(artifact_dir)
@@ -420,35 +500,38 @@ def _save_artifact_sync(
     stored_filename = artifact_dir.name
     content_path = version_dir / stored_filename
 
-    if artifact.inline_data:
-      content_path.write_bytes(artifact.inline_data.data)
-      mime_type = (
-          artifact.inline_data.mime_type
-          if artifact.inline_data.mime_type
-          else "application/octet-stream"
-      )
-    elif artifact.text is not None:
-      content_path.write_text(artifact.text, encoding="utf-8")
-      mime_type = None
-    else:
-      raise InputValidationError(
-          "Artifact must have either inline_data or text content."
-      )
+    # A version directory is only ever observed complete or not at all. A
+    # partially written version -- payload present, metadata missing or
+    # truncated -- is indistinguishable from a valid one on the read path, so
+    # any failure discards the whole directory instead of leaving it behind.
+    try:
+      if artifact.inline_data:
+        content_path.write_bytes(artifact.inline_data.data)
+        mime_type = (
+            artifact.inline_data.mime_type
+            if artifact.inline_data.mime_type
+            else "application/octet-stream"
+        )
+      elif artifact.text is not None:
+        content_path.write_text(artifact.text, encoding="utf-8")
+        mime_type = None
+      else:
+        raise InputValidationError(
+            "Artifact must have either inline_data or text content."
+        )
 
-    canonical_uri = self._canonical_uri(
-        user_id=user_id,
-        session_id=session_id,
-        filename=filename,
-        version=next_version,
-    )
-    _write_metadata(
-        version_dir / "metadata.json",
-        filename=filename,
-        mime_type=mime_type,
-        version=next_version,
-        canonical_uri=canonical_uri,
-        custom_metadata=custom_metadata,
-    )
+      canonical_uri = _canonical_uri(artifact_dir, next_version)
+      _write_metadata(
+          _metadata_path(artifact_dir, next_version),
+          filename=filename,
+          mime_type=mime_type,
+          version=next_version,
+          canonical_uri=canonical_uri,
+          custom_metadata=custom_metadata,
+      )
+    except BaseException:
+      shutil.rmtree(version_dir, ignore_errors=True)
+      raise
 
     logger.debug(
         "Saved artifact %s version %d to %s",
@@ -470,6 +553,7 @@ async def load_artifact(
   ) -> Optional[types.Part]:
     return await asyncio.to_thread(
         self._load_artifact_sync,
+        app_name,
         user_id,
         filename,
         session_id,
@@ -478,6 +562,7 @@ async def load_artifact(
 
   def _load_artifact_sync(
       self,
+      app_name: str,
       user_id: str,
       filename: str,
       session_id: Optional[str],
@@ -485,6 +570,7 @@ def _load_artifact_sync(
   ) -> Optional[types.Part]:
     """Loads an artifact from disk."""
     artifact_dir = self._artifact_dir(
+        app_name=app_name,
         user_id=user_id,
         session_id=session_id,
         filename=filename,
@@ -507,26 +593,28 @@ def _load_artifact_sync(
     metadata = _read_metadata(_metadata_path(artifact_dir, version_to_load))
     mime_type = metadata.mime_type if metadata else None
     stored_filename = artifact_dir.name
+    # The payload location is derived exclusively from the storage layout. It
+    # must never be taken from the metadata document: that document lives in
+    # the artifact tree and is therefore attacker-influenced input, so honoring
+    # a `canonical_uri` from it would turn this into an arbitrary file read.
     content_path = version_dir / stored_filename
-    if metadata and metadata.canonical_uri and not content_path.exists():
-      uri_path = _file_uri_to_path(metadata.canonical_uri)
-      if uri_path and uri_path.exists():
-        content_path = uri_path
 
+    # Read without a preceding `exists()` check. A separate `delete_artifact`
+    # can unlink the payload between the check and the read, and reacting to
+    # that gap is what previously reached the metadata-supplied path.
     if mime_type:
-      if not content_path.exists():
+      data = _read_bytes_if_present(content_path)
+      if data is None:
         logger.warning(
             "Binary artifact %s missing at %s", filename, content_path
         )
         return None
-      data = content_path.read_bytes()
       return types.Part(inline_data=types.Blob(mime_type=mime_type, data=data))
 
-    if not content_path.exists():
+    text = _read_text_if_present(content_path)
+    if text is None:
       logger.warning("Text artifact %s missing at %s", filename, content_path)
       return None
-
-    text = content_path.read_text(encoding="utf-8")
     return types.Part(text=text)
 
   @override
@@ -539,19 +627,21 @@ async def list_artifact_keys(
   ) -> list[str]:
     return await asyncio.to_thread(
         self._list_artifact_keys_sync,
+        app_name,
         user_id,
         session_id,
     )
 
   def _list_artifact_keys_sync(
       self,
+      app_name: str,
       user_id: str,
       session_id: Optional[str],
   ) -> list[str]:
     """Lists artifact filenames for the given session/user."""
     filenames: set[str] = set()
 
-    base_root = self._base_root(user_id)
+    base_root = self._base_root(app_name, user_id)
 
     if session_id is not None:
       session_root = _session_artifacts_dir(base_root, session_id)
@@ -594,6 +684,7 @@ async def delete_artifact(
     """
     await asyncio.to_thread(
         self._delete_artifact_sync,
+        app_name,
         user_id,
         filename,
         session_id,
@@ -601,11 +692,13 @@ async def delete_artifact(
 
   def _delete_artifact_sync(
       self,
+      app_name: str,
       user_id: str,
       filename: str,
       session_id: Optional[str],
   ) -> None:
     artifact_dir = self._artifact_dir(
+        app_name=app_name,
         user_id=user_id,
         session_id=session_id,
         filename=filename,
@@ -626,6 +719,7 @@ async def list_versions(
     """Lists all versions stored for an artifact."""
     return await asyncio.to_thread(
         self._list_versions_sync,
+        app_name,
         user_id,
         filename,
         session_id,
@@ -633,11 +727,13 @@ async def list_versions(
 
   def _list_versions_sync(
       self,
+      app_name: str,
       user_id: str,
       filename: str,
       session_id: Optional[str],
   ) -> list[int]:
     artifact_dir = self._artifact_dir(
+        app_name=app_name,
         user_id=user_id,
         session_id=session_id,
         filename=filename,
@@ -656,6 +752,7 @@ async def list_artifact_versions(
     """Lists metadata for each artifact version on disk."""
     return await asyncio.to_thread(
         self._list_artifact_versions_sync,
+        app_name,
         user_id,
         filename,
         session_id,
@@ -663,11 +760,13 @@ async def list_artifact_versions(
 
   def _list_artifact_versions_sync(
       self,
+      app_name: str,
       user_id: str,
       filename: str,
       session_id: Optional[str],
   ) -> list[ArtifactVersion]:
     artifact_dir = self._artifact_dir(
+        app_name=app_name,
         user_id=user_id,
         session_id=session_id,
         filename=filename,
@@ -679,9 +778,7 @@ def _list_artifact_versions_sync(
       metadata = _read_metadata(metadata_path)
       artifact_versions.append(
           self._build_artifact_version(
-              user_id=user_id,
-              session_id=session_id,
-              filename=filename,
+              artifact_dir=artifact_dir,
               version=version,
               metadata=metadata,
           )
@@ -701,6 +798,7 @@ async def get_artifact_version(
     """Gets metadata for a specific artifact version."""
     return await asyncio.to_thread(
         self._get_artifact_version_sync,
+        app_name,
         user_id,
         filename,
         session_id,
@@ -709,12 +807,14 @@ async def get_artifact_version(
 
   def _get_artifact_version_sync(
       self,
+      app_name: str,
       user_id: str,
       filename: str,
       session_id: Optional[str],
       version: Optional[int],
   ) -> Optional[ArtifactVersion]:
     artifact_dir = self._artifact_dir(
+        app_name=app_name,
         user_id=user_id,
         session_id=session_id,
         filename=filename,
@@ -732,9 +832,7 @@ def _get_artifact_version_sync(
     metadata_path = _metadata_path(artifact_dir, version_to_read)
     metadata = _read_metadata(metadata_path)
     return self._build_artifact_version(
-        user_id=user_id,
-        session_id=session_id,
-        filename=filename,
+        artifact_dir=artifact_dir,
         version=version_to_read,
         metadata=metadata,
     )
@@ -759,20 +857,50 @@ def _write_metadata(
       # artifact services (e.g. GCS).
       custom_metadata=dict(custom_metadata or {}),
   )
-  path.write_text(
-      metadata.model_dump_json(by_alias=True, exclude_none=True),
-      encoding="utf-8",
-  )
+  # Serialize before touching the filesystem: serialization is caller-driven
+  # (`custom_metadata` is arbitrary) and can fail, and it must not be able to
+  # leave a truncated document behind.
+  serialized = metadata.model_dump_json(by_alias=True, exclude_none=True)
+
+  # Write via a uniquely named temporary file in the same directory and rename
+  # it into place, so readers never observe a partial document.
+  fd, tmp_name = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
+  tmp_path = Path(tmp_name)
+  try:
+    with os.fdopen(fd, "w", encoding="utf-8") as tmp_file:
+      tmp_file.write(serialized)
+    # `os.replace` carries the temporary file's mode over to the destination,
+    # and mkstemp made it 0600. Restore the mode the payload beside it got.
+    os.chmod(tmp_path, _DEFAULT_FILE_MODE)
+    os.replace(tmp_path, path)
+  except BaseException:
+    tmp_path.unlink(missing_ok=True)
+    raise
 
 
 def _read_metadata(path: Path) -> Optional[FileArtifactVersion]:
-  """Loads a metadata payload from disk."""
-  if not path.exists():
+  """Loads a metadata payload from disk.
+
+  The path is derived from a caller-supplied filename, so it can be made to
+  name a directory rather than a file; that must degrade to "no metadata"
+  instead of raising.
+
+  Args:
+    path: Location of the metadata document.
+
+  Returns:
+    The parsed metadata, or None for anything that is not a readable,
+    well-formed metadata document.
+  """
+  try:
+    raw = path.read_text(encoding="utf-8")
+  except FileNotFoundError:
+    return None
+  except OSError as exc:
+    logger.warning("Unreadable metadata at %s: %s", path, exc)
     return None
   try:
-    return FileArtifactVersion.model_validate_json(
-        path.read_text(encoding="utf-8")
-    )
+    return FileArtifactVersion.model_validate_json(raw)
   except ValidationError as exc:
     logger.warning("Failed to parse metadata at %s: %s", path, exc)
     return None
diff --git a/tests/unittests/artifacts/test_artifact_service.py b/tests/unittests/artifacts/test_artifact_service.py
index 379d2ab8ea..6913b20657 100644
--- a/tests/unittests/artifacts/test_artifact_service.py
+++ b/tests/unittests/artifacts/test_artifact_service.py
@@ -20,6 +20,7 @@
 import enum
 import json
 from pathlib import Path
+import stat
 from typing import Any
 from typing import Optional
 from typing import Union
@@ -28,6 +29,7 @@
 from urllib.parse import unquote
 from urllib.parse import urlparse
 
+from google.adk.artifacts import file_artifact_service
 from google.adk.artifacts.base_artifact_service import ArtifactVersion
 from google.adk.artifacts.base_artifact_service import ensure_part
 from google.adk.artifacts.file_artifact_service import FileArtifactService
@@ -598,6 +600,194 @@ async def test_get_artifact_version_out_of_index(
   )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("filename", "session_id"),
+    [("report.txt", "session"), ("user:profile.txt", None)],
+)
+async def test_file_artifacts_are_isolated_by_app(
+    tmp_path: Path,
+    filename: str,
+    session_id: Optional[str],
+):
+  """Every file-artifact operation stays within its application."""
+  service = FileArtifactService(root_dir=tmp_path / "artifacts")
+  scope = {
+      "user_id": "user",
+      "session_id": session_id,
+      "filename": filename,
+  }
+
+  assert (
+      await service.save_artifact(
+          app_name="app-a", artifact=types.Part(text="secret-a"), **scope
+      )
+      == 0
+  )
+
+  assert await service.load_artifact(app_name="app-b", **scope) is None
+  assert (
+      await service.list_artifact_keys(
+          app_name="app-b",
+          user_id="user",
+          session_id=session_id,
+      )
+      == []
+  )
+  assert await service.list_versions(app_name="app-b", **scope) == []
+  assert await service.list_artifact_versions(app_name="app-b", **scope) == []
+  assert await service.get_artifact_version(app_name="app-b", **scope) is None
+
+  assert (
+      await service.save_artifact(
+          app_name="app-b", artifact=types.Part(text="secret-b"), **scope
+      )
+      == 0
+  )
+  assert await service.load_artifact(app_name="app-a", **scope) == types.Part(
+      text="secret-a"
+  )
+
+  await service.delete_artifact(app_name="app-b", **scope)
+  assert await service.load_artifact(app_name="app-b", **scope) is None
+  assert await service.load_artifact(app_name="app-a", **scope) == types.Part(
+      text="secret-a"
+  )
+
+
+def _write_unscoped_artifact(root: Path, *texts: str) -> None:
+  """Writes an artifact in the layout used before storage was app-scoped."""
+  versions_dir = (
+      root
+      / "users"
+      / "user"
+      / "sessions"
+      / "session"
+      / "artifacts"
+      / "report.txt"
+      / "versions"
+  )
+  for version, text in enumerate(texts):
+    version_dir = versions_dir / str(version)
+    version_dir.mkdir(parents=True)
+    payload_path = version_dir / "report.txt"
+    payload_path.write_text(text, encoding="utf-8")
+    file_artifact_service._write_metadata(
+        version_dir / "metadata.json",
+        filename="report.txt",
+        mime_type=None,
+        version=version,
+        canonical_uri=payload_path.resolve().as_uri(),
+        custom_metadata=None,
+    )
+
+
+_UNSCOPED_SCOPE = {
+    "user_id": "user",
+    "session_id": "session",
+    "filename": "report.txt",
+}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app_name", ["app-a", "app-b"])
+async def test_file_artifact_reads_never_serve_the_unscoped_layout(
+    tmp_path: Path,
+    app_name: str,
+):
+  """A root can be shared, so no app may read the pre-app-scoped tree."""
+  root = tmp_path / "artifacts"
+  _write_unscoped_artifact(root, "older", "legacy")
+  service = FileArtifactService(root_dir=root)
+
+  assert (
+      await service.load_artifact(app_name=app_name, **_UNSCOPED_SCOPE) is None
+  )
+  assert await service.list_versions(app_name=app_name, **_UNSCOPED_SCOPE) == []
+  assert (
+      await service.list_artifact_versions(app_name=app_name, **_UNSCOPED_SCOPE)
+      == []
+  )
+  assert (
+      await service.get_artifact_version(app_name=app_name, **_UNSCOPED_SCOPE)
+      is None
+  )
+  assert (
+      await service.list_artifact_keys(
+          app_name=app_name, user_id="user", session_id="session"
+      )
+      == []
+  )
+
+
+@pytest.mark.asyncio
+async def test_file_artifact_saves_never_reuse_unscoped_layout(
+    tmp_path: Path,
+):
+  """Saving after the upgrade writes app-scoped and ignores the older copy."""
+  root = tmp_path / "artifacts"
+  _write_unscoped_artifact(root, "older", "legacy")
+  service = FileArtifactService(root_dir=root)
+
+  assert (
+      await service.save_artifact(
+          app_name="app-a",
+          artifact=types.Part(text="current"),
+          **_UNSCOPED_SCOPE,
+      )
+      == 0
+  )
+  assert (root / "apps" / "app-a" / "users" / "user").is_dir()
+  assert await service.load_artifact(
+      app_name="app-a", **_UNSCOPED_SCOPE
+  ) == types.Part(text="current")
+  # Version numbering restarts and the older versions stop being served.
+  assert await service.list_versions(app_name="app-a", **_UNSCOPED_SCOPE) == [0]
+  assert (
+      await service.load_artifact(
+          version=1, app_name="app-a", **_UNSCOPED_SCOPE
+      )
+      is None
+  )
+
+  await service.delete_artifact(app_name="app-a", **_UNSCOPED_SCOPE)
+  assert (
+      await service.load_artifact(app_name="app-a", **_UNSCOPED_SCOPE) is None
+  )
+
+
+@pytest.mark.asyncio
+async def test_file_artifact_delete_only_removes_the_calling_apps_copy(
+    tmp_path: Path,
+):
+  """A delete on a shared root never reaches data outside the calling app."""
+  root = tmp_path / "artifacts"
+  _write_unscoped_artifact(root, "legacy")
+  unscoped_dir = (
+      root
+      / "users"
+      / "user"
+      / "sessions"
+      / "session"
+      / "artifacts"
+      / "report.txt"
+  )
+  service = FileArtifactService(root_dir=root)
+  await service.save_artifact(
+      app_name="app-a",
+      artifact=types.Part(text="secret-a"),
+      **_UNSCOPED_SCOPE,
+  )
+
+  await service.delete_artifact(app_name="app-b", **_UNSCOPED_SCOPE)
+
+  assert unscoped_dir.is_dir()
+  assert await service.load_artifact(
+      app_name="app-a", **_UNSCOPED_SCOPE
+  ) == types.Part(text="secret-a")
+  assert await service.list_versions(app_name="app-a", **_UNSCOPED_SCOPE) == [0]
+
+
 @pytest.mark.asyncio
 async def test_file_metadata_camelcase(tmp_path, artifact_service_factory):
   """Ensures FileArtifactService writes camelCase metadata without newlines."""
@@ -616,6 +806,8 @@ async def test_file_metadata_camelcase(tmp_path, artifact_service_factory):
   metadata_path = (
       tmp_path
       / "artifacts"
+      / "apps"
+      / "myapp"
       / "users"
       / "user123"
       / "sessions"
@@ -677,6 +869,8 @@ async def test_file_list_artifact_versions(tmp_path, artifact_service_factory):
   version_payload_path = (
       tmp_path
       / "artifacts"
+      / "apps"
+      / "myapp"
       / "users"
       / "user123"
       / "sessions"
@@ -756,6 +950,64 @@ async def test_file_save_artifact_rejects_out_of_scope_paths(
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "app_name",
+    [
+        "../escape",
+        "../../etc",
+        "foo/../../bar",
+        "valid/../..",
+        "..",
+        ".",
+        "/leading/slash",
+        "\\leading\\backslash",
+        r"C:\absolute",
+        "null\x00byte",
+        "",
+    ],
+)
+async def test_file_save_artifact_rejects_traversal_in_app_name(
+    tmp_path, app_name
+):
+  """FileArtifactService rejects app_name values that escape root_dir."""
+  artifact_service = FileArtifactService(root_dir=tmp_path / "artifacts")
+  part = types.Part(text="content")
+  with pytest.raises(InputValidationError):
+    await artifact_service.save_artifact(
+        app_name=app_name,
+        user_id="user123",
+        session_id="sess123",
+        filename="safe.txt",
+        artifact=part,
+    )
+
+
+@pytest.mark.asyncio
+async def test_file_artifact_service_accepts_nested_app_name(tmp_path):
+  """A Vertex-style resource name is addressable as an app name."""
+  app_name = "projects/p1/locations/us-central1/reasoningEngines/12345"
+  root_dir = tmp_path / "artifacts"
+  artifact_service = FileArtifactService(root_dir=root_dir)
+
+  await artifact_service.save_artifact(
+      app_name=app_name,
+      user_id="user123",
+      session_id="sess123",
+      filename="report.txt",
+      artifact=types.Part(text="content"),
+  )
+  loaded = await artifact_service.load_artifact(
+      app_name=app_name,
+      user_id="user123",
+      session_id="sess123",
+      filename="report.txt",
+  )
+
+  assert loaded.text == "content"
+  assert (root_dir / "apps" / app_name).is_dir()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "user_id",
@@ -1452,3 +1704,305 @@ async def test_save_artifact_with_snake_case_dict(
   assert loaded is not None
   assert loaded.inline_data is not None
   assert loaded.inline_data.mime_type == "text/plain"
+
+
+def _write_tampered_metadata(
+    root: Path,
+    *,
+    artifact_name: str,
+    canonical_uri: str,
+) -> None:
+  """Writes a metadata document naming `canonical_uri`, bypassing the service.
+
+  This reproduces the on-disk state an attacker can otherwise reach by saving
+  an artifact that overwrites its own metadata document, so the load path can
+  be exercised against a tampered artifact tree directly.
+
+  Args:
+    root: Artifact service root directory.
+    artifact_name: Name of the artifact to tamper with.
+    canonical_uri: Value to write into the document's `canonicalUri` field.
+  """
+  version_dir = (
+      root
+      / "apps"
+      / "app"
+      / "users"
+      / "user"
+      / "sessions"
+      / "session"
+      / "artifacts"
+      / artifact_name
+      / "versions"
+      / "0"
+  )
+  version_dir.mkdir(parents=True)
+  (version_dir / "metadata.json").write_text(
+      json.dumps({
+          "fileName": artifact_name,
+          "version": 0,
+          "canonicalUri": canonical_uri,
+          "customMetadata": {},
+      }),
+      encoding="utf-8",
+  )
+
+
+@pytest.mark.asyncio
+async def test_load_artifact_ignores_canonical_uri_from_metadata(tmp_path):
+  """A tampered canonicalUri must not be used to locate the payload."""
+  secret = tmp_path / "secret.txt"
+  secret.write_text("TOP-SECRET", encoding="utf-8")
+  root = tmp_path / "artifacts"
+  service = FileArtifactService(root_dir=root)
+  # The payload is deliberately absent. That is the state the delete/load race
+  # produced, and it is what previously fell through to `canonical_uri`.
+  _write_tampered_metadata(
+      root, artifact_name="poisoned.txt", canonical_uri=secret.as_uri()
+  )
+
+  loaded = await service.load_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="poisoned.txt",
+  )
+
+  assert loaded is None
+
+
+@pytest.mark.asyncio
+async def test_get_artifact_version_ignores_canonical_uri_from_metadata(
+    tmp_path,
+):
+  """A tampered canonicalUri must not be reflected back to callers."""
+  root = tmp_path / "artifacts"
+  service = FileArtifactService(root_dir=root)
+  _write_tampered_metadata(
+      root, artifact_name="poisoned.txt", canonical_uri="file:///etc/passwd"
+  )
+
+  artifact_version = await service.get_artifact_version(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="poisoned.txt",
+      version=0,
+  )
+
+  assert artifact_version is not None
+  assert artifact_version.canonical_uri != "file:///etc/passwd"
+  assert artifact_version.canonical_uri.startswith(root.as_uri())
+
+
+@pytest.mark.parametrize(
+    "filename",
+    [
+        "metadata.json",
+        "nested/metadata.json",
+        "user:metadata.json",
+        # Case variants: on a case-insensitive filesystem these resolve to the
+        # metadata document too, so the name has to be rejected caselessly.
+        "Metadata.json",
+        "METADATA.JSON",
+        "nested/MetaData.Json",
+    ],
+)
+@pytest.mark.asyncio
+async def test_save_artifact_rejects_reserved_metadata_filename(
+    tmp_path, filename
+):
+  """An artifact may not be named so that it overwrites its own metadata."""
+  service = FileArtifactService(root_dir=tmp_path)
+
+  with pytest.raises(InputValidationError):
+    await service.save_artifact(
+        app_name="app",
+        user_id="user",
+        session_id="session",
+        filename=filename,
+        artifact=types.Part(text="payload"),
+    )
+
+
+@pytest.mark.asyncio
+async def test_reserved_metadata_filename_stays_deletable(tmp_path):
+  """A name rejected on write must still be removable.
+
+  The rejection deliberately lives on the save path rather than in
+  `_artifact_dir`, which reads and deletes share. An artifact stored under this
+  name before it was reserved would otherwise be stranded -- unreadable and
+  impossible to delete through the API.
+  """
+  service = FileArtifactService(root_dir=tmp_path)
+  version_dir = (
+      tmp_path
+      / "apps"
+      / "app"
+      / "users"
+      / "user"
+      / "sessions"
+      / "session"
+      / "artifacts"
+      / "metadata.json"
+      / "versions"
+      / "0"
+  )
+  version_dir.mkdir(parents=True)
+  (version_dir / "metadata.json").write_text(
+      json.dumps({"fileName": "metadata.json", "version": 0}), encoding="utf-8"
+  )
+  artifact_dir = version_dir.parent.parent
+
+  # Reading must not raise, and deleting must actually remove it.
+  await service.load_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="metadata.json",
+  )
+  await service.delete_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="metadata.json",
+  )
+
+  assert not artifact_dir.exists()
+
+
+@pytest.mark.asyncio
+async def test_metadata_and_payload_share_permissions(tmp_path):
+  """The metadata document must be as readable as the payload beside it.
+
+  The metadata document is written through `tempfile.mkstemp`, which hardcodes
+  0600, while the payload goes through `open()` and picks up the umask. Left
+  alone the two end up readable by different principals, so a group-readable
+  deployment can read an artifact but not its metadata.
+  """
+  service = FileArtifactService(root_dir=tmp_path)
+  await service.save_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="report.txt",
+      artifact=types.Part(text="payload"),
+  )
+  version_dir = (
+      tmp_path
+      / "apps"
+      / "app"
+      / "users"
+      / "user"
+      / "sessions"
+      / "session"
+      / "artifacts"
+      / "report.txt"
+      / "versions"
+      / "0"
+  )
+
+  payload_mode = stat.S_IMODE((version_dir / "report.txt").stat().st_mode)
+  metadata_mode = stat.S_IMODE((version_dir / "metadata.json").stat().st_mode)
+
+  assert metadata_mode == payload_mode
+
+
+@pytest.mark.asyncio
+async def test_save_artifact_round_trips_explicitly_empty_inline_data(tmp_path):
+  """An empty payload is present, not missing, and must still round-trip.
+
+  The read helpers signal a missing payload with None, so empty bytes have to
+  stay distinguishable from no bytes at all.
+  """
+  service = FileArtifactService(root_dir=tmp_path)
+
+  await service.save_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="empty.png",
+      artifact=types.Part(
+          inline_data=types.Blob(mime_type="image/png", data=b"")
+      ),
+  )
+
+  loaded = await service.load_artifact(
+      app_name="app", user_id="user", session_id="session", filename="empty.png"
+  )
+  assert loaded is not None
+  assert loaded.inline_data is not None
+  assert loaded.inline_data.data is not None
+  assert not loaded.inline_data.data
+
+
+@pytest.mark.asyncio
+async def test_save_artifact_discards_version_when_metadata_write_fails(
+    tmp_path,
+):
+  """A failed save must not leave a payload behind without valid metadata."""
+  service = FileArtifactService(root_dir=tmp_path)
+  await service.save_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="report.txt",
+      artifact=types.Part(text="v0"),
+  )
+
+  # `custom_metadata` is caller-controlled and can be made unserializable by
+  # nesting it beyond the serializer's depth limit.
+  deeply_nested: Any = {"a": 1}
+  for _ in range(500):
+    deeply_nested = {"a": deeply_nested}
+
+  with pytest.raises(Exception):
+    await service.save_artifact(
+        app_name="app",
+        user_id="user",
+        session_id="session",
+        filename="report.txt",
+        artifact=types.Part(text="poison"),
+        custom_metadata=deeply_nested,
+    )
+
+  # The failed version is discarded entirely and the previous one is intact.
+  assert await service.list_versions(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="report.txt",
+  ) == [0]
+  loaded = await service.load_artifact(
+      app_name="app",
+      user_id="user",
+      session_id="session",
+      filename="report.txt",
+  )
+  assert loaded is not None
+  assert loaded.text == "v0"
+
+
+@pytest.mark.asyncio
+async def test_list_artifact_keys_survives_metadata_path_shadowed_by_dir(
+    tmp_path,
+):
+  """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"),
+  )
+
+  keys = await service.list_artifact_keys(
+      app_name="app", user_id="user", session_id="session"
+  )
+
+  # The shadowed artifact has no readable metadata, so it is listed by its
+  # scope-relative path rather than dropped or raised on.
+  assert keys == ["user:a"]