Fix contract leaf type checking and policy dispatch

modules/core/src/main/scala-2/com/flowforge/core/contracts/derive/ShapeDerivation.scala

@@ -1,6 +1,7 @@
 // scalafix:off DisableSyntax.throw
 package com.flowforge.core.contracts.derive
 
+import com.flowforge.core.contracts.internal.TypeShape
 import magnolia1._
 
 import scala.language.experimental.macros
@@ -13,16 +14,32 @@ trait ShapeDerivation {
     new Shape[T] {
       val fields: List[Field] =
         caseClass.parameters.toList.map { p =>
-          val full = p.typeName.full // Magnolia 1 (Scala 2) exposes typeName
+          val typeName = p.typeName // Magnolia 1 (Scala 2) exposes typeName
           Field(
             name = p.label,
-            tpe = full,
+            tpe = render(typeName),
             hasDefault = p.default.isDefined,
-            isOptional = full.startsWith("scala.Option["),
+            isOptional = typeName.full == optionName,
           )
         }
     }
 
+  private val optionName = "scala.Option"
+
+  /**
+   * A field type rendered the way Scala 3 reflection renders it.
+   *
+   * Magnolia reports a type name and its arguments separately, so `Option[String]` arrives as `scala.Option`
+   * carrying one argument rather than as one rendered string. Reading only the name therefore dropped the
+   * element type, and made every field read as required, because the name on its own never contains the
+   * `Option[` that optionality used to be decided by.
+   */
+  private def render(typeName: TypeName): String = {
+    val name = TypeShape.simpleName(typeName.full)
+    if (typeName.typeArguments.isEmpty) name
+    else s"$name[${typeName.typeArguments.map(render).mkString(", ")}]"
+  }
+
   // Keep v1.0 scope to products only; abort on sums for now
   def split[T](ctx: SealedTrait[Typeclass, T]): Typeclass[T] =
     throw new IllegalArgumentException(

modules/core/src/main/scala-2/com/flowforge/core/contracts/internal/ContractMacros.scala

@@ -54,8 +54,30 @@ object ContractMacros {
       def isAtomicKey(t: Type): Boolean =
         t =:= typeOf[String] || t =:= typeOf[Int] || t =:= typeOf[Long] ||
           t =:= typeOf[Short] || t =:= typeOf[Byte] || t =:= typeOf[Boolean]
+
+      def isTuple(t: Type): Boolean =
+        t.typeSymbol.fullName.startsWith("scala.Tuple")
     }
 
+    /**
+     * A type this macro cannot look inside, compared by its name.
+     *
+     * There used to be a closed list of leaf types here, on the reasoning that a type outside it is one no
+     * sink can write. That is not something a contract can know: a sink takes the writer as a function, so
+     * whether a `UUID` or a domain enum can be written is decided by the writer the caller supplies, not by
+     * this list. Rejecting an unlisted leaf therefore turned away pipelines that were fine. Comparing it by
+     * name still catches the drift that is this macro's job, because a leaf that changes type changes its
+     * name.
+     */
+    def opaqueLeaf(t: Type): PrimitiveShape = PrimitiveShape(TypeShape.simpleName(t.toString))
+
+    def unsupportedTuple(t: Type): Nothing =
+      c.abort(
+        c.enclosingPosition,
+        s"Unsupported tuple in SchemaConforms derivation: ${t.toString}. " +
+          "A tuple has no field names to compare, so use a case class instead.",
+      )
+
     // TypeShape builder - pure functional approach
     object ShapeBuilder {
       def buildTypeShape(tpe: Type, inField: Boolean = false): TypeShape = {
@@ -76,13 +98,13 @@ object ContractMacros {
                     s"Unsupported Map key type: ${k.toString}. Allowed: String, Int, Long, Short, Byte, Boolean",
                   )
                 }
-                MapShape(PrimitiveShape(k.toString), buildTypeShape(v))
+                MapShape(PrimitiveShape(TypeShape.simpleName(k.toString)), buildTypeShape(v))
             }.getOrElse {
-              if (isCaseClass(tpe)) {
-                buildStructShape(tpe)
-              } else {
-                PrimitiveShape(tpe.toString)
-              }
+              // Tuples are checked first because every TupleN is itself a case class. Reading one as a
+              // struct of `_1`, `_2` would make positional junk look like a named schema.
+              if (isTuple(tpe)) unsupportedTuple(tpe)
+              else if (isCaseClass(tpe)) buildStructShape(tpe)
+              else opaqueLeaf(tpe)
             }
           }
         }
