feat: pause a workflow for user input

PiperOrigin-RevId: 1004727066

Author: damianmomotgoogle

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

@@ -323,9 +323,9 @@ public final void routes(List<? extends Route> routes) {
   }
 
   /**
-   * Returns the user's answers to the long-running calls that this workflow node or its descendants
-   * made in an earlier turn, keyed by interrupt ID. A node that reruns on resume reads these
-   * answers. A null answer has a null value in the map.
+   * Returns the user's answers to the requests for input and long-running calls that this workflow
+   * node or its descendants made in an earlier turn, keyed by interrupt ID. A node that reruns on
+   * resume reads these answers. A null answer has a null value in the map.
    *
    * @throws IllegalStateException if this is not the context of a workflow node.
    */

core/src/main/java/com/google/adk/workflow/BaseNode.java

@@ -119,21 +119,27 @@ public Optional<Schema> stateSchema() {
 
   /**
    * Runs the node and emits its events. Each emission of {@link #runNode} that is an {@link Event}
-   * passes through, and any other value becomes the output of a new event.
+   * passes through, a {@link RequestInput} becomes the event that asks for it, and any other value
+   * becomes the output of a new event.
    */
   final Flowable<Event> run(Context context, @Nullable Object nodeInput) {
     InvocationContext invocationContext = checkNotNull(context).invocationContext();
     // The node runner fills in the author, as only it knows the node's place in the graph.
     return Flowable.<Object>defer(() -> runNode(context, nodeInput))
         .map(
-            item ->
-                item instanceof Event event
-                    ? event
-                    : Event.builder()
-                        .id(invocationContext.newUuid())
-                        .timestamp(invocationContext.now().toEpochMilli())
-                        .author("")
-                        .output(item)
-                        .build());
+            item -> {
+              if (item instanceof Event event) {
+                return event;
+              }
+              if (item instanceof RequestInput request) {
+                return request.toEvent(invocationContext);
+              }
+              return Event.builder()
+                  .id(invocationContext.newUuid())
+                  .timestamp(invocationContext.now().toEpochMilli())
+                  .author("")
+                  .output(item)
+                  .build();
+            });
   }
 }

core/src/main/java/com/google/adk/workflow/Node.java

