feat: Integrate event compaction in Java ADK runner

This change integrates the event compaction mechanism into the ADK Runner. The Contents class is updated to filter out events that fall within the timestamp range of a compaction event, replacing them with the compacted summary. The Runner now uses a SlidingWindowEventCompactor to periodically summarize events based on the provided EventsCompactionConfig. Tests are added to verify the compaction logic in both Contents and Runner.

PiperOrigin-RevId: 851559226

Author: google-genai-bot

core/src/main/java/com/google/adk/flows/llmflows/Contents.java

@@ -23,9 +23,11 @@
 import com.google.adk.agents.InvocationContext;
 import com.google.adk.agents.LlmAgent;
 import com.google.adk.events.Event;
+import com.google.adk.events.EventCompaction;
 import com.google.adk.models.LlmRequest;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import com.google.genai.types.Content;
 import com.google.genai.types.FunctionCall;
 import com.google.genai.types.FunctionResponse;
@@ -36,6 +38,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.ListIterator;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -99,24 +102,23 @@ private ImmutableList<Content> getCurrentTurnContents(
   private ImmutableList<Content> getContents(
       Optional<String> currentBranch, List<Event> events, String agentName, String modelName) {
     List<Event> filteredEvents = new ArrayList<>();
+    boolean hasCompactEvent = false;
 
     // Filter the events, leaving the contents and the function calls and responses from the current
     // agent.
     for (Event event : events) {
-      // Skip events without content, or generated neither by user nor by model or has empty text.
-      // E.g. events purely for mutating session states.
-      if (event.content().isEmpty()) {
+      if (event.actions().compaction().isPresent()) {
+        // Always include the compaction event
+        hasCompactEvent = true;
+        filteredEvents.add(event);
         continue;
       }
-      var content = event.content().get();
-      if (content.role().isEmpty()
-          || content.role().get().isEmpty()
-          || content.parts().isEmpty()
-          || content.parts().get().isEmpty()
-          || content.parts().get().get(0).text().map(String::isEmpty).orElse(false)) {
+
+      // Skip events without content, or generated neither by user nor by model or has empty text.
+      // E.g. events purely for mutating session states.
+      if (isEmptyContent(event)) {
         continue;
       }
-
       if (!isEventBelongsToBranch(currentBranch, event)) {
         continue;
       }
@@ -133,6 +135,10 @@ private ImmutableList<Content> getContents(
       }
     }
 
+    if (hasCompactEvent) {
+      filteredEvents = processCompactionEvent(filteredEvents);
+    }
+
     List<Event> resultEvents = rearrangeEventsForLatestFunctionResponse(filteredEvents);
     resultEvents = rearrangeEventsForAsyncFunctionResponsesInHistory(resultEvents, modelName);
 
@@ -142,6 +148,67 @@ private ImmutableList<Content> getContents(
         .collect(toImmutableList());
   }
 
+  /**
+   * Check if an event has missing or empty content.
+   *
+   * <p>This can happen to the events that only changed session state. When both content and
+   * transcriptions are empty, the event will be considered as empty. The content is considered
+   * empty if none of its parts contain text, inline data, file data, function call, or function
+   * response. Parts with only thoughts are also considered empty.
+   *
+   * @param event the event to check.
+   * @return {@code true} if the event is considered to have empty content, {@code false} otherwise.
+   */
+  private boolean isEmptyContent(Event event) {
+    if (event.content().isEmpty()) {
+      return true;
+    }
+    var content = event.content().get();
+    return (content.role().isEmpty()
+        || content.role().get().isEmpty()
+        || content.parts().isEmpty()
+        || content.parts().get().isEmpty()
+        || content.parts().get().get(0).text().map(String::isEmpty).orElse(false));
+  }
+
+  /**
+   * Filters events that are covered by compaction events.
+   *
+   * @param events the list of event to filter.
+   * @return a new list with compaction applied.
+   */
+  private List<Event> processCompactionEvent(List<Event> events) {
+    List<Event> result = new ArrayList<>();
+    ListIterator<Event> iter = events.listIterator(events.size());
+    Long lastCompactionStartTime = null;
+
+    while (iter.hasPrevious()) {
+      Event event = iter.previous();
+      EventCompaction compaction = event.actions().compaction().orElse(null);
+      if (compaction == null) {
+        if (lastCompactionStartTime == null || event.timestamp() < lastCompactionStartTime) {
+          result.add(event);
+        }
+        continue;
+      }
+      // Create a new event for the compaction event in the result.
+      result.add(
+          Event.builder()
+              .timestamp(compaction.endTimestamp())
+              .author("model")
+              .content(compaction.compactedContent())
+              .branch(event.branch())
+              .invocationId(event.invocationId())
+              .actions(event.actions())
+              .build());
+      lastCompactionStartTime =
+          lastCompactionStartTime == null
+              ? compaction.startTimestamp()
+              : Long.min(lastCompactionStartTime, compaction.startTimestamp());
+    }
+    return Lists.reverse(result);
+  }
+
   /** Whether the event is a reply from another agent. */
   private static boolean isOtherAgentReply(String agentName, Event event) {
     return !agentName.isEmpty()

core/src/main/java/com/google/adk/runner/Runner.java

@@ -29,11 +29,15 @@
 import com.google.adk.events.EventActions;
 import com.google.adk.flows.llmflows.ResumabilityConfig;
 import com.google.adk.memory.BaseMemoryService;
+import com.google.adk.models.Model;
 import com.google.adk.plugins.BasePlugin;
 import com.google.adk.plugins.PluginManager;
 import com.google.adk.sessions.BaseSessionService;
 import com.google.adk.sessions.InMemorySessionService;
 import com.google.adk.sessions.Session;
+import com.google.adk.summarizer.EventsCompactionConfig;
+import com.google.adk.summarizer.LlmEventSummarizer;
+import com.google.adk.summarizer.SlidingWindowEventCompactor;
 import com.google.adk.tools.BaseTool;
 import com.google.adk.tools.FunctionTool;
 import com.google.adk.utils.CollectionUtils;
@@ -68,6 +72,7 @@ public class Runner {
   @Nullable private final BaseMemoryService memoryService;
   private final PluginManager pluginManager;
   private final ResumabilityConfig resumabilityConfig;
+  @Nullable private final EventsCompactionConfig eventsCompactionConfig;
 
   /** Builder for {@link Runner}. */
   public static class Builder {
@@ -78,6 +83,7 @@ public static class Builder {
     @Nullable private BaseMemoryService memoryService = null;
     private List<BasePlugin> plugins = ImmutableList.of();
     private ResumabilityConfig resumabilityConfig = new ResumabilityConfig();
+    @Nullable private EventsCompactionConfig eventsCompactionConfig;
 
     @CanIgnoreReturnValue
     public Builder agent(BaseAgent agent) {
@@ -121,6 +127,12 @@ public Builder resumabilityConfig(ResumabilityConfig resumabilityConfig) {
       return this;
     }
 
+    @CanIgnoreReturnValue
+    public Builder eventsCompactionConfig(EventsCompactionConfig eventsCompactionConfig) {
+      this.eventsCompactionConfig = eventsCompactionConfig;
+      return this;
+    }
+
     public Runner build() {
       if (agent == null) {
         throw new IllegalStateException("Agent must be provided.");
@@ -141,7 +153,8 @@ public Runner build() {
           sessionService,
           memoryService,
           plugins,
-          resumabilityConfig);
+          resumabilityConfig,
+          eventsCompactionConfig);
     }
   }
 
@@ -208,13 +221,43 @@ public Runner(
       @Nullable BaseMemoryService memoryService,
       List<BasePlugin> plugins,
       ResumabilityConfig resumabilityConfig) {
+    this(
+        agent,
+        appName,
+        artifactService,
+        sessionService,
+        memoryService,
+        plugins,
+        resumabilityConfig,
+        null);
+  }
+
+  /**
+   * Creates a new {@code Runner} with a list of plugins and resumability config.
+   *
+   * @deprecated Use {@link Runner.Builder} instead.
+   */
+  @Deprecated
+  protected Runner(
+      BaseAgent agent,
+      String appName,
+      BaseArtifactService artifactService,
+      BaseSessionService sessionService,
+      @Nullable BaseMemoryService memoryService,
+      List<BasePlugin> plugins,
+      ResumabilityConfig resumabilityConfig,
+      @Nullable EventsCompactionConfig eventsCompactionConfig) {
     this.agent = agent;
     this.appName = appName;
     this.artifactService = artifactService;
     this.sessionService = sessionService;
     this.memoryService = memoryService;
     this.pluginManager = new PluginManager(plugins);
     this.resumabilityConfig = resumabilityConfig;
+    this.eventsCompactionConfig =
+        Optional.ofNullable(eventsCompactionConfig)
+            .map(c -> populateBaseEventSummarizer(agent, c))
+            .orElse(null);
   }
 
   /**
@@ -491,7 +534,10 @@ public Flowable<Event> runAsync(
                                                       Completable.defer(
                                                           () ->
                                                               pluginManager.runAfterRunCallback(
-                                                                  contextWithUpdatedSession)));
+                                                                  contextWithUpdatedSession)))
+                                                  .concatWith(
+                                                      Completable.defer(
+                                                          () -> compactEvents(updatedSession)));
                                             });
                                   }))
                   .doOnError(
@@ -507,6 +553,28 @@ public Flowable<Event> runAsync(
     }
   }
 
+  private Completable compactEvents(Session session) {
+    EventsCompactionConfig config = eventsCompactionConfig;
+    if (config == null) {
+      return Completable.complete();
+    }
+    config =
+        new EventsCompactionConfig(
+            config.compactionInterval(),
+            config.overlapSize(),
+            config
+                .summarizer()
+                .or(
+                    () ->
+                        Optional.of(agent())
+                            .filter(LlmAgent.class::isInstance)
+                            .map(LlmAgent.class::cast)
+                            .map(LlmAgent::resolvedModel)
+                            .flatMap(Model::model)
+                            .map(LlmEventSummarizer::new)));
+    return new SlidingWindowEventCompactor(config).compact(session, sessionService);
+  }
+
   private void copySessionStates(Session source, Session target) {
     // TODO: remove this hack when deprecating all runAsync with Session.
     for (var entry : source.state().entrySet()) {
@@ -761,5 +829,27 @@ private boolean hasLiveRequestQueueParameter(FunctionTool functionTool) {
         .anyMatch(parameter -> parameter.getType().equals(LiveRequestQueue.class));
   }
 
+  /**
+   * Creates a new {@link EventsCompactionConfig} based on the given configuration. If the {@link
+   * com.google.adk.summarizer.BaseEventSummarizer} is missing, it will be default to the {@link
+   * LlmEventSummarizer} using the same model as the LLM base agent.
+   */
+  private static EventsCompactionConfig populateBaseEventSummarizer(
+      BaseAgent agent, EventsCompactionConfig config) {
+    return new EventsCompactionConfig(
+        config.compactionInterval(),
+        config.overlapSize(),
+        config
+            .summarizer()
+            .or(
+                () ->
+                    Optional.of(agent)
+                        .filter(LlmAgent.class::isInstance)
+                        .map(LlmAgent.class::cast)
+                        .map(LlmAgent::resolvedModel)
+                        .flatMap(Model::model)
+                        .map(LlmEventSummarizer::new)));
+  }
+
   // TODO: run statelessly
 }

core/src/main/java/com/google/adk/summarizer/SlidingWindowEventCompactor.java

@@ -12,13 +12,17 @@
 import java.util.List;
 import java.util.ListIterator;
 import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * This class performs events compaction in a sliding window fashion based on the {@link
  * EventsCompactionConfig}.
  */
 public final class SlidingWindowEventCompactor implements EventCompactor {
 
+  private static final Logger logger = LoggerFactory.getLogger(SlidingWindowEventCompactor.class);
+
   private final EventsCompactionConfig config;
   private final BaseEventSummarizer summarizer;
 
@@ -80,6 +84,8 @@ public SlidingWindowEventCompactor(EventsCompactionConfig config) {
    */
   @Override
   public Completable compact(Session session, BaseSessionService sessionService) {
+    logger.debug("Running event compaction for session {}", session.id());
+
     return Completable.fromMaybe(
         getCompactionEvents(session)
             .flatMap(summarizer::summarizeEvents)