chore: Introduce _BranchPath utility class for dynamic execution branches

Implement _BranchPath, a hierarchical path representation for dot-separated dynamic execution branch strings (e.g. 'parent@1.child@2.node').

This class provides a unified, segment-based API for:
- Parsing and serialization of branch strings.
- Extracting all run IDs across the entire branch chain.
- Robust hierarchy checks (is_descendant_of) that are immune to partial prefix matches.

Also add comprehensive unit tests covering all features and boundary cases.

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 938209288

Author: DeanChensj

src/google/adk/events/_branch_path.py

@@ -0,0 +1,101 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Hierarchical path representation for dynamic execution branches."""
+
+from __future__ import annotations
+
+
+class _BranchPath:
+  """Represents a hierarchical path for execution branches.
+
+  A path consists of dot-separated segments (e.g., 'segment1.segment2'),
+  where each segment represents a node run and is typically formatted
+  as 'node_name@run_id' or 'node_name'.
+
+  Example:
+    'parent_agent@1.collect_user_info_tool@2.sub_workflow'
+  """
+
+  def __init__(self, segments: list[str]):
+    """Initializes a _BranchPath with a list of segments."""
+    self._segments = list(segments)
+
+  @classmethod
+  def from_string(cls, path_str: str | None) -> _BranchPath:
+    """Parses a _BranchPath from a dot-separated string representation."""
+    if not path_str:
+      return cls([])
+    return cls(path_str.split('.'))
+
+  def __str__(self) -> str:
+    """Returns the dot-separated string representation of the path."""
+    return '.'.join(self._segments)
+
+  def __eq__(self, other: object) -> bool:
+    """Returns True if segments are equal."""
+    if not isinstance(other, _BranchPath):
+      return NotImplemented
+    return self._segments == other._segments
+
+  @property
+  def segments(self) -> list[str]:
+    """Returns a copy of the path segments."""
+    return list(self._segments)
+
+  @property
+  def run_ids(self) -> set[str]:
+    """Extracts all run IDs (the part after '@') from all segments in the path.
+
+    Example:
+      - Path: 'parent@1.child@2.node'
+      - Returns: {'1', '2'}
+    """
+    ids = set()
+    for segment in self._segments:
+      parts = segment.rsplit('@', 1)
+      if len(parts) > 1 and parts[1]:
+        ids.add(parts[1])
+    return ids
+
+  @property
+  def parent(self) -> _BranchPath | None:
+    """Returns the parent _BranchPath, or None if this is a root path."""
+    if len(self._segments) <= 1:
+      return None
+    return _BranchPath(self._segments[:-1])
+
+  def is_descendant_of(self, ancestor: _BranchPath) -> bool:
+    """Checks if this path is a descendant of the ancestor path.
+
+    A path is a descendant if it starts with all segments of the ancestor path
+    and has additional segments.
+    """
+    if len(self._segments) <= len(ancestor._segments):
+      return False
+    return self._segments[: len(ancestor._segments)] == ancestor._segments
+
+  @staticmethod
+  def common_prefix(paths: list[_BranchPath]) -> _BranchPath:
+    """Finds the common prefix of a list of _BranchPath objects."""
+    if not paths:
+      return _BranchPath([])
+
+    common_segments = []
+    for segments in zip(*[p.segments for p in paths]):
+      if len(set(segments)) == 1:
+        common_segments.append(segments[0])
+      else:
+        break
+    return _BranchPath(common_segments)

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

@@ -23,6 +23,7 @@
 from typing_extensions import override
 
 from ...agents.invocation_context import InvocationContext
+from ...events._branch_path import _BranchPath
 from ...events.event import Event
 from ...models.llm_request import LlmRequest
 from ._base_llm_processor import BaseLlmRequestProcessor
@@ -900,12 +901,10 @@ def _is_event_belongs_to_branch(
   """
   if not invocation_branch or not event.branch:
     return True
