After Runner.rewind_async, FileArtifactService and GcsArtifactService load a removed artifact as an empty file instead of None #7482

🔴 Required Information

Describe the Bug:
When Runner.rewind_async rewinds past the point where an artifact was first saved, compute_artifact_delta_for_rewind (sessions/_rewind_utils.py) removes it by saving an empty application/octet-stream part as a new version. Only InMemoryArtifactService recognizes that part and loads it as None, which is what the rewind code ("Mark it as inaccessible") and test_runner_rewind.py ("f2 should not exist") expect. FileArtifactService and GcsArtifactService return the empty part as if it were real content. The in-memory service's own comment notes the gap: "This tombstone convention is in-memory only; other artifact services (such as GcsArtifactService) do not perform this empty-payload check."

The rewind documentation says rewind "restores the session's state and artifacts to their condition before the rewind point", with no exception for particular artifact services.

adk web and adk api_server store artifacts with the file service by default (cli/utils/local_storage.py), so on those servers, and on GCS in production, a rewind does not remove the artifact:

• load_artifact returns a 0-byte part instead of None.
• GET /apps/{app}/users/{user}/sessions/{session}/artifacts/{name} returns 200 with an empty body instead of 404, because the endpoint checks if not artifact: and a Part is always truthy.
• When the agent calls load_artifacts for it, the model receives Artifact report.txt is: followed by an empty text part, so it is told the artifact exists and is empty.

test_runner_rewind.py only uses InMemoryArtifactService, so the existing tests do not cover this.

Steps to Reproduce:

1. pip install google-adk==2.11.0 (also reproduced on main at 3d11f9fe7).
2. Save the script below as repro.py and run python repro.py. No model or Google Cloud access is needed.

import asyncio
import tempfile

from google.adk.agents.base_agent import BaseAgent
from google.adk.artifacts.file_artifact_service import FileArtifactService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.events.event import Event
from google.adk.events.event_actions import EventActions
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types


async def check(name, artifact_service):
  runner = Runner(
      app_name="app",
      agent=BaseAgent(name="a"),
      session_service=InMemorySessionService(),
      artifact_service=artifact_service,
  )
  session = await runner.session_service.create_session(
      app_name="app", user_id="u", session_id="s"
  )
  scope = dict(app_name="app", user_id="u", session_id="s")
  await runner.session_service.append_event(
      session=session, event=Event(invocation_id="inv1", author="user")
  )
  # inv2 saves report.txt for the first time.
  await artifact_service.save_artifact(
      **scope, filename="report.txt", artifact=types.Part.from_text(text="draft")
  )
  await runner.session_service.append_event(
      session=session,
      event=Event(
          invocation_id="inv2",
          author="a",
          actions=EventActions(artifact_delta={"report.txt": 0}),
      ),
  )

  await runner.rewind_async(
      user_id="u", session_id="s", rewind_before_invocation_id="inv2"
  )

  loaded = await artifact_service.load_artifact(**scope, filename="report.txt")
  print(f"{name:7} load_artifact after rewind: {loaded!r}")


async def main():
  await check("memory", InMemoryArtifactService())
  await check("file", FileArtifactService(root_dir=tempfile.mkdtemp()))


asyncio.run(main())

Expected Behavior:
Both services print None: report.txt did not exist before inv2, so after rewinding to before inv2 it should not load.

Observed Behavior:

memory  load_artifact after rewind: None
file    load_artifact after rewind: Part(
  inline_data=Blob(
    data=b'',
    mime_type='application/octet-stream'
  )
)

GcsArtifactService behaves like the file service. I checked it with the MockClient from tests/unittests/artifacts/test_artifact_service.py.

Environment Details:

• ADK Library Version: 2.11.0, and main at 3d11f9fe7
• Desktop OS: Windows 11, and Linux (Ubuntu on WSL2)
• Python Version: 3.11

Model Information:

• Are you using LiteLLM: No
• Which model is being used: N/A (no model call needed; with a scripted model, load_artifacts hands the empty part to the model as shown above)

🟡 Optional Information

Regression:
No. Rewind has handled the removal marker only in InMemoryArtifactService since rewind_async was added. GcsArtifactService was never updated, and FileArtifactService was added later.

Additional Context:
I'm working on a fix and will open a PR shortly. The plan is to keep the existing marker, because sessions rewound by earlier releases already store it, define it once in artifact_util, and have FileArtifactService and GcsArtifactService load it as None the way InMemoryArtifactService does. Tests will cover all three services and a Runner.rewind_async round trip on the file service.

Two related points, outside that fix:

• All three services still list the removed artifact in list_artifact_keys after a rewind, so the model's list of available artifacts still names it. I'm leaving that out of the fix to keep it small, and can follow up if you'd like it changed.
• As with the in-memory service today, an artifact a user saves with exactly empty application/octet-stream content would also load as None from the file and GCS services. A narrower marker would avoid that, but it would no longer match sessions that were rewound before the change.

How often has this issue occurred?:

• Always (100%) with the steps above.