Remove Any from public APIs

.github/workflows/ci.yml

@@ -61,6 +61,11 @@ jobs:
             exit 1
           fi
 
+      - name: Guard - ban Any and asInstanceOf in production sources
+        run: |
+          chmod +x scripts/check-type-safety.sh
+          ./scripts/check-type-safety.sh
+
       - name: compile all
         run: sbt clean compileAll
 

.scalafix.conf

@@ -13,7 +13,8 @@ DisableSyntax.noThrows = true
 DisableSyntax.noNulls = true
 DisableSyntax.noReturns = true
 DisableSyntax.noWhileLoops = true
-# asInstanceOf checked with whitelist in CI (scalafix doesn't support per-file excludes)
+# asInstanceOf and Any are banned by scripts/check-type-safety.sh, which runs in CI. They need a per-file
+# allowlist for the reflection boundaries, and DisableSyntax has no per-file excludes.
 DisableSyntax.noIsInstanceOf = true
 DisableSyntax.noXml = true
 DisableSyntax.regex = [
@@ -67,6 +68,5 @@ OrganizeImports.groupedImports = Merge
 # default Ascii order the two formatters each undo the other's selector order and neither check can pass.
 OrganizeImports.importSelectorsOrder = SymbolsFirst
 
-# Note: Scalafix DisableSyntax doesn't support per-file excludes
-# Whitelists for asInstanceOf and scala.util.Try are enforced via shell scripts in CI
-# Examples directory is excluded from CI checks via grep filter (demo code, not production)
+# Note: Scalafix DisableSyntax doesn't support per-file excludes, so the checks that need an allowlist live
+# in scripts/check-type-safety.sh instead. Examples are demo code and excluded from both.

modules/connectors/src/main/scala/com/flowforge/connectors/safety/ConnectorErrorMapper.scala

@@ -2,7 +2,7 @@ package com.flowforge.connectors.safety
 
 import com.flowforge.core.safety.ErrorMapper
 import com.flowforge.core.types.FlowForgeError.{ ConfigurationError, ValidationError }
-import com.flowforge.core.types.SystemError
+import com.flowforge.core.types.{ ContextValue, SystemError }
 
 /**
  * Connector-focused ErrorMapper.
@@ -13,17 +13,25 @@ import com.flowforge.core.types.SystemError
 object ConnectorErrorMapper {
   implicit val connectorMapper: ErrorMapper = {
     case e: java.nio.file.NoSuchFileException =>
-      ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "NoSuchFile"))
+      ValidationError(
+        s"File not found: ${e.getMessage}",
+        None,
+        context = Map("cause" -> ContextValue.Text("NoSuchFile")),
+      )
         .withCause(e)
     case e: java.io.FileNotFoundException =>
-      ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "FileNotFound"))
+      ValidationError(
+        s"File not found: ${e.getMessage}",
+        None,
+        context = Map("cause" -> ContextValue.Text("FileNotFound")),
+      )
         .withCause(e)
     case e: java.io.IOException =>
       SystemError.ServiceUnavailable(serviceName = "filesystem", message = e.getMessage, cause = Some(e))
     case e: java.sql.SQLException =>
       ConfigurationError(
         s"JDBC error: ${e.getMessage}",
-        context = Map("sqlState" -> String.valueOf(e.getSQLState)),
+        context = Map("sqlState" -> ContextValue.Text(String.valueOf(e.getSQLState))),
       ).withCause(e)
     case other => ErrorMapper.default(other)
   }

modules/contracts/src/main/scala/com/flowforge/contracts/DataContract.scala

@@ -76,12 +76,11 @@ object FieldType {
 /** Algebraic data type for field‑level constraints. */
 sealed trait FieldConstraint
 object FieldConstraint {
-  case class MinLength(value: Int)                           extends FieldConstraint
-  case class MaxLength(value: Int)                           extends FieldConstraint
-  case class Range(min: Double, max: Double)                 extends FieldConstraint
-  case class Pattern(regex: Regex)                           extends FieldConstraint
-  case class OneOf(values: Set[String])                      extends FieldConstraint
-  case class Custom(name: String, validator: Any => Boolean) extends FieldConstraint
+  case class MinLength(value: Int)           extends FieldConstraint
+  case class MaxLength(value: Int)           extends FieldConstraint
+  case class Range(min: Double, max: Double) extends FieldConstraint
+  case class Pattern(regex: Regex)           extends FieldConstraint
+  case class OneOf(values: Set[String])      extends FieldConstraint
 }
 
 /** Composable validation rules evaluated at runtime. */
@@ -100,19 +99,21 @@ object RuleSeverity {
 
 /** Common validation rules provided out of the box. */
 object ValidationRules {
-  def nonNull[A](fieldName: String)(extract: A => Any): ValidationRule[A] =
+  def nonNull[A](fieldName: String)(extract: A => AnyRef): ValidationRule[A] =
     new ValidationRule[A] {
       val name = s"nonNull($fieldName)"
       def validate(data: A): ValidatedNel[ContractViolation, Unit] =
-        // `extract` returns Any because it reaches into caller data this module does not control, so a null
-        // genuinely can arrive here. Option is the narrowest way to ask without naming the literal.
+        // `extract` returns a reference because that is the only thing a null can be, and this rule exists to
+        // catch a null arriving from caller data the module does not control. It was `A => Any`, which also
+        // accepted a primitive the check could never fail for. `Option` is the narrowest way to ask without
+        // naming the literal.
         Option(extract(data)) match {
           case Some(_) => ().validNel
           case None    => ContractViolation.NullValue(fieldName).invalidNel
         }
     }
 
-  def unique[A](fieldName: String)(extract: A => Any): ValidationRule[A] =
+  def unique[A, B](fieldName: String)(@annotation.unused extract: A => B): ValidationRule[A] =
     new ValidationRule[A] {
       val name = s"unique($fieldName)"
       def validate(data: A): ValidatedNel[ContractViolation, Unit] =

modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala

@@ -455,13 +455,18 @@ object DataAlgebra {
 
   /**
    * Data profile for understanding dataset characteristics.
+   *
+   * `statistics` is numeric because a profile's statistics are: min, max, mean, stddev, percentiles. It was
+   * `Map[String, Any]`, and every one of the three implementations passes `Map.empty`, so nothing was relying
+   * on the wider type. A profile that needs to report a non-numeric statistic should say what that is rather
+   * than reopening the map to anything.
    */
   case class DataProfile[A](
     recordCount: Long,
     nullCount: Long,
     distinctCount: Long,
     schema: DataSchema,
-    statistics: Map[String, Any])
+    statistics: Map[String, Double])
 
   /**
    * Schema migration for evolution.

modules/core/src/main/scala/com/flowforge/core/algebra/EnterpriseTableAlgebra.scala

@@ -40,7 +40,7 @@ import cats.data.{ NonEmptyList, ValidatedNel }
 import cats.effect.{ MonadCancel, Resource }
 import cats.implicits._
 import com.flowforge.core.types.RefinedTypes.{ FieldName, TableName }
-import com.flowforge.core.types.{ ErrorCategory, ErrorSeverity, FlowForgeError }
+import com.flowforge.core.types.{ ContextValue, ErrorCategory, ErrorSeverity, FlowForgeError }
 
 import java.time.Instant
 import scala.concurrent.duration.FiniteDuration
@@ -486,31 +486,31 @@ case class TableNotFound(tableName: TableName) extends TableError {
   val message       = s"Table '${tableName.value}' not found"
   val category      = ErrorCategory.System
   val severity      = ErrorSeverity.Error
-  val context       = Map("tableName" -> tableName.value)
+  val context       = Map("tableName" -> ContextValue.Text(tableName.value))
   val cause         = None
   val timestamp     = java.time.Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = false
   val recoveryHints = List("Check table name spelling", "Verify table exists", "Check permissions")
 
-  def withContext(additionalContext: Map[String, Any]) = this
-  def withCause(underlyingCause: Throwable)            = this
+  def withContext(additionalContext: Map[String, ContextValue]) = this
+  def withCause(underlyingCause: Throwable)                     = this
 }
 
 sealed trait PartitionError extends FlowForgeError
 case class PartitionNotFound(partitionSpec: PartitionSpec) extends PartitionError {
   val message       = s"Partition '${partitionSpec.toPartitionPath}' not found"
   val category      = ErrorCategory.System
   val severity      = ErrorSeverity.Error
-  val context       = Map("partitionSpec" -> partitionSpec.toPartitionPath)
+  val context       = Map("partitionSpec" -> ContextValue.Text(partitionSpec.toPartitionPath))
   val cause         = None
   val timestamp     = java.time.Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = false
   val recoveryHints = List("Check partition specification", "Verify partition exists")
 
-  def withContext(additionalContext: Map[String, Any]) = this
-  def withCause(underlyingCause: Throwable)            = this
+  def withContext(additionalContext: Map[String, ContextValue]) = this
+  def withCause(underlyingCause: Throwable)                     = this
 }
 
 sealed trait BlobError extends FlowForgeError
@@ -519,18 +519,22 @@ case class BlobAccessError(
   blobName: String,
   reason: String)
     extends BlobError {
-  val message       = s"Cannot access blob '$blobName' in bucket '$bucketName': $reason"
-  val category      = ErrorCategory.System
-  val severity      = ErrorSeverity.Error
-  val context       = Map("bucketName" -> bucketName, "blobName" -> blobName, "reason" -> reason)
+  val message  = s"Cannot access blob '$blobName' in bucket '$bucketName': $reason"
+  val category = ErrorCategory.System
+  val severity = ErrorSeverity.Error
+  val context = Map(
+    "bucketName" -> ContextValue.Text(bucketName),
+    "blobName"   -> ContextValue.Text(blobName),
+    "reason"     -> ContextValue.Text(reason),
+  )
   val cause         = None
   val timestamp     = java.time.Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = true
   val recoveryHints = List("Check cloud permissions", "Verify bucket exists", "Retry operation")
 
-  def withContext(additionalContext: Map[String, Any]) = this
-  def withCause(underlyingCause: Throwable)            = this
+  def withContext(additionalContext: Map[String, ContextValue]) = this
+  def withCause(underlyingCause: Throwable)                     = this
 }
 
 // ===============================

modules/core/src/main/scala/com/flowforge/core/algebra/TypeClasses.scala

@@ -753,17 +753,17 @@ object EncodingHints {
  */
 sealed trait EncodingError extends FlowForgeError
 case class UnsupportedFormat(format: DataFormat, dataType: String) extends EncodingError {
-  val message       = s"Format $format is not supported for data type $dataType"
-  val category      = ErrorCategory.Validation
-  val severity      = ErrorSeverity.Error
-  val context       = Map("format" -> format.toString, "dataType" -> dataType)
-  val cause         = None
+  val message  = s"Format $format is not supported for data type $dataType"
+  val category = ErrorCategory.Validation
+  val severity = ErrorSeverity.Error
+  val context = Map("format" -> ContextValue.Text(format.toString), "dataType" -> ContextValue.Text(dataType))
+  val cause   = None
   val timestamp     = Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = false
   val recoveryHints = List("Use a supported format", "Implement custom encoder")
 
-  def withContext(additionalContext: Map[String, Any]): EncodingError =
+  def withContext(additionalContext: Map[String, ContextValue]): EncodingError =
     copy() // Simplified for brevity
   def withCause(underlyingCause: Throwable): EncodingError =
     copy() // Simplified for brevity
@@ -777,14 +777,14 @@ case class CorruptedData(details: String) extends DecodingError {
   val message       = s"Data is corrupted: $details"
   val category      = ErrorCategory.Validation
   val severity      = ErrorSeverity.Error
-  val context       = Map("details" -> details)
+  val context       = Map("details" -> ContextValue.Text(details))
   val cause         = None
   val timestamp     = Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = false
   val recoveryHints = List("Check data source", "Re-download data", "Use backup data")
 
-  def withContext(additionalContext: Map[String, Any]): DecodingError =
+  def withContext(additionalContext: Map[String, ContextValue]): DecodingError =
     copy() // Simplified for brevity
   def withCause(underlyingCause: Throwable): DecodingError =
     copy() // Simplified for brevity
@@ -797,16 +797,17 @@ sealed trait SchemaError extends FlowForgeError
 case class SchemaIncompatible(expected: DataSchema, actual: DataSchema) extends SchemaError {
   val message =
     s"Schema incompatible: expected ${expected.fields.length} fields, got ${actual.fields.length}"
-  val category      = ErrorCategory.Validation
-  val severity      = ErrorSeverity.Error
-  val context       = Map("expected" -> expected.toString, "actual" -> actual.toString)
+  val category = ErrorCategory.Validation
+  val severity = ErrorSeverity.Error
+  val context =
+    Map("expected" -> ContextValue.Text(expected.toString), "actual" -> ContextValue.Text(actual.toString))
   val cause         = None
   val timestamp     = Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = false
   val recoveryHints = List("Update schema", "Enable schema evolution", "Transform data")
 
-  def withContext(additionalContext: Map[String, Any]): SchemaError =
+  def withContext(additionalContext: Map[String, ContextValue]): SchemaError =
     copy() // Simplified for brevity
   def withCause(underlyingCause: Throwable): SchemaError =
     copy() // Simplified for brevity
@@ -820,14 +821,14 @@ case class SerializationFailed(reason: String) extends SerializationError {
   val message       = s"Serialization failed: $reason"
   val category      = ErrorCategory.System
   val severity      = ErrorSeverity.Error
-  val context       = Map("reason" -> reason)
+  val context       = Map("reason" -> ContextValue.Text(reason))
   val cause         = None
   val timestamp     = Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = true
   val recoveryHints = List("Retry operation", "Check data format", "Use alternative serializer")
 
-  def withContext(additionalContext: Map[String, Any]): SerializationError =
+  def withContext(additionalContext: Map[String, ContextValue]): SerializationError =
     copy() // Simplified for brevity
   def withCause(underlyingCause: Throwable): SerializationError =
     copy() // Simplified for brevity
@@ -841,14 +842,14 @@ case class RuleViolation(ruleName: String, details: String) extends ContractViol
   val message       = s"Contract rule '$ruleName' violated: $details"
   val category      = ErrorCategory.Business
   val severity      = ErrorSeverity.Error
-  val context       = Map("rule" -> ruleName, "details" -> details)
+  val context       = Map("rule" -> ContextValue.Text(ruleName), "details" -> ContextValue.Text(details))
   val cause         = None
   val timestamp     = Instant.now()
   val errorId       = java.util.UUID.randomUUID().toString
   val isRetryable   = false
   val recoveryHints = List("Fix data quality", "Update contract rules", "Contact data owner")
 
-  def withContext(additionalContext: Map[String, Any]): ContractViolation =
+  def withContext(additionalContext: Map[String, ContextValue]): ContractViolation =
     copy() // Simplified for brevity
   def withCause(underlyingCause: Throwable): ContractViolation =
     copy() // Simplified for brevity

modules/core/src/main/scala/com/flowforge/core/patterns/CommonValidations.scala

@@ -83,7 +83,6 @@ package com.flowforge.core.patterns
 
 import cats.syntax.all._
 import com.flowforge.core.patterns.ValidationTypes._
-import com.flowforge.core.types.RefinedTypes._
 import com.flowforge.core.types._
 
 /**
@@ -146,10 +145,22 @@ object CommonValidations {
     if (data.nonEmpty) {
       valid(data)
     } else {
-      val violation = QualityConstraint.NotNull(
-        FieldName.unsafeFrom("data"),
+      // The cast here was hiding a mismatch in the error type, not the value type: this built a
+      // `QualityConstraint.NotNull`, which is a constraint rather than a violation, so `invalid` produced a
+      // `ValidatedNel[QualityConstraint, _]` and the cast forced it into the declared result. Reporting an
+      // actual `QualityViolation` is what the signature promised all along.
+      //
+      // `severity` is passed because `QualityViolation` defaults it to `Warning`, while the constraint this
+      // replaces defaulted to `Error`. This branch rejects the dataset, so anything routing or alerting on
+      // severity has to see a failure rather than a warning.
+      invalid(
+        ValidationError.QualityViolation(
+          constraint = "nonEmpty",
+          violatedValue = "data",
+          message = "Dataset is empty",
+          severity = ErrorSeverity.Error,
+        ),
       )
-      invalid(violation).asInstanceOf[QualityValidationResult[List[A]]]
     }
 
   /**

modules/core/src/main/scala/com/flowforge/core/patterns/DataQualityValidation.scala

@@ -16,9 +16,15 @@ object DataQualityValidation {
 
   /**
    * Validate data freshness - ensure data is not older than specified duration.
+   *
+   * Takes the data it is vouching for and hands it back, the same way [[completeness]] does, so the check
+   * composes into a validation chain. It used to claim `QualityValidationResult[A]` while having no `A` to
+   * return, and bridged the gap with `().asInstanceOf[A]`. That threw `ClassCastException` on the success
+   * path for every `A` except `Unit`, which is why every caller pinned `freshness[Unit]`.
    */
   def freshness[A](
     fieldName: String,
+    value: A,
     timestamp: Instant,
     maxAge: FiniteDuration,
   ): QualityValidationResult[A] = {
@@ -27,9 +33,7 @@ object DataQualityValidation {
     val maxAgeInMillis = maxAge.toMillis
 
     if (age.toMillis <= maxAgeInMillis) {
-      // Can't return A without having an A - this method signature needs fixing
-      // For now, return a unit value cast to A as placeholder
-      ().asInstanceOf[A].validNel
+      value.validNel
     } else {
       val violation = QualityViolation(
         constraint = "freshness",