fix: classify plugin output before persisting events

Merge #7185

Fixes #7184

PiperOrigin-RevId: 988679252

Author: jaywang172

src/google/adk/runners.py

@@ -764,6 +764,11 @@ async def _consume_event_queue(
             f'Unexpected node event queue item: {type(event_or_done).__name__}'
         )
       event = event_or_done
+      output_event = await self._process_event_with_plugin_callbacks(
+          invocation_context=ic,
+          event=event,
+      )
+
       # When an LlmAgent node uses ``message_as_output`` (no
       # ``output_schema``), the wrapper sets both ``event.content``
       # (the model's text) AND ``event.output`` (the same text) to
@@ -772,17 +777,14 @@ async def _consume_event_queue(
       # surface the same text twice.  Task-mode agents set
       # ``event.output`` from the ``finish_task`` FC args without
       # ``message_as_output``, so this clearing doesn't affect them.
-      if not event.partial:
-        if event.node_info.message_as_output and event.content is not None:
-          event = event.model_copy()
-          event.output = None
-
-      output_event = await self._process_event_with_plugin_callbacks(
-          invocation_context=ic,
-          event=event,
-      )
-
-      if not event.partial:
+      if not output_event.partial:
+        if (
+            output_event.node_info
+            and output_event.node_info.message_as_output
+            and output_event.content is not None
+        ):
+          output_event = output_event.model_copy()
+          output_event.output = None
         await self.session_service.append_event(
             session=ic.session, event=output_event
         )
@@ -1442,7 +1444,7 @@ async def _exec_with_plugin(
             invocation_context=invocation_context,
             event=early_exit_event,
         )
-        if self._should_append_event(early_exit_event, is_live_call):
+        if self._should_append_event(output_event, is_live_call):
           await self.session_service.append_event(
               session=invocation_context.session,
               event=output_event,
@@ -1461,15 +1463,16 @@ async def _exec_with_plugin(
 
             if is_live_call:
               # Skip partial transcriptions for Live
-              if event.partial is not True and self._should_append_event(
-                  event, is_live_call
+              if (
+                  output_event.partial is not True
+                  and self._should_append_event(output_event, is_live_call)
               ):
                 logger.debug('Appending live event: %s', output_event)
                 await self.session_service.append_event(
                     session=invocation_context.session, event=output_event
                 )
             else:
-              if event.partial is not True:
+              if output_event.partial is not True:
                 await self.session_service.append_event(
                     session=invocation_context.session, event=output_event
                 )