@@ -98,9 +98,10 @@ default boolean requiresAllPredecessors() {
   /**
    * Runs the node on {@code nodeInput}, the output it receives from its predecessors.
    *
-   * <p>An emitted {@link Event} passes through, and any other emitted value becomes the node's
-   * output (at most one per run); emit nothing for no output, since RxJava forbids null. The
-   * wildcard lets an implementation return a {@code Flowable} of its own type.
+   * <p>An emitted {@link Event} passes through, an emitted {@link RequestInput} pauses the workflow
+   * until the user answers it, and any other emitted value becomes the node's output (at most one
+   * per run); emit nothing for no output, since RxJava forbids null. The wildcard lets an
+   * implementation return a {@code Flowable} of its own type.
    */
   @Experimental
   Flowable<?> runNode(Context context, @Nullable Object nodeInput);

core/src/main/java/com/google/adk/workflow/NodeExecutionState.java

@@ -26,6 +26,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.reactivex.rxjava3.subjects.SingleSubject;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashMap;
@@ -61,6 +62,7 @@ public final class NodeExecutionState {
   private final ImmutableList<String> outputFor;
   private final Map<String, @Nullable Object> resumeInputs;
   private final ConcurrentMap<String, Integer> childRunCounts = new ConcurrentHashMap<>();
+  private final ConcurrentMap<String, SingleSubject<Context>> childRuns = new ConcurrentHashMap<>();
   // The paths of the node runs started but not finished in this turn, shared under one root.
   private final Set<String> unfinishedNodePaths;
   private final Set<String> interruptIds = new CopyOnWriteArraySet<>();
@@ -194,9 +196,9 @@ public ImmutableSet<String> interruptIds() {
   }
 
   /**
-   * Returns the user's answers to the long-running calls that this node or its descendants raised
-   * in an earlier turn, keyed by interrupt ID. A null answer has a null value in the map. The map
-   * is empty unless the node reruns on resume.
+   * Returns the user's answers to the requests for input and long-running calls that this node or
+   * its descendants raised in an earlier turn, keyed by interrupt ID. A null answer has a null
+   * value in the map. The map is empty unless the node reruns on resume.
    */
   public Map<String, @Nullable Object> resumeInputs() {
     return resumeInputs;
@@ -224,6 +226,14 @@ String nextChildRunId(String name) {
     return String.valueOf(childRunCounts.merge(name, 1, Integer::sum));
   }
 
+  /**
+   * Returns the runs of the children this node runs itself, keyed by {@code name@runId}, which
+   * calls with the same run ID share.
+   */
+  ConcurrentMap<String, SingleSubject<Context>> childRuns() {
+    return childRuns;
+  }
+
   /** Records that this run started; it counts as canceled mid-run until it finishes. */
   void markStarted() {
     unfinishedNodePaths.add(nodePath);
@@ -255,6 +265,11 @@ synchronized ImmutableMap<String, ResumeScan.RecoveredNode> recoveredChildren(
     return recoveredChildren;
   }
 
+  /** Returns the parent's state if this run's output is also the parent's, or null. */
+  @Nullable NodeExecutionState outputParent() {
+    return outputParent;
+  }
+
   /** Returns whether an event has carried the node's own or delegated output. */
   boolean outputEmitted() {
     OutputStatus status = outputState.get().status();

core/src/main/java/com/google/adk/workflow/NodeRunner.java

@@ -40,10 +40,12 @@
 import io.reactivex.rxjava3.processors.FlowableProcessor;
 import io.reactivex.rxjava3.processors.UnicastProcessor;
 import io.reactivex.rxjava3.subjects.CompletableSubject;
+import io.reactivex.rxjava3.subjects.SingleSubject;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
 import org.jspecify.annotations.Nullable;
 
@@ -218,8 +220,7 @@ private static Maybe<Object> startChild(
       parentState.claimOutputDelegation();
     }
     String runId = config.runId().orElseGet(() -> parentState.nextChildRunId(node.name()));
-    // Deferred so that a throw from scan or intercept releases the delegation, as a failure does.
-    return Single.defer(() -> runOrReplayChild(parent, node, nodeInput, runId, config))
+    return sharedChildRun(parent, node, nodeInput, runId, config)
         .doOnError(
             error -> {
               if (delegatesOutput) {
@@ -229,6 +230,50 @@ private static Maybe<Object> startChild(
         .flatMapMaybe(child -> childResult(parent, child, delegatesOutput, config.raiseOnWait()));
   }
 
+  /**
+   * Returns the run of {@code node} under {@code runId} shared by the caller's calls with that run
+   * ID, starting it if needed. A run that fails or waits for an output it did not produce is not
+   * shared, so a later call runs the child again. When the call that started a run is canceled, a
+   * call still waiting on it runs the child itself.
+   */
+  private static Single<Context> sharedChildRun(
+      Context parent, Node node, @Nullable Object nodeInput, String runId, RunNodeConfig config) {
+    return Single.defer(
+        () -> {
+          ConcurrentMap<String, SingleSubject<Context>> childRuns =
+              parent.requireNodeState().childRuns();
+          String childKey = segment(node.name(), runId);
+          SingleSubject<Context> run = SingleSubject.create();
+          SingleSubject<Context> shared = childRuns.putIfAbsent(childKey, run);
+          if (shared != null) {
+            return shared.onErrorResumeNext(
+                error ->
+                    error instanceof ChildRunCanceledException
+                        ? sharedChildRun(parent, node, nodeInput, runId, config)
+                        : Single.error(error));
+          }
+          // Defer so a throw from scan or intercept releases the claimed run instead of leaking it.
+          return Single.defer(() -> runOrReplayChild(parent, node, nodeInput, runId, config))
+              .doOnSuccess(
+                  child -> {
+                    if (!keepsItsRun(child)) {
+                      childRuns.remove(childKey, run);
+                    }
+                    run.onSuccess(child);
+                  })
+              .doOnError(
+                  error -> {
+                    childRuns.remove(childKey, run);
+                    run.onError(error);
+                  })
+              .doOnDispose(
+                  () -> {
+                    childRuns.remove(childKey, run);
+                    run.onError(new ChildRunCanceledException());
+                  });
+        });
+  }
+
   /**
    * Runs {@code node} as a child of {@code parent}, or replays the result it recorded in an earlier
    * turn under the same run ID.
@@ -258,6 +303,22 @@ private static Single<Context> runOrReplayChild(
         .run(nodeInput);
   }
 
+  /**
+   * Returns whether a later call with the same run ID shares this finished run of a child: a failed
+   * child runs again, as does one that waits for an output and left no outcome.
+   */
+  private static boolean keepsItsRun(Context child) {
+    NodeExecutionState state = child.requireNodeState();
+    if (state.failure().isPresent()) {
+      return false;
+    }
+    return !state.interruptIds().isEmpty()
+        || state.output().isPresent()
+        || state.selectedRoutes().isPresent()
+        || child.eventActions().transferToAgent().isPresent()
+        || !state.node().waitForOutput();
+  }
+
   /**
    * Returns the child's output, or fails with the child's failure or with an interrupt, which the
    * caller takes on as its own.
@@ -292,9 +353,11 @@ private static Maybe<Object> childResult(
         && (childNode instanceof Workflow || childNode.waitForOutput())) {
       return Maybe.error(new NodeInterruptedException());
     }
-    // No event of this run carries a delegate's replayed output, so the caller emits it.
+    // No event of this run carries a delegate's replayed or shared output, so the caller emits it.
     Optional<Object> output = childState.output();
-    if (delegatesOutput && output.isPresent() && !childState.outputEmitted()) {
+    boolean emittedForCaller =
+        childState.outputEmitted() && childState.outputParent() == parentState;
+    if (delegatesOutput && output.isPresent() && !emittedForCaller) {
       parentState.adoptReplayedOutput(output.get());
     }
     return Maybe.fromOptional(output);
@@ -532,6 +595,13 @@ private Event.Builder newEvent() {
     }
   }
 
+  /** Ends the waiting calls of a shared child run whose first caller was canceled. */
+  private static final class ChildRunCanceledException extends RuntimeException {
+    ChildRunCanceledException() {
+      super("The call that ran this child was canceled.");
+    }
+  }
+
   /**
    * An event queued for the runner, paired with the signal that completes once the caller receives
    * the event and, for an error event, with the failure it reports.

core/src/main/java/com/google/adk/workflow/RequestInput.java

@@ -0,0 +1,127 @@
+/*
+ * 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.workflow;
+
+import com.google.adk.JsonBaseModel;
+import com.google.adk.agents.InvocationContext;
+import com.google.adk.annotations.Experimental;
+import com.google.adk.events.Event;
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableSet;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import com.google.genai.types.Content;
+import com.google.genai.types.FunctionCall;
+import com.google.genai.types.Part;
+import com.google.genai.types.Schema;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * A node's request for user input, which pauses the workflow until the user answers it. A node
+ * emits it from {@link Node#runNode}, and the runtime records it as a long-running {@value
+ * #FUNCTION_CALL_NAME} function call whose ID is {@link #interruptId}. The user answers with a
+ * function response that carries the same ID.
+ */
+@Experimental
+@AutoValue
+public abstract class RequestInput {
+
+  /** The name of the function call that carries a request for input. */
+  public static final String FUNCTION_CALL_NAME = "adk_request_input";
+
+  static final String INTERRUPT_ID_KEY = "interruptId";
+  static final String MESSAGE_KEY = "message";
+  static final String PAYLOAD_KEY = "payload";
+  static final String RESPONSE_SCHEMA_KEY = "response_schema";
+
+  /** Returns the ID of the request, which the answering function response carries. */
+  public abstract String interruptId();
+
+  /** Returns the message to show the user, if any. */
+  public abstract Optional<String> message();
+
+  /** Returns additional data for rendering the request, if any. */
+  public abstract Optional<Object> payload();
+
+  /**
+   * Returns the schema the answer must match, if any. An answer of the form {@code {"result":
+   * value}} is unwrapped first, and a string value is read as JSON unless the schema accepts a
+   * string.
+   */
+  public abstract Optional<Schema> responseSchema();
+
+  /** Returns a builder whose interrupt ID is a new random UUID. */
+  public static Builder builder() {
+    return new AutoValue_RequestInput.Builder().interruptId(UUID.randomUUID().toString());
+  }
+
+  public abstract Builder toBuilder();
+
+  /**
+   * Returns the long-running function-call event that records this request. It has the model role,
+   * so history rewriting keeps it as the call that the answer responds to.
+   *
+   * @throws IllegalArgumentException if the payload has no JSON form
+   */
+  Event toEvent(InvocationContext invocationContext) {
+    Map<String, @Nullable Object> args = new LinkedHashMap<>();
+    args.put(INTERRUPT_ID_KEY, interruptId());
+    args.put(MESSAGE_KEY, message().orElse(null));
+    // Converted here so a payload with no JSON form fails the asking node, not the session service.
+    args.put(PAYLOAD_KEY, payload().map(RequestInput::toJsonValue).orElse(null));
+    args.put(RESPONSE_SCHEMA_KEY, responseSchema().map(RequestInput::toJsonValue).orElse(null));
+    FunctionCall call =
+        FunctionCall.builder().name(FUNCTION_CALL_NAME).id(interruptId()).args(args).build();
+    return Event.builder()
+        .id(invocationContext.newUuid())
+        .timestamp(invocationContext.now().toEpochMilli())
+        .author("")
+        .content(
+            Content.builder()
+                .role("model")
+                .parts(Part.builder().functionCall(call).build())
+                .build())
+        .longRunningToolIds(ImmutableSet.of(interruptId()))
+        .build();
+  }
+
+  private static @Nullable Object toJsonValue(Object value) {
+    return JsonBaseModel.getMapper().convertValue(value, Object.class);
+  }
+
+  /** Builder for {@link RequestInput}. */
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+    @CanIgnoreReturnValue
+    public abstract Builder interruptId(String interruptId);
+
+    @CanIgnoreReturnValue
+    public abstract Builder message(@Nullable String message);
+
+    @CanIgnoreReturnValue
+    public abstract Builder payload(@Nullable Object payload);
+
+    @CanIgnoreReturnValue
+    public abstract Builder responseSchema(@Nullable Schema responseSchema);
+
+    public abstract RequestInput build();
+  }
+}