refactor: yield the sliding-window compaction event to the runner

Move event persistence to the main runner loop, the runtime's
synchronization point: sliding-window compaction now yields its event
instead of appending it, and the runner appends it like any other event.

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

Author: GWeale

src/google/adk/apps/compaction.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import logging
+from typing import AsyncGenerator
 
 from google.genai import types
 
@@ -443,7 +444,7 @@ async def _run_compaction_for_sliding_window(
     session_service: BaseSessionService,
     *,
     skip_token_compaction: bool = False,
-):
+) -> AsyncGenerator[Event, None]:
   """Runs compaction for SlidingWindowCompactor.
 
   This method implements the sliding window compaction logic. It determines
@@ -521,30 +522,35 @@ async def _run_compaction_for_sliding_window(
   Args:
     app: The application instance.
     session: The session containing events to compact.
-    session_service: The session service for appending events.
+    session_service: The session service, used by the token-threshold fallback.
     skip_token_compaction: Whether to skip token-threshold compaction.
+
+  Yields:
+    The sliding-window compaction event, if one is produced. The caller (the
+    runner loop) is responsible for appending it to the session, so that
+    persistence of this event stays at the runtime's synchronization point.
   """
   events = session.events
   if not events:
-    return None
+    return
 
   config = app.events_compaction_config
   if config is None:
-    return None
+    return
 
   # Prefer token-threshold compaction if configured and triggered.
   if not skip_token_compaction and _has_token_threshold_config(config):
     token_compacted = await _run_compaction_for_token_threshold(
         app, session, session_service
     )
     if token_compacted:
-      return None
+      return
 
   if not _has_sliding_window_config(config):
-    return None
+    return
 
   if config.compaction_interval is None or config.overlap_size is None:
-    return None
+    return
 
   # Find the last compaction event and its range.
   last_compacted_end_timestamp = 0.0
@@ -573,7 +579,7 @@ async def _run_compaction_for_sliding_window(
   ]
 
   if len(new_invocation_ids) < config.compaction_interval:
-    return None  # Not enough new invocations to trigger compaction.
+    return  # Not enough new invocations to trigger compaction.
 
   # Determine the range of invocations to compact.
   # The end of the compaction range is the last of the new invocations.
@@ -613,20 +619,20 @@ async def _run_compaction_for_sliding_window(
       events_to_compact = _longest_self_contained_prefix(events_to_compact)
 
   if not events_to_compact:
-    return None
+    return
 
   if app.root_agent is None:
-    return None
+    return
   _ensure_compaction_summarizer(config=config, agent=app.root_agent)
   if config.summarizer is None:
-    return None
+    return
 
   compaction_event = await _summarize_events_with_trace(
       session=session,
       config=config,
       events_to_compact=events_to_compact,
       trigger='sliding_window',
   )
-  if compaction_event:
-    await session_service.append_event(session=session, event=compaction_event)
   logger.debug('Event compactor finished.')
+  if compaction_event:
+    yield compaction_event

src/google/adk/runners.py

@@ -584,12 +584,18 @@ async def _drive_root_node():
           logger.debug('Running event compactor.')
           from google.adk.apps.compaction import _run_compaction_for_sliding_window
 
-          await _run_compaction_for_sliding_window(
-              self.app,
-              session,
-              self.session_service,
-              skip_token_compaction=ic.token_compaction_checked,
-          )
+          async with aclosing(
+              _run_compaction_for_sliding_window(
+                  self.app,
+                  session,
+                  self.session_service,
+                  skip_token_compaction=ic.token_compaction_checked,
+              )
+          ) as compaction_events:
+            async for compaction_event in compaction_events:
+              await self.session_service.append_event(
+                  session=session, event=compaction_event
+              )
 
   async def _run_node_live(
       self,
@@ -1130,12 +1136,18 @@ async def execute(ctx: InvocationContext) -> AsyncGenerator[Event]:
           logger.debug('Running event compactor.')
           from google.adk.apps.compaction import _run_compaction_for_sliding_window
 
-          await _run_compaction_for_sliding_window(
-              self.app,
-              invocation_context.session,
-              self.session_service,
-              skip_token_compaction=invocation_context.token_compaction_checked,
-          )
+          async with aclosing(
+              _run_compaction_for_sliding_window(
+                  self.app,
+                  invocation_context.session,
+                  self.session_service,
+                  skip_token_compaction=invocation_context.token_compaction_checked,
+              )
+          ) as compaction_events:
+            async for compaction_event in compaction_events:
+              await self.session_service.append_event(
+                  session=invocation_context.session, event=compaction_event
+              )
 
     async with aclosing(_run_with_trace(new_message, invocation_id)) as agen:
       async for event in agen: