fix: check that a transfer target is a sibling agent (v1)

Port of "check if transfer target is a sibling agent" from main, originally
contributed as #3862 and closing #3850.

Before, `disallow_transfer_to_peers=True` only shaped what the model was
told: it kept peers out of the transfer instruction and out of the
`transfer_to_agent` enum. If the model named a sibling anyway, having picked
the name out of the conversation history or the user's prompt,
`_get_agent_to_run` looked it up in the agent tree and handed control over.
The setting was a hint, not a rule. Now `_get_agent_to_run` raises
`ValueError` when an `LlmAgent` with `disallow_transfer_to_peers` set resolves
a target that shares its parent and is not itself.

Behaviour change: an agent that sets `disallow_transfer_to_peers=True` and
still transfers to a peer now raises instead of transferring. Transfer to
self, transfer to a parent, and transfers from a caller that is not an
`LlmAgent` are unchanged.

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

@@ -1364,6 +1364,16 @@ class BaseLlmFlow(ABC):
     agent_to_run = root_agent.find_agent(agent_name)
     if not agent_to_run:
       raise ValueError(f'Agent {agent_name} not found in the agent tree.')
+
+    from ...agents.llm_agent import LlmAgent
+
+    if (
+        isinstance(invocation_context.agent, LlmAgent)
+        and invocation_context.agent.disallow_transfer_to_peers
+        and agent_to_run.parent_agent == invocation_context.agent.parent_agent
+        and agent_to_run != invocation_context.agent
+    ):
+      raise ValueError(f'Transfer to sibling agent {agent_name} is disallowed.')
     return agent_to_run
 
   async def _call_llm_async(

tests/unittests/flows/llm_flows/test_base_llm_flow.py

@@ -20,6 +20,7 @@ from unittest.mock import AsyncMock
 
 from google.adk.agents.live_request_queue import LiveRequestQueue
 from google.adk.agents.llm_agent import Agent
+from google.adk.agents.loop_agent import LoopAgent
 from google.adk.agents.run_config import RunConfig
 from google.adk.events.event import Event
 from google.adk.flows.llm_flows.base_llm_flow import _handle_after_model_callback
@@ -2049,3 +2050,104 @@ async def test_send_to_model_rejects_function_call():
       ValueError, match='User message cannot contain function calls'
   ):
     await flow._send_to_model(mock_connection, invocation_context)
+
+
+def _make_agent_tree():
+  root = Agent(name='root')
+  child1 = Agent(name='child1')
+  child2 = Agent(name='child2')
+
+  child1.parent_agent = root
+  child2.parent_agent = root
+  root.sub_agents = [child1, child2]
+  return root, child1, child2
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_sibling_disallowed_raises_value_error():
+  """Transfer to sibling raises ValueError when disallow_transfer_to_peers is True."""
+  # Arrange
+  root, child1, child2 = _make_agent_tree()
+  caller = child1
+  caller.disallow_transfer_to_peers = True
+  ctx = await testing_utils.create_invocation_context(caller)
+  flow = BaseLlmFlowForTesting()
+
+  # Act & Assert
+  with pytest.raises(
+      ValueError, match='Transfer to sibling agent child2 is disallowed'
+  ):
+    flow._get_agent_to_run(ctx, 'child2')
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_sibling_allowed_returns_agent():
+  """Transfer to sibling returns the agent when disallow_transfer_to_peers is False."""
+  # Arrange
+  root, child1, child2 = _make_agent_tree()
+  caller = child1
+  caller.disallow_transfer_to_peers = False
+  ctx = await testing_utils.create_invocation_context(caller)
+  flow = BaseLlmFlowForTesting()
+
+  # Act
+  agent = flow._get_agent_to_run(ctx, 'child2')
+
+  # Assert
+  assert agent is not None
+  assert agent.name == 'child2'
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_unknown_agent_raises_value_error():
+  """Transfer to unknown agent name raises ValueError."""
+  # Arrange
+  root, child1, child2 = _make_agent_tree()
+  caller = child1
+  ctx = await testing_utils.create_invocation_context(caller)
+  flow = BaseLlmFlowForTesting()
+
+  # Act & Assert
+  with pytest.raises(ValueError, match='not found in the agent tree'):
+    flow._get_agent_to_run(ctx, 'not_in_tree')
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_self_allowed_when_peers_disallowed():
+  """Transfer to self is allowed even when disallow_transfer_to_peers is True."""
+  # Arrange
+  root, child1, child2 = _make_agent_tree()
+  caller = child1
+  caller.disallow_transfer_to_peers = True
+  ctx = await testing_utils.create_invocation_context(caller)
+  flow = BaseLlmFlowForTesting()
+
+  # Act
+  agent = flow._get_agent_to_run(ctx, 'child1')
+
+  # Assert
+  assert agent is not None
+  assert agent.name == 'child1'
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_sibling_from_non_llm_agent_allowed():
+  """Transfer to sibling is allowed when the caller is not an LlmAgent."""
+  # Arrange
+  root = Agent(name='root')
+  child1 = LoopAgent(name='child1')
+  child2 = Agent(name='child2')
+
+  child1.parent_agent = root
+  child2.parent_agent = root
+  root.sub_agents = [child1, child2]
+
+  ctx = await testing_utils.create_invocation_context(child1)
+  flow = BaseLlmFlowForTesting()
+
+  # Act
+  agent = flow._get_agent_to_run(ctx, 'child2')
+
+  # Assert
+  assert agent is not None
+  assert agent.name == 'child2'