From 145901db1e7142cf15c66b7298c8116c14111883 Mon Sep 17 00:00:00 2001 From: Damian Momot Date: Thu, 22 Oct 2026 04:27:37 -0700 Subject: [PATCH] refactor: add the runtime that runs child nodes from a workflow node PiperOrigin-RevId: 1004670046 --- .../workflow/DynamicNodeFailedException.java | 48 ++ .../adk/workflow/NodeExecutionState.java | 162 ++++- .../com/google/adk/workflow/NodeInfo.java | 8 +- .../workflow/NodeInterruptedException.java | 32 + .../com/google/adk/workflow/NodeRunner.java | 110 +++- .../google/adk/workflow/RunNodeConfig.java | 121 ++++ .../com/google/adk/workflow/Scheduler.java | 8 +- .../google/adk/testing/WorkflowTestNodes.java | 121 ++++ .../adk/workflow/DynamicDispatchTest.java | 582 ++++++++++++++++++ .../adk/workflow/RunNodeConfigTest.java | 115 ++++ 10 files changed, 1279 insertions(+), 28 deletions(-) create mode 100644 core/src/main/java/com/google/adk/workflow/DynamicNodeFailedException.java create mode 100644 core/src/main/java/com/google/adk/workflow/NodeInterruptedException.java create mode 100644 core/src/main/java/com/google/adk/workflow/RunNodeConfig.java create mode 100644 core/src/test/java/com/google/adk/testing/WorkflowTestNodes.java create mode 100644 core/src/test/java/com/google/adk/workflow/DynamicDispatchTest.java create mode 100644 core/src/test/java/com/google/adk/workflow/RunNodeConfigTest.java diff --git a/core/src/main/java/com/google/adk/workflow/DynamicNodeFailedException.java b/core/src/main/java/com/google/adk/workflow/DynamicNodeFailedException.java new file mode 100644 index 000000000..4c85d9b0a --- /dev/null +++ b/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; + } +} diff --git a/core/src/main/java/com/google/adk/workflow/NodeExecutionState.java b/core/src/main/java/com/google/adk/workflow/NodeExecutionState.java index 3c191ff67..f1c53dab1 100644 --- a/core/src/main/java/com/google/adk/workflow/NodeExecutionState.java +++ b/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 outputFor; - private final Map childRunCounts = new HashMap<>(); + private final ConcurrentMap childRunCounts = new ConcurrentHashMap<>(); + private final Set interruptIds = new CopyOnWriteArraySet<>(); + private final AtomicReference outputState = new AtomicReference<>(NO_OUTPUT); private String eventAuthor; - private final Set interruptIds = new LinkedHashSet<>(); - private @Nullable Object output; - private boolean outputEmitted; private @Nullable ImmutableList 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,6 +72,7 @@ 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, @@ -71,8 +80,8 @@ private NodeExecutionState(Builder builder) { runId, /* separator= */ '/'); this.outputFor = - builder.useAsOutput && parentState != null - ? ImmutableList.builder().add(nodePath).addAll(parentState.outputFor).build() + outputParent != null + ? ImmutableList.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 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 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 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 { diff --git a/core/src/main/java/com/google/adk/workflow/NodeInfo.java b/core/src/main/java/com/google/adk/workflow/NodeInfo.java index 9b8afc564..9016f6a79 100644 --- a/core/src/main/java/com/google/adk/workflow/NodeInfo.java +++ b/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) diff --git a/core/src/main/java/com/google/adk/workflow/NodeInterruptedException.java b/core/src/main/java/com/google/adk/workflow/NodeInterruptedException.java new file mode 100644 index 000000000..026b66f7f --- /dev/null +++ b/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."); + } +} diff --git a/core/src/main/java/com/google/adk/workflow/NodeRunner.java b/core/src/main/java/com/google/adk/workflow/NodeRunner.java index e2c6ae389..3eeb721b3 100644 --- a/core/src/main/java/com/google/adk/workflow/NodeRunner.java +++ b/core/src/main/java/com/google/adk/workflow/NodeRunner.java @@ -16,6 +16,7 @@ package com.google.adk.workflow; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; import static com.google.common.base.Strings.isNullOrEmpty; @@ -26,15 +27,19 @@ import com.google.adk.annotations.Experimental; import com.google.adk.events.Event; import com.google.adk.events.EventActions; +import com.google.common.base.CharMatcher; +import com.google.common.collect.ImmutableSet; import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.genai.types.FinishReason; import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.core.Maybe; import io.reactivex.rxjava3.core.Single; import io.reactivex.rxjava3.processors.FlowableProcessor; import io.reactivex.rxjava3.processors.UnicastProcessor; import io.reactivex.rxjava3.subjects.CompletableSubject; import java.util.Map; +import java.util.Optional; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.function.Function; @@ -44,11 +49,14 @@ * The runner for one workflow node. It stamps every emitted event with its author, node path, and * branch before forwarding it, and records the node's output, routes, interrupts, and failure on * the node state of the returned context. {@link #runRoot} runs a node as the root of an - * invocation. + * invocation, and {@code runChild} runs one as a child that another node starts itself. */ @Experimental public final class NodeRunner { + // '/' and '@' form node-path segments and '.' separates branch segments, so disallow all three. + static final CharMatcher PATH_SEPARATORS = CharMatcher.anyOf("/@."); + /** Sink that receives the events of a workflow run. */ @Experimental interface EventSink { @@ -170,6 +178,97 @@ public static Flowable runRoot( }); } + /** + * Returns a {@link Maybe} that, when subscribed, runs {@code node} as a child of the node whose + * context is {@code parent} and emits the child's output, or completes empty if it has none. The + * caller includes it in its run and catches only failures it handles, such as {@link + * DynamicNodeFailedException}: when the child waits, it fails with an internal {@link + * RuntimeException} that must propagate to end the caller's run, so that the caller waits too. + * + *

