refactor: add the runtime that runs child nodes from a workflow node

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

@@ -0,0 +1,48 @@
+/*
+ * 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.annotations.Experimental;
+
+/**
+ * An exception that {@link NodeRunner#runChild} signals when the child node fails. If the caller
+ * does not handle this exception, the caller fails with the child's error without emitting an error
+ * event of its own.
+ */
+@Experimental
+final class DynamicNodeFailedException extends RuntimeException {
+
+  private final String errorNodePath;
+
+  DynamicNodeFailedException(String nodeName, Exception error, String errorNodePath) {
+    super("Dynamic node '" + nodeName + "' failed.", error);
+    this.errorNodePath = errorNodePath;
+  }
+
+  /** Returns the exception the child failed with, which is also this exception's cause. */
+  public Exception error() {
+    return (Exception) getCause();
+  }
+
+  /**
+   * Returns the path of the node that raised {@link #error()}, which is either the child or one of
+   * its descendants.
+   */
+  public String errorNodePath() {
+    return errorNodePath;
+  }
+}

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

@@ -25,35 +25,43 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import java.util.Collection;
-import java.util.HashMap;
-import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicReference;
 import org.jspecify.annotations.Nullable;
 
 /**
  * The state of one workflow node run, held by the run's {@link Context}. Only the workflow runtime
- * creates it and updates the bookkeeping that records which results an event has carried.
+ * creates it and updates the bookkeeping that records which results an event has carried. Its
+ * output, interrupt IDs and child run IDs are safe to update from several threads, since children
+ * that a node runs itself may start and finish on other threads.
  */
 @Experimental
 public final class NodeExecutionState {
 
+  private static final OutputState NO_OUTPUT = new OutputState(OutputStatus.NONE, null);
+  private static final OutputState DELEGATED_OUTPUT = new OutputState(OutputStatus.DELEGATED, null);
+
   private final Node node;
   private final NodeRunner.EventSink eventSink;
   private final @Nullable Context parent;
+  // The parent's state when this run's output is also the parent's.
+  private final @Nullable NodeExecutionState outputParent;
   private final String runId;
   private final String nodePath;
   private final ImmutableList<String> outputFor;
-  private final Map<String, Integer> childRunCounts = new HashMap<>();
+  private final ConcurrentMap<String, Integer> childRunCounts = new ConcurrentHashMap<>();
+  private final Set<String> interruptIds = new CopyOnWriteArraySet<>();
+  private final AtomicReference<OutputState> outputState = new AtomicReference<>(NO_OUTPUT);
   private String eventAuthor;
-  private final Set<String> interruptIds = new LinkedHashSet<>();
-  private @Nullable Object output;
-  private boolean outputEmitted;
   private @Nullable ImmutableList<Route> selectedRoutes;
   private boolean routesEmitted;
   private @Nullable Exception failure;
+  private @Nullable String failureNodePath;
 
   private NodeExecutionState(Builder builder) {
     checkState(builder.node != null, "A node state needs a node.");
@@ -64,15 +72,16 @@ private NodeExecutionState(Builder builder) {
     this.parent = builder.parent;
     this.runId = builder.runId;
     NodeExecutionState parentState = parent == null ? null : parent.requireNodeState();
+    this.outputParent = builder.useAsOutput ? parentState : null;
     this.nodePath =
         NodeRunner.appendSegment(
             parentState == null ? null : parentState.nodePath,
             node.name(),
             runId,
             /* separator= */ '/');
     this.outputFor =
-        builder.useAsOutput && parentState != null
-            ? ImmutableList.<String>builder().add(nodePath).addAll(parentState.outputFor).build()
+        outputParent != null
+            ? ImmutableList.<String>builder().add(nodePath).addAll(outputParent.outputFor).build()
             : ImmutableList.of(nodePath);
     this.eventAuthor = parentState == null ? "" : parentState.eventAuthor;
   }
@@ -118,24 +127,36 @@ public void eventAuthor(String eventAuthor) {
     this.eventAuthor = checkNotNull(eventAuthor);
   }
 
-  /** Returns the node's output, or empty if it has produced none. */
+  /**
+   * Returns the node's output, which is the output of its {@code useAsOutput} child once that child
+   * emits one, or empty if it has none.
+   */
   public Optional<Object> output() {
-    return Optional.ofNullable(output);
+    return Optional.ofNullable(outputState.get().value());
   }
 
   /**
    * Records the node's output, which an event carries when the node finishes unless an earlier
    * event already carried it. A node produces at most one output.
    *
-   * @throws IllegalStateException if the node already has an output.
+   * @throws IllegalStateException if the node already has an output, or if it delegated its output
+   *     to a {@code useAsOutput} child.
    */
   public void produceOutput(Object output) {
     checkNotNull(output);
-    checkState(
-        this.output == null,
-        "Node '%s' produced a second output; a node produces at most one output.",
-        node.name());
-    this.output = output;
+    outputState.updateAndGet(
+        current -> {
+          checkState(
+              !current.status().isDelegated(),
+              "Node '%s' delegated its output to a useAsOutput child and must not produce one of"
+                  + " its own.",
+              node.name());
+          checkState(
+              current.status() == OutputStatus.NONE,
+              "Node '%s' produced a second output; a node produces at most one output.",
+              node.name());
+          return new OutputState(OutputStatus.PRODUCED, output);
+        });
   }
 
   /**
@@ -177,13 +198,73 @@ String nextChildRunId(String name) {
     return String.valueOf(childRunCounts.merge(name, 1, Integer::sum));
   }
 
-  /** Returns whether an event has carried the node's output. */
+  /** Returns whether an event has carried the node's own or delegated output. */
   boolean outputEmitted() {
-    return outputEmitted;
+    OutputStatus status = outputState.get().status();
+    return status == OutputStatus.EMITTED || status == OutputStatus.DELEGATE_EMITTED;
   }
 
+  /**
+   * Records that an event carried the output the node produced, which also makes it the output of
+   * each ancestor that delegated its output to this run.
+   *
+   * @throws IllegalStateException if the node has no unsent output.
+   */
   void markOutputEmitted() {
-    outputEmitted = true;
+    Object emitted =
+        outputState
+            .updateAndGet(
+                current -> {
+                  checkState(
+                      current.status() == OutputStatus.PRODUCED,
+                      "Node '%s' has no unsent output to mark as emitted.",
+                      node.name());
+                  return new OutputState(OutputStatus.EMITTED, current.value());
+                })
+            .value();
+    for (NodeExecutionState ancestor = outputParent;
+        ancestor != null;
+        ancestor = ancestor.outputParent) {
+      ancestor.recordDelegatedOutput(emitted);
+    }
+  }
+
+  /**
+   * Records {@code output}, emitted by a {@code useAsOutput} descendant, as the node's output if
+   * the node delegated its output. A later output replaces an earlier one, as when a workflow
+   * records its own output after its terminal node's.
+   */
+  private void recordDelegatedOutput(@Nullable Object output) {
+    outputState.updateAndGet(
+        current ->
+            current.status().isDelegated()
+                ? new OutputState(OutputStatus.DELEGATE_EMITTED, output)
+                : current);
+  }
+
+  /**
+   * Delegates the node's output to a {@code useAsOutput} child.
+   *
+   * @throws IllegalStateException if the node already has an output or a delegate.
+   */
+  void claimOutputDelegation() {
+    outputState.updateAndGet(
+        current -> {
+          checkState(
+              !current.status().isDelegated(),
+              "Node '%s' already delegated its output to a useAsOutput child.",
+              node.name());
+          checkState(
+              current.status() == OutputStatus.NONE,
+              "Node '%s' already produced an output and cannot delegate it to a useAsOutput child.",
+              node.name());
+          return DELEGATED_OUTPUT;
+        });
+  }
+
+  /** Withdraws the delegation of the node's output, unless the delegate has emitted an output. */
+  void releaseOutputDelegation() {
+    outputState.compareAndSet(DELEGATED_OUTPUT, NO_OUTPUT);
   }
 
   /** Returns whether an event has carried the node's current route selection. */
@@ -204,10 +285,49 @@ Optional<Exception> failure() {
     return Optional.ofNullable(failure);
   }
 
+  /** Records that the node failed with {@code failure}, which it raised itself. */
   void failure(Exception failure) {
+    failure(failure, nodePath);
+  }
+
+  /**
+   * Records that the node failed with {@code failure}, which the node at {@code failureNodePath}
+   * raised.
+   */
+  void failure(Exception failure, String failureNodePath) {
     this.failure = checkNotNull(failure);
+    this.failureNodePath = checkNotNull(failureNodePath);
   }
 
+  /**
+   * Returns the path of the node that raised {@link #failure()}: this run's path, or the path of a
+   * descendant whose failure the node carried up. Empty if the node has not failed.
+   */
+  Optional<String> failureNodePath() {
+    return Optional.ofNullable(failureNodePath);
+  }
+
+  /** Status of a run's output. */
+  private enum OutputStatus {
+    /** The run has no output and no delegate. */
+    NONE,
+    /** A {@code useAsOutput} child owns the run's output and has emitted none yet. */
+    DELEGATED,
+    /** The run produced an output that no event has carried yet. */
+    PRODUCED,
+    /** An event carried the output the run produced. */
+    EMITTED,
+    /** An event of a {@code useAsOutput} descendant carried the run's output. */
+    DELEGATE_EMITTED;
+
+    boolean isDelegated() {
+      return this == DELEGATED || this == DELEGATE_EMITTED;
+    }
+  }
+
+  /** A run's output status and its output, held together so that each update is atomic. */
+  private record OutputState(OutputStatus status, @Nullable Object value) {}
+
   /** Builder for {@link NodeExecutionState}. */
   @Experimental
   static final class Builder {

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

@@ -46,9 +46,11 @@ public abstract class NodeInfo {
 
   /**
    * Returns the node paths this event's output counts for: the emitting node's path, followed by
-   * the paths of the ancestor workflows that use this event as their output. It is set on an event
-   * that carries the node's output, either in its {@code output} field or as a message-as-output
-   * event whose content is the output; an empty list means the output counts for no node.
+   * the paths of the ancestors that use this output as their own, such as the workflow of a
+   * terminal node or a node that ran the emitting node with {@code useAsOutput}. It is set on an
+   * event that carries the node's output, either in its {@code output} field or as a
+   * message-as-output event whose content is the output; an empty list means the output counts for
+   * no node.
    */
   @JsonProperty("outputFor")
   @JsonInclude(JsonInclude.Include.NON_EMPTY)

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

@@ -0,0 +1,32 @@
+/*
+ * 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.annotations.Experimental;
+
+/**
+ * A signal that ends a node's run when a child it ran with {@link NodeRunner#runChild} waits, so
+ * the node waits too. Node code that catches this signal (for example, by catching every {@link
+ * RuntimeException}) keeps running past the wait, and anything it then emits is still recorded.
+ */
+@Experimental
+final class NodeInterruptedException extends RuntimeException {
+
+  NodeInterruptedException() {
+    super("Node interrupted.");
+  }
+}