refactor: Use artifact_service.load_artifact during rewind

During session rewind, when restoring an artifact to a previous version, the runner now uses `artifact_service.load_artifact` to fetch the artifact's content. Previously, it would construct a `types.Part` with `file_data` using the artifact URI. This change is necessary because some artifact services, such as those backed by GCS or local files, do not support saving `types.Part` objects that contain `file_data`

Close #4932

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 905223401

Author: GWeale

src/google/adk/runners.py

@@ -29,7 +29,6 @@
 import warnings
 
 from google.adk.apps.compaction import _run_compaction_for_sliding_window
-from google.adk.artifacts import artifact_util
 from google.genai import types
 
 from .agents.base_agent import BaseAgent
@@ -754,15 +753,27 @@ async def _compute_artifact_delta_for_rewind(
         )
       else:
         # Artifact version changed after rewind point. Restore to version at
-        # rewind point.
-        artifact_uri = artifact_util.get_artifact_uri(
+        # rewind point by loading the actual data via the artifact service.
+        artifact = await self.artifact_service.load_artifact(
             app_name=self.app_name,
             user_id=session.user_id,
             session_id=session.id,
             filename=filename,
             version=vt,
         )
-        artifact = types.Part(file_data=types.FileData(file_uri=artifact_uri))
+        if artifact is None:
+          logger.warning(
+              'Artifact %s version %d not found during rewind for'
+              ' session %s. Replacing with empty data.',
+              filename,
+              vt,
+              session.id,
+          )
+          artifact = types.Part(
+              inline_data=types.Blob(
+                  mime_type='application/octet-stream', data=b''
+              )
+          )
       await self.artifact_service.save_artifact(
           app_name=self.app_name,
           user_id=session.user_id,

tests/unittests/runners/test_runner_rewind.py

@@ -14,7 +14,12 @@
 
 """Tests for runner.rewind_async."""
 
+from typing import Any
+from typing import Optional
+from typing import Union
+
 from google.adk.agents.base_agent import BaseAgent
+from google.adk.artifacts.base_artifact_service import ensure_part
 from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
 from google.adk.events.event import Event
 from google.adk.events.event import EventActions
@@ -24,6 +29,34 @@
 import pytest
 
 
+class _NoFileDataArtifactService(InMemoryArtifactService):
+  """Artifact service that rejects file_data parts, like GCS/File services."""
+
+  async def save_artifact(
+      self,
+      *,
+      app_name: str,
+      user_id: str,
+      filename: str,
+      artifact: Union[types.Part, dict[str, Any]],
+      session_id: Optional[str] = None,
+      custom_metadata: Optional[dict[str, Any]] = None,
+  ) -> int:
+    artifact = ensure_part(artifact)
+    if artifact.file_data:
+      raise NotImplementedError(
+          "Saving artifact with file_data is not supported."
+      )
+    return await super().save_artifact(
+        app_name=app_name,
+        user_id=user_id,
+        filename=filename,
+        artifact=artifact,
+        session_id=session_id,
+        custom_metadata=custom_metadata,
+    )
+
+
 class TestRunnerRewind:
   """Tests for runner.rewind_async."""
 
@@ -246,3 +279,85 @@ async def test_rewind_async_not_first_invocation(self):
         session_id=session_id,
         filename="f2",
     ) == types.Part.from_text(text="f2v0")
+
+
+class TestRunnerRewindNoFileData:
+  """Tests that rewind works with artifact services that reject file_data."""
+
+  @pytest.mark.asyncio
+  async def test_rewind_uses_load_artifact_not_file_data(self):
+    """Rewind must not construct file_data parts for artifact restoration.
+
+    GCS and File artifact services reject file_data parts. The runner
+    should use load_artifact to get inline_data instead.
+    """
+    root_agent = BaseAgent(name="test_agent")
+    session_service = InMemorySessionService()
+    artifact_service = _NoFileDataArtifactService()
+    runner = Runner(
+        app_name="test_app",
+        agent=root_agent,
+        session_service=session_service,
+        artifact_service=artifact_service,
+    )
+    user_id = "test_user"
+    session_id = "test_session"
+
+    session = await runner.session_service.create_session(
+        app_name=runner.app_name, user_id=user_id, session_id=session_id
+    )
+
+    # invocation1: create artifact f1 v0
+    await runner.artifact_service.save_artifact(
+        app_name=runner.app_name,
+        user_id=user_id,
+        session_id=session_id,
+        filename="f1",
+        artifact=types.Part.from_text(text="f1v0"),
+    )
+    event1 = Event(
+        invocation_id="invocation1",
+        author="agent",
+        content=types.Content(parts=[types.Part.from_text(text="e1")]),
+        actions=EventActions(
+            state_delta={"k1": "v1"}, artifact_delta={"f1": 0}
+        ),
+    )
+    await runner.session_service.append_event(session=session, event=event1)
+
+    # invocation2: update artifact f1 to v1
+    await runner.artifact_service.save_artifact(
+        app_name=runner.app_name,
+        user_id=user_id,
+        session_id=session_id,
+        filename="f1",
+        artifact=types.Part.from_text(text="f1v1"),
+    )
+    event2 = Event(
+        invocation_id="invocation2",
+        author="agent",
+        content=types.Content(parts=[types.Part.from_text(text="e2")]),
+        actions=EventActions(artifact_delta={"f1": 1}),
+    )
+    await runner.session_service.append_event(session=session, event=event2)
+
+    session = await runner.session_service.get_session(
+        app_name=runner.app_name, user_id=user_id, session_id=session_id
+    )
+
+    # Rewind before invocation2 — this would raise NotImplementedError
+    # with the old code that constructed file_data parts.
+    await runner.rewind_async(
+        user_id=user_id,
+        session_id=session_id,
+        rewind_before_invocation_id="invocation2",
+    )
+
+    # f1 should be restored to v0 content
+    restored = await runner.artifact_service.load_artifact(
+        app_name=runner.app_name,
+        user_id=user_id,
+        session_id=session_id,
+        filename="f1",
+    )
+    assert restored == types.Part.from_text(text="f1v0")