diff --git a/core/src/main/java/com/google/adk/runner/Runner.java b/core/src/main/java/com/google/adk/runner/Runner.java
index 89e8e3c15..d58ca2e4d 100644
--- a/core/src/main/java/com/google/adk/runner/Runner.java
+++ b/core/src/main/java/com/google/adk/runner/Runner.java
@@ -344,9 +344,61 @@ public Completable close() {
return Completable.mergeDelayError(completables);
}
+ /**
+ * A user message prepared for persistence, together with the artifact saves it scheduled.
+ *
+ * @param message the message to append: a copy when inline blobs were replaced by placeholders,
+ * otherwise the original
+ * @param artifactSaves the saves to run before the event is appended
+ */
+ private record OffloadedMessage(Content message, Completable artifactSaves) {
+
+ /** Nothing to offload: append the caller's message as-is. */
+ static OffloadedMessage unchanged(Content message) {
+ return new OffloadedMessage(message, Completable.complete());
+ }
+ }
+
+ /**
+ * Saves each inline blob in {@code message} as an artifact and returns a copy of the
+ * message in which every blob is replaced by a file name placeholder.
+ *
+ * The given message is never modified: its parts list may be immutable, and a caller does not
+ * expect the message it passed to {@code runAsync} to be rewritten.
+ *
+ *
Only called when {@link #artifactService} is non-null.
+ */
+ private OffloadedMessage offloadInputBlobs(
+ Session session, Content message, String invocationId) {
+ // Save each inline blob as an artifact and replace it with a file name placeholder in a
+ // copy of the parts list. The caller's message is never modified.
+ List parts = new ArrayList<>(message.parts().get());
+ Completable saveArtifactsFlow = Completable.complete();
+ for (int i = 0; i < parts.size(); i++) {
+ Part part = parts.get(i);
+ if (part.inlineData().isEmpty()) {
+ continue;
+ }
+ String fileName = "artifact_" + invocationId + "_" + i;
+ saveArtifactsFlow =
+ saveArtifactsFlow.andThen(
+ this.artifactService
+ .saveArtifact(this.appName, session.userId(), session.id(), fileName, part)
+ .ignoreElement());
+
+ parts.set(
+ i, Part.fromText("Uploaded file: " + fileName + ". It has been saved to the artifacts"));
+ }
+ return new OffloadedMessage(
+ message.toBuilder().parts(ImmutableList.copyOf(parts)).build(), saveArtifactsFlow);
+ }
+
/**
* Appends a new user message to the session history with optional state delta.
*
+ * The given {@code newMessage} is never modified; when inline blobs are saved as artifacts,
+ * the appended event carries a copy in which the blob data is replaced by placeholders.
+ *
* @throws IllegalArgumentException if message has no parts.
*/
private Single appendNewMessageToSession(
@@ -357,38 +409,18 @@ private Single appendNewMessageToSession(
@Nullable Map stateDelta) {
checkArgument(newMessage.parts().isPresent(), "No parts in the new_message.");
- Completable saveArtifactsFlow = Completable.complete();
- if (this.artifactService != null && saveInputBlobsAsArtifacts) {
- // The runner directly saves the artifacts (if applicable) in the user message and replaces
- // the artifact data with a file name placeholder.
- for (int i = 0; i < newMessage.parts().get().size(); i++) {
- Part part = newMessage.parts().get().get(i);
- if (part.inlineData().isEmpty()) {
- continue;
- }
- String fileName = "artifact_" + invocationContext.invocationId() + "_" + i;
- saveArtifactsFlow =
- saveArtifactsFlow.andThen(
- this.artifactService
- .saveArtifact(this.appName, session.userId(), session.id(), fileName, part)
- .ignoreElement());
-
- newMessage
- .parts()
- .get()
- .set(
- i,
- Part.fromText(
- "Uploaded file: " + fileName + ". It has been saved to the artifacts"));
- }
- }
+ OffloadedMessage offloaded =
+ this.artifactService != null && saveInputBlobsAsArtifacts
+ ? offloadInputBlobs(session, newMessage, invocationContext.invocationId())
+ : OffloadedMessage.unchanged(newMessage);
+
// Appends only. We do not yield the event because it's not from the model.
Event.Builder eventBuilder =
Event.builder()
.id(Event.generateEventId())
.invocationId(invocationContext.invocationId())
.author("user")
- .content(newMessage);
+ .content(offloaded.message());
// Add state delta if provided
if (stateDelta != null && !stateDelta.isEmpty()) {
@@ -396,8 +428,9 @@ private Single appendNewMessageToSession(
EventActions.builder().stateDelta(new ConcurrentHashMap<>(stateDelta)).build());
}
- return saveArtifactsFlow.andThen(
- this.sessionService.appendEvent(session, eventBuilder.build()));
+ return offloaded
+ .artifactSaves()
+ .andThen(this.sessionService.appendEvent(session, eventBuilder.build()));
}
/** See {@link #runAsync(String, String, Content, RunConfig, Map)}. */
diff --git a/core/src/test/java/com/google/adk/runner/RunnerTest.java b/core/src/test/java/com/google/adk/runner/RunnerTest.java
index 38485a5a7..c6f1a3314 100644
--- a/core/src/test/java/com/google/adk/runner/RunnerTest.java
+++ b/core/src/test/java/com/google/adk/runner/RunnerTest.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2025 Google LLC
+ * Copyright 2026 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -50,6 +50,7 @@
import com.google.adk.apps.App;
import com.google.adk.apps.ResumabilityConfig;
import com.google.adk.artifacts.BaseArtifactService;
+import com.google.adk.artifacts.InMemoryArtifactService;
import com.google.adk.events.Event;
import com.google.adk.flows.llmflows.Functions;
import com.google.adk.models.LlmRequest;
@@ -3042,6 +3043,233 @@ public void runner_executesSaveArtifactFlow() {
assertThat(simplifyEvents(events.values())).containsExactly("test agent: from llm");
}
+ private static final String BLOB_MIME_TYPE = "example/octet-stream";
+ private static final String BLOB_PAYLOAD = "blob payload";
+ private static final String PLACEHOLDER_PREFIX = "Uploaded file: ";
+
+ private static Part blobPart() {
+ return Part.fromBytes(BLOB_PAYLOAD.getBytes(UTF_8), BLOB_MIME_TYPE);
+ }
+
+ /**
+ * A message whose parts list is immutable: {@code Content.Builder.parts(List)} stores the
+ * caller's list without copying it.
+ */
+ private static Content immutablePartsMessage() {
+ return Content.builder()
+ .role("user")
+ .parts(ImmutableList.of(Part.fromText("hello"), blobPart()))
+ .build();
+ }
+
+ /** A message whose parts list genai itself collected into an {@code ImmutableList}. */
+ private static Content partBuilderPartsMessage() {
+ return Content.builder()
+ .role("user")
+ .parts(Part.fromText("hello").toBuilder(), blobPart().toBuilder())
+ .build();
+ }
+
+ /**
+ * A message whose parts list accepts {@code set}. Used where the assertion is that the runner
+ * leaves the caller's message alone: with an immutable list the runner could not have modified it
+ * either way, so only a mutable one distinguishes copying from rewriting in place.
+ */
+ private static Content mutablePartsMessage() {
+ return Content.builder()
+ .role("user")
+ .parts(new ArrayList<>(ImmutableList.of(Part.fromText("hello"), blobPart())))
+ .build();
+ }
+
+ private static RunConfig saveInputBlobs(boolean enabled) {
+ return RunConfig.builder().saveInputBlobsAsArtifacts(enabled).build();
+ }
+
+ private Runner runnerWithArtifactService(BaseArtifactService artifactService) {
+ return Runner.builder()
+ .app(App.builder().name("test").rootAgent(agent).build())
+ .artifactService(artifactService)
+ .build();
+ }
+
+ private static ImmutableList artifactNames(
+ BaseArtifactService artifactService, String sessionId) {
+ return ImmutableList.copyOf(
+ artifactService.listArtifactKeys("test", "user", sessionId).blockingGet().filenames());
+ }
+
+ /** The parts of the user message that was actually appended to the session. */
+ private static List appendedUserParts(Runner runner, String sessionId) {
+ Session stored =
+ runner
+ .sessionService()
+ .getSession("test", "user", sessionId, Optional.empty())
+ .blockingGet();
+ return stored.events().stream()
+ .filter(event -> event.author().equals("user"))
+ .findFirst()
+ .flatMap(Event::content)
+ .flatMap(Content::parts)
+ .orElseThrow(() -> new AssertionError("No user message was appended to the session."));
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_immutablePartsList_savesArtifactAndCompletes() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+
+ var events =
+ runner
+ .runAsync("user", testSession.id(), immutablePartsMessage(), saveInputBlobs(true))
+ .test();
+
+ events.assertComplete();
+ assertThat(artifactNames(artifactService, testSession.id())).hasSize(1);
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_partBuilderPartsList_savesArtifactAndCompletes() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+
+ var events =
+ runner
+ .runAsync("user", testSession.id(), partBuilderPartsMessage(), saveInputBlobs(true))
+ .test();
+
+ events.assertComplete();
+ assertThat(artifactNames(artifactService, testSession.id())).hasSize(1);
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_doesNotModifyCallerMessage() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+ Content callerMessage = mutablePartsMessage();
+
+ var events =
+ runner.runAsync("user", testSession.id(), callerMessage, saveInputBlobs(true)).test();
+
+ events.assertComplete();
+ assertThat(callerMessage.parts().get().get(1).inlineData()).isPresent();
+ assertThat(callerMessage.parts().get().get(1).text()).isEmpty();
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_appendedEventReplacesBlobWithPlaceholder() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+
+ var unused =
+ runner
+ .runAsync("user", testSession.id(), immutablePartsMessage(), saveInputBlobs(true))
+ .test();
+
+ List appended = appendedUserParts(runner, testSession.id());
+ assertThat(appended).hasSize(2);
+ assertThat(appended.get(1).inlineData()).isEmpty();
+ assertThat(appended.get(1).text().orElse("")).startsWith(PLACEHOLDER_PREFIX);
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_storesBlobVerbatim() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+
+ var events =
+ runner
+ .runAsync("user", testSession.id(), immutablePartsMessage(), saveInputBlobs(true))
+ .test();
+
+ // Assert completion too: the artifact is written eagerly, before the parts list is rewritten,
+ // so
+ // without this the test would pass even on a run that died partway through.
+ events.assertComplete();
+ String fileName = artifactNames(artifactService, testSession.id()).get(0);
+ Part stored =
+ artifactService.loadArtifact("test", "user", testSession.id(), fileName).blockingGet();
+ assertThat(new String(stored.inlineData().get().data().get(), UTF_8)).isEqualTo(BLOB_PAYLOAD);
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_textOnlyMessage_passesThroughUnchanged() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+ Content callerMessage = Content.fromParts(Part.fromText("hello"));
+
+ var events =
+ runner.runAsync("user", testSession.id(), callerMessage, saveInputBlobs(true)).test();
+
+ events.assertComplete();
+ assertThat(artifactNames(artifactService, testSession.id())).isEmpty();
+ List appended = appendedUserParts(runner, testSession.id());
+ assertThat(appended).hasSize(1);
+ assertThat(appended.get(0).text()).hasValue("hello");
+ assertThat(callerMessage.parts().get().get(0).text()).hasValue("hello");
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_disabledWithTextOnlyMessage_passesThroughUnchanged() {
+ // The default path for every ordinary agent call: no blob, and the option at its default false.
+ // The runner must not touch the message at all.
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+ Content callerMessage = Content.fromParts(Part.fromText("hello"));
+
+ var events =
+ runner.runAsync("user", testSession.id(), callerMessage, saveInputBlobs(false)).test();
+
+ events.assertComplete();
+ assertThat(artifactNames(artifactService, testSession.id())).isEmpty();
+ List appended = appendedUserParts(runner, testSession.id());
+ assertThat(appended).hasSize(1);
+ assertThat(appended.get(0).text()).hasValue("hello");
+ assertThat(callerMessage.parts().get().get(0).text()).hasValue("hello");
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_disabled_keepsBlobAndSavesNothing() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+
+ var events =
+ runner
+ .runAsync("user", testSession.id(), immutablePartsMessage(), saveInputBlobs(false))
+ .test();
+
+ events.assertComplete();
+ assertThat(artifactNames(artifactService, testSession.id())).isEmpty();
+ assertThat(appendedUserParts(runner, testSession.id()).get(1).inlineData()).isPresent();
+ }
+
+ @Test
+ public void saveInputBlobsAsArtifacts_fromPartsConstruction_savesArtifactAndCompletes() {
+ BaseArtifactService artifactService = new InMemoryArtifactService();
+ Runner runner = runnerWithArtifactService(artifactService);
+ Session testSession = runner.sessionService().createSession("test", "user").blockingGet();
+
+ Content fromPartsMessage = Content.fromParts(Part.fromText("hello"), blobPart());
+
+ var events =
+ runner.runAsync("user", testSession.id(), fromPartsMessage, saveInputBlobs(true)).test();
+
+ events.assertComplete();
+ assertThat(artifactNames(artifactService, testSession.id())).hasSize(1);
+ List appended = appendedUserParts(runner, testSession.id());
+ assertThat(appended).hasSize(2);
+ assertThat(appended.get(1).inlineData()).isEmpty();
+ assertThat(appended.get(1).text().orElse("")).startsWith(PLACEHOLDER_PREFIX);
+ }
+
@Test
public void runAsync_partialEvent_streamedButNotPassedToSessionService() {
// The model streams a partial event followed by the final aggregated event in one turn.