feat: Add OpenTelemetry tracing for event compaction

This change introduces a new helper function `_summarize_events_with_trace` to wrap the event summarization process. This function creates a trace span and records metadata about the compaction trigger, summarizer type, event count, and configuration parameters both before and after the summarization occurs. The existing compaction logic is updated to use this new traced function.

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

Author: GWeale

src/google/adk/apps/compaction.py

@@ -22,13 +22,47 @@
 from ..events.event import Event
 from ..sessions.base_session_service import BaseSessionService
 from ..sessions.session import Session
+from ..telemetry.tracing import _build_compaction_attributes
+from ..telemetry.tracing import _build_compaction_result_attributes
+from ..telemetry.tracing import tracer
 from .app import App
 from .app import EventsCompactionConfig
 from .llm_event_summarizer import LlmEventSummarizer
 
 logger = logging.getLogger('google_adk.' + __name__)
 
 
+async def _summarize_events_with_trace(
+    *,
+    session: Session,
+    config: EventsCompactionConfig,
+    events_to_compact: list[Event],
+    trigger: str,
+) -> Event | None:
+  """Summarizes events within a trace span labeled for compaction."""
+  if config.summarizer is None:
+    return None
+
+  attributes = _build_compaction_attributes(
+      session_id=session.id,
+      trigger=trigger,
+      summarizer_type=type(config.summarizer).__name__,
+      event_count=len(events_to_compact),
+      token_threshold=config.token_threshold,
+      event_retention_size=config.event_retention_size,
+      compaction_interval=config.compaction_interval,
+      overlap_size=config.overlap_size,
+  )
+
+  with tracer.start_as_current_span(f'compact_events {trigger}') as span:
+    span.set_attributes(attributes)
+    compaction_event = await config.summarizer.maybe_summarize_events(
+        events=events_to_compact
+    )
+    span.set_attributes(_build_compaction_result_attributes(compaction_event))
+    return compaction_event
+
+
 def _count_text_chars_in_content(content: types.Content | None) -> int:
   """Returns the number of text characters in a content object."""
   total_chars = 0
