refactor(interactions): extract previous-interaction-id lookup helper

Extract the previous-interaction-id lookup into reusable, private module-level
helpers. No behavior change.

Co-authored-by: Haran Rajkumar <haranrk@google.com>
PiperOrigin-RevId: 937636496

Author: haranrk

src/google/adk/flows/llm_flows/interactions_processor.py

@@ -29,6 +29,56 @@
 logger = logging.getLogger('google_adk.' + __name__)
 
 
+def _is_event_in_branch(current_branch: Optional[str], event: Event) -> bool:
+  """Return True if ``event`` belongs to ``current_branch`` (or the root)."""
+  if not current_branch:
+    # No branch means we're at the root; include all events without a branch.
+    return not event.branch
+  return event.branch == current_branch or not event.branch
+
+
+def _find_previous_interaction_id(
+    events: list[Event],
+    *,
+    agent_name: str,
+    current_branch: Optional[str],
+) -> Optional[str]:
+  """Find the most recent interaction_id authored by ``agent_name``.
+
+  Scans ``events`` in reverse, skipping events outside ``current_branch``, and
+  returns the first ``interaction_id`` from an event authored by this agent.
+  """
+  logger.debug(
+      'Finding previous_interaction_id: agent=%s, branch=%s, num_events=%d',
+      agent_name,
+      current_branch,
+      len(events),
+  )
+  for event in reversed(events):
+    if not _is_event_in_branch(current_branch, event):
+      logger.debug(
+          'Skipping event not in branch: author=%s, branch=%s, current=%s',
+          event.author,
+          event.branch,
+          current_branch,
+      )
+      continue
+    logger.debug(
+        'Checking event: author=%s, interaction_id=%s, branch=%s',
+        event.author,
+        event.interaction_id,
+        event.branch,
+    )
+    if event.author == agent_name and event.interaction_id:
+      logger.debug(
+          'Found interaction_id from agent %s: %s',
+          agent_name,
+          event.interaction_id,
+      )
+      return event.interaction_id
+  return None
+
+
 class InteractionsRequestProcessor(BaseLlmRequestProcessor):
   """Request processor for Interactions API stateful conversations.
   This processor extracts the previous_interaction_id from session events
@@ -75,66 +125,12 @@ async def run_async(
   def _find_previous_interaction_id(
       self, invocation_context: 'InvocationContext'
   ) -> Optional[str]:
-    """Find the previous interaction ID from session events.
-    For interactions API stateful mode, we need to find the most recent
-    interaction_id from model responses to chain interactions.
-    Args:
-        invocation_context: The invocation context containing session events.
-    Returns:
-        The previous interaction ID if found, None otherwise.
-    """
-    events = invocation_context.session.events
-    current_branch = invocation_context.branch
-    agent_name = invocation_context.agent.name
-    logger.debug(
-        'Finding previous_interaction_id: agent=%s, branch=%s, num_events=%d',
-        agent_name,
-        current_branch,
-        len(events),
+    """Find the previous interaction ID from session events."""
+    return _find_previous_interaction_id(
+        invocation_context.session.events,
+        agent_name=invocation_context.agent.name,
+        current_branch=invocation_context.branch,
     )
-    # Iterate backwards through events to find the most recent interaction_id
-    for event in reversed(events):
-      # Skip events not in current branch
-      if not self._is_event_in_branch(current_branch, event):
-        logger.debug(
-            'Skipping event not in branch: author=%s, branch=%s, current=%s',
-            event.author,
-            event.branch,
-            current_branch,
-        )
-        continue
-      # Look for model responses with interaction_id from this agent
-      logger.debug(
-          'Checking event: author=%s, interaction_id=%s, branch=%s',
-          event.author,
-          event.interaction_id,
-          event.branch,
-      )
-      # Only consider events from this agent (skip sub-agent events)
-      if event.author == agent_name and event.interaction_id:
-        logger.debug(
-            'Found interaction_id from agent %s: %s',
-            agent_name,
-            event.interaction_id,
-        )
-        return event.interaction_id
-    return None
-
-  def _is_event_in_branch(
-      self, current_branch: Optional[str], event: Event
-  ) -> bool:
-    """Check if an event belongs to the current branch.
-    Args:
-        current_branch: The current branch name.
-        event: The event to check.
-    Returns:
-        True if the event belongs to the current branch.
-    """
-    if not current_branch:
-      # No branch means we're at the root, include all events without branch
-      return not event.branch
-    # Event must be in the same branch or have no branch (root level)
-    return event.branch == current_branch or not event.branch
 
 
 # Module-level processor instance for use in flow configuration

tests/unittests/flows/llm_flows/test_interactions_processor.py

@@ -168,15 +168,13 @@ def test_find_previous_interaction_id_skips_user_events(self):
 
   def test_is_event_in_branch_no_branch(self):
     """Test branch filtering with no current branch."""
-    processor = interactions_processor.InteractionsRequestProcessor()
-
     # Event without branch should be included when no current branch
     event = Event(
         invocation_id="inv1",
         author="test",
         content=types.ModelContent("test"),
     )
-    assert processor._is_event_in_branch(None, event) is True
+    assert interactions_processor._is_event_in_branch(None, event) is True
 
     # Event with branch should be excluded when no current branch
     event_with_branch = Event(
@@ -185,39 +183,84 @@ def test_is_event_in_branch_no_branch(self):
         content=types.ModelContent("test"),
         branch="some_branch",
     )
-    assert processor._is_event_in_branch(None, event_with_branch) is False
+    assert (
+        interactions_processor._is_event_in_branch(None, event_with_branch)
+        is False
+    )
 
   def test_is_event_in_branch_same_branch(self):
     """Test that events in the same branch are included."""
-    processor = interactions_processor.InteractionsRequestProcessor()
-
     event = Event(
         invocation_id="inv1",
         author="test",
         content=types.ModelContent("test"),
         branch="root.child",
     )
-    assert processor._is_event_in_branch("root.child", event) is True
+    assert (
+        interactions_processor._is_event_in_branch("root.child", event) is True
+    )
 
   def test_is_event_in_branch_different_branch(self):
     """Test that events in different branches are excluded."""
-    processor = interactions_processor.InteractionsRequestProcessor()
-
     event = Event(
         invocation_id="inv1",
         author="test",
         content=types.ModelContent("test"),
         branch="root.other",
     )
-    assert processor._is_event_in_branch("root.child", event) is False
+    assert (
+        interactions_processor._is_event_in_branch("root.child", event) is False
+    )
 
   def test_is_event_in_branch_root_events_included(self):
     """Test that root events (no branch) are included in child branches."""
-    processor = interactions_processor.InteractionsRequestProcessor()
-
     event = Event(
         invocation_id="inv1",
         author="test",
         content=types.ModelContent("test"),
     )
-    assert processor._is_event_in_branch("root.child", event) is True
+    assert (
+        interactions_processor._is_event_in_branch("root.child", event) is True
+    )
+
+
+def _evt(author: str, interaction_id: str | None, branch: str | None) -> Event:
+  return Event(author=author, interaction_id=interaction_id, branch=branch)
+
+
+def test_find_previous_interaction_id_returns_latest_for_agent():
+  events = [
+      _evt("my_agent", "int_1", None),
+      _evt("user", None, None),
+      _evt("my_agent", "int_2", None),
+      _evt("other_agent", "int_3", None),
+  ]
+
+  result = interactions_processor._find_previous_interaction_id(
+      events, agent_name="my_agent", current_branch=None
+  )
+
+  assert result == "int_2"
+
+
+def test_find_previous_interaction_id_respects_branch():
+  events = [
+      _evt("my_agent", "int_main", None),
+      _evt("my_agent", "int_other_branch", "branch_b"),
+  ]
+
+  result = interactions_processor._find_previous_interaction_id(
+      events, agent_name="my_agent", current_branch="branch_a"
+  )
+
+  assert result == "int_main"
+
+
+def test_find_previous_interaction_id_none_when_absent():
+  events = [_evt("user", None, None)]
+
+  result = interactions_processor._find_previous_interaction_id(
+      events, agent_name="my_agent", current_branch=None
+  )
+
+  assert result is None