Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 68 additions & 25 deletions core/src/main/java/com/google/adk/agents/LlmAgent.java
Original file line number Diff line number Diff line change
Expand Up @@ -677,8 +677,16 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
() -> {
// Resumed after a transfer: continue the sub-agent instead of re-invoking the model.
if (invocationContext.agentStates().containsKey(name())) {
Optional<BaseAgent> resumeTarget = findSubAgentToResume(invocationContext);
if (resumeTarget.isPresent()) {
Optional<String> resumeTargetName = findTransferToResume(invocationContext);
if (resumeTargetName.isPresent()) {
Optional<BaseAgent> resumeTarget = declaredTransferTarget(resumeTargetName.get());
// End this agent without running anything if the recorded target is undeclared
// (rejected) or already ended after this agent's latest transfer (e.g. on the
// target's own rejected transfer).
if (resumeTarget.isEmpty()
|| endedAfterLatestTransfer(invocationContext, resumeTarget.get())) {
return endOfAgentAndRecord(invocationContext);
}
AtomicBoolean resumePaused = new AtomicBoolean(false);
return resumeTarget
.get()
Expand Down Expand Up @@ -710,6 +718,7 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
paused.set(true);
}
})
.onErrorResumeNext(error -> endIfOwnRejectedTransfer(invocationContext, error))
.concatWith(
Flowable.defer(
() -> {
Expand All @@ -724,6 +733,20 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
});
}

/**
* Emits an end-of-agent marker before {@code error} when it is a {@link
* TransferRejectedException} from this agent, so a resume does not re-run it. Any other error,
* including a sub-agent's rejection, passes through and leaves this agent open.
*/
private Flowable<Event> endIfOwnRejectedTransfer(
InvocationContext invocationContext, Throwable error) {
if (error instanceof TransferRejectedException rejected
&& rejected.agentName().equals(name())) {
return endOfAgentAndRecord(invocationContext).concatWith(Flowable.error(error));
}
return Flowable.error(error);
}

/**
* Returns the name of the agent this agent transferred to in {@code event}, or empty when {@code
* event} is not such a transfer. A {@code transfer_to_agent} function response is required, as in
Expand All @@ -740,41 +763,56 @@ private Optional<String> transferTargetNameFrom(Event event) {
}

/**
* Resolves a transfer target recorded in history back to its agent. A live transfer fails in
* {@code BaseLlmFlow} when the target is gone; a resumed one has no such check, and returning
* empty here would silently re-run this agent instead of the sub-agent it handed off to.
* Returns whether {@code target} ended after this agent's latest transfer, as it does on its own
* rejected transfer. An end marker from before that transfer, such as a past loop iteration, does
* not count.
*/
private BaseAgent resolveTransferTarget(String targetName) {
// Declared targets first, as in Python, so a same-named agent on an unrelated branch is not
// resumed into; then the same tree-wide search the live transfer path uses.
for (BaseAgent target : transferTargets()) {
if (target.name().equals(targetName)) {
return target;
private boolean endedAfterLatestTransfer(InvocationContext context, BaseAgent target) {
// All branches: under a ParallelAgent, the target writes its marker on a descendant branch.
for (Event event :
context.events(/* currentInvocation= */ true, /* currentBranch= */ false).reverse()) {
if (transferTargetNameFrom(event).isPresent()) {
return false;
}
if (isEndMarker(event, target.name())) {
return true;
}
}
return rootAgent()
.findAgent(targetName)
.orElseThrow(
() ->
new IllegalStateException(
"Cannot resume agent "
+ name()
+ ": the agent it transferred to is no longer in the agent tree."));
return false;
}

/**
* When this agent is being resumed, returns the sub-agent it had transferred to (so the resume
* continues that sub-agent), or empty when this agent should continue itself.
* Returns whether {@code event} is {@code author}'s end-of-agent marker, which has no content.
*/
private Optional<BaseAgent> findSubAgentToResume(InvocationContext context) {
private static boolean isEndMarker(Event event, String author) {
return author.equals(event.author())
&& event.actions().endOfAgent()
&& event.content().isEmpty();
}

/**
* Returns the declared target named {@code targetName}, or empty when it is not one, so a
* same-named agent elsewhere in the tree is never resumed in its place.
*/
private Optional<BaseAgent> declaredTransferTarget(String targetName) {
return transferTargets().stream()
.filter(target -> target.name().equals(targetName))
.findFirst();
}

/**
* When this agent is being resumed, returns the name of the agent it had transferred to, or empty
* when this agent should continue itself. The name need not be a declared target.
*/
private Optional<String> findTransferToResume(InvocationContext context) {
ImmutableList<Event> events =
context.events(/* currentInvocation= */ true, /* currentBranch= */ true);
if (events.isEmpty()) {
return Optional.empty();
}
Event lastEvent = Iterables.getLast(events);
if (name().equals(lastEvent.author())) {
return transferTargetNameFrom(lastEvent).map(this::resolveTransferTarget);
return transferTargetNameFrom(lastEvent);
}
if (Objects.equals(lastEvent.author(), Role.USER) && !lastEvent.functionResponses().isEmpty()) {
// IAE (not ISE): an unresolvable resume surfaces through Runner.runAsync's IAE contract.
Expand All @@ -793,9 +831,14 @@ private Optional<BaseAgent> findSubAgentToResume(InvocationContext context) {
}
// A user event answering no call falls through; Python fails the resume instead.
for (int i = events.size() - 2; i >= 0; i--) {
Optional<String> targetName = transferTargetNameFrom(events.get(i));
Event event = events.get(i);
// Transfers before this agent's own end marker belong to a finished run, such as a past loop.
if (isEndMarker(event, name())) {
break;
}
Optional<String> targetName = transferTargetNameFrom(event);
if (targetName.isPresent()) {
return Optional.of(resolveTransferTarget(targetName.get()));
return targetName;
}
}
return Optional.empty();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
/*
* 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.
*/

package com.google.adk.agents;

/**
* Thrown when an {@link LlmAgent} calls {@code transfer_to_agent} with an agent it may not transfer
* to: itself, an agent other than its sub-agents, parent, and peers, or one its {@code
* disallowTransferToParent} or {@code disallowTransferToPeers} setting excludes.
*/
public final class TransferRejectedException extends IllegalStateException {

private final String agentName;

/** Creates an exception for a transfer rejected from {@code agentName} (the calling agent). */
public TransferRejectedException(String agentName, String message) {
super(message);
this.agentName = agentName;
}

/** Returns the name of the agent whose transfer was rejected. */
public String agentName() {
return agentName;
}
}
55 changes: 30 additions & 25 deletions core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import com.google.adk.agents.LlmAgent;
import com.google.adk.agents.ReadonlyContext;
import com.google.adk.agents.RunConfig.StreamingMode;
import com.google.adk.agents.TransferRejectedException;
import com.google.adk.events.Event;
import com.google.adk.flows.BaseFlow;
import com.google.adk.flows.llmflows.RequestProcessor.RequestProcessingResult;
Expand Down Expand Up @@ -424,9 +425,9 @@ private Single<LlmResponse> handleAfterModelCallback(
*
* <p>Handles early termination, LLM call limits, and agent transfer if needed.
*
* @return A {@link Flowable} of {@link Event} objects from this step.
* @return A {@link Flowable} of {@link Event} objects from this step, or an error with {@link
* TransferRejectedException} if a transfer names an agent this agent may not transfer to.
* @throws LlmCallsLimitExceededException if the agent exceeds allowed LLM invocations.
* @throws IllegalStateException if a transfer agent is specified but not found.
*/
private Flowable<Event> runOneStep(Context spanContext, InvocationContext context) {
AtomicReference<LlmRequest> llmRequestRef = new AtomicReference<>(LlmRequest.builder().build());
Expand Down Expand Up @@ -802,34 +803,17 @@ private Flowable<Event> followTransfer(
return self;
}
String agentToTransfer = event.actions().transferToAgent().get();
Optional<BaseAgent> nextAgent = context.agent().rootAgent().findAgent(agentToTransfer);
if (nextAgent.isEmpty()) {
logger.error("Agent not found: {}", agentToTransfer);
return self.concatWith(
Flowable.error(new IllegalStateException("Agent not found: " + agentToTransfer)));
}
BaseAgent target;
// Resumable only: restricting the default flow too would move behavior for existing callers.
if (context.isResumable() && context.agent() instanceof LlmAgent llmAgent) {
Optional<BaseAgent> declared = declaredTransferTarget(llmAgent, agentToTransfer);
if (declared.isEmpty()) {
return self.concatWith(
Flowable.error(
new IllegalStateException(
"Agent "
+ llmAgent.name()
+ " is not allowed to transfer to the requested agent.")));
}
target = declared.get();
} else {
target = nextAgent.get();
LlmAgent agent = (LlmAgent) context.agent();
Optional<BaseAgent> target = declaredTransferTarget(agent, agentToTransfer);
if (target.isEmpty()) {
return self.concatWith(Flowable.error(transferRejected(agent, agentToTransfer)));
}
// Otherwise the target can build its request from a session missing the transfer event.
return self.concatWith(
PersistBarrier.awaitPersisted(context, ImmutableList.of(event))
.andThen(
// Not around the barrier: andThen subscribes the target on the append's thread.
Flowable.defer(() -> target.runAsync(context))
Flowable.defer(() -> target.get().runAsync(context))
.compose(Tracing.withContext(spanContext))));
}

Expand All @@ -840,13 +824,34 @@ private Flowable<Event> followTransfer(
*/
private static Optional<BaseAgent> declaredTransferTarget(LlmAgent from, String name) {
if (name.equals(from.name())) {
return Optional.of(from);
return Optional.empty();
}
return AgentTransfer.transferTargets(from).stream()
.filter(candidate -> candidate.name().equals(name))
.findFirst();
}

/**
* Returns the error for a transfer {@link #declaredTransferTarget} rejected. The message names
* only {@code from}, since {@code name} is model output.
*/
private static TransferRejectedException transferRejected(LlmAgent from, String name) {
if (name.equals(from.name())) {
return new TransferRejectedException(
from.name(), "Agent '" + from.name() + "' cannot transfer to itself.");
}
if (from.rootAgent().findAgent(name).isEmpty()) {
return new TransferRejectedException(
from.name(),
"Agent '"
+ from.name()
+ "' requested a transfer to an agent that is not in the agent tree.");
}
return new TransferRejectedException(
from.name(),
"Agent '" + from.name() + "' is not allowed to transfer to the requested agent.");
}

/**
* Runs {@code callEvent}'s function calls and emits what follows from them: the tool-confirmation
* request, the function response, and any structured final response. Shared by the normal
Expand Down
Loading
Loading