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..d30fdabc6 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, workflow {@code FunctionNode} bodies, 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..991352ee2 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, workflow {@code FunctionNode} bodies, + * 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..4f99619bd --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/FunctionNode.java @@ -0,0 +1,254 @@ +/* + * 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()) + // Only explicit schemas carry over; the copy infers the others from its ports. + .inputSchema(super.inputSchema().orElse(null)) + .outputSchema(super.outputSchema().orElse(null)) + .stateSchema(stateSchema().orElse(null)) + .body(body); + } + + /** + * Returns the schema that describes the input port: the one set on the builder or, if none, one + * inferred from the input type. An {@code Object}, {@code Void} or other type with no fixed JSON + * shape infers none. + */ + @Override + public Optional inputSchema() { + return super.inputSchema().or(() -> Optional.ofNullable(inputCodec.schema())); + } + + /** + * Returns the schema that describes the output port: the one set on the builder or, if none, one + * inferred from the output type. A returned {@code Event} is sent as it is, so an {@code Event} + * output infers none. + */ + @Override + public Optional outputSchema() { + return super.outputSchema().or(() -> Optional.ofNullable(outputCodec.schema())); + } + + /** Whether {@link #inputSchema} was inferred rather than set on the builder. */ + boolean inputSchemaInferred() { + return super.inputSchema().isEmpty() && inputCodec.schema() != null; + } + + /** Whether {@link #outputSchema} was inferred rather than set on the builder. */ + boolean outputSchemaInferred() { + return super.outputSchema().isEmpty() && outputCodec.schema() != null; + } + + @Override + public Flowable runNode(Context context, @Nullable Object nodeInput) { + return Flowable.defer( + () -> { + I input = inputCodec.decode(nodeInput, "input of node '" + name() + "'"); + WorkerScope scope = new WorkerScope(context); + O result; + try { + result = body.apply(scope, input); + } finally { + scope.close(); + } + return Flowable.fromOptional(Optional.ofNullable(output(result))); + }) + .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); + } + + /** The scope of one body run, closed once the body returns. */ + private static final class WorkerScope implements NodeScope { + private final Context context; + private volatile boolean open = true; + + WorkerScope(Context context) { + this.context = context; + } + + @Override + public Context context() { + checkState(open, "A NodeScope works only while its body runs."); + return context; + } + + void close() { + open = false; + } + } + + /** + * 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 that describes the node's input, in place of one inferred from its type. */ + @CanIgnoreReturnValue + public Builder inputSchema(@Nullable Schema inputSchema) { + this.inputSchema = inputSchema; + return this; + } + + /** Sets the schema that describes the node's output, in place of one inferred from its type. */ + @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..d1ab39f7f --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/NodeBody.java @@ -0,0 +1,36 @@ +/* + * 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; + +/** + * The body of a {@link FunctionNode}, which takes the node's single input rather than parameters + * from state, as a Python function node does; 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 as it is, or the output, which is + * sent in its JSON form; a returned {@code Content} is the output, not a message. + */ + O apply(NodeScope scope, 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/NodeScope.java b/core/src/main/java/com/google/adk/workflow/NodeScope.java new file mode 100644 index 000000000..52e9a4afb --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/NodeScope.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.agents.Context; +import com.google.adk.annotations.Experimental; + +/** + * What a {@link NodeBody} can use besides its input. A scope works only while its body runs, and + * throws {@link IllegalStateException} once the body has returned. + */ +@Experimental +public interface NodeScope { + + /** Returns the node run's context, with its state, resume inputs and routes. */ + Context context(); +} 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..114174e8c --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/PortCodec.java @@ -0,0 +1,458 @@ +/* + * 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.common.base.Supplier; +import com.google.common.base.Suppliers; +import com.google.genai.JsonSerializable; +import com.google.genai.types.Content; +import com.google.genai.types.Schema; +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. + */ + 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 final Supplier<@Nullable Schema> schema; + + private PortCodec(JavaType type, boolean output) { + this.type = type; + // An event output passes through as it is, so no JSON form describes it. + this.schema = + output && type.getRawClass() == Event.class + ? () -> null + : Suppliers.memoize(() -> PortSchemas.infer(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, /* output= */ false); + } + + /** + * 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, /* output= */ true); + } + + /** 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 + && beanSerializer(type) != null + && !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."); + } + } + + static @Nullable TypeSerializer findTypeSerializer(JavaType type) { + try { + return MAPPER.getSerializerProviderInstance().findTypeSerializer(type); + } catch (JsonMappingException | RuntimeException e) { + return null; + } + } + + /** Returns the serializer that writes {@code type} as a bean, or {@code null} if none does. */ + static @Nullable BeanSerializerBase beanSerializer(JavaType type) { + try { + return MAPPER.getSerializerProviderInstance().findValueSerializer(type) + instanceof BeanSerializerBase bean + ? bean + : null; + } catch (JsonMappingException | RuntimeException e) { + return null; + } + } + + /** 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 the schema inferred for this port's JSON form, or {@code null} if it has none. */ + @Nullable Schema schema() { + return schema.get(); + } + + /** 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/main/java/com/google/adk/workflow/PortSchemas.java b/core/src/main/java/com/google/adk/workflow/PortSchemas.java new file mode 100644 index 000000000..0b04ad26a --- /dev/null +++ b/core/src/main/java/com/google/adk/workflow/PortSchemas.java @@ -0,0 +1,271 @@ +/* + * 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.ImmutableList.toImmutableList; + +import com.fasterxml.jackson.annotation.JsonFormat; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.OptBoolean; +import com.fasterxml.jackson.databind.AnnotationIntrospector; +import com.fasterxml.jackson.databind.BeanDescription; +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.SerializationConfig; +import com.fasterxml.jackson.databind.introspect.AnnotatedMember; +import com.fasterxml.jackson.databind.introspect.AnnotationMap; +import com.fasterxml.jackson.databind.ser.BeanPropertyWriter; +import com.fasterxml.jackson.databind.ser.PropertyWriter; +import com.fasterxml.jackson.databind.ser.std.BeanSerializerBase; +import com.fasterxml.jackson.databind.type.TypeFactory; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Streams; +import com.google.genai.types.Schema; +import com.google.genai.types.Type; +import java.lang.annotation.Annotation; +import java.lang.invoke.MethodType; +import java.lang.reflect.AnnotatedElement; +import java.lang.reflect.Method; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.stream.Stream; +import org.jspecify.annotations.Nullable; + +/** + * Infers the schema of a {@link FunctionNode} port's JSON form from the port's type. Records, POJOs + * and enums are read through the port's Jackson mapper, so names and values are the ones written. + */ +final class PortSchemas { + + // Jackson writes a char[] as a string, like a byte[] in base64. + private static final ImmutableSet> STRINGS = + ImmutableSet.of( + String.class, CharSequence.class, Character.class, byte[].class, char[].class); + private static final ImmutableSet> INTEGERS = + ImmutableSet.of(Integer.class, Long.class, Short.class, Byte.class, BigInteger.class); + private static final ImmutableSet> NUMBERS = + ImmutableSet.of(Float.class, Double.class, BigDecimal.class, Number.class); + private static final SerializationConfig CONFIG = PortCodec.MAPPER.getSerializationConfig(); + private static final AnnotationIntrospector ANNOTATIONS = CONFIG.getAnnotationIntrospector(); + + /** The records and POJOs being described, to find a type that contains itself. */ + private final Set enclosing = new HashSet<>(); + + private boolean recursive; + + private PortSchemas() {} + + /** + * Returns the schema of {@code type}'s JSON form as a port writes it, or {@code null} if that + * form has no fixed shape or {@code type} contains itself. + */ + static @Nullable Schema infer(JavaType type) { + PortSchemas inference = new PortSchemas(); + Schema schema = inference.describe(type); + return inference.recursive ? null : schema; + } + + private @Nullable Schema describe(JavaType type) { + // Boxes a primitive, so each set lists only the boxed type. + Class raw = MethodType.methodType(type.getRawClass()).wrap().returnType(); + if (STRINGS.contains(raw)) { + return typed(Type.Known.STRING); + } + if (raw == Boolean.class) { + return typed(Type.Known.BOOLEAN); + } + if (INTEGERS.contains(raw)) { + return typed(Type.Known.INTEGER); + } + if (NUMBERS.contains(raw)) { + return typed(Type.Known.NUMBER); + } + if (raw.isEnum()) { + return describeEnum(raw); + } + if (type.isReferenceType()) { + Schema content = describe(type.getContentType()); + return content == null ? null : content.toBuilder().nullable(true).build(); + } + if (type.isArrayType() || type.isCollectionLikeType()) { + Schema items = describe(type.getContentType()); + Schema.Builder array = Schema.builder().type(Type.Known.ARRAY); + return (items == null ? array : array.items(items)).build(); + } + if (type.isMapLikeType()) { + return typed(Type.Known.OBJECT); + } + return describeObject(type); + } + + /** Returns a string enum of the values the mapper writes, or {@code null} if one is not text. */ + private static @Nullable Schema describeEnum(Class type) { + ImmutableList values; + try { + values = + Arrays.stream(type.getEnumConstants()) + .map(constant -> PortCodec.MAPPER.valueToTree(constant)) + .collect(toImmutableList()); + } catch (RuntimeException e) { + return null; + } + return values.stream().allMatch(JsonNode::isTextual) + ? Schema.builder() + .type(Type.Known.STRING) + .enum_(values.stream().map(JsonNode::textValue).collect(toImmutableList())) + .build() + : null; + } + + /** Returns the schema of a record or POJO, or {@code null} if its keys are not fixed. */ + private @Nullable Schema describeObject(JavaType type) { + // A polymorphic type also writes its subtypes' properties. + BeanSerializerBase serializer = + PortCodec.findTypeSerializer(type) == null ? PortCodec.beanSerializer(type) : null; + if (serializer == null) { + return null; + } + ImmutableList writers = ImmutableList.copyOf(serializer.properties()); + if (!hasFixedKeys(type, writers)) { + return null; + } + if (!enclosing.add(type)) { + recursive = true; + return null; + } + ImmutableMap.Builder properties = ImmutableMap.builder(); + ImmutableList.Builder required = ImmutableList.builder(); + for (PropertyWriter property : writers) { + JavaType propertyType = property.getType(); + boolean nullable = + !propertyType.isPrimitive() + && (propertyType.isReferenceType() || hasNullableAnnotation(property.getMember())); + if (hasElementTypeId(property)) { + propertyType = propertyType.withContentType(TypeFactory.unknownType()); + } + // A serializer, type id, object id or shape of the property's own may write another type. + Schema schema = hasOwnForm(property) ? null : describe(propertyType); + if (schema == null) { + // A property of no fixed shape takes any value. + schema = Schema.builder().build(); + } else if (nullable) { + schema = schema.toBuilder().nullable(true).build(); + } + properties.put(property.getName(), schema); + if (!nullable && !isMarkedNotRequired(property)) { + required.add(property.getName()); + } + } + enclosing.remove(type); + ImmutableList requiredNames = required.build(); + Schema.Builder object = + Schema.builder().type(Type.Known.OBJECT).properties(properties.buildOrThrow()); + return (requiredNames.isEmpty() ? object : object.required(requiredNames)).build(); + } + + /** + * Whether the keys {@code type} writes are the names of its {@code writers}, and it has no + * any-setter to take other keys. An any-getter, an unwrapped property, an object id or a shape + * other than an object writes other keys, or no object. + */ + private static boolean hasFixedKeys(JavaType type, ImmutableList writers) { + try { + BeanDescription description = CONFIG.introspect(type); + JsonFormat.Value format = description.findExpectedFormat(); + return description.findAnyGetter() == null + && description.findAnySetterAccessor() == null + && description.getObjectIdInfo() == null + && (!format.hasShape() || format.getShape() == JsonFormat.Shape.OBJECT) + && writers.stream() + .noneMatch(writer -> writer instanceof BeanPropertyWriter w && w.isUnwrapping()); + } catch (RuntimeException e) { + // Jackson rejects a malformed or second any-getter or any-setter. + return false; + } + } + + /** + * Whether {@code property} has a serializer, content serializer, type id, object id or shape of + * its own. A type id on a container goes on each element instead; see {@link #hasElementTypeId}. + */ + private static boolean hasOwnForm(PropertyWriter property) { + if (property instanceof BeanPropertyWriter writer && writer.getTypeSerializer() != null) { + return true; + } + AnnotatedMember member = property.getMember(); + if (member == null) { + return false; + } + JsonFormat.Value format = ANNOTATIONS.findFormat(member); + return ANNOTATIONS.findSerializer(member) != null + || ANNOTATIONS.findContentSerializer(member) != null + || ANNOTATIONS.findObjectIdInfo(member) != null + || (format != null && format.hasShape()); + } + + /** Whether the {@code @JsonTypeInfo} of a container property puts a type id on each element. */ + private static boolean hasElementTypeId(PropertyWriter property) { + JavaType type = property.getType(); + return property.getMember() != null + && (type.isContainerType() || type.isReferenceType()) + && ANNOTATIONS.findPropertyContentTypeResolver(CONFIG, property.getMember(), type) != null; + } + + /** Whether {@code member} is annotated {@code @Nullable}, from whichever package. */ + private static boolean hasNullableAnnotation(@Nullable AnnotatedMember member) { + if (member == null || member.getAnnotated() == null) { + return false; + } + AnnotationMap declared = member.getAllAnnotations(); + return Stream.concat( + declared == null ? Stream.empty() : Streams.stream(declared.annotations()), + Arrays.stream(typeAnnotations(member.getAnnotated()))) + .anyMatch(annotation -> annotation.annotationType().getSimpleName().equals("Nullable")); + } + + /** + * Returns the annotations on the type of a getter or field, where a type-use annotation such as + * JSpecify's sits. Android has no {@code AnnotatedType}, so this calls the JDK reflectively. + */ + private static Annotation[] typeAnnotations(AnnotatedElement element) { + String accessor = element instanceof Method ? "getAnnotatedReturnType" : "getAnnotatedType"; + try { + return ((AnnotatedElement) element.getClass().getMethod(accessor).invoke(element)) + .getAnnotations(); + } catch (ReflectiveOperationException e) { + return new Annotation[0]; + } + } + + /** + * Whether {@code property} has {@code @JsonProperty(isRequired = OptBoolean.FALSE)}; Jackson + * cannot tell {@code required = false} from that attribute's default. + */ + private static boolean isMarkedNotRequired(PropertyWriter property) { + JsonProperty annotation = property.getAnnotation(JsonProperty.class); + return annotation != null && annotation.isRequired() == OptBoolean.FALSE; + } + + private static Schema typed(Type.Known type) { + return Schema.builder().type(type).build(); + } +} 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..d8c815d77 --- /dev/null +++ b/core/src/test/java/com/google/adk/workflow/FunctionNodeTest.java @@ -0,0 +1,1493 @@ +/* + * 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.ImmutableMap.toImmutableMap; +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.JsonAnyGetter; +import com.fasterxml.jackson.annotation.JsonAnySetter; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonFormat; +import com.fasterxml.jackson.annotation.JsonIdentityInfo; +import com.fasterxml.jackson.annotation.JsonIdentityReference; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import com.fasterxml.jackson.annotation.JsonUnwrapped; +import com.fasterxml.jackson.annotation.JsonValue; +import com.fasterxml.jackson.annotation.ObjectIdGenerators; +import com.fasterxml.jackson.annotation.OptBoolean; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.ser.std.ToStringSerializer; +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.common.collect.Maps; +import com.google.genai.types.Content; +import com.google.genai.types.FinishReason; +import com.google.genai.types.Part; +import com.google.genai.types.Schema; +import com.google.genai.types.Type; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.core.Scheduler; +import io.reactivex.rxjava3.schedulers.Schedulers; +import java.io.IOException; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.reflect.RecordComponent; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +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"; + private static final Schema ANY = Schema.builder().build(); + private static final Schema STRING = typed(Type.Known.STRING); + private static final Schema INTEGER = typed(Type.Known.INTEGER); + private static final Schema ORDER = + object(ImmutableMap.of("id", STRING, "quantity", INTEGER), "id", "quantity"); + + 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; + } + } + + record Note( + String title, + int pages, + Optional subtitle, + @Nullable String body, + @OtherNullness.Nullable String footer) {} + + /** A record whose JSON form renames one property, ignores one and does not require two. */ + record Draft( + @JsonProperty("section_name") String sectionName, + @JsonProperty(isRequired = OptBoolean.FALSE) String content, + @JsonProperty(isRequired = OptBoolean.FALSE) int priority, + @JsonIgnore String notes) {} + + enum Status { + OPEN, + SHIPPED + } + + record Shipment(Status status, Order order, List history, Map byId) {} + + enum Color { + @JsonProperty("red") + RED, + @JsonProperty("dark-blue") + BLUE + } + + /** An enum that Jackson writes as a number. */ + enum Level { + LOW(1), + HIGH(2); + + private final int code; + + Level(int code) { + this.code = code; + } + + @JsonValue + int code() { + return code; + } + } + + /** A record that also writes the entries of a map as keys of its own. */ + record Tagged(String name) { + @JsonAnyGetter + ImmutableMap extra() { + return ImmutableMap.of("k", 1); + } + } + + /** A class whose any-setter takes keys it does not declare. */ + static final class Catchall { + public String id; + + @JsonAnySetter + void put(String key, Object value) {} + } + + /** A class with two any-setters, which Jackson rejects. */ + static final class TwoCatchalls { + public String id; + + @JsonAnySetter + void put(String key, Object value) {} + + @JsonAnySetter + void add(String key, Object value) {} + } + + record Flat(String label, @JsonUnwrapped Order order) {} + + @JsonFormat(shape = JsonFormat.Shape.ARRAY) + record Pair(int x, int y) {} + + /** A polymorphic base type with a property of its own. */ + @JsonTypeInfo(use = JsonTypeInfo.Id.NAME) + abstract static class Pet { + public String name; + } + + @JsonIdentityInfo(generator = ObjectIdGenerators.IntSequenceGenerator.class) + record Ref(String label) {} + + /** A record whose properties have no fixed JSON type. */ + record Irregular( + @JsonSerialize(using = ToStringSerializer.class) long id, + @JsonSerialize(contentUsing = ToStringSerializer.class) List codes, + @JsonFormat(shape = JsonFormat.Shape.STRING) int count, + @JsonTypeInfo(use = JsonTypeInfo.Id.NAME) Order typed, + @JsonTypeInfo(use = JsonTypeInfo.Id.NAME) List typedOrders, + @JsonIdentityInfo(generator = ObjectIdGenerators.IntSequenceGenerator.class) + @JsonIdentityReference(alwaysAsId = true) + Order byId, + Tagged tagged, + Catchall catchall, + TwoCatchalls twoCatchalls, + Flat flat, + Pair pair, + Ref ref, + Pet pet, + Level level, + Instant at, + Object raw, + List pets) {} + + record Measure(Number amount, CharSequence unit) {} + + record TreeNode(String label, List children) {} + + record Box(T item) {} + + /** Holds a declaration annotation named like the {@code @Nullable} of other packages. */ + static final class OtherNullness { + @Retention(RetentionPolicy.RUNTIME) + @interface Nullable {} + + private OtherNullness() {} + } + + @Test + public void of_recordPortFedAMap_readsTheRecord() { + // Arrange: a map output feeds a record port (Python's input injection into a model). + FunctionNode makeOrder = + FunctionNodes.of( + "make_order", + Content.class, + Object.class, + (scope, message) -> ImmutableMap.of("id", "a1", "quantity", 3)); + FunctionNode describe = + FunctionNodes.of( + "describe", + Order.class, + String.class, + (scope, 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, (scope, message) -> new Order("a1", 3)); + FunctionNode describe = + FunctionNodes.of( + "describe", + Order.class, + String.class, + (scope, 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, (scope, message) -> 3); + FunctionNode increment = + FunctionNodes.of("increment", int.class, Integer.class, (scope, 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, + (scope, message) -> Content.fromParts(Part.fromText("hi"))); + + // Act + RunResult result = run(chain(reply)); + + // Assert: unlike Python, 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, + (scope, 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, + (scope, 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, + (scope, message) -> { + leftStarted.countDown(); + leftSawRight.set(rightStarted.await(5, TimeUnit.SECONDS)); + return null; + }); + FunctionNode right = + FunctionNodes.of( + "right", + Content.class, + Void.class, + (scope, 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: Python's input injection into a dict. + FunctionNode makeParams = + FunctionNodes.of( + "make_params", + Content.class, + Object.class, + (scope, message) -> ImmutableMap.of("p1", "value1", "p2", 100)); + FunctionNode describe = + FunctionNodes.of( + "describe", + Map.class, + String.class, + (scope, 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, (scope, 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: Python's various function nodes, minus generators. + FunctionNode hello = + FunctionNodes.of("hello", Content.class, String.class, (scope, message) -> "Hello"); + FunctionNode silent = + FunctionNodes.of("silent", String.class, Void.class, (scope, 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, (scope, message) -> ImmutableList.of(1, 2)); + FunctionNode describe = + FunctionNodes.of("describe", List.class, String.class, (scope, 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, (scope, 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, + (scope, 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, + (scope, 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, + (scope, message) -> { + scope.context().state().put("my_key", "my_value"); + return null; + }); + FunctionNode read = + FunctionNodes.of( + "read", + Void.class, + String.class, + (scope, ignored) -> "my_key=" + scope.context().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, + (scope, message) -> { + scope.context().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, (scope, message) -> "s3cr3t"); + FunctionNode describe = + FunctionNodes.of("describe", Order.class, String.class, (scope, 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, (scope, message) -> "s3cr3t"); + FunctionNode read = + FunctionNodes.of("read", Content.class, String.class, (scope, 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, + (scope, 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, (scope, order) -> order.id()); + FunctionNode makeOrder = + FunctionNodes.of( + "make_order", + Content.class, + Object.class, + (scope, 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 toBuilder_copy_keepsOnlyTheExplicitSchema() { + // Arrange + Schema explicit = Schema.builder().type(Type.Known.OBJECT).build(); + FunctionNode node = + FunctionNodes.builder(Order.class, Order.class) + .name("n") + .inputSchema(explicit) + .body((scope, order) -> order) + .build(); + + // Act + FunctionNode copy = node.toBuilder().name("copy").build(); + + // Assert: the explicit schema wins over the inferred one in the node and in its copy. + assertThat(node.inputSchema()).hasValue(explicit); + assertThat(copy.inputSchema()).hasValue(explicit); + assertThat(copy.inputSchemaInferred()).isFalse(); + assertThat(copy.outputSchema()).hasValue(ORDER); + assertThat(copy.outputSchemaInferred()).isTrue(); + } + + @Test + public void of_inputThatBreaksTheInferredSchema_stillRuns() { + // Arrange: the inferred Order schema requires a quantity and has no note. + Node makeOrder = rawOutput("make_order", ImmutableMap.of("id", "a1", "note", "rush")); + FunctionNode describe = + FunctionNodes.of( + "describe", + Order.class, + String.class, + (scope, order) -> order.id() + " x" + order.quantity()); + + // Act + RunResult result = run(chain(makeOrder, describe)); + + // Assert: guards against enforcing inferred schemas, which only describe a port. + assertThat(describe.inputSchemaInferred()).isTrue(); + result.assertCompleted(); + assertThat(result.outputAt("wf@1/describe@1")).isEqualTo("a1 x0"); + } + + @Test + public void of_recordWithOrWithoutAnOptionalKey_readsTheValueOrEmpty() { + // Arrange: the inferred Note schema does not require subtitle, so it may arrive without one. + FunctionNode read = + FunctionNodes.of( + "read", Note.class, String.class, (scope, note) -> String.valueOf(note.subtitle())); + Node without = rawOutput("make", ImmutableMap.of("title", "t", "pages", 1)); + Node with = rawOutput("make", ImmutableMap.of("title", "t", "pages", 1, "subtitle", "s")); + + // Act + RunResult missing = run(chain(without, read)); + RunResult present = run(chain(with, read)); + + // Assert + missing.assertCompleted(); + present.assertCompleted(); + assertThat(missing.outputAt("wf@1/read@1")).isEqualTo("Optional.empty"); + assertThat(present.outputAt("wf@1/read@1")).isEqualTo("Optional[s]"); + } + + @Test + public void inputSchema_scalarAndArrayTypes_mapToTheirJsonTypes() { + // Arrange: as in ADK Kotlin, a long is an INTEGER, unlike in a tool declaration. + Schema number = typed(Type.Known.NUMBER); + ImmutableMap, Schema> expected = + ImmutableMap., Schema>builder() + .put(String.class, STRING) + .put(byte[].class, STRING) + .put(char[].class, STRING) + .put(String[].class, Schema.builder().type(Type.Known.ARRAY).items(STRING).build()) + .put(boolean.class, typed(Type.Known.BOOLEAN)) + .put(int.class, INTEGER) + .put(long.class, INTEGER) + .put(Double.class, number) + .put( + Measure.class, + object(ImmutableMap.of("amount", number, "unit", STRING), "amount", "unit")) + .buildOrThrow(); + + // Act + ImmutableMap, Schema> actual = + expected.keySet().stream() + .collect(toImmutableMap(type -> type, type -> inputSchema(type).orElseThrow())); + + // Assert + assertThat(actual).isEqualTo(expected); + } + + @Test + public void inputSchema_optionalOrNullableProperties_areNullableAndNotRequired() { + // Act + Optional schema = inputSchema(Note.class); + + // Assert: any annotation named Nullable counts, a type-use one included. + Schema text = nullable(STRING); + ImmutableMap properties = + ImmutableMap.of( + "title", STRING, "pages", INTEGER, "subtitle", text, "body", text, "footer", text); + assertThat(schema).hasValue(object(properties, "title", "pages")); + } + + @Test + public void outputSchema_jacksonAnnotations_renameIgnoreOrUnrequireProperties() { + // Act + Optional schema = outputSchema(Draft.class); + + // Assert: a primitive marked not required is optional too, as a missing one reads as zero. + assertThat(schema) + .hasValue( + object( + ImmutableMap.of("section_name", STRING, "content", STRING, "priority", INTEGER), + "section_name")); + } + + @Test + public void inputSchema_nestedRecordEnumCollectionAndMap_describeEachLevel() { + // Act + Optional shipment = inputSchema(Shipment.class); + Optional list = inputSchema(new TypeReference>() {}); + Optional set = inputSchema(new TypeReference>() {}); + Optional objects = inputSchema(new TypeReference>() {}); + + // Assert + Schema status = Schema.builder().type(Type.Known.STRING).enum_("OPEN", "SHIPPED").build(); + Schema orders = Schema.builder().type(Type.Known.ARRAY).items(ORDER).build(); + ImmutableMap properties = + ImmutableMap.of( + "status", status, "order", ORDER, "history", orders, "byId", typed(Type.Known.OBJECT)); + assertThat(shipment).hasValue(object(properties, "status", "order", "history", "byId")); + assertThat(list).hasValue(orders); + assertThat(set).hasValue(orders); + assertThat(objects).hasValue(typed(Type.Known.ARRAY)); + } + + @Test + public void inputSchema_enums_listTheValuesTheMapperWrites() { + // Act + Optional color = inputSchema(Color.class); + Optional level = inputSchema(Level.class); + + // Assert: an enum written as a number has no string values to list, so it has no schema. + assertThat(color) + .hasValue(Schema.builder().type(Type.Known.STRING).enum_("red", "dark-blue").build()); + assertThat(level).isEmpty(); + } + + @Test + public void outputSchema_propertiesOfNoFixedJsonType_takeAnyValue() { + // Act + Optional irregular = outputSchema(Irregular.class); + Optional pair = outputSchema(Pair.class); + + // Assert: each property takes any value, but a list whose elements carry type ids is a list. + ImmutableList names = + Arrays.stream(Irregular.class.getRecordComponents()) + .map(RecordComponent::getName) + .collect(ImmutableList.toImmutableList()); + ImmutableMap properties = + Maps.toMap( + names, + name -> + name.equals("typedOrders") || name.equals("pets") ? typed(Type.Known.ARRAY) : ANY); + assertThat(irregular).hasValue(object(properties, names.toArray(String[]::new))); + assertThat(pair).isEmpty(); + } + + @Test + public void inputSchema_recursiveTypes_areEmptyUnlikeNestedGenerics() { + // Act + Optional tree = inputSchema(TreeNode.class); + Optional box = inputSchema(new TypeReference>>() {}); + + // Assert: the Box inside a Box> is another type, not a cycle. + Schema intBox = object(ImmutableMap.of("item", INTEGER), "item"); + assertThat(tree).isEmpty(); + assertThat(box).hasValue(object(ImmutableMap.of("item", intBox), "item")); + } + + @Test + public void of_contentPorts_inferContentsSchemaForBothPorts() { + // Act + FunctionNode node = + FunctionNodes.of("n", Content.class, Content.class, (scope, content) -> content); + + // Assert: a returned Content is the output in its JSON form. + Schema schema = node.inputSchema().orElseThrow(); + assertThat(schema.type()).hasValue(new Type(Type.Known.OBJECT)); + assertThat(schema.properties().orElseThrow()).containsEntry("role", nullable(STRING)); + assertThat(schema.properties().orElseThrow().keySet()).containsExactly("parts", "role"); + assertThat(node.outputSchema()).hasValue(schema); + assertThat(node.inputSchemaInferred()).isTrue(); + assertThat(node.outputSchemaInferred()).isTrue(); + } + + @Test + public void of_eventAndObjectPorts_inferASchemaOnlyForAnEventInput() { + // Act + FunctionNode node = + FunctionNodes.of("n", Event.class, Event.class, (scope, event) -> event); + Optional object = inputSchema(Object.class); + + // Assert: a returned Event is sent as it is, and an Object takes any JSON value. + assertThat(node.inputSchema().orElseThrow().properties().orElseThrow()) + .containsEntry("author", STRING); + assertThat(node.inputSchemaInferred()).isTrue(); + assertThat(node.outputSchema()).isEmpty(); + assertThat(node.outputSchemaInferred()).isFalse(); + assertThat(object).isEmpty(); + } + + @Test + public void scope_usedAfterItsBodyReturned_throws() { + // Arrange + AtomicReference kept = new AtomicReference<>(); + FunctionNode keep = + FunctionNodes.of( + "keep", + Content.class, + Void.class, + (scope, message) -> { + kept.set(scope); + return null; + }); + run(chain(keep)).assertCompleted(); + + // Act + IllegalStateException error = + assertThrows(IllegalStateException.class, () -> kept.get().context()); + + // Assert + assertThat(error).hasMessageThat().contains("only while its body runs"); + } + + @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: Python's list conversion into models. + AtomicReference> received = new AtomicReference<>(); + FunctionNode upstream = + FunctionNodes.of( + "upstream", + Content.class, + Object.class, + (scope, 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, + (scope, 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: Python's dict conversion into models. + AtomicReference> received = new AtomicReference<>(); + FunctionNode upstream = + FunctionNodes.of( + "upstream", + Content.class, + Object.class, + (scope, 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( + (scope, 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: Python's list port fed a non-list. + FunctionNode number = + FunctionNodes.of("number", Content.class, Integer.class, (scope, message) -> 123); + FunctionNode, String> first = + FunctionNodes.of( + "first", + new TypeReference>() {}, + String.class, + (scope, 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, (scope, 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, (scope, 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, (scope, 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, (scope, 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, (scope, 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, (scope, m) -> new SimpleRange(2, 5)); + FunctionNode place = + FunctionNodes.of( + "place", Range.class, Point.class, (scope, r) -> Point.create(r.high() - r.low())); + FunctionNode readX = + FunctionNodes.of("read_x", Point.class, Integer.class, (scope, 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, (scope, m) -> new OrderId("A-1")); + FunctionNode readId = + FunctionNodes.of("read_id", OrderId.class, String.class, (scope, 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>() {}, + (scope, 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, (scope, 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, (scope, 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, (scope, 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, + (scope, 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, + (scope, 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, + (scope, m) -> new Bag(ImmutableMap.of("name", "x"))); + FunctionNode readBag = + FunctionNodes.of( + "read_bag", Bag.class, String.class, (scope, 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, (scope, 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, (scope, message) -> message.text()); + FunctionNode sink = + FunctionNodes.of("sink", Object.class, Void.class, (scope, in) -> null); + + // Act + RunResult result = run(chain(echo, sink)); + + // Assert + result.assertCompleted(); + assertThat(result.outputAt("wf@1/echo@1")).isEqualTo("go"); + } + + private static Optional inputSchema(Class type) { + return FunctionNodes.of("n", type, Void.class, (scope, input) -> null).inputSchema(); + } + + private static Optional inputSchema(TypeReference type) { + return FunctionNodes.of("n", type, Void.class, (scope, input) -> null).inputSchema(); + } + + private static Optional outputSchema(Class type) { + return FunctionNodes.of("n", Void.class, type, (scope, input) -> null).outputSchema(); + } + + private static Schema typed(Type.Known type) { + return Schema.builder().type(type).build(); + } + + private static Schema nullable(Schema schema) { + return schema.toBuilder().nullable(true).build(); + } + + private static Schema object(ImmutableMap properties, String... required) { + Schema.Builder object = Schema.builder().type(Type.Known.OBJECT).properties(properties); + return (required.length == 0 ? object : object.required(required)).build(); + } + + /** 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(""); + } + } +}