-  # We use dot to delimit branch nodes. To avoid simple prefix match
-  # (e.g. agent_0 unexpectedly matching agent_00), require either perfect branch
-  # match, or match prefix with an additional explicit '.'
-  return invocation_branch == event.branch or invocation_branch.startswith(
-      f'{event.branch}.'
-  )
+
+  inv_path = _BranchPath.from_string(invocation_branch)
+  evt_path = _BranchPath.from_string(event.branch)
+  return inv_path == evt_path or inv_path.is_descendant_of(evt_path)
 
 
 def _is_function_call_event(event: Event, function_name: str) -> bool:

src/google/adk/workflow/_join_node.py

@@ -23,6 +23,7 @@
 from typing_extensions import override
 
 from ..agents.context import Context
+from ..events._branch_path import _BranchPath
 from ..events.event import Event
 from ._base_node import BaseNode
 
@@ -33,15 +34,8 @@ def _get_common_branch_prefix(branches: list[str]) -> str:
   """Find the common prefix of dot-separated branch strings."""
   if not branches:
     return ''
-  split_branches = [b.split('.') if b else [] for b in branches]
-
-  common = []
-  for segments in zip(*split_branches):
-    if len(set(segments)) == 1:
-      common.append(segments[0])
-    else:
-      break
-  return '.'.join(common)
+  paths = [_BranchPath.from_string(b) for b in branches]
+  return str(_BranchPath.common_prefix(paths))
 
 
 class JoinNode(BaseNode):

src/google/adk/workflow/_workflow.py

@@ -30,6 +30,7 @@
 
 from pydantic import Field
 
+from ..events._branch_path import _BranchPath
 from ._base_node import BaseNode
 from ._base_node import START
 from ._dynamic_node_scheduler import DynamicNodeScheduler
@@ -58,15 +59,8 @@ def get_common_branch_prefix(branches: list[str]) -> str:
   """Find the common prefix of dot-separated branch strings."""
   if not branches:
     return ''
-  split_branches = [b.split('.') if b else [] for b in branches]
-
-  common = []
-  for segments in zip(*split_branches):
-    if len(set(segments)) == 1:
-      common.append(segments[0])
-    else:
-      break
-  return '.'.join(common)
+  paths = [_BranchPath.from_string(b) for b in branches]
+  return str(_BranchPath.common_prefix(paths))
 
 
 # ---------------------------------------------------------------------------

tests/unittests/events/test_branch_path.py