@@ -383,8 +417,11 @@ async def _run_compaction_for_token_threshold_config(
   if config.summarizer is None:
     return False
 
-  compaction_event = await config.summarizer.maybe_summarize_events(
-      events=events_to_compact
+  compaction_event = await _summarize_events_with_trace(
+      session=session,
+      config=config,
+      events_to_compact=events_to_compact,
+      trigger='token_threshold',
   )
   if compaction_event:
     await session_service.append_event(session=session, event=compaction_event)
@@ -602,8 +639,11 @@ async def _run_compaction_for_sliding_window(
   if config.summarizer is None:
     return None
 
-  compaction_event = await config.summarizer.maybe_summarize_events(
-      events=events_to_compact
+  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)

src/google/adk/telemetry/tracing.py

@@ -445,6 +445,58 @@ def trace_send_data(
     span.set_attribute('gcp.vertex.agent.data', '{}')
 
 
+def _build_compaction_attributes(
+    *,
+    session_id: str,
+    trigger: str,
+    summarizer_type: str,
+    event_count: int,
+    token_threshold: int | None = None,
+    event_retention_size: int | None = None,
+    compaction_interval: int | None = None,
+    overlap_size: int | None = None,
+) -> dict[str, AttributeValue]:
+  """Builds span attributes for event compaction tracing."""
+  attributes: dict[str, AttributeValue] = {
+      GEN_AI_SYSTEM: _guess_gemini_system_name(),
+      GEN_AI_OPERATION_NAME: 'compact_events',
+      GEN_AI_CONVERSATION_ID: session_id,
+      'gen_ai.compaction.trigger': trigger,
+      'gen_ai.compaction.summarizer_type': summarizer_type,
+      'gen_ai.compaction.event_count': event_count,
+  }
+  if token_threshold is not None:
+    attributes['gen_ai.compaction.token_threshold'] = token_threshold
+  if event_retention_size is not None:
+    attributes['gen_ai.compaction.event_retention_size'] = event_retention_size
+  if compaction_interval is not None:
+    attributes['gen_ai.compaction.compaction_interval'] = compaction_interval
+  if overlap_size is not None:
+    attributes['gen_ai.compaction.overlap_size'] = overlap_size
+  return attributes
+
+
+def _build_compaction_result_attributes(
+    compacted_event: Event | None,
+) -> dict[str, AttributeValue]:
+  """Builds span attributes for compaction result."""
+  if (
+      compacted_event is None
+      or compacted_event.actions is None
+      or compacted_event.actions.compaction is None
+  ):
+    return {}
+
+  attributes: dict[str, AttributeValue] = {}
+  compaction = compacted_event.actions.compaction
+  attributes['gen_ai.compaction.result_event_id'] = compacted_event.id
+  if compaction.start_timestamp is not None:
+    attributes['gen_ai.compaction.start_timestamp'] = compaction.start_timestamp
+  if compaction.end_timestamp is not None:
+    attributes['gen_ai.compaction.end_timestamp'] = compaction.end_timestamp
+  return attributes
+
+
 def _build_llm_request_for_trace(llm_request: LlmRequest) -> dict[str, Any]:
   """Builds a dictionary representation of the LLM request for tracing.
 

tests/unittests/apps/test_compaction.py

@@ -19,6 +19,7 @@
 from google.adk.agents.base_agent import BaseAgent
 from google.adk.apps.app import App
 from google.adk.apps.app import EventsCompactionConfig
+from google.adk.apps.base_events_summarizer import BaseEventsSummarizer
 from google.adk.apps.compaction import _run_compaction_for_sliding_window
 import google.adk.apps.compaction as compaction_module
 from google.adk.apps.llm_event_summarizer import LlmEventSummarizer
@@ -31,10 +32,78 @@
 from google.genai import types
 from google.genai.types import Content
 from google.genai.types import Part
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 from pydantic import ValidationError
 import pytest
 
 
+class _StubSummarizer(BaseEventsSummarizer):
+
+  def __init__(self, compacted_event: Event | None):
+    self._compacted_event = compacted_event
+
+  async def maybe_summarize_events(
+      self, *, events: list[Event]
+  ) -> Event | None:
+    del events
+    return self._compacted_event
+
+
+def _create_trace_test_event(
+    *,
+    timestamp: float,
+    invocation_id: str,
+    text: str,
+    prompt_token_count: int | None = None,
+) -> Event:
+  usage_metadata = None
+  if prompt_token_count is not None:
+    usage_metadata = types.GenerateContentResponseUsageMetadata(
+        prompt_token_count=prompt_token_count
+    )
+  return Event(
+      timestamp=timestamp,
+      invocation_id=invocation_id,
+      author='user',
+      content=Content(role='user', parts=[Part(text=text)]),
+      usage_metadata=usage_metadata,
+  )
+
+
+def _create_trace_compacted_event(
+    *, start_ts: float, end_ts: float, summary_text: str
+) -> Event:
+  compaction = EventCompaction(
+      start_timestamp=start_ts,
+      end_timestamp=end_ts,
+      compacted_content=Content(role='model', parts=[Part(text=summary_text)]),
+  )
+  return Event(
+      id='compacted-event-id',
+      timestamp=end_ts,
+      author='compactor',
+      content=compaction.compacted_content,
+      actions=EventActions(compaction=compaction),
+      invocation_id='compacted-invocation-id',
+  )
+
+
+@pytest.fixture
+def span_exporter(monkeypatch: pytest.MonkeyPatch) -> InMemorySpanExporter:
+  tracer_provider = TracerProvider()
+  span_exporter = InMemorySpanExporter()
+  tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
+  real_tracer = tracer_provider.get_tracer(__name__)
+  monkeypatch.setattr(
+      compaction_module.tracer,
+      'start_as_current_span',
+      real_tracer.start_as_current_span,
+  )
+  return span_exporter
+
+
 @pytest.mark.parametrize(
     'env_variables', ['GOOGLE_AI', 'VERTEX'], indirect=True
 )
@@ -923,6 +992,128 @@ def test_get_contents_compaction_at_beginning(self):
     actual_texts = [c.parts[0].text for c in result_contents]
     self.assertEqual(actual_texts, expected_texts)
 
+
+@pytest.mark.asyncio
+async def test_run_compaction_for_token_threshold_adds_summary_trace(
+    span_exporter: InMemorySpanExporter,
+):
+  session = Session(
+      app_name='app',
+      user_id='user',
+      id='session-id',
+      events=[
+          _create_trace_test_event(
+              timestamp=1.0, invocation_id='inv1', text='e1'
+          ),
+          _create_trace_test_event(
+              timestamp=2.0, invocation_id='inv2', text='e2'
+          ),
+          _create_trace_test_event(
+              timestamp=3.0,
+              invocation_id='inv3',
+              text='e3',
+              prompt_token_count=100,
+          ),
+      ],
+  )
+  session_service = AsyncMock(spec=BaseSessionService)
+  compacted_event = _create_trace_compacted_event(
+      start_ts=1.0, end_ts=2.0, summary_text='summary'
+  )
+  summarizer = _StubSummarizer(compacted_event)
+  config = EventsCompactionConfig(
+      summarizer=summarizer,
+      compaction_interval=999,
+      overlap_size=0,
+      token_threshold=50,
+      event_retention_size=1,
+  )
+
+  compacted = (
+      await (
+          compaction_module._run_compaction_for_token_threshold_config(
+              config=config,
+              session=session,
+              session_service=session_service,
+              agent=Mock(spec=BaseAgent),
+          )
+      )
+  )
+
+  assert compacted is True
+  spans = span_exporter.get_finished_spans()
+  summary_span = next(
+      span for span in spans if span.name == 'compact_events token_threshold'
+  )
+  assert summary_span.attributes['gen_ai.conversation.id'] == 'session-id'
+  assert (
+      summary_span.attributes['gen_ai.compaction.trigger'] == 'token_threshold'
+  )
+  assert summary_span.attributes['gen_ai.compaction.event_count'] == 2
+  assert summary_span.attributes['gen_ai.compaction.token_threshold'] == 50
+  assert summary_span.attributes['gen_ai.compaction.event_retention_size'] == 1
+  assert (
+      summary_span.attributes['gen_ai.compaction.result_event_id']
+      == 'compacted-event-id'
+  )
+
+
+@pytest.mark.asyncio
+async def test_run_compaction_for_sliding_window_adds_summary_trace(
+    span_exporter: InMemorySpanExporter,
+):
+  compacted_event = _create_trace_compacted_event(
+      start_ts=1.0, end_ts=4.0, summary_text='summary'
+  )
+  summarizer = _StubSummarizer(compacted_event)
+  app = App(
+      name='test',
+      root_agent=Mock(spec=BaseAgent),
+      events_compaction_config=EventsCompactionConfig(
+          summarizer=summarizer,
+          compaction_interval=2,
+          overlap_size=1,
+      ),
+  )
+  session = Session(
+      app_name='test',
+      user_id='u1',
+      id='session-id',
+      events=[
+          _create_trace_test_event(
+              timestamp=1.0, invocation_id='inv1', text='e1'
+          ),
+          _create_trace_test_event(
+              timestamp=2.0, invocation_id='inv2', text='e2'
+          ),
+          _create_trace_test_event(
+              timestamp=3.0, invocation_id='inv3', text='e3'
+          ),
+          _create_trace_test_event(
+              timestamp=4.0, invocation_id='inv4', text='e4'
+          ),
+      ],
+  )
+  session_service = AsyncMock(spec=BaseSessionService)
+
+  await _run_compaction_for_sliding_window(app, session, session_service)
+
+  spans = span_exporter.get_finished_spans()
+  summary_span = next(
+      span for span in spans if span.name == 'compact_events sliding_window'
+  )
+  assert summary_span.attributes['gen_ai.conversation.id'] == 'session-id'
+  assert (
+      summary_span.attributes['gen_ai.compaction.trigger'] == 'sliding_window'
+  )
+  assert summary_span.attributes['gen_ai.compaction.event_count'] == 4
+  assert summary_span.attributes['gen_ai.compaction.compaction_interval'] == 2
+  assert summary_span.attributes['gen_ai.compaction.overlap_size'] == 1
+  assert (
+      summary_span.attributes['gen_ai.compaction.result_event_id']
+      == 'compacted-event-id'
+  )
+
   async def test_sliding_window_excludes_pending_function_call_events(self):
     """Sliding-window compaction stops before pending function calls."""
     app = App(