It fails with {@link DynamicNodeFailedException} if the child fails; with {@link + * IllegalStateException} if {@code parent} is not a workflow node's context, if the caller's + * {@link Node#rerunOnResume} is false, or if {@code config} delegates an output that the caller + * already has or has delegated; and with {@link IllegalArgumentException} if the child's name + * cannot form a node path. + */ + static Maybe runChild( + Context parent, Node node, @Nullable Object nodeInput, RunNodeConfig config) { + checkNotNull(parent); + checkNotNull(node); + checkNotNull(config); + return Maybe.defer(() -> startChild(parent, node, nodeInput, config)); + } + + /** Runs {@code node} as a child of {@code parent} and turns the child's run into its result. */ + private static Maybe startChild( + Context parent, Node node, @Nullable Object nodeInput, RunNodeConfig config) { + NodeExecutionState parentState = parent.requireNodeState(); + checkState( + parentState.node().rerunOnResume(), + "Node '%s' must rerun on resume to run child nodes: a child it runs may be interrupted, and" + + " the workflow reruns the caller on resume to collect the child's output.", + parentState.node().name()); + // No graph validates a child that a node runs itself, so check that it keeps paths intact. + checkArgument(!node.name().isEmpty(), "A node name must not be empty."); + checkArgument( + PATH_SEPARATORS.matchesNoneOf(node.name()), + "A node name must not contain '/', '@', or '.', but '%s' does. Agent names allow these" + + " characters, so an agent run as a node needs a name that a node path can carry.", + node.name()); + // A workflow takes its output from its terminal node, so it never delegates it to a child. + boolean delegatesOutput = config.useAsOutput() && !(parentState.node() instanceof Workflow); + if (delegatesOutput) { + parentState.claimOutputDelegation(); + } + return builder() + .node(node) + .parent(parent) + .runId(config.runId().orElseGet(() -> parentState.nextChildRunId(node.name()))) + .useAsOutput(config.useAsOutput()) + .useSubBranch(config.useSubBranch()) + .overrideBranch(config.overrideBranch().orElse(null)) + .build() + .run(nodeInput) + .flatMapMaybe(child -> childResult(parent, child, delegatesOutput, config.raiseOnWait())); + } + + /** + * Returns the child's output, or fails with the child's failure or with an interrupt, which the + * caller takes on as its own. + */ + private static Maybe childResult( + Context parent, Context child, boolean delegatesOutput, boolean raiseOnWait) { + NodeExecutionState parentState = parent.requireNodeState(); + NodeExecutionState childState = child.requireNodeState(); + if (child.eventActions().skipSummarization().orElse(false)) { + parent.eventActions().setSkipSummarization(true); + } + Optional failure = childState.failure(); + if (failure.isPresent()) { + // A delegate that fails before it emits an output leaves the caller free to produce one. + if (delegatesOutput) { + parentState.releaseOutputDelegation(); + } + return Maybe.error( + new DynamicNodeFailedException( + childState.node().name(), failure.get(), childState.failureNodePath().orElseThrow())); + } + ImmutableSet interruptIds = childState.interruptIds(); + if (!interruptIds.isEmpty()) { + // The caller waits as well, and reruns on resume to collect the child's output. + parentState.addInterruptIds(interruptIds); + return Maybe.error(new NodeInterruptedException()); + } + Node childNode = childState.node(); + if (raiseOnWait + && childState.output().isEmpty() + && child.eventActions().transferToAgent().isEmpty() + && (childNode instanceof Workflow || childNode.waitForOutput())) { + return Maybe.error(new NodeInterruptedException()); + } + return Maybe.fromOptional(childState.output()); + } + /** * Runs the node on {@code nodeInput} and returns its context when the node finishes. An exception * from the node is recorded on the context's node state rather than signaled, while an {@code @@ -231,6 +330,15 @@ Completable run(@Nullable Object nodeInput) { } private Completable onError(Throwable error) { + if (error instanceof NodeInterruptedException) { + // A child waits, or raiseOnWait ended the run; any interrupt IDs are already recorded. + return flushPending(); + } + if (error instanceof DynamicNodeFailedException childFailure) { + // The child already sent an error event, so the node takes on its failure without one. + nodeState.failure(childFailure.error(), childFailure.errorNodePath()); + return Completable.complete(); + } // Only an Exception fails the node; an Error is fatal and propagates. return error instanceof Exception exception ? fail(exception) : Completable.error(error); } diff --git a/core/src/main/java/com/google/adk/workflow/RunNodeConfig.java b/core/src/main/java/com/google/adk/workflow/RunNodeConfig.java new file mode 100644 index 000000000..ccf0adfb1 --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/RunNodeConfig.java @@ -0,0 +1,121 @@ +/* + * 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 static com.google.common.base.Preconditions.checkArgument; + +import com.google.adk.annotations.Experimental; +import com.google.auto.value.AutoValue; +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import java.util.Optional; +import org.jspecify.annotations.Nullable; + +/** Configuration for running a child node with {@link NodeRunner#runChild}. */ +@Experimental +@AutoValue +abstract class RunNodeConfig { + + /** + * Returns the ID of the child's run, which names the run in node paths and branches. When empty, + * the child receives the next number among the caller's runs of nodes with the same name. + */ + public abstract Optional runId(); + + /** + * Returns whether the child's output is also the caller's output. When true, the caller must not + * produce an output of its own, even if the child produces none, unless the child fails before + * emitting its output. + */ + public abstract boolean useAsOutput(); + + /** Returns whether the child runs on its own sub-branch, {@code .@}. */ + public abstract boolean useSubBranch(); + + /** Returns the branch the child runs on instead of its caller's, if any. */ + public abstract Optional overrideBranch(); + + /** + * Returns whether a {@link Workflow} or wait-for-output child that finishes without an output + * ends its caller's run, as an interrupt does, instead of completing the call empty. + */ + public abstract boolean raiseOnWait(); + + public static Builder builder() { + return new AutoValue_RunNodeConfig.Builder() + .useAsOutput(false) + .useSubBranch(false) + .raiseOnWait(false); + } + + public abstract Builder toBuilder(); + + /** Builder for {@link RunNodeConfig}. */ + @AutoValue.Builder + public abstract static class Builder { + + /** + * Sets the ID of the child's run, which must contain a non-digit character and must not contain + * {@code /}, {@code @}, or {@code .}; null or empty generates an ID. + */ + @CanIgnoreReturnValue + public abstract Builder runId(@Nullable String runId); + + abstract Optional runId(); + + @CanIgnoreReturnValue + public abstract Builder useAsOutput(boolean useAsOutput); + + @CanIgnoreReturnValue + public abstract Builder useSubBranch(boolean useSubBranch); + + @CanIgnoreReturnValue + public abstract Builder overrideBranch(@Nullable String overrideBranch); + + @CanIgnoreReturnValue + public abstract Builder raiseOnWait(boolean raiseOnWait); + + abstract RunNodeConfig autoBuild(); + + /** + * Builds the config. + * + * @throws IllegalArgumentException if the run ID has no non-digit character or contains {@code + * /}, {@code @}, or {@code .}. + */ + public final RunNodeConfig build() { + if (runId().map(String::isEmpty).orElse(false)) { + runId(null); + } + RunNodeConfig config = autoBuild(); + config + .runId() + .ifPresent( + runId -> { + // A run ID of digits only could collide with an ID the runtime generates. + checkArgument( + !runId.chars().allMatch(Character::isDigit), + "runId \"%s\" must contain a non-digit character.", + runId); + checkArgument( + NodeRunner.PATH_SEPARATORS.matchesNoneOf(runId), + "runId \"%s\" must not contain '/', '@', or '.'.", + runId); + }); + return config; + } + } +} diff --git a/core/src/main/java/com/google/adk/workflow/Scheduler.java b/core/src/main/java/com/google/adk/workflow/Scheduler.java index cc335c906..2f77f884b 100644 --- a/core/src/main/java/com/google/adk/workflow/Scheduler.java +++ b/core/src/main/java/com/google/adk/workflow/Scheduler.java @@ -146,9 +146,10 @@ private void handleCompletions() { */ private boolean handleFailures(List batch) { for (Completion completion : batch) { - Optional failure = completion.context().requireNodeState().failure(); + NodeExecutionState childState = completion.context().requireNodeState(); + Optional failure = childState.failure(); if (failure.isPresent()) { - workflowState.failure(failure.get()); + workflowState.failure(failure.get(), childState.failureNodePath().orElseThrow()); failed = true; childRuns.dispose(); wakeUps.onComplete(); @@ -287,7 +288,8 @@ private void enqueueBarrierTrigger(String target) { * records the output of the sole terminal node that produced one as the workflow's output. */ private Completable finalizeRun() { - if (!interruptIds.isEmpty()) { + // A child run from the workflow's own context records its interrupts on the workflow. + if (!interruptIds.isEmpty() || !workflowState.interruptIds().isEmpty()) { workflowState.addInterruptIds(interruptIds); return Completable.complete(); } diff --git a/core/src/test/java/com/google/adk/testing/WorkflowTestNodes.java b/core/src/test/java/com/google/adk/testing/WorkflowTestNodes.java new file mode 100644 index 000000000..7672087ef --- /dev/null +++ b/core/src/test/java/com/google/adk/testing/WorkflowTestNodes.java @@ -0,0 +1,121 @@ +/* + * 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.testing; + +import com.google.adk.agents.Context; +import com.google.adk.events.Event; +import com.google.adk.workflow.Edge; +import com.google.adk.workflow.Node; +import com.google.adk.workflow.Workflow; +import com.google.common.collect.ImmutableSet; +import io.reactivex.rxjava3.core.Flowable; +import java.util.function.BiFunction; +import org.jspecify.annotations.Nullable; + +/** Workflows and workflow nodes with common test behaviors, so a test can declare them inline. */ +public final class WorkflowTestNodes { + + /** Returns a workflow named {@code wf} that runs {@code edges}. */ + public static Workflow workflow(Edge... edges) { + return workflow("wf", edges); + } + + /** Returns a workflow named {@code name} that runs {@code edges}. */ + public static Workflow workflow(String name, Edge... edges) { + return Workflow.builder().name(name).edges(edges).build(); + } + + /** + * Returns a node that runs {@code body} and reruns on resume, as required for nodes that run + * child nodes. + */ + public static Node dispatcher( + String name, BiFunction> body) { + return new DispatchingNode(name, body); + } + + /** Returns a node that emits {@code output}. */ + public static Node emitter(String name, Object output) { + return new TestNode(name, (ctx, input) -> Flowable.just(output)); + } + + /** Returns a node that outputs its input prefixed with {@code prefix} and a colon. */ + public static Node echo(String name, String prefix) { + return new TestNode(name, (ctx, input) -> Flowable.just(prefix + ":" + input)); + } + + /** Returns a node that fails with {@code failure}. */ + public static Node throwing(String name, Exception failure) { + return new TestNode(name, (ctx, input) -> Flowable.error(failure)); + } + + /** Returns a node that emits {@code output} and then fails. */ + public static Node emitThenFail(String name, Object output) { + return new TestNode( + name, + (ctx, input) -> + Flowable.just(output) + .concatWith(Flowable.error(new IllegalStateException("failed after its output")))); + } + + /** Returns a node that waits on the long-running tool call {@code interruptId}. */ + public static Node interrupting(String name, String interruptId) { + return new TestNode( + name, + (ctx, input) -> + Flowable.just( + Event.builder() + .id(Event.generateEventId()) + .author("") + .longRunningToolIds(ImmutableSet.of(interruptId)) + .build())); + } + + /** Returns a node that waits for an output and produces none. */ + public static Node waitingForOutput(String name) { + return new WaitingNode(name); + } + + private record DispatchingNode( + String name, BiFunction> body) implements Node { + + @Override + public boolean rerunOnResume() { + return true; + } + + @Override + public Flowable runNode(Context context, @Nullable Object nodeInput) { + return body.apply(context, nodeInput); + } + } + + private record WaitingNode(String name) implements Node { + + @Override + public boolean waitForOutput() { + return true; + } + + @Override + public Flowable runNode(Context context, @Nullable Object nodeInput) { + return Flowable.empty(); + } + } + + private WorkflowTestNodes() {} +} diff --git a/core/src/test/java/com/google/adk/workflow/DynamicDispatchTest.java b/core/src/test/java/com/google/adk/workflow/DynamicDispatchTest.java new file mode 100644 index 000000000..5707955f7 --- /dev/null +++ b/core/src/test/java/com/google/adk/workflow/DynamicDispatchTest.java @@ -0,0 +1,582 @@ +/* + * 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 static com.google.adk.testing.WorkflowTestNodes.dispatcher; +import static com.google.adk.testing.WorkflowTestNodes.echo; +import static com.google.adk.testing.WorkflowTestNodes.emitThenFail; +import static com.google.adk.testing.WorkflowTestNodes.emitter; +import static com.google.adk.testing.WorkflowTestNodes.interrupting; +import static com.google.adk.testing.WorkflowTestNodes.throwing; +import static com.google.adk.testing.WorkflowTestNodes.waitingForOutput; +import static com.google.adk.testing.WorkflowTestNodes.workflow; +import static com.google.common.collect.ImmutableList.toImmutableList; +import static com.google.common.collect.MoreCollectors.onlyElement; +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.adk.agents.CallbackContext; +import com.google.adk.agents.Context; +import com.google.adk.events.Event; +import com.google.adk.testing.TestNode; +import com.google.adk.testing.TestUtils; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import io.reactivex.rxjava3.core.Completable; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.core.Maybe; +import io.reactivex.rxjava3.schedulers.Schedulers; +import io.reactivex.rxjava3.subjects.MaybeSubject; +import java.util.List; +import java.util.Optional; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; +import org.jspecify.annotations.Nullable; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** + * Tests for the runtime that runs a child node from a workflow node ({@link NodeRunner#runChild}): + * output delegation, failure paths and interrupts, checked on the events and node state of each + * run. + */ +@RunWith(JUnit4.class) +public final class DynamicDispatchTest { + + private static final Start START = Start.INSTANCE; + private static final RunNodeConfig DEFAULTS = RunNodeConfig.builder().build(); + private static final RunNodeConfig USE_AS_OUTPUT = + RunNodeConfig.builder().useAsOutput(true).build(); + private static final RunNodeConfig RAISE_ON_WAIT = + RunNodeConfig.builder().raiseOnWait(true).build(); + + // Children that run concurrently send their events on several threads. + private final Queue events = new ConcurrentLinkedQueue<>(); + private final NodeRunner.EventSink recordingSink = + (event, nodeFailure) -> Completable.fromAction(() -> events.add(event)); + + @Test + public void runChild_concurrentRuns_getDistinctRunIdsAndTheirOwnOutputs() { + int count = 64; + Node child = echo("child", "echo"); + Node dispatcher = + dispatcher( + "dispatcher", + (ctx, input) -> + Flowable.range(1, count) + .concatMapEager( + i -> + NodeRunner.runChild(ctx, child, i, DEFAULTS) + .subscribeOn(Schedulers.io()) + .toFlowable()) + .toList() + .toFlowable()); + + run(workflow(new Edge(START, dispatcher))); + + assertThat(pathsContaining("/child@")) + .containsExactlyElementsIn( + IntStream.rangeClosed(1, count) + .mapToObj(i -> "wf@1/dispatcher@1/child@" + i) + .collect(toImmutableList())); + assertThat(outputEventAt("wf@1/dispatcher@1").output()) + .hasValue( + IntStream.rangeClosed(1, count).mapToObj(i -> "echo:" + i).collect(toImmutableList())); + } + + @Test + public void runChild_useAsOutputOnATerminalNode_countsTheOutputForEveryLevel() { + Node step1 = emitter("step_1", "step_1_done"); + Node step2 = echo("step_2", "final"); + Workflow inner = workflow("inner_wf", new Edge(START, step1), new Edge(step1, step2)); + Node nodeA = delegating("node_a", inner); + + Context context = run(workflow(new Edge(START, nodeA))); + + ImmutableList outputEvents = + recorded().stream().filter(event -> event.output().isPresent()).collect(toImmutableList()); + assertThat(paths(outputEvents)) + .containsExactly("wf@1/node_a@1/inner_wf@1/step_1@1", "wf@1/node_a@1/inner_wf@1/step_2@1") + .inOrder(); + assertThat(outputFor(outputEvents.get(0))).containsExactly("wf@1/node_a@1/inner_wf@1/step_1@1"); + assertThat(outputFor(outputEvents.get(1))) + .containsExactly( + "wf@1/node_a@1/inner_wf@1/step_2@1", + "wf@1/node_a@1/inner_wf@1", + "wf@1/node_a@1", + "wf@1") + .inOrder(); + assertThat(context.requireNodeState().output()).hasValue("final:step_1_done"); + } + + @Test + public void runChild_useAsOutputFromAWorkflowsContext_delegatesNothing() { + Node helper = + new TestNode( + "helper", + (ctx, input) -> + NodeRunner.runChild( + workflowContext(ctx), emitter("first", "A"), null, USE_AS_OUTPUT) + .zipWith( + NodeRunner.runChild( + workflowContext(ctx), emitter("second", "B"), null, USE_AS_OUTPUT), + ImmutableList::of) + .toFlowable()); + + run(workflow(new Edge(START, helper))); + + assertThat(outputEventAt("wf@1/helper@1").output()).hasValue(ImmutableList.of("A", "B")); + } + + @Test + public void runChild_secondUseAsOutput_failsTheCaller() { + Node dispatcher = + dispatcher( + "dispatcher", + (ctx, input) -> + NodeRunner.runChild(ctx, emitter("first", "A"), null, USE_AS_OUTPUT) + .ignoreElement() + .andThen( + NodeRunner.runChild(ctx, emitter("second", "B"), null, USE_AS_OUTPUT) + .ignoreElement()) + .toFlowable()); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure).hasMessageThat().contains("already delegated its output"); + } + + @Test + public void runChild_useAsOutputThenCallerProducesOutputOnItsState_failsTheCaller() { + Node dispatcher = + dispatcher( + "dispatcher", + (ctx, input) -> + NodeRunner.runChild(ctx, emitter("child", "CHILD_OUT"), null, USE_AS_OUTPUT) + .ignoreElement() + .andThen( + Completable.fromAction(() -> ctx.requireNodeState().produceOutput("OWN"))) + .toFlowable()); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure).hasMessageThat().contains("delegated its output"); + } + + @Test + public void runChild_useAsOutputAfterTheCallerProducedAnOutput_failsTheCaller() { + Node dispatcher = + dispatcher( + "dispatcher", + (ctx, input) -> + Completable.fromAction(() -> ctx.requireNodeState().produceOutput("OWN")) + .andThen( + NodeRunner.runChild( + ctx, emitter("child", "CHILD_OUT"), null, USE_AS_OUTPUT)) + .ignoreElement() + .toFlowable()); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure) + .hasMessageThat() + .contains("already produced an output and cannot delegate it"); + } + + @Test + public void runChild_failingChild_recordsTheChildsPathAsWhereTheCallerFailed() { + IllegalStateException childFailure = new IllegalStateException("child boom"); + Node dispatcher = returning("dispatcher", throwing("child", childFailure), DEFAULTS); + + Context context = run(workflow(new Edge(START, dispatcher))); + + assertThat(context.requireNodeState().failure()).hasValue(childFailure); + assertThat(context.requireNodeState().failureNodePath()).hasValue("wf@1/dispatcher@1/child@1"); + } + + @Test + public void runChild_delegateWithARejectedName_letsTheCallerProduceItsOwnOutput() { + Node dispatcher = + handling( + "dispatcher", + emitter("a/b", "C"), + USE_AS_OUTPUT, + IllegalArgumentException.class, + "FALLBACK"); + + run(workflow(new Edge(START, dispatcher))); + + assertThat(outputEventAt("wf@1/dispatcher@1").output()).hasValue("FALLBACK"); + } + + @Test + public void runChild_delegateThatFailsBeforeEmittingItsOutput_letsTheCallerProduceItsOwn() { + Node child = + new TestNode( + "child", + (ctx, input) -> { + ctx.requireNodeState().produceOutput("CHILD_OUT"); + return Flowable.error(new IllegalStateException("failed before emitting")); + }); + Node dispatcher = + handling("dispatcher", child, USE_AS_OUTPUT, DynamicNodeFailedException.class, "FALLBACK"); + Node after = echo("after", "after"); + + run(workflow(new Edge(START, dispatcher), new Edge(dispatcher, after))); + + assertThat(outputEventAt("wf@1/after@1").output()).hasValue("after:FALLBACK"); + assertThat(outputs()).doesNotContain("CHILD_OUT"); + } + + @Test + public void runChild_delegateThatFailsBeforeEmitting_letsTheCallerDelegateAgain() { + Node dispatcher = + dispatcher( + "dispatcher", + (ctx, input) -> + NodeRunner.runChild( + ctx, + throwing("first", new IllegalStateException("first failed")), + null, + USE_AS_OUTPUT) + .ignoreElement() + .onErrorResumeNext( + error -> + error instanceof DynamicNodeFailedException + ? NodeRunner.runChild( + ctx, emitter("second", "SECOND"), null, USE_AS_OUTPUT) + .ignoreElement() + : Completable.error(error)) + .toFlowable()); + Node after = echo("after", "after"); + + run(workflow(new Edge(START, dispatcher), new Edge(dispatcher, after))); + + assertThat(outputEventAt("wf@1/after@1").output()).hasValue("after:SECOND"); + } + + @Test + public void runChild_delegateCanceledBeforeEmitting_keepsTheCallersOutputDelegated() { + MaybeSubject cancel = MaybeSubject.create(); + Node child = + new TestNode( + "child", + (ctx, input) -> + Completable.fromAction(() -> cancel.onSuccess(true)).andThen(Flowable.never())); + Node dispatcher = cancelingAndFallingBack("dispatcher", child, cancel); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure).hasMessageThat().contains("delegated its output"); + } + + @Test + public void runChild_delegateCanceledAfterEmittingItsOutput_failsACallerProducingItsOwn() { + MaybeSubject cancel = MaybeSubject.create(); + Node child = + new TestNode( + "child", + (ctx, input) -> + Flowable.just("CHILD_OUT") + .concatWith(Completable.fromAction(() -> cancel.onSuccess(true))) + .concatWith(Flowable.never())); + Node dispatcher = cancelingAndFallingBack("dispatcher", child, cancel); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure).hasMessageThat().contains("delegated its output"); + assertThat(eventAt("wf@1/dispatcher@1/child@1").output()).hasValue("CHILD_OUT"); + } + + @Test + public void runChild_delegateThatFailsAfterEmittingItsOutput_failsACallerProducingItsOwn() { + Node dispatcher = + handling( + "dispatcher", + emitThenFail("child", "CHILD_OUT"), + USE_AS_OUTPUT, + DynamicNodeFailedException.class, + "FALLBACK"); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure).hasMessageThat().contains("delegated its output"); + } + + @Test + public void runChild_delegateWorkflowFailingAfterItsOutput_failsACallerProducingItsOwn() { + Workflow inner = workflow("inner_wf", new Edge(START, emitThenFail("term", "CHILD_OUT"))); + Node dispatcher = + handling("dispatcher", inner, USE_AS_OUTPUT, DynamicNodeFailedException.class, "FALLBACK"); + + Exception failure = runToFailure(workflow(new Edge(START, dispatcher))); + + assertThat(failure).isInstanceOf(IllegalStateException.class); + assertThat(failure).hasMessageThat().contains("delegated its output"); + } + + @Test + public void runChild_workflowChildWhoseNodeFails_recordsThatNodesPathAsWhereTheCallerFailed() { + Workflow inner = + workflow( + "inner_wf", new Edge(START, throwing("leaf", new IllegalStateException("leaf boom")))); + Node dispatcher = returning("dispatcher", inner, DEFAULTS); + + Context context = run(workflow(new Edge(START, dispatcher))); + + assertThat(context.requireNodeState().failureNodePath()) + .hasValue("wf@1/dispatcher@1/inner_wf@1/leaf@1"); + } + + @Test + public void runChild_delegateThatFailsAfterEmittingItsOutput_keepsThatOutputAsTheCallers() { + Node dispatcher = + handling( + "dispatcher", + emitThenFail("child", "CHILD_OUT"), + USE_AS_OUTPUT, + DynamicNodeFailedException.class, + /* fallback= */ null); + + Context context = run(workflow(new Edge(START, dispatcher))); + + assertThat(context.requireNodeState().failure()).isEmpty(); + assertThat(context.requireNodeState().output()).hasValue("CHILD_OUT"); + } + + @Test + public void runChild_childWaitingOnAnInterrupt_leavesTheCallerWaiting() { + Node dispatcher = returning("dispatcher", interrupting("child", "ask"), DEFAULTS); + Node after = emitter("after", "AFTER"); + + Context context = run(workflow(new Edge(START, dispatcher), new Edge(dispatcher, after))); + + assertThat(eventAt("wf@1/dispatcher@1/child@1").longRunningToolIds()) + .hasValue(ImmutableSet.of("ask")); + assertThat(context.requireNodeState().interruptIds()).containsExactly("ask"); + assertThat(eventsAt("wf@1/after@1")).isEmpty(); + } + + @Test + public void runChild_callerCatchingTheWaitSignal_keepsRunningButStillWaits() { + Node dispatcher = + handling("dispatcher", interrupting("child", "ask"), DEFAULTS, RuntimeException.class, "X"); + Node after = emitter("after", "AFTER"); + + Context context = run(workflow(new Edge(START, dispatcher), new Edge(dispatcher, after))); + + assertThat(outputEventAt("wf@1/dispatcher@1").output()).hasValue("X"); + assertThat(context.requireNodeState().interruptIds()).containsExactly("ask"); + assertThat(eventsAt("wf@1/after@1")).isEmpty(); + } + + @Test + public void runChild_childInterruptingOnTheWorkflowsContext_keepsTheWorkflowWithoutAnOutput() { + Node helper = + new TestNode( + "helper", + (ctx, input) -> + NodeRunner.runChild( + workflowContext(ctx), interrupting("child", "ask"), null, DEFAULTS) + .toFlowable()); + Node after = emitter("after", "AFTER"); + + Context context = run(workflow(new Edge(START, helper), new Edge(helper, after))); + + assertThat(outputEventAt("wf@1/after@1").output()).hasValue("AFTER"); + assertThat(context.requireNodeState().interruptIds()).containsExactly("ask"); + assertThat(context.requireNodeState().output()).isEmpty(); + } + + @Test + public void runChild_raiseOnWaitWithAWaitForOutputChildWithoutAnOutput_endsTheCallersRun() { + Node dispatcher = continuing("dispatcher", waitingForOutput("child"), RAISE_ON_WAIT); + + Context context = run(workflow(new Edge(START, dispatcher))); + + assertThat(context.requireNodeState().failure()).isEmpty(); + assertThat(context.requireNodeState().interruptIds()).isEmpty(); + assertThat(context.requireNodeState().output()).isEmpty(); + assertThat(outputs()).doesNotContain("AFTER_WAIT"); + } + + @Test + public void runChild_raiseOnWaitWithAChildThatHasAnOutput_letsTheCallerCarryOn() { + Node dispatcher = continuing("dispatcher", emitter("child", "C"), RAISE_ON_WAIT); + + Context context = run(workflow(new Edge(START, dispatcher))); + + assertThat(context.requireNodeState().output()).hasValue("AFTER_WAIT"); + } + + @Test + public void runChild_onAContextThatIsNotAWorkflowNodes_fails() { + Context context = + new CallbackContext( + TestUtils.createInvocationContext(TestUtils.createRootAgent()), + /* eventActions= */ null); + + Maybe result = NodeRunner.runChild(context, emitter("child", "C"), null, DEFAULTS); + + IllegalStateException thrown = assertThrows(IllegalStateException.class, result::blockingGet); + + assertThat(thrown).hasMessageThat().contains("workflow node"); + } + + /** Runs {@code root} at the root of an invocation and returns its context. */ + @CanIgnoreReturnValue + private Context run(Node root) { + return NodeRunner.builder() + .node(root) + .invocationContext(TestUtils.createInvocationContext(TestUtils.createRootAgent())) + .eventSink(recordingSink) + .build() + .run(/* nodeInput= */ null) + .timeout(30, TimeUnit.SECONDS) + .blockingGet(); + } + + /** Runs {@code root} like {@link #run} and returns the exception it failed with. */ + private Exception runToFailure(Node root) { + Optional failure = run(root).requireNodeState().failure(); + assertThat(failure).isPresent(); + return failure.get(); + } + + private ImmutableList recorded() { + return ImmutableList.copyOf(events); + } + + private Event eventAt(String nodePath) { + return eventsAt(nodePath).stream().collect(onlyElement()); + } + + private ImmutableList eventsAt(String nodePath) { + return recorded().stream() + .filter(event -> pathOf(event).equals(nodePath)) + .collect(toImmutableList()); + } + + private Event outputEventAt(String nodePath) { + return outputEventsAt(nodePath).stream().collect(onlyElement()); + } + + private ImmutableList outputEventsAt(String nodePath) { + return eventsAt(nodePath).stream() + .filter(event -> event.output().isPresent()) + .collect(toImmutableList()); + } + + /** Returns the paths of the recorded events that contain {@code fragment}, in event order. */ + private ImmutableList pathsContaining(String fragment) { + return recorded().stream() + .map(DynamicDispatchTest::pathOf) + .filter(path -> path.contains(fragment)) + .collect(toImmutableList()); + } + + private ImmutableList outputs() { + return recorded().stream().flatMap(event -> event.output().stream()).collect(toImmutableList()); + } + + private static String pathOf(Event event) { + return event.nodeInfo().map(NodeInfo::path).orElse(""); + } + + private static ImmutableList paths(List events) { + return events.stream().map(DynamicDispatchTest::pathOf).collect(toImmutableList()); + } + + private static ImmutableList outputFor(Event event) { + return event.nodeInfo().map(NodeInfo::outputFor).orElse(ImmutableList.of()); + } + + /** Returns the context of the workflow that runs the node whose context is {@code context}. */ + private static Context workflowContext(Context context) { + return context.requireNodeState().parent().orElseThrow(); + } + + /** Returns a node that runs {@code child} with {@code config} and emits the child's output. */ + private static Node returning(String name, Node child, RunNodeConfig config) { + return dispatcher( + name, (ctx, input) -> NodeRunner.runChild(ctx, child, null, config).toFlowable()); + } + + /** Returns a node that delegates its output to {@code child} and emits nothing itself. */ + private static Node delegating(String name, Node child) { + return dispatcher( + name, + (ctx, input) -> + NodeRunner.runChild(ctx, child, null, USE_AS_OUTPUT).ignoreElement().toFlowable()); + } + + /** Returns a node that runs {@code child} with {@code config}, then emits {@code AFTER_WAIT}. */ + private static Node continuing(String name, Node child, RunNodeConfig config) { + return dispatcher( + name, + (ctx, input) -> + NodeRunner.runChild(ctx, child, null, config) + .ignoreElement() + .andThen(Flowable.just("AFTER_WAIT"))); + } + + /** + * Returns a node that runs {@code child} with {@code config} and emits the child's output, or + * emits {@code fallback}, if not null, when the run fails with a {@code handled} exception. + */ + private static Node handling( + String name, + Node child, + RunNodeConfig config, + Class handled, + @Nullable Object fallback) { + return dispatcher( + name, + (ctx, input) -> + NodeRunner.runChild(ctx, child, null, config) + .onErrorResumeNext( + error -> + handled.isInstance(error) + ? Maybe.fromOptional(Optional.ofNullable(fallback)) + : Maybe.error(error)) + .toFlowable()); + } + + /** + * Returns a node that delegates its output to {@code child}, stops waiting for it once {@code + * cancel} succeeds, and then emits {@code FALLBACK}. + */ + private static Node cancelingAndFallingBack(String name, Node child, MaybeSubject cancel) { + return dispatcher( + name, + (ctx, input) -> + NodeRunner.runChild(ctx, child, null, USE_AS_OUTPUT) + .takeUntil(cancel) + .defaultIfEmpty("FALLBACK") + .toFlowable()); + } +} diff --git a/core/src/test/java/com/google/adk/workflow/RunNodeConfigTest.java b/core/src/test/java/com/google/adk/workflow/RunNodeConfigTest.java new file mode 100644 index 000000000..6497adfa2 --- /dev/null +++ b/core/src/test/java/com/google/adk/workflow/RunNodeConfigTest.java @@ -0,0 +1,115 @@ +/* + * 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 static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class RunNodeConfigTest { + + @Test + public void builder_defaultsEveryProperty() { + RunNodeConfig config = RunNodeConfig.builder().build(); + + assertThat(config.runId()).isEmpty(); + assertThat(config.useAsOutput()).isFalse(); + assertThat(config.useSubBranch()).isFalse(); + assertThat(config.overrideBranch()).isEmpty(); + assertThat(config.raiseOnWait()).isFalse(); + } + + @Test + public void builder_setsEveryProperty() { + RunNodeConfig config = + RunNodeConfig.builder() + .runId("lookup") + .useAsOutput(true) + .useSubBranch(true) + .overrideBranch("tool.call@1") + .raiseOnWait(true) + .build(); + + assertThat(config.runId()).hasValue("lookup"); + assertThat(config.useAsOutput()).isTrue(); + assertThat(config.useSubBranch()).isTrue(); + assertThat(config.overrideBranch()).hasValue("tool.call@1"); + assertThat(config.raiseOnWait()).isTrue(); + } + + @Test + public void toBuilder_copiesEveryProperty() { + RunNodeConfig config = + RunNodeConfig.builder() + .runId("lookup") + .useAsOutput(true) + .useSubBranch(true) + .overrideBranch("tool.call@1") + .raiseOnWait(true) + .build(); + + assertThat(config.toBuilder().build()).isEqualTo(config); + } + + @Test + public void runId_nullClearsIt() { + RunNodeConfig config = RunNodeConfig.builder().runId("lookup").build(); + + assertThat(config.toBuilder().runId(null).build().runId()).isEmpty(); + } + + @Test + public void build_treatsAnEmptyRunIdAsNone() { + RunNodeConfig config = RunNodeConfig.builder().runId("").build(); + + assertThat(config.runId()).isEmpty(); + } + + @Test + public void build_rejectsAnAllDigitRunId() { + RunNodeConfig.Builder builder = RunNodeConfig.builder().runId("12"); + + IllegalArgumentException thrown = assertThrows(IllegalArgumentException.class, builder::build); + + assertThat(thrown).hasMessageThat().contains("runId \"12\" must contain a non-digit character"); + } + + @Test + public void build_rejectsARunIdWithASlash() { + assertThat(buildFailure("a/b")).contains("runId \"a/b\" must not contain '/', '@', or '.'"); + } + + @Test + public void build_rejectsARunIdWithAnAtSign() { + assertThat(buildFailure("a@b")).contains("runId \"a@b\" must not contain '/', '@', or '.'"); + } + + @Test + public void build_rejectsARunIdWithADot() { + assertThat(buildFailure("a.b")).contains("runId \"a.b\" must not contain '/', '@', or '.'"); + } + + /** Returns the message of the exception that building a config with {@code runId} throws. */ + private static String buildFailure(String runId) { + RunNodeConfig.Builder builder = RunNodeConfig.builder().runId(runId); + return assertThrows(IllegalArgumentException.class, builder::build).getMessage(); + } +}