@@ -0,0 +1,160 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Unit tests for _BranchPath.
+
+Verifies that _BranchPath correctly parses, serializes, and manipulates
+hierarchical dynamic execution branch paths.
+"""
+
+from __future__ import annotations
+
+from google.adk.events._branch_path import _BranchPath
+import pytest
+
+
+def test_from_string_with_empty_string_returns_empty_path():
+  """Parsing an empty string returns a _BranchPath with no segments."""
+  path = _BranchPath.from_string("")
+
+  assert path.segments == []
+  assert str(path) == ""
+
+
+def test_from_string_with_single_segment_returns_path_with_one_segment():
+  """Parsing a single name returns a _BranchPath with one segment."""
+  path = _BranchPath.from_string("agent_0")
+
+  assert path.segments == ["agent_0"]
+  assert str(path) == "agent_0"
+
+
+def test_from_string_with_multiple_segments_returns_path_with_all_segments():
+  """Parsing a dot-separated string returns a _BranchPath with all segments."""
+  path = _BranchPath.from_string("parent.child.node")
+
+  assert path.segments == ["parent", "child", "node"]
+  assert str(path) == "parent.child.node"
+
+
+def test_equality_compares_path_segments():
+  """Two _BranchPath objects are equal if and only if their segments match."""
+  path1 = _BranchPath.from_string("parent.child")
+  path2 = _BranchPath.from_string("parent.child")
+  path3 = _BranchPath.from_string("parent.other")
+
+  assert path1 == path2
+  assert path1 != path3
+  assert path1 != "parent.child"  # Different type
+
+
+def test_run_ids_extracts_all_run_ids_from_path():
+  """run_ids extracts all run IDs (the part after '@') from all segments."""
+  # Given paths with various run ID patterns
+  path_with_ids = _BranchPath.from_string("parent@1.child@2.node")
+  path_no_ids = _BranchPath.from_string("parent.child")
+  path_mixed = _BranchPath.from_string("parent@1.child.node@3")
+
+  # Then the extracted run IDs match expectations
+  assert path_with_ids.run_ids == {"1", "2"}
+  assert path_no_ids.run_ids == set()
+  assert path_mixed.run_ids == {"1", "3"}
+
+
+def test_parent_returns_parent_path_or_none_for_root():
+  """parent returns a new _BranchPath excluding the leaf segment, or None."""
+  path = _BranchPath.from_string("parent.child.node")
+
+  assert path.parent == _BranchPath.from_string("parent.child")
+  assert path.parent.parent == _BranchPath.from_string("parent")
+  assert path.parent.parent.parent is None
+
+
+def test_is_descendant_of_verifies_path_hierarchy_safely():
+  """is_descendant_of returns True if the path is a strict sub-path of ancestor."""
+  # Given an ancestor and various comparison paths
+  ancestor = _BranchPath.from_string("parent.child")
+  descendant = _BranchPath.from_string("parent.child.node.leaf")
+  not_descendant = _BranchPath.from_string("parent.other")
+  same = _BranchPath.from_string("parent.child")
+
+  # Then descendant checks match expectations
+  assert descendant.is_descendant_of(ancestor)
+  assert not ancestor.is_descendant_of(descendant)
+  assert not not_descendant.is_descendant_of(ancestor)
+  assert not same.is_descendant_of(ancestor)
+
+
+def test_is_descendant_of_is_immune_to_partial_name_prefix_match():
+  """is_descendant_of compares segments, avoiding partial string prefix bugs."""
+  # Given an ancestor and a path that has a partial string prefix but different segment
+  ancestor = _BranchPath.from_string("agent_0")
+  descendant_with_prefix = _BranchPath.from_string("agent_00.child")
+
+  # Then it is not recognized as a descendant because segments don't match
+  assert not descendant_with_prefix.is_descendant_of(ancestor)
+
+
+def test_common_prefix_finds_longest_shared_path():
+  """common_prefix returns the longest common prefix of a list of paths."""
+  # Given a list of paths sharing a common prefix
+  paths = [
+      _BranchPath.from_string("parent.child.node1"),
+      _BranchPath.from_string("parent.child.node2.leaf"),
+      _BranchPath.from_string("parent.child.node3"),
+  ]
+
+  # When finding the common prefix
+  result = _BranchPath.common_prefix(paths)
+
+  # Then the result matches the shared parent path
+  assert result == _BranchPath.from_string("parent.child")
+
+
+def test_common_prefix_with_no_shared_path_returns_empty():
+  """common_prefix returns an empty path if there is no shared prefix."""
+  paths = [
+      _BranchPath.from_string("parent.child"),
+      _BranchPath.from_string("other.child"),
+  ]
+
+  result = _BranchPath.common_prefix(paths)
+
+  assert result == _BranchPath.from_string("")
+
+
+def test_common_prefix_with_empty_list_returns_empty():
+  """common_prefix returns an empty path if the input list is empty."""
+  result = _BranchPath.common_prefix([])
+
+  assert result == _BranchPath.from_string("")
+
+
+def test_constructor_copies_segments_list():
+  """_BranchPath copies the input segments list to ensure immutability."""
+  segments = ["parent", "child"]
+  path = _BranchPath(segments)
+
+  # Mutate the original list
+  segments.append("grandchild")
+
+  # The path segments should remain unchanged
+  assert path.segments == ["parent", "child"]
+
+
+def test_run_ids_filters_out_empty_run_ids():
+  """run_ids filters out segments with empty run IDs (e.g. ending with '@')."""
+  path = _BranchPath.from_string("parent@.child@2.node@")
+
+  assert path.run_ids == {"2"}