feat: add the experimental Node and BaseNode workflow types

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

@@ -0,0 +1,139 @@
+/*
+ * 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.checkNotNull;
+
+import com.google.adk.agents.Context;
+import com.google.adk.agents.InvocationContext;
+import com.google.adk.annotations.Experimental;
+import com.google.adk.events.Event;
+import com.google.genai.types.Schema;
+import io.reactivex.rxjava3.core.Flowable;
+import java.util.Optional;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Base class for nodes in a workflow graph. It is meant for the framework's own nodes; to write a
+ * node, implement {@link Node}. Subclasses implement {@link #runNode}, and the framework turns each
+ * emission into an {@link Event}.
+ */
+@Experimental
+public abstract class BaseNode implements Node {
+
+  private final String name;
+  private final String description;
+  private final boolean rerunOnResume;
+  private final boolean waitForOutput;
+  private final NodeConfig config;
+  private final @Nullable Schema inputSchema;
+  private final @Nullable Schema outputSchema;
+  private final @Nullable Schema stateSchema;
+
+  /** Creates a node with every property other than its name and description at its default. */
+  protected BaseNode(String name, String description) {
+    this(
+        name,
+        description,
+        /* rerunOnResume= */ false,
+        /* waitForOutput= */ false,
+        NodeConfig.builder().build(),
+        /* inputSchema= */ null,
+        /* outputSchema= */ null,
+        /* stateSchema= */ null);
+  }
+
+  /** Creates a node; each parameter sets the {@link Node} property of the same name. */
+  protected BaseNode(
+      String name,
+      String description,
+      boolean rerunOnResume,
+      boolean waitForOutput,
+      NodeConfig config,
+      @Nullable Schema inputSchema,
+      @Nullable Schema outputSchema,
+      @Nullable Schema stateSchema) {
+    this.name = checkNotNull(name);
+    this.description = checkNotNull(description);
+    this.rerunOnResume = rerunOnResume;
+    this.waitForOutput = waitForOutput;
+    this.config = checkNotNull(config);
+    this.inputSchema = inputSchema;
+    this.outputSchema = outputSchema;
+    this.stateSchema = stateSchema;
+  }
+
+  @Override
+  public final String name() {
+    return name;
+  }
+
+  @Override
+  public String description() {
+    return description;
+  }
+
+  @Override
+  public boolean rerunOnResume() {
+    return rerunOnResume;
+  }
+
+  @Override
+  public boolean waitForOutput() {
+    return waitForOutput;
+  }
+
+  @Override
+  public NodeConfig config() {
+    return config;
+  }
+
+  @Override
+  public Optional<Schema> inputSchema() {
+    return Optional.ofNullable(inputSchema);
+  }
+
+  @Override
+  public Optional<Schema> outputSchema() {
+    return Optional.ofNullable(outputSchema);
+  }
+
+  @Override
+  public Optional<Schema> stateSchema() {
+    return Optional.ofNullable(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.
+   */
+  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());
+  }
+}

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

@@ -0,0 +1,107 @@
+/*
+ * 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.agents.Context;
+import com.google.adk.annotations.Experimental;
+import com.google.adk.events.Event;
+import com.google.genai.types.Schema;
+import io.reactivex.rxjava3.core.Flowable;
+import java.util.Optional;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * A unit of work in a workflow graph. Node names must be unique within a graph. Every member except
+ * {@link #name} and {@link #description} is experimental, so implementing this interface is too.
+ */
+public interface Node {
+
+  /** Returns the name that identifies the node within its graph. */
+  String name();
+
+  /** Returns what the node does, for humans and for a model that may call it. */
+  default String description() {
+    return "";
+  }
+
+  /**
+   * Returns {@code true} if a resumed node runs again from scratch, rather than completing with the
+   * resuming answer as its output.
+   */
+  @Experimental
+  default boolean rerunOnResume() {
+    return false;
+  }
+
+  /**
+   * Returns {@code true} if the node stays re-triggerable until it produces an output or a route,
+   * instead of completing when its run completes.
+   */
+  @Experimental
+  default boolean waitForOutput() {
+    return false;
+  }
+
+  /** Returns the node's retry policy and execution timeout. */
+  @Experimental
+  default NodeConfig config() {
+    return NodeConfig.builder().build();
+  }
+
+  /** Returns the schema the node's input must match. */
+  @Experimental
+  default Optional<Schema> inputSchema() {
+    return Optional.empty();
+  }
+
+  /**
+   * Returns the schema that the node's output value must match. The content of a message-as-output
+   * event is not checked against this schema.
+   */
+  @Experimental
+  default Optional<Schema> outputSchema() {
+    return Optional.empty();
+  }
+
+  /**
+   * Returns the schema declaring the state keys the node uses. Child nodes inherit it unless they
+   * declare their own.
+   */
+  @Experimental
+  default Optional<Schema> stateSchema() {
+    return Optional.empty();
+  }
+
+  /**
+   * Returns {@code true} if the node runs only once every predecessor has completed, receiving all
+   * their outputs keyed by node name, as a fan-in node does.
+   */
+  @Experimental
+  default boolean requiresAllPredecessors() {
+    return false;
+  }
+
+  /**
+   * 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.
+   */
+  @Experimental
+  Flowable<?> runNode(Context context, @Nullable Object nodeInput);
+}

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

@@ -0,0 +1,70 @@
+/*
+ * 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.time.Duration;
+import java.util.Optional;
+import org.jspecify.annotations.Nullable;
+
+/** A node's retry policy and execution timeout. */
+@Experimental
+@AutoValue
+public abstract class NodeConfig {
+
+  /** Returns how the node is retried when it fails; empty does not retry. */
+  public abstract Optional<RetryConfig> retryConfig();
+
+  /**
+   * Returns how long an attempt may run before it is cancelled and treated as a failure; empty
+   * imposes no limit.
+   */
+  public abstract Optional<Duration> timeout();
+
+  public static Builder builder() {
+    return new AutoValue_NodeConfig.Builder();
+  }
+
+  public abstract Builder toBuilder();
+
+  /** Builder for {@link NodeConfig}. */
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+    @CanIgnoreReturnValue
+    public abstract Builder retryConfig(@Nullable RetryConfig retryConfig);
+
+    /** Sets the timeout, which must be positive; null imposes no limit. */
+    @CanIgnoreReturnValue
+    public abstract Builder timeout(@Nullable Duration timeout);
+
+    abstract NodeConfig autoBuild();
+
+    public final NodeConfig build() {
+      NodeConfig config = autoBuild();
+      // A zero or negative timeout would fire at once.
+      checkArgument(
+          config.timeout().map(timeout -> timeout.compareTo(Duration.ZERO) > 0).orElse(true),
+          "timeout must be positive, or null for no timeout.");
+      return config;
+    }
+  }
+}