fix(eval): count logical Live requests across response segments

src/google/adk/evaluation/_efficiency_evaluators.py

@@ -199,6 +199,8 @@ def __init__(
     self._eval_metric = eval_metric
 
   def _compute_value(self, invocation: Invocation) -> Optional[float]:
+    if invocation.inference_call_count is not None:
+      return float(invocation.inference_call_count)
     if not isinstance(invocation.intermediate_data, InvocationEvents):
       return None
     model_events = [

src/google/adk/evaluation/eval_case.py

@@ -116,6 +116,14 @@ class Invocation(EvalBaseModel):
   creation_timestamp: float = 0.0
   """Timestamp for the current invocation, primarily intended for debugging purposes."""
 
+  inference_call_count: Optional[int] = Field(default=None, ge=0)
+  """Number of model requests reconstructed from streamed event boundaries.
+
+  Set during session-event conversion, before turn completion markers are
+  removed. None keeps event-based counting for older eval files and unary
+  invocations. Usage and intermediate content remain on their original events.
+  """
+
   duration: Optional[float] = None
   """Wall-clock seconds this invocation took, measured while it ran.
 

src/google/adk/evaluation/evaluation_generator.py

@@ -1028,7 +1028,10 @@ def convert_events_to_eval_invocations(
               final_response = event.content
               final_event = event
 
-          should_add_event = event.grounding_metadata is not None
+          should_add_event = (
+              event.grounding_metadata is not None
+              or event.usage_metadata is not None
+          )
           for p in event.content.parts:
             if (
                 p.function_call
@@ -1040,7 +1043,10 @@ def convert_events_to_eval_invocations(
               break
           if should_add_event:
             events_to_add.append(event)
-        elif event.grounding_metadata is not None:
+        elif (
+            event.grounding_metadata is not None
+            or event.usage_metadata is not None
+        ):
           events_to_add.append(event)
 
       invocation_events = []
@@ -1062,22 +1068,138 @@ def convert_events_to_eval_invocations(
                 model_version=e.model_version,
             )
         )
+      event_pairs = list(zip(events_to_add, invocation_events))
+      merged_events = []
+      merged_event_sources: list[Event] = []
+      for index, (event, invocation_event) in enumerate(event_pairs):
+        if (
+            not (event.content and event.content.parts)
+            and event.grounding_metadata is None
+            and event.usage_metadata is not None
+        ):
+          # Live usage can arrive before or after content. Merge into an event
+          # already counted as a model call, without changing the input events
+          # or overwriting usage reported by another call.
+          # TODO: Index by author, model and Live session if long invocations
+          # make this quadratic search expensive.
+          model_event = next(
+              (
+                  candidate
+                  for candidate_source, candidate in (
+                      event_pairs[:index][::-1] + event_pairs[index + 1 :]
+                  )
+                  if candidate_source.live_session_id == event.live_session_id
+                  and candidate.model_version is not None
+                  and candidate.author == event.author
+                  and candidate.usage_metadata is None
+                  and (
+                      event.model_version is None
+                      or candidate.model_version == event.model_version
+                  )
+              ),
+              None,
+          )
+          if model_event is not None:
+            model_event.usage_metadata = event.usage_metadata
+            continue
+        merged_events.append(invocation_event)
+        merged_event_sources.append(event)
       invocations.append(
           Invocation(
               invocation_id=invocation_id,
               user_content=user_content,
               final_response=final_response,
               intermediate_data=InvocationEvents(
-                  invocation_events=invocation_events
+                  invocation_events=merged_events
               ),
               creation_timestamp=invocation_timestamp,
+              inference_call_count=EvaluationGenerator._count_streamed_inference_calls(
+                  events, merged_event_sources
+              ),
               duration=(durations_per_invocation or {}).get(invocation_id),
               app_details=app_details,
           )
       )
 
     return invocations
 
+  @staticmethod
+  def _count_streamed_inference_calls(
+      events: list[Event], merged_event_sources: list[Event]
+  ) -> Optional[int]:
+    """Counts requests before stream completion markers are projected out."""
+    stream_keys = {
+        (event.author, event.live_session_id)
+        for event in events
+        if event.live_session_id is not None
+        or event.turn_complete is not None
+        or event.partial
+    }
+    if not stream_keys:
+      # Old eval files and ordinary unary responses retain event-based counting.
+      return None
+
+    # Standalone usage merged into a model event must not start another call.
+    # Keep the raw sequence for boundaries, including markers absent from evals.
+    model_event_ids = {id(event) for event in merged_event_sources}
+    count = 0
+    active_sessions: set[tuple[str, Optional[str]]] = set()
+    for event in events:
+      if event.author.lower() == _USER_AUTHOR:
+        # Live input transcriptions can interleave with the model response.
+        # Their session ID survives normalization into user text content.
+        if event.input_transcription is None and event.live_session_id is None:
+          active_sessions.clear()
+        continue
+
+      has_model_data = (
+          event.model_version is not None or event.usage_metadata is not None
+      )
+      # Completion/interrupt metadata does not start a new request. Text flushes
+      # carry model_version so sessions that omit partial chunks still count.
+      has_payload = (
+          bool(event.content and event.content.parts)
+          or event.usage_metadata is not None
+          or (
+              event.grounding_metadata is not None
+              and not event.turn_complete
+              and not event.interrupted
+          )
+      )
+      key = (event.author, event.live_session_id)
+      if id(event) in model_event_ids and has_model_data and has_payload:
+        if key not in stream_keys:
+          # Unary responses still count individually in mixed invocations,
+          # including usage-only responses with no content.
+          count += 1
+        elif key not in active_sessions:
+          count += 1
+          active_sessions.add(key)
+
+      if event.get_function_responses() and event.live_session_id is None:
+        # Legacy tool responses omit connection identity. Preserve their
+        # author-level boundary without closing other authors' requests.
+        active_sessions = {
+            active_key
+            for active_key in active_sessions
+            if active_key[0] != event.author
+        }
+      elif (
+          event.turn_complete
+          or event.interrupted
+          or event.get_function_responses()
+          or (
+              event.live_session_id is None
+              and event.turn_complete is None
+              and not event.partial
+          )
+      ):
+        # SSE has a final response, possibly only usage/error metadata. Live
+        # can flush text before tools, so it needs a real connection boundary.
+        active_sessions.discard(key)
+
+    return count
+
   @staticmethod
   def _get_app_details_by_invocation_id(
       events: list[Event], request_intercepter: _RequestIntercepterPlugin

src/google/adk/live/_live_llm_flow.py

@@ -435,12 +435,14 @@ async def postprocess_live_flow(
 
   # Handle transcription events ONCE per llm_response, outside the event loop
   if llm_response.input_transcription:
+    model_response_event.live_session_id = llm_response.live_session_id
     model_response_event.input_transcription = llm_response.input_transcription
     model_response_event.partial = llm_response.partial
     yield model_response_event
     return
 
   if llm_response.output_transcription:
+    model_response_event.live_session_id = llm_response.live_session_id
     model_response_event.output_transcription = (
         llm_response.output_transcription
     )

src/google/adk/models/gemini_llm_connection.py

@@ -280,6 +280,7 @@ def __build_full_text_response(
         grounding_metadata=grounding_metadata,
         interrupted=interrupted,
         partial=False,
+        model_version=self._model_version,
         live_session_id=self._gemini_session.session_id,
     )