diff --git a/core/src/main/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessor.java b/core/src/main/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessor.java index a93eb3cb4..6f73a0a8d 100644 --- a/core/src/main/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessor.java +++ b/core/src/main/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessor.java @@ -43,6 +43,7 @@ import io.reactivex.rxjava3.core.Single; import java.util.Collection; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; import java.util.Objects; import java.util.Optional; @@ -75,6 +76,28 @@ public Single processRequest( int finalConfirmationEventIndex = confirmationResult.get().eventIndex(); ImmutableMap requestConfirmationFunctionResponses = confirmationResult.get().responses(); + String agentName = invocationContext.agent().name(); + ImmutableMap functionCallsById = + functionCallsById(events, agentName); + ImmutableSet confirmationRequestedIds = confirmationRequestedIds(events); + // A tool has been confirmed, but it might already have been executed by a subsequent processor + // or in a subsequent turn: such calls have a function response after the user confirmation + // event. This is applied before the resumability check rather than after, because + // findMostRecentConfirmations re-matches the same stale user event on every later LLM call, so + // a settled confirmation would otherwise be re-examined - and re-logged - for the rest of the + // session. + // + // Only responses this agent produced count. A peer event landing after the approval that + // reuses the pending call's ID would otherwise convince this scan the tool had already run, + // silently dropping the approval - and it short-circuits before the resumability check, so + // that would not even leave a log line. + ImmutableSet alreadyResumedIds = + events.subList(finalConfirmationEventIndex + 1, events.size()).stream() + .filter(event -> Objects.equals(event.author(), agentName)) + .flatMap(event -> event.functionResponses().stream()) + .map(FunctionResponse::id) + .flatMap(Optional::stream) + .collect(toImmutableSet()); // Search backwards from the event before confirmation for the corresponding // request_confirmation function calls emitted by the model. @@ -83,6 +106,13 @@ public Single processRequest( if (event.functionCalls().isEmpty()) { continue; } + // Only this agent can ask this agent's user for confirmation. Function call parts also reach + // the session from an A2A peer response - ResponseConverter turns one into a model-role event + // authored by the local RemoteA2AAgent - and honouring a confirmation call from there would + // let the peer choose which local tool runs. + if (!Objects.equals(event.author(), agentName)) { + continue; + } Map toolsToResumeWithConfirmation = new HashMap<>(); Map toolsToResumeWithArgs = new HashMap<>(); @@ -95,6 +125,11 @@ public Single processRequest( .forEach( fc -> getOriginalFunctionCall(fc) + .filter(ofc -> !alreadyResumedIds.contains(ofc.id().get())) + .filter( + ofc -> + isResumableFunctionCall( + ofc, functionCallsById, confirmationRequestedIds, agentName)) .ifPresent( ofc -> { toolsToResumeWithConfirmation.put( @@ -103,23 +138,6 @@ public Single processRequest( toolsToResumeWithArgs.put(ofc.id().get(), ofc); })); - if (toolsToResumeWithConfirmation.isEmpty()) { - continue; - } - - // If a tool has been confirmed, it might have been executed by a subsequent - // processor, or in a subsequent turn. We identify tools that have already been - // executed by checking for function responses with matching IDs in events that - // occurred *after* the user confirmation event. - ImmutableSet alreadyConfirmedIds = - events.subList(finalConfirmationEventIndex + 1, events.size()).stream() - .flatMap(e -> e.functionResponses().stream()) - .map(FunctionResponse::id) - .flatMap(Optional::stream) - .collect(toImmutableSet()); - toolsToResumeWithConfirmation.keySet().removeAll(alreadyConfirmedIds); - toolsToResumeWithArgs.keySet().removeAll(alreadyConfirmedIds); - // If all confirmed tools in this event have already been processed, continue // searching in older events. if (toolsToResumeWithConfirmation.isEmpty()) { @@ -173,6 +191,126 @@ private static Optional findMostRecentConfirmations( return Optional.empty(); } + /** + * Indexes the tool function calls in session history by ID, keeping the most recent one per ID. + * + *

Confirmation calls are excluded: a confirmation resumes a real tool call, never another + * confirmation. + * + *

Collisions resolve last-wins, so a re-issue of an ID by {@code agentName} supersedes an + * earlier one - except that a foreign author may never displace a call {@code agentName} emitted. + * IDs are not globally unique and anyone can put an event in the session, so without that + * precedence a peer could reuse the ID of a call this agent is waiting on, shadow it, and have + * the author check in {@code isResumableFunctionCall} reject the legitimate confirmation + * - turning that check into a way for a peer to veto any pending tool call. + */ + private static ImmutableMap functionCallsById( + ImmutableList events, String agentName) { + Map byId = new LinkedHashMap<>(); + for (Event event : events) { + for (FunctionCall functionCall : event.functionCalls()) { + if (functionCall.id().isEmpty() + || Objects.equals( + functionCall.name().orElse(null), REQUEST_CONFIRMATION_FUNCTION_CALL_NAME)) { + continue; + } + String id = functionCall.id().get(); + AuthoredFunctionCall existing = byId.get(id); + if (existing == null + || Objects.equals(event.author(), agentName) + || !Objects.equals(existing.author(), agentName)) { + byId.put(id, new AuthoredFunctionCall(event.author(), functionCall)); + } + } + } + return ImmutableMap.copyOf(byId); + } + + /** + * Collects the IDs of function calls that a tool actually asked the user to confirm. + * + *

Covers both ways a confirmation is requested: a tool calling {@link + * com.google.adk.tools.ToolContext#requestConfirmation}, and a {@link + * com.google.adk.tools.FunctionTool} created with {@code requireConfirmation}, which routes + * through the same call. Accumulates over all events rather than keeping one event per ID: + * re-executing a confirmed tool emits a second function response with the same ID and no + * requested confirmations, which would otherwise shadow the original request. + */ + private static ImmutableSet confirmationRequestedIds(ImmutableList events) { + ImmutableSet.Builder ids = ImmutableSet.builder(); + for (Event event : events) { + Map requested = event.actions().requestedToolConfirmations(); + if (requested.isEmpty()) { + continue; + } + for (FunctionResponse functionResponse : event.functionResponses()) { + functionResponse.id().filter(requested::containsKey).ifPresent(ids::add); + } + } + return ids.build(); + } + + /** + * Returns whether {@code originalFunctionCall} faithfully reproduces a tool call {@code + * agentName} emitted and was genuinely awaiting confirmation. + * + *

The resumed call is read out of the {@code originalFunctionCall} argument of an {@code + * adk_request_confirmation} call found in session history, and function call parts reach the + * session from places other than the local model - notably an A2A peer response, which {@code + * ResponseConverter} turns into a model-role event. Resuming such a call unchecked would let + * whoever authored that event pick both the tool and its arguments, so only resume a call that + * matches one this agent emitted, by ID, author, name and arguments, and that a tool actually + * asked to have confirmed. + */ + private static boolean isResumableFunctionCall( + FunctionCall originalFunctionCall, + ImmutableMap functionCallsById, + ImmutableSet confirmationRequestedIds, + String agentName) { + String id = originalFunctionCall.id().get(); + AuthoredFunctionCall emitted = functionCallsById.get(id); + if (emitted == null) { + logger.warn( + "Ignoring tool confirmation for function call ID {}: no such function call in the session" + + " history.", + id); + return false; + } + if (!Objects.equals(emitted.author(), agentName)) { + // Another agent emitted the call; leave it for that agent's own processor. + logger.debug( + "Skipping tool confirmation for function call ID {}: emitted by {}, not by {}.", + id, + emitted.author(), + agentName); + return false; + } + if (!Objects.equals(emitted.functionCall().name(), originalFunctionCall.name())) { + logger.warn( + "Ignoring tool confirmation for function call ID {}: tool name does not match the" + + " function call this agent emitted.", + id); + return false; + } + if (!Objects.equals( + emitted.functionCall().args().orElse(ImmutableMap.of()), + originalFunctionCall.args().orElse(ImmutableMap.of()))) { + logger.warn( + "Ignoring tool confirmation for function call ID {}: arguments do not match the function" + + " call this agent emitted.", + id); + return false; + } + if (!confirmationRequestedIds.contains(id)) { + logger.warn( + "Ignoring tool confirmation for function call ID {}: no tool requested confirmation for" + + " it.", + id); + return false; + } + return true; + } + private Optional getOriginalFunctionCall(FunctionCall functionCall) { if (!functionCall.args().orElse(ImmutableMap.of()).containsKey(ORIGINAL_FUNCTION_CALL)) { return Optional.empty(); @@ -252,4 +390,7 @@ private static Optional> maybeCreateToolConf private record ConfirmationResult( ImmutableMap responses, int eventIndex) {} + + /** A tool function call from session history, together with the author of its event. */ + private record AuthoredFunctionCall(String author, FunctionCall functionCall) {} } diff --git a/core/src/test/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessorTest.java b/core/src/test/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessorTest.java index 55adeb39e..c8da89026 100644 --- a/core/src/test/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessorTest.java +++ b/core/src/test/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessorTest.java @@ -25,6 +25,8 @@ import com.google.adk.agents.InvocationContext; import com.google.adk.agents.LlmAgent; import com.google.adk.events.Event; +import com.google.adk.events.EventActions; +import com.google.adk.events.ToolConfirmation; import com.google.adk.models.LlmRequest; import com.google.adk.plugins.PluginManager; import com.google.adk.sessions.InMemorySessionService; @@ -44,6 +46,7 @@ @RunWith(JUnit4.class) public class RequestConfirmationLlmRequestProcessorTest { + private static final String AGENT_NAME = "test agent"; private static final String ECHO_TOOL_NAME = "echo_tool"; private static final String ORIGINAL_FUNCTION_CALL_ID = "original_fc_id"; private static final ImmutableMap ORIGINAL_FUNCTION_CALL_ARGS = @@ -60,15 +63,53 @@ public class RequestConfirmationLlmRequestProcessorTest { "args", Optional.of(ORIGINAL_FUNCTION_CALL_ARGS))); private static final FunctionCall FUNCTION_CALL = - FunctionCall.builder().id(FUNCTION_CALL_ID).name(ECHO_TOOL_NAME).args(ARGS).build(); + FunctionCall.builder() + .id(FUNCTION_CALL_ID) + .name(REQUEST_CONFIRMATION_FUNCTION_CALL_NAME) + .args(ARGS) + .build(); private static final InMemorySessionService sessionService = new InMemorySessionService(); - private static final Event REQUEST_CONFIRMATION_EVENT = + /** The tool call the agent itself emitted, which the confirmation later resumes. */ + private static final Event ORIGINAL_FUNCTION_CALL_EVENT = + functionCallEvent( + AGENT_NAME, + FunctionCall.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name(ECHO_TOOL_NAME) + .args(ORIGINAL_FUNCTION_CALL_ARGS) + .build()); + + /** + * The tool's own response asking for the call to be confirmed. This is what {@link + * com.google.adk.tools.ToolContext#requestConfirmation} produces, and what a {@code + * requireConfirmation} FunctionTool routes through. + */ + private static final Event CONFIRMATION_REQUESTED_EVENT = Event.builder() - .author("model") - .content(Content.fromParts(Part.builder().functionCall(FUNCTION_CALL).build())) + .author(AGENT_NAME) + .content( + Content.fromParts( + Part.builder() + .functionResponse( + FunctionResponse.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name(ECHO_TOOL_NAME) + .response(ImmutableMap.of("error", "requires confirmation")) + .build()) + .build())) + .actions( + EventActions.builder() + .requestedToolConfirmations( + ImmutableMap.of( + ORIGINAL_FUNCTION_CALL_ID, + ToolConfirmation.builder().hint("please confirm").build())) + .build()) .build(); + private static final Event REQUEST_CONFIRMATION_EVENT = + functionCallEvent(AGENT_NAME, FUNCTION_CALL); + private static final Event USER_CONFIRMATION_EVENT = Event.builder() .author("user") @@ -84,16 +125,21 @@ public class RequestConfirmationLlmRequestProcessorTest { .build())) .build(); + /** The full, legitimate lead-up to a user confirmation. */ + private static final ImmutableList CONFIRMED_CALL_EVENTS = + ImmutableList.of( + ORIGINAL_FUNCTION_CALL_EVENT, + CONFIRMATION_REQUESTED_EVENT, + REQUEST_CONFIRMATION_EVENT, + USER_CONFIRMATION_EVENT); + private static final RequestConfirmationLlmRequestProcessor processor = new RequestConfirmationLlmRequestProcessor(); @Test public void runAsync_withConfirmation_callsOriginalFunction() { LlmAgent agent = createAgentWithEchoTool(); - Session session = - Session.builder("session_id") - .events(ImmutableList.of(REQUEST_CONFIRMATION_EVENT, USER_CONFIRMATION_EVENT)) - .build(); + Session session = Session.builder("session_id").events(CONFIRMED_CALL_EVENTS).build(); InvocationContext context = buildInvocationContext(agent, session); @@ -113,9 +159,11 @@ public void runAsync_withConfirmation_callsOriginalFunction() { @Test public void runAsync_withConfirmationAndToolAlreadyCalled_doesNotCallOriginalFunction() { LlmAgent agent = createAgentWithEchoTool(); + // Authored by the agent, matching Functions.java:740 which builds real tool response events + // with invocationContext.agent().name(). Event toolResponseEvent = Event.builder() - .author("model") + .author(AGENT_NAME) .content( Content.fromParts( Part.builder() @@ -130,8 +178,10 @@ public void runAsync_withConfirmationAndToolAlreadyCalled_doesNotCallOriginalFun Session session = Session.builder("session_id") .events( - ImmutableList.of( - REQUEST_CONFIRMATION_EVENT, USER_CONFIRMATION_EVENT, toolResponseEvent)) + ImmutableList.builder() + .addAll(CONFIRMED_CALL_EVENTS) + .add(toolResponseEvent) + .build()) .build(); InvocationContext context = buildInvocationContext(agent, session); @@ -172,6 +222,206 @@ public void runAsync_noUserConfirmationEvent_empty() { .isEmpty(); } + @Test + public void runAsync_peerReusesPendingCallId_stillCallsOriginalFunction() { + // A peer must not be able to veto a pending confirmation by reusing the ID of a call this + // agent is waiting on. The history index resolves collisions last-wins, so without author + // precedence the peer's entry shadows the agent's, the author check rejects the legitimate + // confirmation, and the user's approval silently does nothing. + LlmAgent agent = createAgentWithEchoTool(); + Event peerNoise = + functionCallEvent( + "remote_a2a_agent", + FunctionCall.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name("peer_noise") + .args(ImmutableMap.of("x", "y")) + .build()); + Session session = + Session.builder("session_id") + .events( + ImmutableList.of( + ORIGINAL_FUNCTION_CALL_EVENT, + CONFIRMATION_REQUESTED_EVENT, + REQUEST_CONFIRMATION_EVENT, + peerNoise, + USER_CONFIRMATION_EVENT)) + .build(); + + assertThat(resumedEvents(agent, session)).hasSize(1); + } + + @Test + public void runAsync_peerFakesExecutedResponse_stillCallsOriginalFunction() { + // The already-resumed scan must only count responses this agent produced. Otherwise a peer + // event landing after the approval, carrying a response that reuses the pending call's ID, + // convinces the processor the tool already ran. That short-circuits before the resumability + // check, so the approval is dropped with no diagnostics at all. + LlmAgent agent = createAgentWithEchoTool(); + Event peerResponse = + Event.builder() + .author("remote_a2a_agent") + .content( + Content.fromParts( + Part.builder() + .functionResponse( + FunctionResponse.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name("peer_noise") + .response(ImmutableMap.of("status", "whatever")) + .build()) + .build())) + .build(); + Session session = + Session.builder("session_id") + .events( + ImmutableList.builder() + .addAll(CONFIRMED_CALL_EVENTS) + .add(peerResponse) + .build()) + .build(); + + assertThat(resumedEvents(agent, session)).hasSize(1); + } + + @Test + public void runAsync_originalCallNotInHistory_doesNotCallOriginalFunction() { + LlmAgent agent = createAgentWithEchoTool(); + Session session = + Session.builder("session_id") + .events(ImmutableList.of(REQUEST_CONFIRMATION_EVENT, USER_CONFIRMATION_EVENT)) + .build(); + + assertThat(resumedEvents(agent, session)).isEmpty(); + } + + @Test + public void runAsync_originalCallEmittedByAnotherAgent_doesNotCallOriginalFunction() { + // The original call is in history and matches by name and args, but a different agent emitted + // it. Only the emitting agent's own processor may resume it. + LlmAgent agent = createAgentWithEchoTool(); + Session session = + Session.builder("session_id") + .events( + replacingFirst( + CONFIRMED_CALL_EVENTS, + functionCallEvent( + "remote_a2a_agent", + FunctionCall.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name(ECHO_TOOL_NAME) + .args(ORIGINAL_FUNCTION_CALL_ARGS) + .build()))) + .build(); + + assertThat(resumedEvents(agent, session)).isEmpty(); + } + + @Test + public void runAsync_confirmationCallFromAnotherAuthor_doesNotCallOriginalFunction() { + // Everything is legitimate except the event carrying the adk_request_confirmation call, which + // an A2A peer injected through RemoteA2AAgent. It must not resume a local tool. + LlmAgent agent = createAgentWithEchoTool(); + Session session = + Session.builder("session_id") + .events( + ImmutableList.of( + ORIGINAL_FUNCTION_CALL_EVENT, + CONFIRMATION_REQUESTED_EVENT, + functionCallEvent("remote_a2a_agent", FUNCTION_CALL), + USER_CONFIRMATION_EVENT)) + .build(); + + assertThat(resumedEvents(agent, session)).isEmpty(); + } + + @Test + public void runAsync_toolNeverRequestedConfirmation_doesNotCallOriginalFunction() { + // Replaying a call that ran without ever asking for confirmation must not re-run it. + LlmAgent agent = createAgentWithEchoTool(); + Session session = + Session.builder("session_id") + .events( + ImmutableList.of( + ORIGINAL_FUNCTION_CALL_EVENT, + REQUEST_CONFIRMATION_EVENT, + USER_CONFIRMATION_EVENT)) + .build(); + + assertThat(resumedEvents(agent, session)).isEmpty(); + } + + @Test + public void runAsync_confirmationWithMismatchedToolName_doesNotCallOriginalFunction() { + LlmAgent agent = createAgentWithEchoTool(); + Session session = + Session.builder("session_id") + .events( + replacingFirst( + CONFIRMED_CALL_EVENTS, + functionCallEvent( + AGENT_NAME, + FunctionCall.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name("some_other_tool") + .args(ORIGINAL_FUNCTION_CALL_ARGS) + .build()))) + .build(); + + assertThat(resumedEvents(agent, session)).isEmpty(); + } + + @Test + public void runAsync_confirmationWithMismatchedArgs_doesNotCallOriginalFunction() { + LlmAgent agent = createAgentWithEchoTool(); + Session session = + Session.builder("session_id") + .events( + replacingFirst( + CONFIRMED_CALL_EVENTS, + functionCallEvent( + AGENT_NAME, + FunctionCall.builder() + .id(ORIGINAL_FUNCTION_CALL_ID) + .name(ECHO_TOOL_NAME) + .args(ImmutableMap.of("say", "something else")) + .build()))) + .build(); + + assertThat(resumedEvents(agent, session)).isEmpty(); + } + + @Test + public void testAgentNameMatchesFixtures() { + // The fixtures hard-code the author, so catch a rename in TestUtils rather than silently + // turning every negative test into a false pass. + assertThat(createAgentWithEchoTool().name()).isEqualTo(AGENT_NAME); + } + + private static ImmutableList resumedEvents(LlmAgent agent, Session session) { + return ImmutableList.copyOf( + processor + .processRequest(buildInvocationContext(agent, session), LlmRequest.builder().build()) + .blockingGet() + .events()); + } + + /** Returns {@code events} with its first element swapped for {@code replacement}. */ + private static ImmutableList replacingFirst( + ImmutableList events, Event replacement) { + return ImmutableList.builder() + .add(replacement) + .addAll(events.subList(1, events.size())) + .build(); + } + + private static Event functionCallEvent(String author, FunctionCall functionCall) { + return Event.builder() + .author(author) + .content(Content.fromParts(Part.builder().functionCall(functionCall).build())) + .build(); + } + private static InvocationContext buildInvocationContext(LlmAgent agent, Session session) { return InvocationContext.builder() .pluginManager(new PluginManager())