fix(agents): only stop ParallelAgent on direct sub-agent escalation

[mithun-sudo]

Match adk-python: nested escalate events must not cancel sibling branches. Adds regression test for LoopAgent nested under ParallelAgent.

Link to Issue or Description of Change

1. Link to an existing issue (if applicable):

• Closes: ParallelAgent stops sibling branches on nested sub-agent escalation #1490
• Related: ParallelAgent stops sibling branches on nested sub-agent escalation #1490

Problem:
ParallelAgent stops all parallel branches when any event has escalate=true, including escalation from nested sub-agents (e.g. an agent inside a LoopAgent sub-branch). adk-python only stops when the escalating event's author is a direct sub-agent (_asks_this_agent_to_exit).

Solution:
Filter takeUntil with asksThisAgentToExit(), which requires both escalate=true and event.author() in the set of direct sub-agent names. Matches adk-python parity without changing behavior for direct sub-agent escalation (existing test still passes).

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Summary:

• ./mvnw -pl core clean compile test -Dtest=ParallelAgentEscalationTest — 2 tests passed
• ./mvnw -Prelease clean package — BUILD SUCCESS
• Verified regression: runAsync_nestedLoopEscalation_keepsSiblingBranchesRunning fails on main (stream completes after 1 nested escalate) and passes with this change
• runAsync_escalationEvent_shortCircuitsOtherAgents still passes (direct sub-agent escalate still short-circuits siblings)

Manual End-to-End (E2E) Tests:

Not applicable — behavior covered by unit tests; no user-facing API or runtime config change.

Checklist

• I have read the CONTRIBUTING.md document.
• My pull request contains a single commit.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end.
• Any dependent changes have been merged and published in downstream modules.

Additional context

Python reference:

• src/google/adk/agents/parallel_agent.py — _asks_this_agent_to_exit
• tests/unittests/agents/test_parallel_agent.py — test_run_async_keeps_siblings_when_a_nested_loop_ends_itself

[hemasekhar-p]

Hi @mithun-sudo, thank you for your contribution. we appreciate you taking the time to submit this PR. It is currently under review by our team, and we will keep you updated if any additional information is required.

[wikaaaaa]

Thank you for the contribution! LGTM, two minor comments.

[wikaaaaa]

Can we add .filter(Objects::nonNull) here?

ImmutableSet<String> directSubAgentNames =
    currentSubAgents.stream()
        .map(BaseAgent::name)
        .filter(Objects::nonNull)
        .collect(ImmutableSet.toImmutableSet());

Mocked agents have null names because BaseAgent.name() is a final method that cannot be mocked. And ImmutableSet.toImmutableSet() throws NullPointerException if any stream element is null.

BaseAgent already has a workaround like this -

adk-java/core/src/main/java/com/google/adk/agents/BaseAgent.java

Lines 160 to 164 in c1a2960

	// NOTE: Mocked agents have null names because BaseAgent.name() is a final method that
	// cannot be mocked.
	if (subAgentName != null && !subAgentNames.add(subAgentName)) {
	duplicateSubAgentNames.add(subAgentName);
	}

[wikaaaaa]

Can you please add a comment with a brief explanation why we check directSubAgentNames?

Sth like python has -
https://github.com/google/adk-python/blob/7ae1c9b026c84bf8a65921003f71b0f30c8e3166/src/google/adk/agents/parallel_agent.py#L70