fix(adk): improve event handling and HITL tool processing

- Fix EventTranslator to close active streams on empty final responses
- Add HITL/frontend tool scenario handling in message processing loop
- Skip historical backend tool interactions to prevent duplicate processing
- Change state update author from "system" to "user" to avoid ADK warnings
- Add debug logging for message loop and background execution

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: mefogle

integrations/adk-middleware/python/src/ag_ui_adk/adk_agent.py

@@ -375,13 +375,11 @@ async def run(self, input: RunAgentInput) -> AsyncGenerator[BaseEvent, None]:
         skip_tool_message_batch = False
 
         # Check if there are pending tool calls AND tool results in unseen messages
-        # If so, we should skip to the tool results first
-        pending_tool_call_ids = await self._get_pending_tool_call_ids(input.thread_id)
-        has_pending_tools = pending_tool_call_ids is not None and len(pending_tool_call_ids) > 0
+        has_pending_tools = await self._has_pending_tool_calls(input.thread_id)
         has_tool_results_in_unseen = any(getattr(msg, "role", None) == "tool" for msg in unseen_messages)
 
         if has_pending_tools and has_tool_results_in_unseen:
-            # Find the index of the first tool result and skip to it
+            # HITL/Frontend tool scenario: skip to the tool results first
             for i, msg in enumerate(unseen_messages):
                 if getattr(msg, "role", None) == "tool":
                     # Mark all messages before the tool result as processed (they're already in the ADK session)
@@ -395,6 +393,8 @@ async def run(self, input: RunAgentInput) -> AsyncGenerator[BaseEvent, None]:
                     index = i
                     break
 
+        logger.debug(f"[RUN_LOOP] Starting message loop for thread={input.thread_id}, total_unseen={total_unseen}, starting_index={index}")
+
         while index < total_unseen:
             current = unseen_messages[index]
             role = getattr(current, "role", None)
@@ -508,6 +508,38 @@ async def run(self, input: RunAgentInput) -> AsyncGenerator[BaseEvent, None]:
                 else:
                     skip_tool_message_batch = False
 
+                # Check if there's an upcoming tool batch that will be skipped
+                # If so, this non-tool batch is part of historical backend tool interaction
+                # and should also be skipped
+                upcoming_tool_batch_skipped = False
+                if index < total_unseen and getattr(unseen_messages[index], "role", None) == "tool":
+                    # Peek at the upcoming tool batch
+                    peek_idx = index
+                    upcoming_tool_call_ids = []
+                    while peek_idx < total_unseen and getattr(unseen_messages[peek_idx], "role", None) == "tool":
+                        tool_call_id = getattr(unseen_messages[peek_idx], "tool_call_id", None)
+                        if tool_call_id:
+                            upcoming_tool_call_ids.append(tool_call_id)
+                        peek_idx += 1
+
+                    if upcoming_tool_call_ids:
+                        pending_ids = await self._get_pending_tool_call_ids(input.thread_id)
+                        if pending_ids is not None:
+                            pending_set = set(pending_ids)
+                            # If NONE of the upcoming tool results match pending, they're historical
+                            if not any(tc_id in pending_set for tc_id in upcoming_tool_call_ids):
+                                upcoming_tool_batch_skipped = True
+
+                if upcoming_tool_batch_skipped:
+                    # Skip this message batch - it's part of historical backend tool interaction
+                    # Mark the messages as processed
+                    logger.debug(f"[RUN_LOOP] Skipping message batch (upcoming tool batch will be skipped)")
+                    batch_ids = self._collect_message_ids(message_batch)
+                    if batch_ids:
+                        self._session_manager.mark_messages_processed(app_name, input.thread_id, batch_ids)
+                    continue
+
+                logger.debug(f"[RUN_LOOP] Calling _start_new_execution with message_batch of {len(message_batch)} messages")
                 async for event in self._start_new_execution(input, message_batch=message_batch):
                     yield event
 
@@ -1049,6 +1081,8 @@ async def _run_adk_in_background(
             event_queue: Queue for emitting events
         """
         runner: Optional[Runner] = None
+        logger.debug(f"[BG_EXEC] _run_adk_in_background called for thread={input.thread_id}")
+        logger.debug(f"[BG_EXEC]   tool_results={len(tool_results) if tool_results else 0}, message_batch={len(message_batch) if message_batch else 0}")
         try:
             # Agent is already prepared with tools and SystemMessage instructions (if any)
             # from _start_background_execution, so no additional agent copying needed here

integrations/adk-middleware/python/src/ag_ui_adk/event_translator.py

@@ -279,23 +279,10 @@ async def _translate_text_content(
             return
 
         combined_text = "".join(text_parts)
-        if not combined_text:
-            return
-
-        # Use proper ADK streaming detection (handle None values)
-        is_partial = getattr(adk_event, 'partial', False)
-        turn_complete = getattr(adk_event, 'turn_complete', False)
-        
-        # (is_final_response is already calculated above)
-        
-        # Handle None values: if a turn is complete or a final chunk arrives, end streaming
-        has_finish_reason = bool(getattr(adk_event, 'finish_reason', None))
-        should_send_end = (
-            (turn_complete and not is_partial)
-            or (is_final_response and not is_partial)
-            or (has_finish_reason and self._is_streaming)
-        )
 
+        # Handle is_final_response BEFORE the empty text early return.
+        # An empty final response is a valid stream-closing signal that must close
+        # any active stream, even if there's no new text content.
         if is_final_response:
             # This is the final, complete message event.
 
@@ -365,7 +352,23 @@ async def _translate_text_content(
             self._last_streamed_run_id = None
             return
 
-        
+        # Early return for empty text (non-final responses only).
+        # Final responses with empty text are handled above to close active streams.
+        if not combined_text:
+            return
+
+        # Use proper ADK streaming detection (handle None values)
+        is_partial = getattr(adk_event, 'partial', False)
+        turn_complete = getattr(adk_event, 'turn_complete', False)
+
+        # Handle None values: if a turn is complete or a final chunk arrives, end streaming
+        has_finish_reason = bool(getattr(adk_event, 'finish_reason', None))
+        should_send_end = (
+            (turn_complete and not is_partial)
+            or (is_final_response and not is_partial)
+            or (has_finish_reason and self._is_streaming)
+        )
+
         # Handle streaming logic (if not is_final_response)
         if not self._is_streaming:
             # Start of new message - emit START event

integrations/adk-middleware/python/src/ag_ui_adk/session_manager.py

@@ -196,10 +196,12 @@ async def update_session_state(
                 # This depends on ADK's behavior - may need to explicitly clear
 
             # Create event with state changes
+            # Use "user" as author since state updates come from the frontend
+            # Note: Using "system" causes ADK runner warnings in _find_agent_to_run
             actions = EventActions(state_delta=state_delta)
             event = Event(
                 invocation_id=f"state_update_{int(time.time())}",
-                author="system",
+                author="user",
                 actions=actions,
                 timestamp=time.time()
             )