Nested LoopAgents will escalate all LoopAgents

[jhonratiu]

A nested LoopAgent will escalate and break all LoopAgents

SequentialAgent(
    sub_agents=[
        LoopAgent(
            sub_agents=[
                LoopAgent(
                    sub_agents=[
                        section_evaluator,
                        EscalationChecker(name="IBreakAfter5Loops"),
                    ]
                ),
                EscalationChecker(name="IAlsoBreakAfter5Loops"),
            ],
        ),
    ],
)

Expecting to loop 25 times, but the inner loop will reach five and the outer loop will exit after running once.

  @override
  async def _run_async_impl(
      self, ctx: InvocationContext
  ) -> AsyncGenerator[Event, None]:
    times_looped = 0
    while not self.max_iterations or times_looped < self.max_iterations:
      for sub_agent in self.sub_agents:
        should_exit = False
        async for event in sub_agent.run_async(ctx):
          yield event
          if event.actions.escalate:
            should_exit = True

        if should_exit:
          return

      times_looped += 1
    return

In the LoopAgent _run_async_impl method every event appears to be passed up first, triggering an escalate on every loop agent that is part of the invocation. Something like this would stop the event from propagating further upwards but may impact the reporting.

async for event in sub_agent.run_async(ctx):
  if event.actions.escalate:
    should_exit = True
    event.actions.escalate = False
  yield event

Desktop:

• ADK version: 1.10.0

[Jacksunwei]

Proposed Enhancement: Add EscalationContext for Nested Workflow Control

Use Case

With complex nested workflow agents, users need finer control over escalation scope:

SequentialAgent([
    LoopAgent(  # Outer: should continue after inner exits
        sub_agents=[
            LoopAgent(  # Inner: should exit independently
                sub_agents=[validation_agent]
            ),
            next_step_agent
        ]
    )
])

Current behavior: exit_loop() in inner loop exits BOTH loops
Desired: Option to exit only the inner loop, letting outer loop continue

---

Proposed Design

Add EscalationContext for scoped escalation control:

class EscalationContext(BaseModel):
  type: Literal['parent', 'root'] = 'root'
  """
  - 'root' (default): Escalate through all ancestors. Preserves current behavior.
  - 'parent' (new): Escalate to nearest workflow parent only.
  """
  
  handled_by: List[str] = Field(default_factory=list)
  """Track which agents processed this escalation."""
  
  target_agent: Optional[str] = None
  """Optional: Escalate to specific named agent."""

Backward Compatibility:

# Old code - continues to work exactly as before
tool_context.actions.escalate = True  
# → Treated as type='root', exits all nested workflows

# New code - explicit control
tool_context.actions.escalation_context = EscalationContext(type='parent')
# → Exits only nearest workflow parent

tool_context.actions.escalation_context = EscalationContext(type='root')
# → Exits all workflows (same as escalate=True)

Example Tool Implementations:

# Current exit_loop (preserves behavior)
def exit_loop(tool_context: ToolContext):
  """Exits all enclosing loops."""
  tool_context.actions.escalate = True  # → type='root'
  tool_context.actions.skip_summarization = True

# New scoped exit tool
def exit_current_loop(tool_context: ToolContext):
  """Exits only the immediate loop."""
  tool_context.actions.escalation_context = EscalationContext(type='parent')
  tool_context.actions.skip_summarization = True

# Or update exit_loop for better semantics (would be breaking change)
def exit_loop(tool_context: ToolContext):
  """Exits the current loop only."""
  tool_context.actions.escalation_context = EscalationContext(type='parent')
  tool_context.actions.skip_summarization = True
  
def exit_all_loops(tool_context: ToolContext):
  """Exits all nested loops."""
  tool_context.actions.escalation_context = EscalationContext(type='root')
  tool_context.actions.skip_summarization = True

---

Benefits

✅ Backward compatible - no breaking changes
✅ Enables new nested workflow patterns
✅ Clear semantics for complex agent hierarchies
✅ Future-proof: can add more escalation types if needed

---

@jhonratiu @hangfei @boyangsvl - Thoughts on this enhancement? Does adding scoped escalation (type='parent') address the nested workflow use cases you're seeing?

[klateefa]

@jhonratiu Thank you for reporting this. I understand the request is currently under review. Please keep me posted if anything else is needed. Thanks for your patience.

[dShvetsov]

I replaced the LoopAgent with the following wrapper. This is not a solution, just a workaround while proper fix is baking.

from typing_extensions import override
from google.adk.agents import LoopAgent

class InnerLoopAgent(LoopAgent):
    @override
    async def _run_async_impl(self, *args, **kwargs):
        async for event in super()._run_async_impl(*args, **kwargs):
            event.actions.escalate = False
            yield event

[a2105z]

Taking this.

@Jacksunwei @klateefa @hangfei @boyangsvl — claiming #2808. This has been open since September (almost a year). The nested-loop case is still broken, people are shipping wrappers that wipe escalate on the way up, and your scoped-escalation design from November is still the right fix.

Why this matters: exit_loop() / actions.escalate = True is the only way a LoopAgent stops early. Nested loops (inner validation loop inside an outer retry loop) are a normal workflow. Today the inner escalate is yielded unchanged, so every ancestor LoopAgent also returns. You asked for 25 iterations and got 5.

Fix (your design, backward compatible):

• escalate=True with no context stays root — exits every enclosing loop (current behavior, including exit_loop)
• EventActions.escalation_context = EscalationContext(type='parent') exits only the nearest LoopAgent; that agent records itself on handled_by so outer loops continue
• New exit_current_loop tool for the parent-scoped case

PR incoming.

[a2105z]

#6842