diff --git a/core/src/main/java/com/google/adk/agents/InvocationContext.java b/core/src/main/java/com/google/adk/agents/InvocationContext.java index 3a761dedc..4ed184330 100644 --- a/core/src/main/java/com/google/adk/agents/InvocationContext.java +++ b/core/src/main/java/com/google/adk/agents/InvocationContext.java @@ -314,9 +314,10 @@ public String userId() { /** * Returns the {@link Scheduler} on which this invocation runs the work ADK hands to worker * threads: the sub-agents of a {@code ParallelAgent}, tool calls in {@code - * ToolExecutionMode.PARALLEL_SUBSCRIBE} mode, and the hop that starts the live send loop. An - * agent-level setting ({@code ParallelAgent.Builder#scheduler}, {@code - * LlmAgent.Builder#executor}) takes precedence over it. Defaults to {@link Schedulers#io()}. + * ToolExecutionMode.PARALLEL_SUBSCRIBE} mode, the plain bodies of workflow {@code FunctionNode}s, + * and the hop that starts the live send loop. An agent-level setting ({@code + * ParallelAgent.Builder#scheduler}, {@code LlmAgent.Builder#executor}) takes precedence over it. + * Defaults to {@link Schedulers#io()}. */ public Scheduler scheduler() { return scheduler != null ? scheduler : Schedulers.io(); diff --git a/core/src/main/java/com/google/adk/runner/Runner.java b/core/src/main/java/com/google/adk/runner/Runner.java index d80f60d4b..be8cde376 100644 --- a/core/src/main/java/com/google/adk/runner/Runner.java +++ b/core/src/main/java/com/google/adk/runner/Runner.java @@ -184,10 +184,11 @@ public Builder plugins(Plugin... plugins) { /** * Sets the {@link Scheduler} for the work ADK hands off to worker threads in this runner's * invocations: the sub-agents of a {@code ParallelAgent}, tool calls in {@link - * RunConfig.ToolExecutionMode#PARALLEL_SUBSCRIBE} mode, and the hop that starts the live send - * loop. The runs that an {@code AgentTool} nests inherit it. An agent-level {@code - * ParallelAgent.Builder#scheduler} or {@code LlmAgent.Builder#executor} takes precedence. - * {@code null} (the default) selects {@code Schedulers.io()}. + * RunConfig.ToolExecutionMode#PARALLEL_SUBSCRIBE} mode, the plain bodies of workflow {@code + * FunctionNode}s, and the hop that starts the live send loop. The runs that an {@code + * AgentTool} nests inherit it. An agent-level {@code ParallelAgent.Builder#scheduler} or {@code + * LlmAgent.Builder#executor} takes precedence. {@code null} (the default) selects {@code + * Schedulers.io()}. * *

Only these hand-offs are affected; work that completes asynchronously elsewhere (model * clients, session services, MCP, plugins) stays on its own threads. {@code diff --git a/core/src/main/java/com/google/adk/workflow/FunctionNode.java b/core/src/main/java/com/google/adk/workflow/FunctionNode.java new file mode 100644 index 000000000..0f97bf7ff --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/FunctionNode.java @@ -0,0 +1,196 @@ +/* + * 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 static com.google.common.base.Preconditions.checkState; + +import com.google.adk.agents.Context; +import com.google.adk.annotations.Experimental; +import com.google.adk.events.Event; +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import com.google.genai.types.Schema; +import io.reactivex.rxjava3.core.Flowable; +import java.util.Optional; +import org.jspecify.annotations.Nullable; + +/** + * A workflow node that runs a Java function on each input, with a type per port. Jackson reads each + * input into the input type and writes each output in JSON form, so records and POJOs work as + * ports, and a successor reads the same value whether it comes live or from a stored session. Build + * one with {@link FunctionNodes#of} or {@link FunctionNodes#builder}. + */ +@Experimental +public final class FunctionNode extends BaseNode { + + private static final PortCodec ANY = + PortCodec.output(PortCodec.javaType(Object.class), "an event output"); + + private final PortCodec inputCodec; + private final PortCodec outputCodec; + private final NodeBody body; + + private FunctionNode(Builder builder) { + super( + checkNotNull(builder.name), + builder.description, + builder.rerunOnResume, + builder.waitForOutput, + builder.config, + builder.inputSchema, + builder.outputSchema, + builder.stateSchema); + this.inputCodec = builder.inputCodec; + this.outputCodec = builder.outputCodec; + this.body = checkNotNull(builder.body); + } + + /** Returns a builder initialized with this node's properties. */ + public Builder toBuilder() { + return new Builder<>(inputCodec, outputCodec) + .name(name()) + .description(description()) + .rerunOnResume(rerunOnResume()) + .waitForOutput(waitForOutput()) + .config(config()) + .inputSchema(inputSchema().orElse(null)) + .outputSchema(outputSchema().orElse(null)) + .stateSchema(stateSchema().orElse(null)) + .body(body); + } + + @Override + public Flowable runNode(Context context, @Nullable Object nodeInput) { + return Flowable.defer( + () -> { + I input = inputCodec.decode(nodeInput, "input of node '" + name() + "'"); + return Flowable.fromOptional(Optional.ofNullable(output(body.apply(context, input)))); + }) + .subscribeOn(context.invocationContext().scheduler()); + } + + /** Returns the event or JSON value to send for {@code result}, or {@code null} for none. */ + private @Nullable Object output(@Nullable O result) { + String what = "output of node '" + name() + "'"; + if (result instanceof Event event) { + // A returned event's output is sent in JSON form too, as a stored session hands it back. + return event.output().isEmpty() + ? event + : event.toBuilder().output(ANY.encode(event.output().get(), what)).build(); + } + return outputCodec.encode(result, what); + } + + /** + * Fluent builder for {@link FunctionNode}. Any property left unset falls back to the default of + * the matching {@link Node} property; name and body are required. + */ + public static final class Builder { + private final PortCodec inputCodec; + private final PortCodec outputCodec; + private @Nullable String name; + private String description = ""; + private boolean rerunOnResume; + private boolean waitForOutput; + private NodeConfig config = NodeConfig.builder().build(); + private @Nullable Schema inputSchema; + private @Nullable Schema outputSchema; + private @Nullable Schema stateSchema; + private @Nullable NodeBody body; + + Builder(PortCodec inputCodec, PortCodec outputCodec) { + this.inputCodec = inputCodec; + this.outputCodec = outputCodec; + } + + /** Sets the node's name, unique within its graph. Required. */ + @CanIgnoreReturnValue + public Builder name(String name) { + this.name = checkNotNull(name); + return this; + } + + /** Sets what the node does, for humans and for a model that may call it. */ + @CanIgnoreReturnValue + public Builder description(String description) { + this.description = checkNotNull(description); + return this; + } + + /** Sets whether a resume runs the node again rather than using the resuming answer. */ + @CanIgnoreReturnValue + public Builder rerunOnResume(boolean rerunOnResume) { + this.rerunOnResume = rerunOnResume; + return this; + } + + /** Sets whether the node stays re-triggerable until it produces an output or a route. */ + @CanIgnoreReturnValue + public Builder waitForOutput(boolean waitForOutput) { + this.waitForOutput = waitForOutput; + return this; + } + + /** Sets the node's retry policy and execution timeout. */ + @CanIgnoreReturnValue + public Builder config(NodeConfig config) { + this.config = checkNotNull(config); + return this; + } + + /** Sets the schema the node's input must match. */ + @CanIgnoreReturnValue + public Builder inputSchema(@Nullable Schema inputSchema) { + this.inputSchema = inputSchema; + return this; + } + + /** Sets the schema the node's output must match. */ + @CanIgnoreReturnValue + public Builder outputSchema(@Nullable Schema outputSchema) { + this.outputSchema = outputSchema; + return this; + } + + /** + * Sets the state keys the node uses, which child nodes inherit unless they declare their own. + */ + @CanIgnoreReturnValue + public Builder stateSchema(@Nullable Schema stateSchema) { + this.stateSchema = stateSchema; + return this; + } + + /** Sets the function the node runs on each input. Required. */ + @CanIgnoreReturnValue + public Builder body(NodeBody body) { + this.body = checkNotNull(body); + return this; + } + + /** + * Builds the node. + * + * @throws IllegalStateException if the name or the body is not set + */ + public FunctionNode build() { + checkState(name != null, "FunctionNode.Builder requires name to be set."); + checkState(body != null, "FunctionNode.Builder for '%s' requires body to be set.", name); + return new FunctionNode<>(this); + } + } +} diff --git a/core/src/main/java/com/google/adk/workflow/FunctionNodes.java b/core/src/main/java/com/google/adk/workflow/FunctionNodes.java new file mode 100644 index 000000000..0a893e68f --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/FunctionNodes.java @@ -0,0 +1,108 @@ +/* + * 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.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JavaType; +import com.google.adk.annotations.Experimental; + +/** + * Creates {@link FunctionNode}s with a type per port, as a {@code Class} or, for a generic type + * such as {@code List}, a {@code TypeReference}. An {@code Object} input port takes its + * input as it is, a {@code Content} or {@code Event} port reads its own type, a {@code Void} port + * ignores its input, and Jackson reads any other type; every output is written in JSON form. + */ +@Experimental +public final class FunctionNodes { + + private static final String BUILDER = "a FunctionNodes.builder node"; + + /** + * Returns a node named {@code name} that runs {@code body} on each input read as {@code + * inputType} and writes its {@code outputType} output in JSON form. + * + * @throws WorkflowConfigurationException if Jackson cannot read and write {@code inputType} or + * write {@code outputType}, or if a port type has no single concrete JSON shape; a field type + * Jackson cannot handle fails on the first value instead + */ + public static FunctionNode of( + String name, Class inputType, Class outputType, NodeBody body) { + return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body); + } + + /** Like {@link #of(String, Class, Class, NodeBody)}, for generic port types. */ + public static FunctionNode of( + String name, TypeReference inputType, TypeReference outputType, NodeBody body) { + return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body); + } + + /** Like {@link #of(String, Class, Class, NodeBody)}, for a generic output type. */ + public static FunctionNode of( + String name, Class inputType, TypeReference outputType, NodeBody body) { + return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body); + } + + /** Like {@link #of(String, Class, Class, NodeBody)}, for a generic input type. */ + public static FunctionNode of( + String name, TypeReference inputType, Class outputType, NodeBody body) { + return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body); + } + + /** + * Returns a builder for a node with these port types, for properties {@link #of} does not take. + * + * @throws WorkflowConfigurationException as {@link #of} does + */ + public static FunctionNode.Builder builder(Class inputType, Class outputType) { + return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER); + } + + /** Like {@link #builder(Class, Class)}, for generic port types. */ + public static FunctionNode.Builder builder( + TypeReference inputType, TypeReference outputType) { + return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER); + } + + /** Like {@link #builder(Class, Class)}, for a generic output type. */ + public static FunctionNode.Builder builder( + Class inputType, TypeReference outputType) { + return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER); + } + + /** Like {@link #builder(Class, Class)}, for a generic input type. */ + public static FunctionNode.Builder builder( + TypeReference inputType, Class outputType) { + return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER); + } + + private static FunctionNode node( + String name, JavaType inputType, JavaType outputType, NodeBody body) { + return FunctionNodes.builder(inputType, outputType, "node '" + name + "'") + .name(name) + .body(body) + .build(); + } + + private static FunctionNode.Builder builder( + JavaType inputType, JavaType outputType, String node) { + return new FunctionNode.Builder<>( + PortCodec.input(inputType, "the input of " + node), + PortCodec.output(outputType, "the output of " + node)); + } + + private FunctionNodes() {} +} diff --git a/core/src/main/java/com/google/adk/workflow/NodeBody.java b/core/src/main/java/com/google/adk/workflow/NodeBody.java new file mode 100644 index 000000000..950b25fab --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/NodeBody.java @@ -0,0 +1,38 @@ +/* + * 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; + +/** + * The body of a {@link FunctionNode}, which takes the node run's context and its single input; the + * input may be shared with other nodes, so the body must not mutate it. It runs on the {@code + * Runner}'s scheduler, so it may block, and a small fixed-size scheduler can deadlock bodies that + * wait for one another. + */ +@Experimental +@FunctionalInterface +public interface NodeBody { + + /** + * Returns {@code null} for no output, an {@code Event} to send with any output in JSON form, or + * the output, which is sent in its JSON form; a returned {@code Content} is the output, not a + * message. + */ + O apply(Context context, I input) throws Exception; +} diff --git a/core/src/main/java/com/google/adk/workflow/NodeInputValidationException.java b/core/src/main/java/com/google/adk/workflow/NodeInputValidationException.java new file mode 100644 index 000000000..44feeb009 --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/NodeInputValidationException.java @@ -0,0 +1,31 @@ +/* + * 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; + +/** + * Thrown when a node's input does not fit its input port. The message names the port and the + * expected type, never the value. + */ +@Experimental +public final class NodeInputValidationException extends IllegalArgumentException { + + public NodeInputValidationException(String message) { + super(message); + } +} diff --git a/core/src/main/java/com/google/adk/workflow/PortCodec.java b/core/src/main/java/com/google/adk/workflow/PortCodec.java new file mode 100644 index 000000000..c67368e2b --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/PortCodec.java @@ -0,0 +1,441 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.JsonDeserializer; +import com.fasterxml.jackson.databind.JsonMappingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.JsonSerializer; +import com.fasterxml.jackson.databind.KeyDeserializer; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.databind.deser.AbstractDeserializer; +import com.fasterxml.jackson.databind.deser.BeanDeserializerBase; +import com.fasterxml.jackson.databind.deser.DefaultDeserializationContext; +import com.fasterxml.jackson.databind.deser.ValueInstantiator; +import com.fasterxml.jackson.databind.deser.impl.UnsupportedTypeDeserializer; +import com.fasterxml.jackson.databind.json.JsonMapper; +import com.fasterxml.jackson.databind.jsontype.TypeSerializer; +import com.fasterxml.jackson.databind.ser.impl.UnknownSerializer; +import com.fasterxml.jackson.databind.ser.impl.UnsupportedTypeSerializer; +import com.fasterxml.jackson.databind.ser.std.BeanSerializerBase; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import com.google.adk.events.Event; +import com.google.genai.JsonSerializable; +import com.google.genai.types.Content; +import java.lang.invoke.MethodType; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.jspecify.annotations.Nullable; + +/** + * Reads a {@link FunctionNode} port's input into its type and writes its output in JSON form. An + * {@code Object} port takes its input as it is, a {@code Content} or {@code Event} port reads its + * own type, a {@code Void} port ignores its input, and any other port reads through Jackson. + */ +final class PortCodec { + + /** + * One mapper for every port: unknown properties are ignored, null fields are written, and {@code + * java.time} values travel as ISO-8601 strings. + */ + private static final ObjectMapper MAPPER = + JsonMapper.builder() + .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) + .addModule(new Jdk8Module()) + .addModule(new JavaTimeModule()) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS) + .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS) + .build(); + + private final JavaType type; + + private PortCodec(JavaType type) { + this.type = type; + } + + static JavaType javaType(Class type) { + return MAPPER.getTypeFactory().constructType(MethodType.methodType(type).wrap().returnType()); + } + + static JavaType javaType(TypeReference type) { + return MAPPER.getTypeFactory().constructType(type); + } + + /** + * Returns the codec of an input port, which Jackson must both read and write. + * + * @throws WorkflowConfigurationException if Jackson cannot read and write {@code type}, or if it + * has no single concrete JSON shape + */ + static PortCodec input(JavaType type, String what) { + if (!passesAsIs(type)) { + requireReadable(type, what); + requireWritable(type, what); + } + return new PortCodec<>(type); + } + + /** + * Returns the codec of an output port, which Jackson must write. + * + * @throws WorkflowConfigurationException if Jackson cannot write {@code type}, or if it has no + * single concrete JSON shape + */ + static PortCodec output(JavaType type, String what) { + if (!passesAsIs(type)) { + requireWritable(type, what); + } + return new PortCodec<>(type); + } + + /** Whether a port of {@code type} handles values without a Jackson check. */ + private static boolean passesAsIs(JavaType type) { + Class raw = type.getRawClass(); + return raw == Object.class || raw == Void.class || raw == Content.class || raw == Event.class; + } + + /** A port has no "absent" value: a missing input arrives as null, and null sends no output. */ + private static WorkflowConfigurationException optionalPort(JavaType type, String what) { + return new WorkflowConfigurationException( + capitalized(what) + + " has type " + + type.toCanonical() + + ". Declare it without Optional: a missing input arrives as null, and a null output" + + " sends none."); + } + + private static void requireReadable(JavaType type, String what) { + if (type.getRawClass() == Object.class) { + return; + } + if (type.isReferenceType()) { + throw optionalPort(type, what); + } + if (type.isContainerType()) { + if (type.getKeyType() != null) { + requireKeyReadable(type.getKeyType(), "a key of " + what); + } + requireReadable(type.getContentType(), "an element of " + what); + } else { + requireConcrete(type, what); + } + JsonDeserializer deserializer = null; + String reason = ""; + try { + deserializer = deserializationContext().findRootValueDeserializer(type); + } catch (JsonMappingException | RuntimeException e) { + reason = reason(e); + } + boolean readable = + deserializer != null + && !(deserializer instanceof AbstractDeserializer) + && !(deserializer instanceof UnsupportedTypeDeserializer) + && !(deserializer instanceof ValueInstantiator.Gettable gettable + && gettable.getValueInstantiator() != null + && !gettable.getValueInstantiator().canInstantiate()); + if (!readable) { + throw new WorkflowConfigurationException( + capitalized(what) + + " has type " + + type.toCanonical() + + ", which Jackson cannot create from JSON" + + reason + + ". Use a record, a class with a constructor Jackson can call, or Object.class."); + } + // A @JsonValue type writes a single value, so only an object-writing type needs this. + if (deserializer instanceof BeanDeserializerBase bean + && bean.getValueInstantiator() != null + && writesObject(type) + && !createsFromObject(bean.getValueInstantiator())) { + throw new WorkflowConfigurationException( + capitalized(what) + + " has type " + + type.toCanonical() + + ", which Jackson can create only from a string or number, not from the JSON object" + + " it writes. Use a record, a class with a no-argument constructor, or a" + + " @JsonCreator constructor with @JsonProperty parameters."); + } + } + + private static void requireWritable(JavaType type, String what) { + if (type.getRawClass() == Object.class) { + return; + } + if (type.isReferenceType()) { + throw optionalPort(type, what); + } + if (type.isContainerType()) { + requireWritable(type.getContentType(), "an element of " + what); + return; + } + requireConcrete(type, what); + JsonSerializer serializer = null; + String reason = ""; + try { + serializer = MAPPER.getSerializerProviderInstance().findValueSerializer(type); + } catch (JsonMappingException | RuntimeException e) { + reason = reason(e); + } + // An UnknownSerializer stands for a class with no properties, which fails when written. + if (serializer == null + || serializer instanceof UnsupportedTypeSerializer + || serializer instanceof UnknownSerializer) { + throw new WorkflowConfigurationException( + capitalized(what) + + " has type " + + type.toCanonical() + + ", which Jackson cannot write as JSON" + + reason + + ". Use a record, a class with getters, or Object.class."); + } + } + + /** + * Throws unless {@code type} has one concrete JSON shape: a polymorphic type is rejected, and so + * is an interface or abstract class without a creator Jackson can call. + */ + private static void requireConcrete(JavaType type, String what) { + String problem; + if (findTypeSerializer(type) != null) { + problem = "is polymorphic (@JsonTypeInfo)"; + } else if (type.isAbstract() + && !type.isEnumType() + && !type.isPrimitive() + && !hasBeanCreator(type)) { + problem = "is an interface or abstract class without a Jackson creator"; + } else { + return; + } + String fix; + if (Number.class.isAssignableFrom(type.getRawClass())) { + fix = "Declare the port as Long, Double or BigDecimal, or use Object.class."; + } else if (CharSequence.class.isAssignableFrom(type.getRawClass())) { + fix = "Declare the port as String."; + } else { + fix = + "Declare the port as a concrete record, wrap the hierarchy in one, or use Object.class."; + } + throw new WorkflowConfigurationException( + capitalized(what) + " has type " + type.toCanonical() + ", which " + problem + ". " + fix); + } + + private static void requireKeyReadable(JavaType type, String what) { + KeyDeserializer deserializer = null; + String reason = ""; + try { + deserializer = deserializationContext().findKeyDeserializer(type, null); + } catch (JsonMappingException | RuntimeException e) { + reason = reason(e); + } + if (deserializer == null) { + throw new WorkflowConfigurationException( + capitalized(what) + + " has type " + + type.toCanonical() + + ", which Jackson cannot read as a map key" + + reason + + ". Use a String, a number, an enum, or Object.class."); + } + } + + private static @Nullable TypeSerializer findTypeSerializer(JavaType type) { + try { + return MAPPER.getSerializerProviderInstance().findTypeSerializer(type); + } catch (JsonMappingException | RuntimeException e) { + return null; + } + } + + private static boolean writesObject(JavaType type) { + try { + return MAPPER.getSerializerProviderInstance().findValueSerializer(type) + instanceof BeanSerializerBase; + } catch (JsonMappingException | RuntimeException e) { + return false; + } + } + + /** Whether {@code instantiator} builds a bean from a JSON object, the form a port writes. */ + private static boolean createsFromObject(ValueInstantiator instantiator) { + return instantiator.canCreateUsingDefault() + || instantiator.canCreateFromObjectWith() + || instantiator.canCreateUsingDelegate(); + } + + /** + * Whether Jackson creates {@code type} as a bean through a creator, such as an AutoValue factory. + */ + private static boolean hasBeanCreator(JavaType type) { + try { + return deserializationContext().findRootValueDeserializer(type) + instanceof BeanDeserializerBase bean + && bean.getValueInstantiator() != null + && bean.getValueInstantiator().canInstantiate(); + } catch (JsonMappingException | RuntimeException e) { + return false; + } + } + + private static DefaultDeserializationContext deserializationContext() { + return ((DefaultDeserializationContext) MAPPER.getDeserializationContext()) + .createDummyInstance(MAPPER.getDeserializationConfig()); + } + + /** The first line of Jackson's message about a type, which names the type and never a value. */ + private static String reason(Exception e) { + String message = + e instanceof JsonMappingException mapping ? mapping.getOriginalMessage() : e.getMessage(); + return message == null ? "" : " (" + message.lines().findFirst().orElse("") + ")"; + } + + private static String capitalized(String what) { + return Character.toUpperCase(what.charAt(0)) + what.substring(1); + } + + /** Returns {@code value} read as this port's type; {@code what} names the port in errors. */ + @SuppressWarnings("unchecked") // The checks and Jackson guarantee the type. + @Nullable T decode(@Nullable Object value, String what) { + Class raw = type.getRawClass(); + if (raw == Void.class || value == null) { + return null; + } + if (raw == Object.class || hasType(value, type)) { + return (T) value; + } + if (value instanceof Content || value instanceof Event) { + throw mismatch(what, value.getClass().getSimpleName()); + } + try { + if (raw == Content.class) { + return (T) Content.fromJson(MAPPER.writeValueAsString(value)); + } + if (raw == Event.class) { + return (T) Event.fromJson(MAPPER.writeValueAsString(value)); + } + return MAPPER.convertValue(value, type); + } catch (RuntimeException | JsonProcessingException e) { + // Jackson's and genai's messages can quote the value, so no cause. + throw new NodeInputValidationException( + "validation error: " + what + " does not match " + type.toCanonical() + "."); + } + } + + /** + * Returns {@code value} in JSON form: maps, lists, {@code Long}, {@code Double}, strings, + * booleans and nulls. + * + * @throws IllegalArgumentException if {@code value} is not of this port's type or has no JSON + * form + */ + @Nullable Object encode(@Nullable Object value, String what) { + if (value == null) { + return null; + } + if (!type.getRawClass().isInstance(value)) { + throw new IllegalArgumentException( + what + + " expected " + + type.toCanonical() + + ", but got " + + value.getClass().getSimpleName() + + "."); + } + try { + // genai types serialize through their own mapper, as a session stores them. + JsonNode tree = + value instanceof JsonSerializable genai + ? MAPPER.readTree(genai.toJson()) + : MAPPER.valueToTree(value); + return jsonNative(tree); + } catch (IllegalArgumentException | JsonProcessingException e) { + // Jackson's message can quote the value, so no cause. + throw new IllegalArgumentException(what + " has no JSON form as " + type.toCanonical() + "."); + } + } + + private NodeInputValidationException mismatch(String what, String got) { + return new NodeInputValidationException( + "validation error: " + what + " expected " + type.toCanonical() + ", but got " + got + "."); + } + + /** Whether {@code value} already has {@code type}, down to each element and map entry. */ + private static boolean hasType(Object value, JavaType type) { + if (!type.getRawClass().isInstance(value)) { + return false; + } + if (type.isCollectionLikeType()) { + return ((Collection) value) + .stream().allMatch(item -> item == null || hasType(item, type.getContentType())); + } + if (type.isMapLikeType()) { + return ((Map) value) + .entrySet().stream() + .allMatch( + entry -> + (entry.getKey() == null || hasType(entry.getKey(), type.getKeyType())) + && (entry.getValue() == null + || hasType(entry.getValue(), type.getContentType()))); + } + if (type.isArrayType()) { + return type.getContentType().isPrimitive() + || Arrays.stream((Object[]) value) + .allMatch(item -> item == null || hasType(item, type.getContentType())); + } + return !type.isContainerType(); + } + + private static @Nullable Object jsonNative(JsonNode node) { + if (node.isObject()) { + Map map = new LinkedHashMap<>(); + node.fields() + .forEachRemaining(field -> map.put(field.getKey(), jsonNative(field.getValue()))); + return map; + } + if (node.isArray()) { + List list = new ArrayList<>(node.size()); + node.forEach(item -> list.add(jsonNative(item))); + return list; + } + if (node.isIntegralNumber() && node.canConvertToLong()) { + return node.longValue(); + } + if (node.isFloat()) { + // Widening a float adds digits it never had. + return Double.valueOf(Float.toString(node.floatValue())); + } + if (node.isNumber()) { + return node.doubleValue(); + } + if (node.isBoolean()) { + return node.booleanValue(); + } + if (node.isNull() || node.isMissingNode()) { + return null; + } + return node.asText(); + } +} diff --git a/core/src/test/java/com/google/adk/workflow/FunctionNodeTest.java b/core/src/test/java/com/google/adk/workflow/FunctionNodeTest.java new file mode 100644 index 000000000..439bf7473 --- /dev/null +++ b/core/src/test/java/com/google/adk/workflow/FunctionNodeTest.java @@ -0,0 +1,1081 @@ +/* + * 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.collect.MoreCollectors.onlyElement; +import static com.google.common.truth.Truth.assertThat; +import static com.google.common.truth.Truth.assertWithMessage; +import static org.junit.Assert.assertThrows; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import com.fasterxml.jackson.annotation.JsonValue; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.google.adk.agents.Context; +import com.google.adk.apps.App; +import com.google.adk.events.Event; +import com.google.adk.events.EventActions; +import com.google.adk.runner.Runner; +import com.google.adk.sessions.Session; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.genai.types.Content; +import com.google.genai.types.FinishReason; +import com.google.genai.types.Part; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.core.Scheduler; +import io.reactivex.rxjava3.schedulers.Schedulers; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.jspecify.annotations.Nullable; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests {@link FunctionNode} through a {@link Runner}, with the workflow as the app's root. */ +@RunWith(JUnit4.class) +public final class FunctionNodeTest { + + private static final String APP = "app"; + private static final String USER = "user"; + + record Order(String id, int quantity) {} + + record Section(@JsonProperty("section_name") String sectionName, String content) {} + + @JsonTypeInfo(use = JsonTypeInfo.Id.NAME) + @JsonSubTypes(@JsonSubTypes.Type(value = Circle.class, name = "circle")) + interface Shape {} + + record Circle(double radius) implements Shape {} + + /** An interface port type without a creator, which has no single JSON shape. */ + interface Decision {} + + /** An Immutables-style value type: abstract, read through the concrete class it names. */ + @JsonDeserialize(as = SimpleRange.class) + abstract static class Range { + @JsonProperty("low") + abstract int low(); + + @JsonProperty("high") + abstract int high(); + } + + static final class SimpleRange extends Range { + private final int low; + private final int high; + + @JsonCreator + SimpleRange(@JsonProperty("low") int low, @JsonProperty("high") int high) { + this.low = low; + this.high = high; + } + + @Override + int low() { + return low; + } + + @Override + int high() { + return high; + } + } + + /** An AutoValue-style value type: abstract, with a static creator. */ + abstract static class Point { + @JsonProperty("x") + abstract int x(); + + @JsonCreator + static Point create(@JsonProperty("x") int x) { + return new AutoPoint(x); + } + } + + static final class AutoPoint extends Point { + private final int x; + + AutoPoint(int x) { + this.x = x; + } + + @Override + int x() { + return x; + } + } + + /** A value type that Jackson writes as a single string through {@code @JsonValue}. */ + static final class OrderId { + private final String value; + + @JsonCreator(mode = JsonCreator.Mode.DELEGATING) + OrderId(String value) { + this.value = value; + } + + @JsonValue + String value() { + return value; + } + } + + /** A class Jackson writes as an object and reads through a delegating map creator. */ + static final class Bag { + private final ImmutableMap values; + + @JsonCreator(mode = JsonCreator.Mode.DELEGATING) + Bag(Map values) { + this.values = ImmutableMap.copyOf(values); + } + + public Object getName() { + return values.get("name"); + } + } + + /** A class Jackson reads but cannot write, since it has no getters. */ + static final class Token { + @SuppressWarnings("unused") + private final String secret; + + @JsonCreator + Token(@JsonProperty("secret") String secret) { + this.secret = secret; + } + } + + /** A class Jackson cannot create: its only constructor takes unannotated parameters. */ + static final class NoCreator { + public final String label; + + public NoCreator(String label, int unused) { + this.label = label; + } + } + + /** A class Jackson cannot write, since it has no properties. */ + static final class NoProperties {} + + /** A class Jackson writes as an object but can create only from a string. */ + static final class StringOnly { + private final String value; + + StringOnly(String value) { + this.value = value; + } + + public String getValue() { + return value; + } + } + + @Test + public void of_recordPortFedAMap_readsTheRecord() { + // Arrange: a map output feeds a record port. + FunctionNode makeOrder = + FunctionNodes.of( + "make_order", + Content.class, + Object.class, + (ctx, message) -> ImmutableMap.of("id", "a1", "quantity", 3)); + FunctionNode describe = + FunctionNodes.of( + "describe", + Order.class, + String.class, + (ctx, order) -> order.id() + " x" + order.quantity()); + + // Act + RunResult result = run(chain(makeOrder, describe)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/describe@1")).isEqualTo("a1 x3"); + assertThat(result.outputAt("wf@1/make_order@1")) + .isEqualTo(ImmutableMap.of("id", "a1", "quantity", 3L)); + } + + @Test + public void of_recordPort_readsTheSameRecordFromTheStoredJsonForm() { + // Arrange: a session read back from JSON hands a successor its own number types. + FunctionNode makeOrder = + FunctionNodes.of( + "make_order", Content.class, Order.class, (ctx, message) -> new Order("a1", 3)); + FunctionNode describe = + FunctionNodes.of( + "describe", + Order.class, + String.class, + (ctx, order) -> order.id() + " x" + order.quantity()); + RunResult live = run(chain(makeOrder, describe)); + Event liveOutput = live.outputEventAt("wf@1/make_order@1"); + Object stored = Event.fromJson(liveOutput.toJson()).output().get(); + Node replay = rawOutput("make_order", stored); + + // Act + RunResult fromStored = run(chain(replay, describe)); + + // Assert + fromStored.assertCompleted(); + assertThat(fromStored.outputAt("wf@1/describe@1")).isEqualTo(live.outputAt("wf@1/describe@1")); + } + + @Test + public void of_primitivePort_readsTheBoxedValue() { + // Arrange + FunctionNode three = + FunctionNodes.of("three", Content.class, Integer.class, (ctx, message) -> 3); + FunctionNode increment = + FunctionNodes.of("increment", int.class, Integer.class, (ctx, n) -> n + 1); + + // Act + RunResult result = run(chain(three, increment)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/increment@1")).isEqualTo(4L); + } + + @Test + public void of_bodyReturningAContent_outputsItsJsonForm() { + // Arrange + FunctionNode reply = + FunctionNodes.of( + "reply", + Content.class, + Content.class, + (ctx, message) -> Content.fromParts(Part.fromText("hi"))); + + // Act + RunResult result = run(chain(reply)); + + // Assert: a returned Content is the output, not a message. + result.assertCompleted(); + Event event = result.outputEventAt("wf@1/reply@1"); + assertThat(event.content()).isEmpty(); + assertThat(event.output()) + .hasValue( + ImmutableMap.of( + "parts", ImmutableList.of(ImmutableMap.of("text", "hi")), "role", "user")); + } + + @Test + public void of_bodyReturningAnEventWithARecordOutput_sendsItsJsonForm() { + // Arrange + FunctionNode order = + FunctionNodes.of( + "order", + Content.class, + Event.class, + (ctx, message) -> Event.builder().output(new Order("a1", 3)).build()); + + // Act + RunResult result = run(chain(order)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/order@1")) + .isEqualTo(ImmutableMap.of("id", "a1", "quantity", 3L)); + } + + @Test + public void of_bodyThatThrows_failsTheNodeWithItsException() { + // Arrange + FunctionNode broken = + FunctionNodes.of( + "broken", + Content.class, + String.class, + (ctx, message) -> { + throw new IOException("down"); + }); + + // Act + RunResult result = run(chain(broken)); + + // Assert + Event error = + result.events().stream().filter(e -> e.errorCode().isPresent()).collect(onlyElement()); + assertThat(error.errorCode()).hasValue(new FinishReason("IOException")); + assertThat(error.errorMessage()).hasValue("down"); + } + + @Test + public void of_blockingSiblingsWaitingForEachOther_bothFinish() { + // Arrange: each body blocks until the other has started, which needs two threads. + CountDownLatch leftStarted = new CountDownLatch(1); + CountDownLatch rightStarted = new CountDownLatch(1); + AtomicBoolean leftSawRight = new AtomicBoolean(); + AtomicBoolean rightSawLeft = new AtomicBoolean(); + FunctionNode left = + FunctionNodes.of( + "left", + Content.class, + Void.class, + (ctx, message) -> { + leftStarted.countDown(); + leftSawRight.set(rightStarted.await(5, TimeUnit.SECONDS)); + return null; + }); + FunctionNode right = + FunctionNodes.of( + "right", + Content.class, + Void.class, + (ctx, message) -> { + rightStarted.countDown(); + rightSawLeft.set(leftStarted.await(5, TimeUnit.SECONDS)); + return null; + }); + Workflow workflow = + Workflow.builder() + .name("wf") + .edges(new Edge(Start.INSTANCE, left), new Edge(Start.INSTANCE, right)) + .build(); + + // Act + RunResult result = run(workflow); + + // Assert + result.assertCompleted(); + assertThat(leftSawRight.get()).isTrue(); + assertThat(rightSawLeft.get()).isTrue(); + } + + @Test + @SuppressWarnings("rawtypes") // A Class port for a map is raw. + public void of_mapPortFedAMap_getsTheJsonObject() { + // Arrange: a map output feeds a map port. + FunctionNode makeParams = + FunctionNodes.of( + "make_params", + Content.class, + Object.class, + (ctx, message) -> ImmutableMap.of("p1", "value1", "p2", 100)); + FunctionNode describe = + FunctionNodes.of( + "describe", + Map.class, + String.class, + (ctx, params) -> "p1=" + params.get("p1") + ", p2=" + params.get("p2")); + + // Act + RunResult result = run(chain(makeParams, describe)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/describe@1")).isEqualTo("p1=value1, p2=100"); + } + + @Test + public void of_recordOutput_isItsJsonForm() { + // Arrange + FunctionNode makeOrder = + FunctionNodes.of( + "make_order", Content.class, Order.class, (ctx, message) -> new Order("a1", 3)); + + // Act + RunResult result = run(chain(makeOrder)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/make_order@1")) + .isEqualTo(ImmutableMap.of("id", "a1", "quantity", 3L)); + } + + @Test + public void of_bodiesReturningAValueOrNothing_outputWhatTheyReturn() { + // Arrange + FunctionNode hello = + FunctionNodes.of("hello", Content.class, String.class, (ctx, message) -> "Hello"); + FunctionNode silent = + FunctionNodes.of("silent", String.class, Void.class, (ctx, text) -> null); + + // Act + RunResult result = run(chain(hello, silent)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/hello@1")).isEqualTo("Hello"); + assertThat(result.eventsAt("wf@1/silent@1")).isEmpty(); + } + + @Test + @SuppressWarnings("rawtypes") // A Class port for a list is raw. + public void of_listPortWithoutItemType_getsTheJsonList() { + // Arrange + FunctionNode makeList = + FunctionNodes.of( + "make_list", Content.class, List.class, (ctx, message) -> ImmutableList.of(1, 2)); + FunctionNode describe = + FunctionNodes.of("describe", List.class, String.class, (ctx, list) -> "list=" + list); + + // Act + RunResult result = run(chain(makeList, describe)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/describe@1")).isEqualTo("list=[1, 2]"); + } + + @Test + public void of_contentPort_getsTheUserMessage() { + // Arrange + FunctionNode echo = + FunctionNodes.of("echo", Content.class, String.class, (ctx, message) -> message.text()); + + // Act + RunResult result = run(chain(echo)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/echo@1")).isEqualTo("go"); + } + + @Test + public void of_bodyReturningARouteEvent_sendsItWithoutAnOutput() { + // Arrange + FunctionNode route = + FunctionNodes.of( + "route", + Content.class, + Event.class, + (ctx, message) -> + Event.builder() + .actions( + EventActions.builder() + .route(ImmutableList.of(new Route.Tag("some_route"))) + .build()) + .build()); + + // Act + RunResult result = run(chain(route)); + + // Assert + result.assertCompleted(); + Event event = result.eventsAt("wf@1/route@1").stream().collect(onlyElement()); + assertThat(event.output()).isEmpty(); + assertThat(event.actions().route()).hasValue(ImmutableList.of(new Route.Tag("some_route"))); + } + + @Test + public void of_bodyReturningAContentEvent_sendsTheMessageWithoutAnOutput() { + // Arrange + FunctionNode say = + FunctionNodes.of( + "say", + Content.class, + Event.class, + (ctx, message) -> + Event.builder().content(Content.fromParts(Part.fromText("some content"))).build()); + + // Act + RunResult result = run(chain(say)); + + // Assert + result.assertCompleted(); + Event event = result.eventsAt("wf@1/say@1").stream().collect(onlyElement()); + assertThat(event.output()).isEmpty(); + assertThat(event.content().get().text()).isEqualTo("some content"); + } + + @Test + public void of_bodySettingStateAndReturningNothing_sendsTheStateDelta() { + // Arrange + FunctionNode write = + FunctionNodes.of( + "write", + Content.class, + Void.class, + (ctx, message) -> { + ctx.state().put("my_key", "my_value"); + return null; + }); + FunctionNode read = + FunctionNodes.of( + "read", + Void.class, + String.class, + (ctx, ignored) -> "my_key=" + ctx.state().get("my_key")); + + // Act + RunResult result = run(chain(write, read)); + + // Assert + result.assertCompleted(); + Event written = result.eventsAt("wf@1/write@1").stream().collect(onlyElement()); + assertThat(written.actions().stateDelta()).containsExactly("my_key", "my_value"); + assertThat(written.output()).isEmpty(); + assertThat(result.outputAt("wf@1/read@1")).isEqualTo("my_key=my_value"); + } + + @Test + public void of_bodySettingStateAndReturningAnEvent_mergesBothStateDeltas() { + // Arrange + FunctionNode write = + FunctionNodes.of( + "write", + Content.class, + Event.class, + (ctx, message) -> { + ctx.state().put("from_ctx", "ctx_value"); + return Event.builder() + .output("result") + .actions( + EventActions.builder() + .stateDelta(new HashMap<>(ImmutableMap.of("from_event", "event_value"))) + .build()) + .build(); + }); + + // Act + RunResult result = run(chain(write)); + + // Assert + result.assertCompleted(); + Event written = result.eventsAt("wf@1/write@1").stream().collect(onlyElement()); + assertThat(written.output()).hasValue("result"); + assertThat(written.actions().stateDelta()) + .containsExactly("from_event", "event_value", "from_ctx", "ctx_value"); + } + + @Test + public void of_inputThatDoesNotFitThePort_failsTheNodeWithoutQuotingTheValue() { + // Arrange + FunctionNode secret = + FunctionNodes.of("secret", Content.class, String.class, (ctx, message) -> "s3cr3t"); + FunctionNode describe = + FunctionNodes.of("describe", Order.class, String.class, (ctx, order) -> order.id()); + + // Act + RunResult result = run(chain(secret, describe)); + + // Assert + Event error = + result.events().stream().filter(e -> e.errorCode().isPresent()).collect(onlyElement()); + assertThat(error.errorCode()).hasValue(new FinishReason("NodeInputValidationException")); + assertThat(error.errorMessage().get()).contains("input of node 'describe'"); + assertThat(error.errorMessage().get()).doesNotContain("s3cr3t"); + } + + @Test + public void of_valueThatIsNotAContent_failsAContentPortWithoutQuotingTheValue() { + // Arrange + FunctionNode secret = + FunctionNodes.of("secret", Content.class, String.class, (ctx, message) -> "s3cr3t"); + FunctionNode read = + FunctionNodes.of("read", Content.class, String.class, (ctx, content) -> content.text()); + + // Act + RunResult result = run(chain(secret, read)); + + // Assert + Event error = + result.events().stream().filter(e -> e.errorCode().isPresent()).collect(onlyElement()); + assertThat(error.errorCode()).hasValue(new FinishReason("NodeInputValidationException")); + assertThat(error.errorMessage().get()).contains("input of node 'read'"); + assertThat(error.errorMessage().get()).doesNotContain("s3cr3t"); + } + + @Test + public void of_body_runsOnTheRunnersScheduler() { + // Arrange + ExecutorService executor = + Executors.newSingleThreadExecutor(task -> new Thread(task, "app-scheduler")); + FunctionNode where = + FunctionNodes.of( + "where", + Content.class, + String.class, + (ctx, message) -> Thread.currentThread().getName()); + + // Act + RunResult result; + try { + result = run(chain(where), Schedulers.from(executor)); + } finally { + executor.shutdownNow(); + } + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/where@1")).isEqualTo("app-scheduler"); + } + + @Test + public void toBuilder_copy_keepsThePortsAndTheBody() { + // Arrange + FunctionNode describe = + FunctionNodes.of("describe", Order.class, String.class, (ctx, order) -> order.id()); + FunctionNode makeOrder = + FunctionNodes.of( + "make_order", + Content.class, + Object.class, + (ctx, message) -> ImmutableMap.of("id", "a1", "quantity", 3)); + + // Act + RunResult result = run(chain(makeOrder, describe.toBuilder().name("copy").build())); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/copy@1")).isEqualTo("a1"); + } + + @Test + public void builder_withoutABody_failsAtBuild() { + // Arrange + FunctionNode.Builder builder = + FunctionNodes.builder(String.class, String.class).name("empty"); + + // Act + IllegalStateException error = assertThrows(IllegalStateException.class, builder::build); + + // Assert + assertThat(error).hasMessageThat().contains("requires body"); + } + + @Test + public void of_listOfRecordsPort_readsEachElementAsARecord() { + // Arrange: a list of maps feeds a list-of-records port. + AtomicReference> received = new AtomicReference<>(); + FunctionNode upstream = + FunctionNodes.of( + "upstream", + Content.class, + Object.class, + (ctx, message) -> + ImmutableList.of( + ImmutableMap.of("section_name", "s1", "content", "c1"), + ImmutableMap.of("section_name", "s2", "content", "c2"))); + FunctionNode, String> aggregate = + FunctionNodes.of( + "aggregate", + new TypeReference>() {}, + String.class, + (ctx, sections) -> { + received.set(sections); + return "Done"; + }); + + // Act + RunResult result = run(chain(upstream, aggregate)); + + // Assert + result.assertCompleted(); + assertThat(received.get()) + .containsExactly(new Section("s1", "c1"), new Section("s2", "c2")) + .inOrder(); + } + + @Test + public void of_mapOfRecordsPort_readsEachValueAsARecord() { + // Arrange: a map of maps feeds a map-of-records port. + AtomicReference> received = new AtomicReference<>(); + FunctionNode upstream = + FunctionNodes.of( + "upstream", + Content.class, + Object.class, + (ctx, message) -> + ImmutableMap.of( + "one", ImmutableMap.of("section_name", "s1", "content", "c1"), + "two", ImmutableMap.of("section_name", "s2", "content", "c2"))); + FunctionNode, String> aggregate = + FunctionNodes.builder(new TypeReference>() {}, String.class) + .name("aggregate") + .body( + (ctx, sections) -> { + received.set(sections); + return "Done"; + }) + .build(); + + // Act + RunResult result = run(chain(upstream, aggregate)); + + // Assert + result.assertCompleted(); + assertThat(received.get()) + .containsExactly("one", new Section("s1", "c1"), "two", new Section("s2", "c2")); + } + + @Test + public void of_numberIntoAListPort_failsTheNode() { + // Arrange: a number feeds a list port. + FunctionNode number = + FunctionNodes.of("number", Content.class, Integer.class, (ctx, message) -> 123); + FunctionNode, String> first = + FunctionNodes.of( + "first", + new TypeReference>() {}, + String.class, + (ctx, sections) -> sections.get(0).sectionName()); + + // Act + RunResult result = run(chain(number, first)); + + // Assert + Event error = + result.events().stream().filter(e -> e.errorCode().isPresent()).collect(onlyElement()); + assertThat(error.errorCode()).hasValue(new FinishReason("NodeInputValidationException")); + } + + @Test + public void of_polymorphicPort_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", Shape.class, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().contains("is polymorphic (@JsonTypeInfo)"); + } + + @Test + public void of_numberPort_failsAtBuildSuggestingAConcreteNumber() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", Number.class, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().contains("Declare the port as Long, Double or BigDecimal"); + } + + @Test + public void of_charSequencePort_failsAtBuildSuggestingString() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", CharSequence.class, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().contains("Declare the port as String"); + } + + @Test + public void of_outputTypeWithoutProperties_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", Content.class, NoProperties.class, (ctx, in) -> null)); + + // Assert + assertThat(error).hasMessageThat().contains("cannot write as JSON"); + } + + @Test + public void of_inputTypeCreatableOnlyFromAString_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", StringOnly.class, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().contains("only from a string or number"); + } + + @Test + public void builder_mapPortWithAnUnreadableKey_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.builder(new TypeReference>() {}, String.class)); + + // Assert + assertThat(error).hasMessageThat().contains("cannot read as a map key"); + } + + @Test + public void of_abstractTypesWithACreator_roundTrip() { + // Arrange + FunctionNode pick = + FunctionNodes.of("pick", Content.class, Range.class, (ctx, m) -> new SimpleRange(2, 5)); + FunctionNode place = + FunctionNodes.of( + "place", Range.class, Point.class, (ctx, r) -> Point.create(r.high() - r.low())); + FunctionNode readX = + FunctionNodes.of("read_x", Point.class, Integer.class, (ctx, point) -> point.x()); + + // Act + RunResult result = run(chain(pick, place, readX)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/pick@1")).isEqualTo(ImmutableMap.of("low", 2L, "high", 5L)); + assertThat(result.outputAt("wf@1/read_x@1")).isEqualTo(3L); + } + + @Test + public void of_valueTypeWrittenAsAString_roundTrips() { + // Arrange + FunctionNode makeId = + FunctionNodes.of("make_id", Content.class, OrderId.class, (ctx, m) -> new OrderId("A-1")); + FunctionNode readId = + FunctionNodes.of("read_id", OrderId.class, String.class, (ctx, id) -> "id " + id.value()); + + // Act + RunResult result = run(chain(makeId, readId)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/make_id@1")).isEqualTo("A-1"); + assertThat(result.outputAt("wf@1/read_id@1")).isEqualTo("id A-1"); + } + + @Test + public void of_genericOutputPort_writesEachElementInJsonForm() { + // Arrange + FunctionNode> sections = + FunctionNodes.of( + "sections", + Content.class, + new TypeReference>() {}, + (ctx, m) -> ImmutableList.of(new Section("s1", "c1"))); + + // Act + RunResult result = run(chain(sections)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/sections@1")) + .isEqualTo(ImmutableList.of(ImmutableMap.of("section_name", "s1", "content", "c1"))); + } + + @Test + public void of_inputTypeJacksonCannotCreate_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", NoCreator.class, String.class, (ctx, in) -> in.label)); + + // Assert + assertThat(error).hasMessageThat().contains("cannot create from JSON"); + } + + @Test + public void of_inputTypeJacksonCannotWrite_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", Token.class, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().contains("The input of node 'n'"); + assertThat(error).hasMessageThat().contains("cannot write as JSON"); + } + + @Test + public void of_listOfAnUncreatableType_failsAtBuildNamingTheElement() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> + FunctionNodes.of( + "n", new TypeReference>() {}, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().startsWith("An element of the input of node 'n'"); + } + + @Test + public void of_optionalPort_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> + FunctionNodes.of( + "n", new TypeReference>() {}, String.class, (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().contains("Declare it without Optional"); + } + + @Test + public void of_listOfOptionals_failsAtBuildNamingTheElement() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> + FunctionNodes.of( + "n", + new TypeReference>>() {}, + String.class, + (ctx, in) -> "")); + + // Assert + assertThat(error).hasMessageThat().startsWith("An element of the input of node 'n'"); + assertThat(error).hasMessageThat().contains("Declare it without Optional"); + } + + @Test + public void of_typeReadThroughADelegatingMapCreator_roundTrips() { + // Arrange + FunctionNode makeBag = + FunctionNodes.of( + "make_bag", + Content.class, + Bag.class, + (ctx, m) -> new Bag(ImmutableMap.of("name", "x"))); + FunctionNode readBag = + FunctionNodes.of("read_bag", Bag.class, String.class, (ctx, bag) -> "bag " + bag.getName()); + + // Act + RunResult result = run(chain(makeBag, readBag)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/make_bag@1")).isEqualTo(ImmutableMap.of("name", "x")); + assertThat(result.outputAt("wf@1/read_bag@1")).isEqualTo("bag x"); + } + + @Test + public void of_interfaceOutputWithoutACreator_failsAtBuild() { + // Act + WorkflowConfigurationException error = + assertThrows( + WorkflowConfigurationException.class, + () -> FunctionNodes.of("n", String.class, Decision.class, (ctx, in) -> null)); + + // Assert + assertThat(error) + .hasMessageThat() + .contains("interface or abstract class without a Jackson creator"); + } + + @Test + public void of_untypedAndAdkPorts_needNoJacksonShape() { + // Arrange + FunctionNode echo = + FunctionNodes.of("echo", Content.class, Object.class, (ctx, message) -> message.text()); + FunctionNode sink = + FunctionNodes.of("sink", Object.class, Void.class, (ctx, in) -> null); + + // Act + RunResult result = run(chain(echo, sink)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/echo@1")).isEqualTo("go"); + } + + /** Returns a node that emits {@code output} as it is, as a hand-written node does. */ + private static Node rawOutput(String name, Object output) { + return new Node() { + @Override + public String name() { + return name; + } + + @Override + public Flowable runNode(Context context, @Nullable Object nodeInput) { + return Flowable.just(output); + } + }; + } + + /** Returns a workflow named {@code wf} that runs {@code nodes} in sequence from START. */ + private static Workflow chain(Node... nodes) { + List edges = new ArrayList<>(); + Node previous = Start.INSTANCE; + for (Node node : nodes) { + edges.add(new Edge(previous, node)); + previous = node; + } + return Workflow.builder().name("wf").edges(edges.toArray(new Edge[0])).build(); + } + + private static RunResult run(Node root) { + return run(root, null); + } + + /** Runs {@code root} as an app's root node through a {@link Runner} on a "go" message. */ + private static RunResult run(Node root, @Nullable Scheduler scheduler) { + Runner runner = + Runner.builder() + .app(App.builder().name(APP).rootNode(root).build()) + .scheduler(scheduler) + .build(); + Session session = runner.sessionService().createSession(APP, USER).blockingGet(); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + runner + .runAsync(USER, session.id(), Content.fromParts(Part.fromText("go"))) + .timeout(30, TimeUnit.SECONDS) + .blockingSubscribe(events::add, error::set); + Session stored = + runner.sessionService().getSession(APP, USER, session.id(), Optional.empty()).blockingGet(); + return new RunResult(ImmutableList.copyOf(events), error.get(), stored); + } + + /** The events a run streamed, the error it failed with, if any, and the session it stored. */ + private record RunResult( + ImmutableList events, @Nullable Throwable error, Session session) { + + void assertCompleted() { + assertWithMessage("the run's error").that(error).isNull(); + } + + ImmutableList eventsAt(String nodePath) { + return events.stream() + .filter(e -> pathOf(e).equals(nodePath)) + .collect(ImmutableList.toImmutableList()); + } + + @Nullable Object outputAt(String nodePath) { + return outputEventAt(nodePath).output().get(); + } + + Event outputEventAt(String nodePath) { + return events.stream() + .filter(e -> pathOf(e).equals(nodePath) && e.output().isPresent()) + .collect(onlyElement()); + } + + private static String pathOf(Event event) { + return event.nodeInfo().map(NodeInfo::path).orElse(""); + } + } +}