[BUG] GcsArtifactService.saveArtifactAndReturnBlob overwrites version 0 when listVersions fails

[svetanis]

Please make sure you read the contribution guide and file the issues in the right place.
Contribution guide.

🔴 Required Information

Describe the Bug

GcsArtifactService.listVersions returns an empty list when the listing raises StorageException:

// core/src/main/java/com/google/adk/artifacts/GcsArtifactService.java — listVersions
} catch (StorageException e) {
  return ImmutableList.of();
}

saveArtifactAndReturnBlob derives the next version number from that result, and the write that
follows carries no precondition:

// saveArtifactAndReturnBlob
return listVersions(appName, userId, sessionId, filename)
    .map(versions -> versions.isEmpty() ? 0 : max(versions) + 1)
// ...
Blob blob = storageClient.create(blobInfo, dataToSave);

At the point where the version number is chosen, an empty list has two possible meanings: the
artifact has no versions yet, or the listing did not complete. A 503, a 429, a timeout, or IAM
propagation after a binding change produces the second, and the save proceeds as if it were the
first — it computes version 0 and writes over the object already stored at version 0. The call
returns success and version 0. The earlier content is no longer retrievable, and nothing is
logged.

Both conditions are needed for the loss: a save whose listing fails, and at least one version of
that artifact already stored. A listing failure on the first save of a file changes nothing — 0 is
the correct version and the write lands on an unused name. The computed version is always 0, so
higher versions are never overwritten.

Example. An agent saves report.txt at 10:00 and gets version 0. At 10:05 it saves again and
the listing fails for two seconds. Both calls returned success and version 0; loadArtifact(..., 0)
now returns the 10:05 content, and the 10:00 content is gone.

With versions 0 and 1 both stored, the same failure writes the newest content to version 0.
loadArtifact with no version resolves to max(versions), so it then returns version 1 — the older
content.

How the file currently handles StorageException:

Method	Operation	On failure
loadArtifact	objects.get	returns null
listArtifactKeys (session prefix)	objects.list	throws VerifyException
listArtifactKeys (user prefix)	objects.list	throws VerifyException
deleteArtifact	delete	throws VerifyException
listVersions	objects.list	returns ImmutableList.of()
saveArtifactAndReturnBlob	create	throws VerifyException

listArtifactKeys and listVersions issue the same
storageClient.list(bucketName, BlobListOption.prefix(...)) call. listVersions is the one whose
result selects the blob name a save writes to.

Steps to Reproduce

Reproduced against a real GCS bucket. A transient cannot be scheduled, so the run reaches the same
branch deterministically by withholding storage.objects.list from the identity performing the
second save. The catch does not inspect the cause, so a 403 arrives there the same way a 503, a 429
or a socket error does. This is how the run triggers the branch, not a claim about how deployments
are configured — storage.objects.list is part of roles/storage.objectViewer, and artifact
identities normally hold it.

1. A bucket with uniform bucket-level access, and a service account granted storage.objects.create,
   storage.objects.get and storage.objects.delete on it, but not storage.objects.list. (Delete
   is what permits an overwrite in GCS at all; objectAdmin and objectUser both include it.)
2. A GcsArtifactService over a fully-permissioned Storage client:
   saveArtifact("report.txt", "PAYLOAD-A").
3. A second GcsArtifactService over a Storage client authenticated as the service account from
   step 1: saveArtifact("report.txt", "PAYLOAD-B"). A transient affects one save, not every save.
4. loadArtifact("report.txt", 0), then list app/user/session/report.txt/ with the privileged
   credential.

The same path is reachable in a unit test without GCP: stub Storage.list(...) to throw
StorageException, save twice, and both saves compute version 0 and target the same blob name.

Expected Behavior

A save whose version listing did not complete does not write, and the failure reaches the caller —
as it already does for the same operation in listArtifactKeys. The stored artifact is unchanged.

Observed Behavior

Both saves return success and version 0. The bucket holds a single object containing the second
payload, and loadArtifact(..., 0) returns it. The control arm — same code, same payloads, same
bucket, an identity that can list — returns versions 0 and 1 and keeps both objects.

--- ARM 1 - listVersions denied ---
save 1 "report.txt" (client whose listing works) : PAYLOAD-A -> reported version 0
  bucket: 1 object   .../f18-session/report.txt/0 = PAYLOAD-A
save 2 "report.txt" (client under test)          : PAYLOAD-B -> reported version 0
  bucket: 1 object   .../f18-session/report.txt/0 = PAYLOAD-B
loadArtifact(version 0)                          : PAYLOAD-B

--- ARM 2 - control, identity that CAN list ---
save 1 "report.txt"                              : PAYLOAD-A -> reported version 0
save 2 "report.txt"                              : PAYLOAD-B -> reported version 1
  bucket: 2 objects  .../report.txt/0 = PAYLOAD-A, .../report.txt/1 = PAYLOAD-B
loadArtifact(version 0)                          : PAYLOAD-A

The arms differ only in which client performs the second save.

Environment Details

• ADK Library Version: 1.7.2-SNAPSHOT
• OS: not OS-specific

Model Information

N/A — no model is involved.

---

🟡 Optional Information

Related, reading the same result, not covered here

Two other callers derive behavior from the same empty list. Both misreport rather than overwrite,
and changing them is observable to callers, so they are noted here rather than folded into the fix:

Site	Effect of the empty list
loadArtifact, no version given	Maybe.empty() — a stored artifact is reported as absent
deleteArtifact	versions.isEmpty() → Completable.complete() — a delete that removed nothing reports success
dev ArtifactController.listArtifactVersions	the endpoint returns [] and logs "Found 0 versions for artifact X"

Regression

No. The first revision of this file already returns an empty list on a failed listing and already
derives nextVersion from it.

Willingness to contribute

Yes — a PR follows immediately, scoped to the save path. It gives that path its own view of the
listing: a private method that lets StorageException propagate, surfaced as
VerifyException("Failed to list artifact versions from GCS", e), matching the wording
listArtifactKeys already uses. listVersions and its existing test are unchanged, so the three
callers above behave exactly as they do today.

How often has this issue occurred?

Always (100%), when a save's version listing fails and at least one version is already stored.