@@ -107,11 +129,30 @@ object ContractMacros {
       }
     }
 
+    // Matched by subtyping rather than by the rendered type name, so that a policy named as the trait
+    // (SchemaPolicy.Backward) and the same policy named as the case object (SchemaPolicy.Backward.type)
+    // resolve to the same rules. The policy traits are disjoint, so at most one entry can match.
+    val known: List[(Type, SchemaPolicy)] = List(
+      typeOf[SchemaPolicy.Exact]            -> SchemaPolicy.Exact,
+      typeOf[SchemaPolicy.ExactUnordered]   -> SchemaPolicy.ExactUnordered,
+      typeOf[SchemaPolicy.ExactUnorderedCI] -> SchemaPolicy.ExactUnorderedCI,
+      typeOf[SchemaPolicy.ExactOrdered]     -> SchemaPolicy.ExactOrdered,
+      typeOf[SchemaPolicy.ExactOrderedCI]   -> SchemaPolicy.ExactOrderedCI,
+      typeOf[SchemaPolicy.ExactByPosition]  -> SchemaPolicy.ExactByPosition,
+      typeOf[SchemaPolicy.Backward]         -> SchemaPolicy.Backward,
+      typeOf[SchemaPolicy.Forward]          -> SchemaPolicy.Forward,
+      typeOf[SchemaPolicy.Full]             -> SchemaPolicy.Full,
+    )
+
+    val rules = known.collectFirst { case (t, policy) if weakTypeOf[P] <:< t => ComparisonRules.of(policy) }
+      .getOrElse(ComparisonRules.strictest)
+
     ShapeDiff
       .report(
         policyName = weakTypeOf[P].toString,
         outName = weakTypeOf[Out].toString,
         contractName = weakTypeOf[Contract].toString,
+        rules = rules,
         out = ShapeBuilder.buildTypeShape(weakTypeOf[Out]),
         contract = ShapeBuilder.buildTypeShape(weakTypeOf[Contract]),
       )

modules/core/src/main/scala-3/com/flowforge/core/contracts/derive/ShapeDerivation.scala

@@ -1,6 +1,6 @@
 package com.flowforge.core.contracts.derive
 
-import com.flowforge.core.contracts.internal.TypeShapes
+import com.flowforge.core.contracts.internal.{ TypeShape, TypeShapes }
 
 import scala.quoted.*
 
