feat(artifacts): add save_media_frames() for batched media persistence

docs/guides/artifacts/artifact_service/index.md

@@ -195,12 +195,76 @@ and the deployed agent falls back to `InMemoryArtifactService` and loses every
 artifact when it restarts.
 
 Writing your own means subclassing `BaseArtifactService` and implementing its
-seven abstract methods: `save_artifact`, `load_artifact`, `list_artifact_keys`,
+abstract methods: `save_artifact`, `load_artifact`, `list_artifact_keys`,
 `delete_artifact`, `list_versions`, `list_artifact_versions`, and
-`get_artifact_version`. All are keyword-only and take `app_name`, `user_id`, and
-an optional `session_id`, where `None` means the user-scoped namespace. Your
-implementation is responsible for honoring the `user:` prefix, since the routing
-lives in the service and not above it.
+`get_artifact_version`. All are keyword-only and take `app_name`, `user_id`,
+and an optional `session_id`, where `None` means the user-scoped namespace.
+Your implementation is responsible for honoring the `user:`
+prefix, since the routing lives in the service and not above it.
+
+`save_media_frames` is deliberately *not* abstract. It has a default that raises
+`NotImplementedError`, so an existing subclass stays instantiable and only needs
+to override it if it wants to store frame collections.
+
+## Saving media frames (`save_media_frames`)
+
+In addition to individual single-blob artifacts, `BaseArtifactService` supports
+persisting sequences of video/image frames through `save_media_frames`:
+
+```python
+from google.adk.artifacts import MediaFrame
+
+version = await artifact_service.save_media_frames(
+    app_name="vision_app",
+    user_id="u1",
+    session_id="s1",
+    collection_name="input_media_20260101_120000_000000",
+    frames=[
+        MediaFrame(
+            blob=types.Blob(data=frame_bytes_0, mime_type="image/jpeg"),
+            timestamp=0.0,
+        ),
+        MediaFrame(
+            blob=types.Blob(data=frame_bytes_1, mime_type="image/jpeg"),
+            timestamp=0.5,
+        ),
+    ],
+    custom_metadata={"source": "camera_feed"},
+)
+```
+
+Timestamps are seconds and must be non-decreasing; a batch that runs backwards
+is rejected rather than stored with a negative duration.
+
+Each frame is stored in a `frames/` subfolder (e.g. `frame_0000.jpeg`,
+`frame_0001.jpeg`), accompanied by an atomic `metadata.json` document. Your
+`custom_metadata` is merged into the top level of that document, and these keys
+are written over it:
+
+*   `type`: Always `"video_frame_sequence"`.
+*   `frameCount`: Total count of persisted frames.
+*   `startTimestampMs` / `endTimestampMs`: First and last frame timestamps.
+*   `durationMs`: Span from the first frame to the last.
+*   `estimatedFps`: Frames per second derived from those timestamps, or `0.0`
+    for a single frame or a zero-length span.
+*   `frames`: Per-frame `frameIndex`, `offsetMs`, `fileName`, `mimeType`, and
+    `sizeBytes`.
+
+Those keys win a collision, so a caller cannot make the document disagree with
+the frames actually written.
+
+One difference is worth knowing on `GcsArtifactService`: an ordinary artifact's
+`custom_metadata` is stored as Cloud Storage object metadata, which is
+string-valued, so `{"attempt": 3}` reads back as `{"attempt": "3"}`. A
+collection's metadata lives in a JSON sidecar instead -- object metadata is
+capped at a few KiB and a frame index will not fit -- so its values keep their
+original types. `FileArtifactService` and `InMemoryArtifactService` preserve
+types in both cases.
+
+For backward compatibility with preview tools and standard readers, loading the
+collection name directly via `load_artifact(filename=collection_name)` returns the
+initial frame (`frame_0000`) as a `types.Part`.
+
 
 `list_artifact_keys` must be complete: anything readable in scope should be
 returned. Callers (such as `LoadArtifactsTool`) rely on this listing to

src/google/adk/artifacts/__init__.py

@@ -18,6 +18,7 @@
 from typing import TYPE_CHECKING
 
 from .base_artifact_service import BaseArtifactService
+from .base_artifact_service import MediaFrame
 
 if TYPE_CHECKING:
   from .file_artifact_service import FileArtifactService
@@ -29,6 +30,7 @@
     'FileArtifactService',
     'GcsArtifactService',
     'InMemoryArtifactService',
+    'MediaFrame',
 ]
 
 _LAZY_MEMBERS: dict[str, str] = {

src/google/adk/artifacts/artifact_util.py

@@ -21,6 +21,7 @@
 from google.genai import types
 
 from ..errors import input_validation_error
+from .base_artifact_service import MediaFrame
 
 
 class ParsedArtifactUri(NamedTuple):
@@ -54,6 +55,108 @@ class ParsedArtifactUri(NamedTuple):
     rf"artifact://apps/({_PATH_SEGMENT_PATTERN})/users/({_PATH_SEGMENT_PATTERN})/artifacts/(.+)/versions/(\d+)"
 )
 
+# Layout shared by every backend that stores media frame collections.
+FRAMES_DIR_NAME = "frames"
+MEDIA_COLLECTION_TYPE = "video_frame_sequence"
+DEFAULT_FRAME_MIME_TYPE = "image/jpeg"
+DEFAULT_FRAME_EXTENSION = "jpeg"
+
+# Scope marker every backend strips before a name becomes a stored path. Each
+# backend owns its own handling of it; this copy exists so the shared
+# validation below sees the same name storage will.
+_USER_NAMESPACE_PREFIX = "user:"
+
+# A frame extension becomes a path segment, so it is held to the same standard
+# as the caller-supplied identifiers `validate_path_segment` guards: no
+# separators, no traversal, no null bytes. Restricting to the characters real
+# subtypes use is simpler to reason about than enumerating what to reject.
+_FRAME_EXTENSION_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9+._-]*")
+
+
+def frame_extension_from_mime_type(mime_type: str | None) -> str:
+  """Derives the filename extension for a media frame from its mime type.
+
+  The mime type reaches the artifact services from a live model stream, so it
+  is caller-supplied data that ends up inside a path. Anything that is not a
+  plausible subtype -- including values carrying "/" or "\\" separators or ".."
+  traversal segments -- falls back to the default rather than raising, because
+  an odd mime type should not fail a whole frame batch.
+
+  Args:
+    mime_type: The blob mime type, e.g. "image/jpeg" or "image/png;foo=bar".
+
+  Returns:
+    A path-safe extension, defaulting to "jpeg".
+  """
+  subtype = (mime_type or "").split("/")[-1].split(";")[0].strip()
+  if not _FRAME_EXTENSION_RE.fullmatch(subtype):
+    return DEFAULT_FRAME_EXTENSION
+  return subtype
+
+
+def frame_file_name(index: int, mime_type: str | None) -> str:
+  """Returns the `frames/`-relative filename for one frame."""
+  return f"frame_{index:04d}.{frame_extension_from_mime_type(mime_type)}"
+
+
+def validate_media_collection_name(collection_name: str) -> None:
+  """Rejects a collection name that collides with the frames subdirectory.
+
+  `FileArtifactService` stages a version's preview file and its `frames/`
+  subdirectory as siblings, so a collection whose own final segment is "frames"
+  makes them the same path and the save dies partway through with
+  `IsADirectoryError`. The flat-namespace backends have no such collision, but
+  they reject the name too: a caller that can write a collection on one backend
+  and not another has no portable contract to program against, and the failure
+  only shows up after switching backends.
+
+  The name is normalized the way the backends normalize it before it reaches
+  storage -- the `user:` scope marker is not part of the stored path, and the
+  file backend strips surrounding whitespace -- so `user:frames` is caught too.
+
+  Args:
+    collection_name: The caller-supplied media collection name.
+
+  Raises:
+    InputValidationError: If the name's final path segment is "frames".
+  """
+  stripped = collection_name.removeprefix(_USER_NAMESPACE_PREFIX).strip()
+  if stripped.rpartition("/")[2].casefold() == FRAMES_DIR_NAME:
+    raise input_validation_error.InputValidationError(
+        f"Collection filename {collection_name!r} is reserved for media frame"
+        " storage."
+    )
+
+
+def validate_frame_timestamps(
+    frames: list[MediaFrame],
+) -> None:
+  """Rejects a frame batch whose timestamps run backwards.
+
+  Every backend takes `durationMs` from the first and last frame and each
+  frame's `offsetMs` from the first, so an out-of-order batch does not fail --
+  it silently persists a negative duration, negative offsets, and a meaningless
+  `estimatedFps`. Callers buffer frames as they arrive, so out-of-order input
+  means the caller has a bug worth surfacing rather than recording.
+
+  Equal timestamps are allowed: frames captured within the same clock tick are
+  legitimate and yield a zero offset delta.
+
+  Args:
+    frames: The `MediaFrame` batch about to be written.
+
+  Raises:
+    InputValidationError: If any frame predates the frame before it.
+  """
+  for idx in range(1, len(frames)):
+    previous_ts = frames[idx - 1].timestamp
+    current_ts = frames[idx].timestamp
+    if current_ts < previous_ts:
+      raise input_validation_error.InputValidationError(
+          f"Frame timestamps must be non-decreasing, but frame {idx} at"
+          f" {current_ts} precedes frame {idx - 1} at {previous_ts}."
+      )
+
 
 def parse_artifact_uri(uri: str) -> ParsedArtifactUri | None:
   """Parses an artifact URI.

src/google/adk/artifacts/base_artifact_service.py

@@ -64,6 +64,32 @@ class ArtifactVersion(BaseModel):
   )
 
 
+class MediaFrame(BaseModel):
+  """One frame of a media collection passed to `save_media_frames`.
+
+  A bare `(blob, timestamp)` tuple would be cheaper, but it fixes the shape
+  permanently: a per-frame sequence number, capture source or encoding hint
+  could not be added later without breaking every caller. A model can gain
+  optional fields compatibly.
+  """
+
+  model_config = ConfigDict(
+      alias_generator=alias_generators.to_camel,
+      populate_by_name=True,
+  )
+
+  blob: types.Blob = Field(
+      description="The frame payload together with its MIME type."
+  )
+  timestamp: float = Field(
+      description=(
+          "Capture time in seconds. The collection's duration comes from the"
+          " first and last frame and each frame's offset from the first, so"
+          " timestamps must be non-decreasing across the batch."
+      )
+  )
+
+
 def ensure_part(artifact: Union[types.Part, dict[str, Any]]) -> types.Part:
   """Normalizes an artifact to a ``types.Part`` instance.
 
@@ -122,6 +148,33 @@ async def save_artifact(
       This is incremented by 1 after each successful save.
     """
 
+  async def save_media_frames(
+      self,
+      *,
+      app_name: str,
+      user_id: str,
+      collection_name: str,
+      frames: list[MediaFrame],
+      session_id: Optional[str] = None,
+      custom_metadata: Optional[dict[str, Any]] = None,
+  ) -> int:
+    """Saves a sequence of media frames to artifact storage.
+
+    Args:
+      app_name: The app name.
+      user_id: The user ID.
+      collection_name: The name of the collection folder for these frames.
+      frames: The frames to store, in non-decreasing timestamp order.
+      session_id: The session ID. If None, the artifact is user-scoped.
+      custom_metadata: Optional custom metadata to associate with the frames.
+
+    Returns:
+      The revision/version integer of the stored media collection.
+    """
+    raise NotImplementedError(
+        f"{type(self).__name__} does not implement save_media_frames."
+    )
+
   @abstractmethod
   async def load_artifact(
       self,