feat(HITL): Let ADK resume after HITL approval is present

feat(HITL): Declining a proposal now correctly intercepts the run

fix: Events for HITL are now emitted correctly
fix: HITL endless loop when asking for approvals
PiperOrigin-RevId: 839858592

Author: google-genai-bot

core/src/main/java/com/google/adk/agents/LlmAgent.java

@@ -736,8 +736,8 @@ public IncludeContents includeContents() {
     return includeContents;
   }
 
-  public List<BaseTool> tools() {
-    return canonicalTools().toList().blockingGet();
+  public Single<List<BaseTool>> tools() {
+    return canonicalTools().toList();
   }
 
   public List<Object> toolsUnion() {

core/src/main/java/com/google/adk/flows/llmflows/Functions.java

@@ -17,6 +17,7 @@
 
 package com.google.adk.flows.llmflows;
 
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 
 import com.google.adk.Telemetry;
@@ -64,7 +65,7 @@
 public final class Functions {
 
   private static final String AF_FUNCTION_CALL_ID_PREFIX = "adk-";
-  static final String REQUEST_CONFIRMATION_FUNCTION_CALL_NAME = "adk_request_confirmation";
+  public static final String REQUEST_CONFIRMATION_FUNCTION_CALL_NAME = "adk_request_confirmation";
   private static final Logger logger = LoggerFactory.getLogger(Functions.class);
 
   /** Generates a unique ID for a function call. */
@@ -147,12 +148,22 @@ public static Maybe<Event> handleFunctionCalls(
     Function<FunctionCall, Maybe<Event>> functionCallMapper =
         functionCall -> {
           BaseTool tool = tools.get(functionCall.name().get());
+          ToolConfirmation toolConfirmation = toolConfirmations.get(functionCall.id().orElse(null));
           ToolContext toolContext =
               ToolContext.builder(invocationContext)
                   .functionCallId(functionCall.id().orElse(""))
-                  .toolConfirmation(toolConfirmations.get(functionCall.id().orElse(null)))
+                  .toolConfirmation(toolConfirmation)
                   .build();
 
+          if (toolConfirmation != null && !toolConfirmation.confirmed()) {
+            return Maybe.just(
+                buildResponseEvent(
+                    tool,
+                    ImmutableMap.of("error", "User declined tool execution for " + tool.name()),
+                    toolContext,
+                    invocationContext));
+          }
+
           Map<String, Object> functionArgs = functionCall.args().orElse(ImmutableMap.of());
 
           Maybe<Map<String, Object>> maybeFunctionResult =
@@ -241,6 +252,18 @@ public static Maybe<Event> handleFunctionCalls(
    */
   public static Maybe<Event> handleFunctionCallsLive(
       InvocationContext invocationContext, Event functionCallEvent, Map<String, BaseTool> tools) {
+    return handleFunctionCallsLive(invocationContext, functionCallEvent, tools, ImmutableMap.of());
+  }
+
+  /**
+   * Handles function calls in a live/streaming context with tool confirmations, supporting
+   * background execution and stream termination.
+   */
+  public static Maybe<Event> handleFunctionCallsLive(
+      InvocationContext invocationContext,
+      Event functionCallEvent,
+      Map<String, BaseTool> tools,
+      Map<String, ToolConfirmation> toolConfirmations) {
     ImmutableList<FunctionCall> functionCalls = functionCallEvent.functionCalls();
 
     for (FunctionCall functionCall : functionCalls) {
@@ -255,7 +278,9 @@ public static Maybe<Event> handleFunctionCallsLive(
           ToolContext toolContext =
               ToolContext.builder(invocationContext)
                   .functionCallId(functionCall.id().orElse(""))
+                  .toolConfirmation(toolConfirmations.get(functionCall.id().orElse(null)))
                   .build();
+
           Map<String, Object> functionArgs = functionCall.args().orElse(new HashMap<>());
 
           Maybe<Map<String, Object>> maybeFunctionResult =
@@ -664,5 +689,29 @@ public static Optional<Event> generateRequestConfirmationEvent(
             .build());
   }
 
+  /**
+   * Gets the ask user confirmation function calls from the event.
+   *
+   * @param event The event to extract function calls from.
+   * @return A list of function calls for asking user confirmation.
+   */
+  public static ImmutableList<FunctionCall> getAskUserConfirmationFunctionCalls(Event event) {
+    return event
+        .content()
+        .flatMap(Content::parts)
+        .map(
+            parts ->
+                parts.stream()
+                    .flatMap(part -> part.functionCall().stream())
+                    .filter(
+                        functionCall ->
+                            functionCall
+                                .name()
+                                .map(name -> name.equals(REQUEST_CONFIRMATION_FUNCTION_CALL_NAME))
+                                .orElse(false))
+                    .collect(toImmutableList()))
+        .orElse(ImmutableList.of());
+  }
+
   private Functions() {}
 }

core/src/main/java/com/google/adk/flows/llmflows/RequestConfirmationLlmRequestProcessor.java

@@ -19,10 +19,11 @@
 import static com.google.adk.flows.llmflows.Functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import com.google.adk.JsonBaseModel;
 import com.google.adk.agents.InvocationContext;
 import com.google.adk.agents.LlmAgent;
 import com.google.adk.events.Event;
@@ -31,14 +32,15 @@
 import com.google.adk.tools.ToolConfirmation;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.genai.types.Content;
 import com.google.genai.types.FunctionCall;
 import com.google.genai.types.FunctionResponse;
 import com.google.genai.types.Part;
 import io.reactivex.rxjava3.core.Maybe;
 import io.reactivex.rxjava3.core.Single;
 import java.util.Collection;
-import java.util.List;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -49,68 +51,138 @@
 public class RequestConfirmationLlmRequestProcessor implements RequestProcessor {
   private static final Logger logger =
       LoggerFactory.getLogger(RequestConfirmationLlmRequestProcessor.class);
-  private final ObjectMapper objectMapper;
-
-  public RequestConfirmationLlmRequestProcessor() {
-    objectMapper = new ObjectMapper().registerModule(new Jdk8Module());
-  }
+  private static final ObjectMapper OBJECT_MAPPER = JsonBaseModel.getMapper();
+  private static final String ORIGINAL_FUNCTION_CALL = "originalFunctionCall";
 
   @Override
   public Single<RequestProcessor.RequestProcessingResult> processRequest(
       InvocationContext invocationContext, LlmRequest llmRequest) {
-    List<Event> events = invocationContext.session().events();
+    ImmutableList<Event> events = ImmutableList.copyOf(invocationContext.session().events());
     if (events.isEmpty()) {
       logger.info(
           "No events are present in the session. Skipping request confirmation processing.");
       return Single.just(RequestProcessingResult.create(llmRequest, ImmutableList.of()));
     }
 
-    ImmutableMap<String, ToolConfirmation> requestConfirmationFunctionResponses =
-        filterRequestConfirmationFunctionResponses(events);
+    ImmutableMap<String, ToolConfirmation> responses = ImmutableMap.of();
+    int confirmationEventIndex = -1;
+    for (int i = events.size() - 1; i >= 0; i--) {
+      Event event = events.get(i);
+      if (!Objects.equals(event.author(), "user")) {
+        continue;
+      }
+      if (event.functionResponses().isEmpty()) {
+        continue;
+      }
+      responses =
+          event.functionResponses().stream()
+              .filter(functionResponse -> functionResponse.id().isPresent())
+              .filter(
+                  functionResponse ->
+                      Objects.equals(
+                          functionResponse.name().orElse(null),
+                          REQUEST_CONFIRMATION_FUNCTION_CALL_NAME))
+              .map(this::maybeCreateToolConfirmationEntry)
+              .flatMap(Optional::stream)
+              .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
+      confirmationEventIndex = i;
+      break;
+    }
+
+    // Make it final to enable access from lambda expressions.
+    final ImmutableMap<String, ToolConfirmation> requestConfirmationFunctionResponses = responses;
+
     if (requestConfirmationFunctionResponses.isEmpty()) {
       logger.info("No request confirmation function responses found.");
       return Single.just(RequestProcessingResult.create(llmRequest, ImmutableList.of()));
     }
 
-    for (ImmutableList<FunctionCall> functionCalls :
-        events.stream()
-            .map(Event::functionCalls)
-            .filter(fc -> !fc.isEmpty())
-            .collect(toImmutableList())) {
+    for (int i = events.size() - 2; i >= 0; i--) {
+      Event event = events.get(i);
+      if (event.functionCalls().isEmpty()) {
+        continue;
+      }
+
+      Map<String, ToolConfirmation> toolsToResumeWithConfirmation = new HashMap<>();
+      Map<String, FunctionCall> toolsToResumeWithArgs = new HashMap<>();
+
+      event.functionCalls().stream()
+          .filter(
+              fc ->
+                  fc.id().isPresent()
+                      && requestConfirmationFunctionResponses.containsKey(fc.id().get()))
+          .forEach(
+              fc ->
+                  getOriginalFunctionCall(fc)
+                      .ifPresent(
+                          ofc -> {
+                            toolsToResumeWithConfirmation.put(
+                                ofc.id().get(),
+                                requestConfirmationFunctionResponses.get(fc.id().get()));
+                            toolsToResumeWithArgs.put(ofc.id().get(), ofc);
+                          }));
+
+      if (toolsToResumeWithConfirmation.isEmpty()) {
+        continue;
+      }
+
+      // Remove the tools that have already been confirmed.
+      ImmutableSet<String> alreadyConfirmedIds =
+          events.subList(confirmationEventIndex + 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);
 
-      ImmutableMap<String, FunctionCall> toolsToResumeWithArgs =
-          filterToolsToResumeWithArgs(functionCalls, requestConfirmationFunctionResponses);
-      ImmutableMap<String, ToolConfirmation> toolsToResumeWithConfirmation =
-          toolsToResumeWithArgs.keySet().stream()
-              .filter(
-                  id ->
-                      events.stream()
-                          .flatMap(e -> e.functionResponses().stream())
-                          .anyMatch(fr -> Objects.equals(fr.id().orElse(null), id)))
-              .collect(toImmutableMap(k -> k, requestConfirmationFunctionResponses::get));
       if (toolsToResumeWithConfirmation.isEmpty()) {
-        logger.info("No tools to resume with confirmation.");
         continue;
       }
 
       return assembleEvent(
-              invocationContext, toolsToResumeWithArgs.values(), toolsToResumeWithConfirmation)
-          .map(event -> RequestProcessingResult.create(llmRequest, ImmutableList.of(event)))
+              invocationContext,
+              toolsToResumeWithArgs.values(),
+              ImmutableMap.copyOf(toolsToResumeWithConfirmation))
+          .map(e -> RequestProcessingResult.create(llmRequest, ImmutableList.of(e)))
           .toSingle();
     }
 
     return Single.just(RequestProcessingResult.create(llmRequest, ImmutableList.of()));
   }
 
+  private Optional<FunctionCall> getOriginalFunctionCall(FunctionCall functionCall) {
+    if (!functionCall.args().orElse(ImmutableMap.of()).containsKey(ORIGINAL_FUNCTION_CALL)) {
+      return Optional.empty();
+    }
+    try {
+      FunctionCall originalFunctionCall =
+          OBJECT_MAPPER.convertValue(
+              functionCall.args().get().get(ORIGINAL_FUNCTION_CALL), FunctionCall.class);
+      if (originalFunctionCall.id().isEmpty()) {
+        return Optional.empty();
+      }
+      return Optional.of(originalFunctionCall);
+    } catch (IllegalArgumentException e) {
+      logger.warn("Failed to convert originalFunctionCall argument.", e);
+      return Optional.empty();
+    }
+  }
+
   private Maybe<Event> assembleEvent(
       InvocationContext invocationContext,
       Collection<FunctionCall> functionCalls,
       Map<String, ToolConfirmation> toolConfirmations) {
-    ImmutableMap.Builder<String, BaseTool> toolsBuilder = ImmutableMap.builder();
+    Single<ImmutableMap<String, BaseTool>> toolsMapSingle;
     if (invocationContext.agent() instanceof LlmAgent llmAgent) {
-      for (BaseTool tool : llmAgent.tools()) {
-        toolsBuilder.put(tool.name(), tool);
-      }
+      toolsMapSingle =
+          llmAgent
+              .tools()
+              .map(
+                  toolList ->
+                      toolList.stream().collect(toImmutableMap(BaseTool::name, tool -> tool)));
+    } else {
+      toolsMapSingle = Single.just(ImmutableMap.of());
     }
 
     var functionCallEvent =
@@ -124,23 +196,10 @@ private Maybe<Event> assembleEvent(
                     .build())
             .build();
 
-    return Functions.handleFunctionCalls(
-        invocationContext, functionCallEvent, toolsBuilder.buildOrThrow(), toolConfirmations);
-  }
-
-  private ImmutableMap<String, ToolConfirmation> filterRequestConfirmationFunctionResponses(
-      List<Event> events) {
-    return events.stream()
-        .filter(event -> Objects.equals(event.author(), "user"))
-        .flatMap(event -> event.functionResponses().stream())
-        .filter(functionResponse -> functionResponse.id().isPresent())
-        .filter(
-            functionResponse ->
-                Objects.equals(
-                    functionResponse.name().orElse(null), REQUEST_CONFIRMATION_FUNCTION_CALL_NAME))
-        .map(this::maybeCreateToolConfirmationEntry)
-        .flatMap(Optional::stream)
-        .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
+    return toolsMapSingle.flatMapMaybe(
+        toolsMap ->
+            Functions.handleFunctionCalls(
+                invocationContext, functionCallEvent, toolsMap, toolConfirmations));
   }
 
   private Optional<Map.Entry<String, ToolConfirmation>> maybeCreateToolConfirmationEntry(
@@ -150,36 +209,19 @@ private Optional<Map.Entry<String, ToolConfirmation>> maybeCreateToolConfirmatio
       return Optional.of(
           Map.entry(
               functionResponse.id().get(),
-              objectMapper.convertValue(responseMap, ToolConfirmation.class)));
+              OBJECT_MAPPER.convertValue(responseMap, ToolConfirmation.class)));
     }
 
     try {
       return Optional.of(
           Map.entry(
               functionResponse.id().get(),
-              objectMapper.readValue(
+              OBJECT_MAPPER.readValue(
                   (String) responseMap.get("response"), ToolConfirmation.class)));
     } catch (JsonProcessingException e) {
       logger.error("Failed to parse tool confirmation response", e);
     }
 
     return Optional.empty();
   }
-
-  private ImmutableMap<String, FunctionCall> filterToolsToResumeWithArgs(
-      ImmutableList<FunctionCall> functionCalls,
-      Map<String, ToolConfirmation> requestConfirmationFunctionResponses) {
-    return functionCalls.stream()
-        .filter(fc -> fc.id().isPresent())
-        .filter(fc -> requestConfirmationFunctionResponses.containsKey(fc.id().get()))
-        .filter(
-            fc -> Objects.equals(fc.name().orElse(null), REQUEST_CONFIRMATION_FUNCTION_CALL_NAME))
-        .filter(fc -> fc.args().orElse(ImmutableMap.of()).containsKey("originalFunctionCall"))
-        .collect(
-            toImmutableMap(
-                fc -> fc.id().get(),
-                fc ->
-                    objectMapper.convertValue(
-                        fc.args().get().get("originalFunctionCall"), FunctionCall.class)));
-  }
 }

core/src/main/java/com/google/adk/flows/llmflows/SingleFlow.java

@@ -31,6 +31,7 @@ public class SingleFlow extends BaseLlmFlow {
           new Identity(),
           new Contents(),
           new Examples(),
+          new RequestConfirmationLlmRequestProcessor(),
           CodeExecution.requestProcessor);
 
   protected static final ImmutableList<ResponseProcessor> RESPONSE_PROCESSORS =