@@ -24,7 +24,7 @@ object ShapeDerivation {
       case (name, fieldType, hasDefault) =>
         // Read off the type out here: inside the quote a different Quotes instance is in scope, so a
         // `TypeRepr` from this one does not typecheck there.
-        val rendered   = Expr(fieldType.show)
+        val rendered   = Expr(TypeShape.simpleName(fieldType.show))
         val isOptional = Expr(TypeShapes.isOption(fieldType.dealias))
         '{
           Field(

modules/core/src/main/scala-3/com/flowforge/core/contracts/internal/ContractMacros.scala

@@ -24,11 +24,39 @@ object ContractMacros {
         policyName = TypeRepr.of[P].show,
         outName = TypeRepr.of[Out].show,
         contractName = TypeRepr.of[Contract].show,
+        rules = rulesOf[P],
         out = TypeShapes.of(TypeRepr.of[Out]),
         contract = TypeShapes.of(TypeRepr.of[Contract]),
       )
       .foreach(message => report.errorAndAbort(message))
 
     '{ new SchemaConforms[Out, Contract, P] {} }
   }
+
+  /**
+   * The comparison rules the policy type `P` stands for.
+   *
+   * Matched by subtyping rather than by the rendered type name, so that a policy named as the trait
+   * (`SchemaPolicy.Backward`) and the same policy named as the case object (`SchemaPolicy.Backward.type`)
+   * resolve to the same rules. The policy traits are disjoint, so at most one branch can match.
+   */
+  private def rulesOf[P <: SchemaPolicy: Type](using q: Quotes): ComparisonRules = {
+    import q.reflect.*
+    val requested = TypeRepr.of[P]
+
+    val known: List[(TypeRepr, SchemaPolicy)] = List(
+      TypeRepr.of[SchemaPolicy.Exact]            -> SchemaPolicy.Exact,
+      TypeRepr.of[SchemaPolicy.ExactUnordered]   -> SchemaPolicy.ExactUnordered,
+      TypeRepr.of[SchemaPolicy.ExactUnorderedCI] -> SchemaPolicy.ExactUnorderedCI,
+      TypeRepr.of[SchemaPolicy.ExactOrdered]     -> SchemaPolicy.ExactOrdered,
+      TypeRepr.of[SchemaPolicy.ExactOrderedCI]   -> SchemaPolicy.ExactOrderedCI,
+      TypeRepr.of[SchemaPolicy.ExactByPosition]  -> SchemaPolicy.ExactByPosition,
+      TypeRepr.of[SchemaPolicy.Backward]         -> SchemaPolicy.Backward,
+      TypeRepr.of[SchemaPolicy.Forward]          -> SchemaPolicy.Forward,
+      TypeRepr.of[SchemaPolicy.Full]             -> SchemaPolicy.Full,
+    )
+
+    known.collectFirst { case (tpe, policy) if requested <:< tpe => ComparisonRules.of(policy) }
+      .getOrElse(ComparisonRules.strictest)
+  }
 }

modules/core/src/main/scala-3/com/flowforge/core/contracts/internal/TypeShapes.scala

@@ -62,17 +62,48 @@ object TypeShapes {
             s"Unsupported Map key type: ${key.show}. Allowed: String, Int, Long, Short, Byte, Boolean",
           )
         }
-        MapShape(PrimitiveShape(key.show), shapeOf(TypeRepr.of[v], inField = false))
+        MapShape(PrimitiveShape(TypeShape.simpleName(key.show)), shapeOf(TypeRepr.of[v], inField = false))
 
       case '[Seq[a]]   => SequenceShape(shapeOf(TypeRepr.of[a], inField = false))
       case '[Set[a]]   => SequenceShape(shapeOf(TypeRepr.of[a], inField = false))
       case '[Array[a]] => SequenceShape(shapeOf(TypeRepr.of[a], inField = false))
 
-      case _ =>
-        if (isCaseClass(t)) StructShape(fieldsOf(t)) else PrimitiveShape(t.show)
+      case _ => structOrLeaf(t)
     }
   }
 
