feat: check workflow edges by assignability

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

@@ -81,8 +81,11 @@ public Builder<I, O> toBuilder() {
 
   /**
    * 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.
+   * inferred from the input type, where an {@code Object}, {@code Void} or other type with no fixed
+   * JSON shape infers none. An edge into an inferred record or POJO fails at build if the
+   * producer's output schema lists a key the type does not, while a builder schema checks only the
+   * keys it lists and requires. On a {@code Content} port a builder schema describes the message
+   * text, and an edge is checked against the {@code Content} schema instead.
    */
   @Override
   public Optional<Schema> inputSchema() {
@@ -104,6 +107,21 @@ boolean inputSchemaInferred() {
     return super.inputSchema().isEmpty() && inputCodec.schema() != null;
   }
 
+  /** Returns the schema an edge into this node must fit, the inferred one on a Content port. */
+  Optional<Schema> edgeInputSchema() {
+    return inputCodec.takesContent() ? Optional.ofNullable(inputCodec.schema()) : inputSchema();
+  }
+
+  /** Whether a builder schema would let this port take keys its type does not list. */
+  boolean acceptsExtraKeysWithASchema() {
+    return inputSchemaInferred() && !inputCodec.takesContent();
+  }
+
+  /** Whether {@link #edgeInputSchema} was inferred, so its objects with properties are closed. */
+  boolean edgeInputSchemaInferred() {
+    return inputCodec.takesContent() || inputSchemaInferred();
+  }
+
   @Override
   public Flowable<?> runNode(Context context, @Nullable Object nodeInput) {
     return Flowable.defer(
@@ -227,7 +245,10 @@ public Builder<I, O> config(NodeConfig config) {
       return this;
     }
 
-    /** Sets the schema that describes the node's input, in place of one inferred from its type. */
+    /**
+     * Sets the schema that describes the node's input, in place of one inferred from its type. On a
+     * {@code Content} port, the schema describes the message's text.
+     */
     @CanIgnoreReturnValue
     public Builder<I, O> inputSchema(@Nullable Schema inputSchema) {
       this.inputSchema = inputSchema;

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

@@ -24,8 +24,10 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
+import com.google.genai.types.Schema;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -80,6 +82,7 @@ static Graph of(List<Edge> edges) {
       NodeNames.validate(node.name());
     }
     GraphValidation.validate(nodes, edgeList);
+    checkEdgeSchemas(edgeList);
     ImmutableMap<String, Node> nodesByName =
         nodes.stream().collect(toImmutableMap(Node::name, node -> node));
     ImmutableSet<String> sourceNames =
@@ -92,6 +95,41 @@ static Graph of(List<Edge> edges) {
     return new Graph(nodes, edgeList, nodesByName, terminalNodeNames);
   }
 
+  /**
+   * Throws unless the output schema of each edge's source fits the input schema of its target, when
+   * both are set. Each object that lists properties in a schema the target inferred is closed.
+   */
+  private static void checkEdgeSchemas(List<Edge> edges) {
+    for (Edge edge : edges) {
+      FunctionNode<?, ?> function = edge.to() instanceof FunctionNode<?, ?> node ? node : null;
+      Optional<Schema> output = edge.from().outputSchema();
+      Optional<Schema> input =
+          function != null ? function.edgeInputSchema() : edge.to().inputSchema();
+      if (output.isEmpty() || input.isEmpty()) {
+        continue;
+      }
+      boolean closed = function != null && function.edgeInputSchemaInferred();
+      if (!SchemaFit.fits(output.get(), input.get(), closed)) {
+        throw new WorkflowConfigurationException(
+            "Schema mismatch on edge "
+                + edge.from().name()
+                + " -> "
+                + edge.to().name()
+                + ". Output schema "
+                + output.get().toJson()
+                + " does not fit input schema "
+                + input.get().toJson()
+                + "."
+                + (function != null
+                        && function.acceptsExtraKeysWithASchema()
+                        && SchemaFit.fits(output.get(), input.get(), /* closed= */ false)
+                    ? " It was inferred from the port's type and takes no other keys; set"
+                        + " inputSchema on the builder or read the objects as Maps to accept them."
+                    : ""));
+      }
+    }
+  }
+
   ImmutableList<Node> nodes() {
     return nodes;
   }

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

@@ -62,7 +62,13 @@ default NodeConfig config() {
     return NodeConfig.builder().build();
   }
 
-  /** Returns the schema the node's input must match. */
+  /**
+   * Returns the schema the node's input must match. An edge into the node fails at build when its
+   * source's output schema lists a key or type this one rejects, so an open producer such as a map
+   * builds, and a {@link FunctionNode} on a {@code Content} port checks the edge against the {@code
+   * Content} schema. A node that requires all predecessors applies it to each predecessor's output,
+   * not the joined map.
+   */
   @Experimental
   default Optional<Schema> inputSchema() {
     return Optional.empty();

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

@@ -499,6 +499,11 @@ private static String capitalized(String what) {
     return schema.get();
   }
 
+  /** Whether this port reads a {@code Content}. */
+  boolean takesContent() {
+    return type.getRawClass() == Content.class;
+  }
+
   /**
    * Returns {@code value} read as this port's type; {@code what} names the port in errors.
    *

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

@@ -0,0 +1,200 @@
+/*
+ * 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.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.genai.types.Schema;
+import com.google.genai.types.Type;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Decides at build time whether an output schema fits an input schema by types, enums, keys and
+ * nullability; it ignores formats, patterns and bounds. An output with no type or enum fits any
+ * input that accepts its nulls, and an object output that lists no keys fits any object input.
+ */
+final class SchemaFit {
+
+  /**
+   * Whether {@code source} fits {@code target} by types, enums, keys and nullability. With {@code
+   * closed}, each object in {@code target} that lists properties takes no other key, as in a schema
+   * inferred from a type.
+   */
+  static boolean fits(Schema source, Schema target, boolean closed) {
+    if (source.equals(target)) {
+      return true;
+    }
+    if (source.anyOf().isPresent()) {
+      return unionFits(source, source.anyOf().get(), target, closed);
+    }
+    Type.Known type = declaredType(source);
+    if (isNullable(source) || type == Type.Known.NULL) {
+      return acceptsNull(target)
+          && (type == Type.Known.NULL
+              || fits(source.toBuilder().clearNullable().build(), target, closed));
+    }
+    if (type == null && source.enum_().isPresent()) {
+      // An enum lists strings, as a typed one does.
+      return fits(source.toBuilder().type(Type.Known.STRING).build(), target, closed);
+    }
+    if (target.anyOf().isPresent()) {
+      return fits(source, target.toBuilder().clearAnyOf().build(), closed)
+          && target.anyOf().get().stream().anyMatch(branch -> fits(source, branch, closed));
+    }
+    if (type == null) {
+      return true;
+    }
+    if (target.enum_().isPresent()
+        && !(source.enum_().isPresent()
+            && target.enum_().get().containsAll(source.enum_().get()))) {
+      return false;
+    }
+    Type.Known targetType = declaredType(target);
+    if (targetType != null
+        && targetType != type
+        && !(type == Type.Known.INTEGER && targetType == Type.Known.NUMBER)) {
+      return false;
+    }
+    return switch (type) {
+      case ARRAY ->
+          target.items().isEmpty()
+              || source.items().isEmpty()
+              || fits(source.items().get(), target.items().get(), closed);
+      case OBJECT -> objectFits(source, target, closed);
+      default -> true;
+    };
+  }
+
+  /** Whether a union fits: its own part and each of its {@code branches} must. */
+  private static boolean unionFits(
+      Schema source, List<Schema> branches, Schema target, boolean closed) {
+    boolean takesNull = acceptsNull(source);
+    if (takesNull && !acceptsNull(target)) {
+      return false;
+    }
+    // A union that rejects null admits only its branches' other values.
+    List<Schema> values =
+        takesNull
+            ? branches
+            : branches.stream()
+                .filter(branch -> declaredType(branch) != Type.Known.NULL)
+                .map(branch -> branch.toBuilder().clearNullable().build())
+                .collect(toImmutableList());
+    // An untyped union's own keys, items or enum take the one type its branches share, if any.
+    List<Type.@Nullable Known> types =
+        values.stream()
+            .filter(value -> declaredType(value) != Type.Known.NULL)
+            .map(SchemaFit::impliedType)
+            .distinct()
+            .collect(Collectors.toCollection(ArrayList::new));
+    Type.Known type = declaredType(source);
+    Schema.Builder own = source.toBuilder().clearAnyOf();
+    if (type == null
+        && (impliedType(source) != null || source.enum_().isPresent())
+        && types.size() == 1
+        && types.get(0) != null) {
+      own.type(types.get(0));
+    }
+    // An untyped branch takes its union's type.
+    return fits(own.build(), target, closed)
+        && values.stream()
+            .allMatch(
+                value ->
+                    fits(
+                        type != null && declaredType(value) == null
+                            ? value.toBuilder().type(type).build()
+                            : value,
+                        target,
+                        closed));
+  }
+
+  /** Whether {@code source}, an object, fits {@code target}'s keys. */
+  private static boolean objectFits(Schema source, Schema target, boolean closed) {
+    // An object that constrains no keys, such as a map, passes: the build cannot know its keys.
+    if (!declaresKeys(source)) {
+      return true;
+    }
+    Map<String, Schema> given = source.properties().orElse(ImmutableMap.of());
+    Map<String, Schema> wanted = target.properties().orElse(ImmutableMap.of());
+    List<String> required = source.required().orElse(ImmutableList.of());
+    List<String> targetRequired = target.required().orElse(ImmutableList.of());
+    // The source must require each key the target requires, and declare it if the target does.
+    if (!required.containsAll(targetRequired)
+        || targetRequired.stream()
+            .anyMatch(key -> wanted.containsKey(key) && !given.containsKey(key))) {
+      return false;
+    }
+    boolean closedTarget = closed && target.properties().isPresent();
+    if (closedTarget && !wanted.keySet().containsAll(required)) {
+      return false;
+    }
+    return given.entrySet().stream()
+        .allMatch(
+            property ->
+                wanted.containsKey(property.getKey())
+                    ? fits(property.getValue(), wanted.get(property.getKey()), closed)
+                    : !closedTarget);
+  }
+
+  /**
+   * Whether {@code schema} takes null: if nullable, typed NULL, or untyped with no enum, unless it
+   * has anyOf branches and none of them takes null.
+   */
+  private static boolean acceptsNull(Schema schema) {
+    return isNullable(schema)
+        || declaredType(schema) == Type.Known.NULL
+        || (declaredType(schema) == null
+            && schema.enum_().isEmpty()
+            && schema
+                .anyOf()
+                .map(branches -> branches.stream().anyMatch(SchemaFit::acceptsNull))
+                .orElse(true));
+  }
+
+  private static boolean isNullable(Schema schema) {
+    return schema.nullable().orElse(false);
+  }
+
+  private static boolean declaresKeys(Schema schema) {
+    return schema.properties().isPresent() || schema.required().isPresent();
+  }
+
+  /** Returns the type {@code schema} declares, or else the shape its keys or items imply. */
+  private static Type.@Nullable Known impliedType(Schema schema) {
+    Type.Known type = declaredType(schema);
+    if (type != null) {
+      return type;
+    }
+    if (declaresKeys(schema)) {
+      return Type.Known.OBJECT;
+    }
+    return schema.items().isPresent() ? Type.Known.ARRAY : null;
+  }
+
+  private static Type.@Nullable Known declaredType(Schema schema) {
+    Type.Known type = schema.type().map(Type::knownEnum).orElse(null);
+    return type == Type.Known.TYPE_UNSPECIFIED ? null : type;
+  }
+
+  private SchemaFit() {}
+}

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

@@ -210,6 +210,8 @@ public Builder stateSchema(@Nullable Schema stateSchema) {
      *     empty or contains {@code /}, {@code @}, or {@code .}
      * @throws GraphValidationException if the graph is malformed, for example if {@link Start}
      *     cannot reach one of its nodes
+     * @throws WorkflowConfigurationException if the output schema of an edge's source admits a
+     *     value that the input schema of its target does not
      */
     public Workflow build() {
       checkState(name != null, "Workflow.Builder requires name to be set.");