+  private def structOrLeaf(using q: Quotes)(tpe: q.reflect.TypeRepr): TypeShape = {
+    import q.reflect.*
+    // Tuples are checked first because every TupleN is itself a case class. Reading one as a struct of
+    // `_1`, `_2` would make positional junk look like a named schema, so a tuple is rejected rather than
+    // reinterpreted.
+    if (tpe <:< TypeRepr.of[Tuple]) unsupportedTuple(tpe)
+    else if (isCaseClass(tpe)) StructShape(fieldsOf(tpe))
+    else opaqueLeaf(tpe)
+  }
+
+  /**
+   * A type this macro cannot look inside, compared by its name.
+   *
+   * There used to be a closed list of leaf types here, on the reasoning that a type outside it is one no sink
+   * can write. That is not something a contract can know: a sink takes the writer as a function, so whether a
+   * `UUID` or a domain enum can be written is decided by the writer the caller supplies, not by this list.
+   * Rejecting an unlisted leaf therefore turned away pipelines that were fine. Comparing it by name still
+   * catches the drift that is this macro's job, because a leaf that changes type changes its name.
+   */
+  private def opaqueLeaf(using q: Quotes)(tpe: q.reflect.TypeRepr): PrimitiveShape = {
+    import q.reflect.*
+    PrimitiveShape(TypeShape.simpleName(tpe.show))
+  }
+
+  private def unsupportedTuple(using q: Quotes)(tpe: q.reflect.TypeRepr): Nothing = {
+    import q.reflect.*
+    report.errorAndAbort(
+      s"Unsupported tuple in SchemaConforms derivation: ${tpe.show}. " +
+        "A tuple has no field names to compare, so use a case class instead.",
+    )
+  }
+
   private def fieldsOf(using q: Quotes)(tpe: q.reflect.TypeRepr): List[FieldShape] = {
     import q.reflect.*
     params(tpe).map {

modules/core/src/main/scala/com/flowforge/core/contracts/internal/ComparisonRules.scala

@@ -0,0 +1,93 @@
+package com.flowforge.core.contracts.internal
+
+import com.flowforge.core.contracts.SchemaPolicy
+
+/** How producer fields are lined up against contract fields before anything is compared. */
+sealed trait FieldMatching
+
+object FieldMatching {
+  case object ByName        extends FieldMatching
+  case object ByNameOrdered extends FieldMatching
+  case object ByPosition    extends FieldMatching
+}
+
+/** Whether a field name has to match exactly or only up to case. */
+sealed trait NameCasing
+
+object NameCasing {
+  case object Sensitive   extends NameCasing
+  case object Insensitive extends NameCasing
+}
+
+/** Which differences survive once the fields are lined up. */
+sealed trait Tolerance
+
+object Tolerance {
+
+  /** Every difference is drift. */
+  case object Strict extends Tolerance
+
+  /** The producer may add fields, and may omit a contract field that is optional or has a default. */
+  case object Backward extends Tolerance
+
+  /** The producer may omit fields, so it can be a subset of the contract. */
+  case object Forward extends Tolerance
+
+  /** Escape hatch: comparison still runs, but nothing it finds is drift. */
+  case object Permissive extends Tolerance
+}
+
+/**
+ * A policy flattened into the three independent questions comparison actually asks.
+ *
+ * [[com.flowforge.core.contracts.SchemaPolicy]] is the vocabulary users write; this is the vocabulary the
+ * comparison reads. Keeping them apart is what lets the comparison be a pure function, and it makes the
+ * contradictory combinations unconstructible: there is no way to ask for "by position and also ordered by
+ * name", which a set of independent booleans would have allowed.
+ */
+final case class ComparisonRules(
+  matching: FieldMatching,
+  casing: NameCasing,
+  tolerance: Tolerance) {
+
+  /** Field names reduced to the form this policy compares them in. */
+  def normalize(name: String): String = casing match {
+    case NameCasing.Sensitive   => name
+    case NameCasing.Insensitive => name.toLowerCase
+  }
+
+  def sameName(left: String, right: String): Boolean = normalize(left) == normalize(right)
+}
+
+object ComparisonRules {
+
+  import FieldMatching._
+  import NameCasing._
+
+  /**
+   * The rules each policy stands for.
+   *
+   * Total over the policy ADT on purpose: a new policy that nobody mapped is a compile error here, not a
+   * silent fallback to some default at the call site.
+   */
+  def of(policy: SchemaPolicy): ComparisonRules = policy match {
+    case SchemaPolicy.Exact            => ComparisonRules(ByName, Sensitive, Tolerance.Strict)
+    case SchemaPolicy.ExactUnordered   => ComparisonRules(ByName, Sensitive, Tolerance.Strict)
+    case SchemaPolicy.ExactUnorderedCI => ComparisonRules(ByName, Insensitive, Tolerance.Strict)
+    case SchemaPolicy.ExactOrdered     => ComparisonRules(ByNameOrdered, Sensitive, Tolerance.Strict)
+    case SchemaPolicy.ExactOrderedCI   => ComparisonRules(ByNameOrdered, Insensitive, Tolerance.Strict)
+    case SchemaPolicy.ExactByPosition  => ComparisonRules(ByPosition, Sensitive, Tolerance.Strict)
+    case SchemaPolicy.Backward         => ComparisonRules(ByName, Sensitive, Tolerance.Backward)
+    case SchemaPolicy.Forward          => ComparisonRules(ByName, Sensitive, Tolerance.Forward)
+    case SchemaPolicy.Full             => ComparisonRules(ByName, Sensitive, Tolerance.Permissive)
+  }
+
+  /**
+   * What to compare under when the policy type is not one of the known policies.
+   *
+   * Reached only for an abstract `P <: SchemaPolicy`, which is generic code that has not fixed its policy
+   * yet. Strict name matching is the safe default: it can report drift that a looser policy would have
+   * accepted, but it never passes a producer that the requested policy would have rejected.
+   */
+  val strictest: ComparisonRules = ComparisonRules(ByName, Sensitive, Tolerance.Strict)
+}

modules/core/src/main/scala/com/flowforge/core/contracts/internal/SchemaAST.scala

@@ -21,6 +21,51 @@ object TypeShape {
       extends TypeShape
   final case class StructShape(fields: List[FieldShape]) extends TypeShape
 
+  /**
+   * Prefixes that carry no information for a reader of a drift report.
+   *
+   * Matched anywhere in the name rather than only at the front, because a type argument carries its own
+   * prefix: `Option[scala.Predef.String]` has to reduce to `Option[String]`, which stripping a leading prefix
+   * cannot do. The uppercase lookahead is what keeps the reduction to type names only, so
+   * `scala.collection.immutable.Map` is left alone instead of becoming `collection.immutable.Map`. The
+   * alternation is ordered longest-first so that `scala.Predef.` wins over the `scala.` that starts it.
+   */
+  private val redundantPrefixes = """(?:scala\.Predef\.|scala\.|java\.lang\.)(?=[A-Z])""".r
+
+  /**
+   * A rendered type name without the prefixes a reader does not need.
+   *
+   * The two macros print the same type differently - Scala 3 reflection says `scala.Predef.String` where
+   * Scala 2 says `java.lang.String` - so a drift report would read differently depending on which compiler
+   * produced it. Names are reduced here, at the one place both macros build a shape through.
+   */
+  def simpleName(rendered: String): String = redundantPrefixes.replaceAllIn(rendered, "")
+
+  /**
+   * A shape as it reads in a drift report.
+   *
+   * Deliberately not the Scala source spelling of the type: what a report needs to say is what the comparison
+   * actually looked at, which is the normalized shape rather than the collection class the producer happened
+   * to use.
+   */
+  def render(shape: TypeShape): String = shape match {
+    case PrimitiveShape(name)          => simpleName(name)
+    case OptionalShape(inner)          => s"optional ${render(inner)}"
+    case SequenceShape(elem)           => s"seq[${render(elem)}]"
+    case MapShape(key, value)          => s"map[${render(key)} -> ${render(value)}]"
+    case FieldShape(name, inner, _, _) => s"$name: ${render(inner)}"
+    case StructShape(fields) =>
+      fields.map(f => s"${f.name}: ${render(f.shape)}").mkString("{", ", ", "}")
+  }
+
+  /** A field as it reads in a drift report, with what the contract says about absent values. */
+  def renderField(field: FieldShape): String = {
+    val optional = if (field.isOptional) " (optional)" else ""
+    val default  = if (field.hasDefault) " (default)" else ""
+    s"${render(field.shape)}$optional$default"
+  }
+
+  /** Kept for the legacy [[SchemaAST]] alias only; drift reports go through [[render]]. */
   def pretty(shape: TypeShape): String = shape match {
     case PrimitiveShape(name) => name
     case SequenceShape(elem)  => s"List[${pretty(elem)}]"