diff --git a/core/src/main/scala-2/caliban/schema/SchemaDerivation.scala b/core/src/main/scala-2/caliban/schema/SchemaDerivation.scala
index bac5dbf92..a5c595613 100644
--- a/core/src/main/scala-2/caliban/schema/SchemaDerivation.scala
+++ b/core/src/main/scala-2/caliban/schema/SchemaDerivation.scala
@@ -11,6 +11,21 @@ import scala.language.experimental.macros
 
 trait CommonSchemaDerivation[R] {
 
+  case class DerivationConfig(
+    /**
+     * Whether to enable the `SemanticNonNull` feature on derivation.
+     * It is currently disabled by default since it is not yet stable.
+     */
+    enableSemanticNonNull: Boolean = false
+  )
+
+  /**
+   * Returns a configuration object that can be used to customize the derivation behavior.
+   *
+   * Override this method to customize the configuration.
+   */
+  def config: DerivationConfig = DerivationConfig()
+
   /**
    * Default naming logic for input types.
    * This is needed to avoid a name clash between a type used as an input and the same type used as an output.
@@ -80,21 +95,31 @@ trait CommonSchemaDerivation[R] {
           ctx.parameters
             .filterNot(_.annotations.exists(_ == GQLExcluded()))
             .map { p =>
-              val isOptional = {
+              val (isNullable, isSemanticNonNull) = {
                 val hasNullableAnn = p.annotations.contains(GQLNullable())
                 val hasNonNullAnn  = p.annotations.contains(GQLNonNullable())
-                !hasNonNullAnn && (hasNullableAnn || p.typeclass.optional)
+
+                if (hasNonNullAnn) (false, false)
+                else if (hasNullableAnn) (true, false)
+                else if (p.typeclass.optional) (true, !p.typeclass.nullable)
+                else (false, false)
               }
               Types.makeField(
                 getName(p),
                 getDescription(p),
                 p.typeclass.arguments,
                 () =>
-                  if (isOptional) p.typeclass.toType_(isInput, isSubscription)
+                  if (isNullable) p.typeclass.toType_(isInput, isSubscription)
                   else p.typeclass.toType_(isInput, isSubscription).nonNull,
                 p.annotations.collectFirst { case GQLDeprecated(_) => () }.isDefined,
                 p.annotations.collectFirst { case GQLDeprecated(reason) => reason },
-                Option(p.annotations.collect { case GQLDirective(dir) => dir }.toList).filter(_.nonEmpty)
+                Option(
+                  p.annotations.collect { case GQLDirective(dir) => dir }.toList ++ {
+                    if (config.enableSemanticNonNull && isSemanticNonNull)
+                      Some(SchemaUtils.SemanticNonNull)
+                    else None
+                  }
+                ).filter(_.nonEmpty)
               )
             }
             .toList,
diff --git a/core/src/main/scala-3/caliban/schema/DerivationUtils.scala b/core/src/main/scala-3/caliban/schema/DerivationUtils.scala
index a1fd4fe88..b5204610e 100644
--- a/core/src/main/scala-3/caliban/schema/DerivationUtils.scala
+++ b/core/src/main/scala-3/caliban/schema/DerivationUtils.scala
@@ -120,27 +120,37 @@ private object DerivationUtils {
   def mkObject[R](
     annotations: List[Any],
     fields: List[(String, List[Any], Schema[R, Any])],
-    info: TypeInfo
+    info: TypeInfo,
+    enableSemanticNonNull: Boolean
   )(isInput: Boolean, isSubscription: Boolean): __Type = makeObject(
     Some(getName(annotations, info)),
     getDescription(annotations),
     fields.map { (name, fieldAnnotations, schema) =>
-      val deprecatedReason = getDeprecatedReason(fieldAnnotations)
-      val isOptional       = {
+      val deprecatedReason                = getDeprecatedReason(fieldAnnotations)
+      val (isNullable, isSemanticNonNull) = {
         val hasNullableAnn = fieldAnnotations.contains(GQLNullable())
         val hasNonNullAnn  = fieldAnnotations.contains(GQLNonNullable())
-        !hasNonNullAnn && (hasNullableAnn || schema.optional)
+
+        if (hasNonNullAnn) (false, false)
+        else if (hasNullableAnn) (true, false)
+        else if (schema.optional) (true, !schema.nullable)
+        else (false, false)
       }
       Types.makeField(
         name,
         getDescription(fieldAnnotations),
         schema.arguments,
         () =>
-          if (isOptional) schema.toType_(isInput, isSubscription)
+          if (isNullable) schema.toType_(isInput, isSubscription)
           else schema.toType_(isInput, isSubscription).nonNull,
         deprecatedReason.isDefined,
         deprecatedReason,
-        Option(getDirectives(fieldAnnotations)).filter(_.nonEmpty)
+        Option(
+          getDirectives(fieldAnnotations) ++ {
+            if (enableSemanticNonNull && isSemanticNonNull) Some(SchemaUtils.SemanticNonNull)
+            else None
+          }
+        ).filter(_.nonEmpty)
       )
     },
     getDirectives(annotations),
diff --git a/core/src/main/scala-3/caliban/schema/EnumValueSchema.scala b/core/src/main/scala-3/caliban/schema/EnumValueSchema.scala
index aec6e4152..7bd6587cf 100644
--- a/core/src/main/scala-3/caliban/schema/EnumValueSchema.scala
+++ b/core/src/main/scala-3/caliban/schema/EnumValueSchema.scala
@@ -7,12 +7,13 @@ import magnolia1.TypeInfo
 
 final private class EnumValueSchema[R, A](
   info: TypeInfo,
-  anns: List[Any]
+  anns: List[Any],
+  enableSemanticNonNull: Boolean
 ) extends Schema[R, A] {
 
   def toType(isInput: Boolean, isSubscription: Boolean): __Type =
     if (isInput) mkInputObject[R](anns, Nil, info)(isInput, isSubscription)
-    else mkObject[R](anns, Nil, info)(isInput, isSubscription)
+    else mkObject[R](anns, Nil, info, enableSemanticNonNull)(isInput, isSubscription)
 
   private val step               = PureStep(EnumValue(getName(anns, info)))
   def resolve(value: A): Step[R] = step
diff --git a/core/src/main/scala-3/caliban/schema/ObjectSchema.scala b/core/src/main/scala-3/caliban/schema/ObjectSchema.scala
index d0afc06a1..1b5169dab 100644
--- a/core/src/main/scala-3/caliban/schema/ObjectSchema.scala
+++ b/core/src/main/scala-3/caliban/schema/ObjectSchema.scala
@@ -12,7 +12,8 @@ final private class ObjectSchema[R, A](
   _methodFields: => List[(String, List[Any], Schema[R, ?])],
   info: TypeInfo,
   anns: List[Any],
-  paramAnnotations: Map[String, List[Any]]
+  paramAnnotations: Map[String, List[Any]],
+  enableSemanticNonNull: Boolean
 )(using ct: ClassTag[A])
     extends Schema[R, A] {
 
@@ -48,7 +49,7 @@ final private class ObjectSchema[R, A](
   def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
     val _ = resolver // Init the lazy val
     if (isInput) mkInputObject[R](anns, fields.map(_._1), info)(isInput, isSubscription)
-    else mkObject[R](anns, fields.map(_._1), info)(isInput, isSubscription)
+    else mkObject[R](anns, fields.map(_._1), info, enableSemanticNonNull)(isInput, isSubscription)
   }
 
   def resolve(value: A): Step[R] = resolver.resolve(value)
diff --git a/core/src/main/scala-3/caliban/schema/SchemaDerivation.scala b/core/src/main/scala-3/caliban/schema/SchemaDerivation.scala
index fe2c6838e..4262eb5f2 100644
--- a/core/src/main/scala-3/caliban/schema/SchemaDerivation.scala
+++ b/core/src/main/scala-3/caliban/schema/SchemaDerivation.scala
@@ -23,6 +23,21 @@ object PrintDerived {
 trait CommonSchemaDerivation {
   export DerivationUtils.customizeInputTypeName
 
+  case class DerivationConfig(
+    /**
+     * Whether to enable the `SemanticNonNull` feature on derivation.
+     * It is currently disabled by default since it is not yet stable.
+     */
+    enableSemanticNonNull: Boolean = false
+  )
+
+  /**
+   * Returns a configuration object that can be used to customize the derivation behavior.
+   *
+   * Override this method to customize the configuration.
+   */
+  def config: DerivationConfig = DerivationConfig()
+
   inline def recurseSum[R, P, Label, A <: Tuple](
     inline types: List[(String, __Type, List[Any])] = Nil,
     inline schemas: List[Schema[R, Any]] = Nil
@@ -95,7 +110,8 @@ trait CommonSchemaDerivation {
             new EnumValueSchema[R, A](
               MagnoliaMacro.typeInfo[A],
               // Workaround until we figure out why the macro uses the parent's annotations when the leaf is a Scala 3 enum
-              inline if (!MagnoliaMacro.isEnum[A]) MagnoliaMacro.anns[A] else Nil
+              inline if (!MagnoliaMacro.isEnum[A]) MagnoliaMacro.anns[A] else Nil,
+              config.enableSemanticNonNull
             )
           case _ if Macros.hasAnnotation[A, GQLValueType] =>
             new ValueTypeSchema[R, A](
@@ -109,7 +125,8 @@ trait CommonSchemaDerivation {
               Macros.fieldsFromMethods[R, A],
               MagnoliaMacro.typeInfo[A],
               MagnoliaMacro.anns[A],
-              MagnoliaMacro.paramAnns[A].toMap
+              MagnoliaMacro.paramAnns[A].toMap,
+              config.enableSemanticNonNull
             )(using summonInline[ClassTag[A]])
         }
 
diff --git a/core/src/main/scala/caliban/schema/Schema.scala b/core/src/main/scala/caliban/schema/Schema.scala
index 8193ad0c9..24657497f 100644
--- a/core/src/main/scala/caliban/schema/Schema.scala
+++ b/core/src/main/scala/caliban/schema/Schema.scala
@@ -73,9 +73,21 @@ trait Schema[-R, T] { self =>
   def resolve(value: T): Step[R]
 
   /**
-   * Defines if the type is considered optional or non-null. Should be false except for `Option`.
+   * Defines if the type can resolve to null or not.
+   * It is true if the type is nullable or if it can fail.
    */
-  def optional: Boolean = false
+  @deprecatedOverriding("this method will be made final. Override canFail and nullable instead", "2.7.0")
+  def optional: Boolean = canFail || nullable
+
+  /**
+   * Defines if the underlying type represents a nullable value. Should be false except for `Option`.
+   */
+  def nullable: Boolean = false
+
+  /**
+   * Defines if the type can fail during resolution.
+   */
+  def canFail: Boolean = false
 
   /**
    * Defined the arguments of the given type. Should be empty except for `Function`.
@@ -90,7 +102,8 @@ trait Schema[-R, T] { self =>
    * @param f a function from `A` to `T`.
    */
   def contramap[A](f: A => T): Schema[R, A] = new Schema[R, A] {
-    override def optional: Boolean                                         = self.optional
+    override def nullable: Boolean                                         = self.nullable
+    override def canFail: Boolean                                          = self.canFail
     override def arguments: List[__InputValue]                             = self.arguments
     override def toType(isInput: Boolean, isSubscription: Boolean): __Type = self.toType_(isInput, isSubscription)
     override def resolve(value: A): Step[R]                                = self.resolve(f(value))
@@ -102,7 +115,8 @@ trait Schema[-R, T] { self =>
    * @param inputName new name for the type when it's an input type (by default "Input" is added after the name)
    */
   def rename(name: String, inputName: Option[String] = None): Schema[R, T] = new Schema[R, T] {
-    override def optional: Boolean                                         = self.optional
+    override def nullable: Boolean                                         = self.nullable
+    override def canFail: Boolean                                          = self.canFail
     override def arguments: List[__InputValue]                             = self.arguments
     override def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
       val tpe     = self.toType_(isInput, isSubscription)
@@ -342,7 +356,7 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     Schema.stringSchema.contramap(Cursor[Base64Cursor].encode)
 
   implicit def optionSchema[R0, A](implicit ev: Schema[R0, A]): Schema[R0, Option[A]]                                  = new Schema[R0, Option[A]] {
-    override def optional: Boolean                                         = true
+    override def nullable: Boolean                                         = true
     override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
 
     override def resolve(value: Option[A]): Step[R0] =
@@ -354,7 +368,7 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
   implicit def listSchema[R0, A](implicit ev: Schema[R0, A]): Schema[R0, List[A]]                                      = new Schema[R0, List[A]] {
     override def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
       val t = ev.toType_(isInput, isSubscription)
-      (if (ev.optional) t else t.nonNull).list
+      (if (ev.nullable || ev.canFail) t else t.nonNull).list
     }
 
     override def resolve(value: List[A]): Step[R0] = ListStep(value.map(ev.resolve))
@@ -367,14 +381,16 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     listSchema[R0, A].contramap(_.toList)
   implicit def functionUnitSchema[R0, A](implicit ev: Schema[R0, A]): Schema[R0, () => A]                              =
     new Schema[R0, () => A] {
-      override def optional: Boolean                                         = ev.optional
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = ev.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: () => A): Step[R0]                         = FunctionStep(_ => ev.resolve(value()))
     }
   implicit def metadataFunctionSchema[R0, A](implicit ev: Schema[R0, A]): Schema[R0, Field => A]                       =
     new Schema[R0, Field => A] {
       override def arguments: List[__InputValue]                             = ev.arguments
-      override def optional: Boolean                                         = ev.optional
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = ev.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: Field => A): Step[R0]                      = MetadataFunctionStep(field => ev.resolve(value(field)))
     }
@@ -389,12 +405,14 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     val description: String = s"Either $typeAName or $typeBName"
 
     implicit val leftSchema: Schema[RA, A]  = new Schema[RA, A] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = true
+      override def canFail: Boolean                                          = evA.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = evA.toType_(isInput, isSubscription)
       override def resolve(value: A): Step[RA]                               = evA.resolve(value)
     }
     implicit val rightSchema: Schema[RB, B] = new Schema[RB, B] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = true
+      override def canFail: Boolean                                          = evB.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = evB.toType_(isInput, isSubscription)
       override def resolve(value: B): Step[RB]                               = evB.resolve(value)
     }
@@ -462,14 +480,15 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
               __InputValue(
                 unwrappedArgumentName,
                 None,
-                () => if (ev1.optional) inputType else inputType.nonNull,
+                () => if (ev1.nullable || ev1.canFail) inputType else inputType.nonNull,
                 None
               )
             )
           )
       }
 
-      override def optional: Boolean                                         = ev2.optional
+      override def nullable: Boolean                                         = ev2.nullable
+      override def canFail: Boolean                                          = ev2.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev2.toType_(isInput, isSubscription)
 
       override def resolve(f: A => B): Step[RB]                         =
@@ -497,7 +516,8 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     effectSchema[R0, R0, R0, Throwable, A].contramap[Future[A]](future => ZIO.fromFuture(_ => future)(Trace.empty))
   implicit def infallibleEffectSchema[R0, R1 >: R0, R2 >: R0, A](implicit ev: Schema[R2, A]): Schema[R0, URIO[R1, A]] =
     new Schema[R0, URIO[R1, A]] {
-      override def optional: Boolean                                         = ev.optional
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = ev.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: URIO[R1, A]): Step[R0]                     = QueryStep(ZQuery.fromZIONow(value.map(ev.resolve)))
     }
@@ -505,7 +525,8 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZIO[R1, E, A]] =
     new Schema[R0, ZIO[R1, E, A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: ZIO[R1, E, A]): Step[R0]                   = QueryStep(ZQuery.fromZIONow(value.map(ev.resolve)))
     }
@@ -513,7 +534,8 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZIO[R1, E, A]] =
     new Schema[R0, ZIO[R1, E, A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: ZIO[R1, E, A]): Step[R0]                   = QueryStep(
         ZQuery.fromZIONow(value.mapBoth(convertError, ev.resolve))
@@ -523,7 +545,8 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZQuery[R1, Nothing, A]] =
     new Schema[R0, ZQuery[R1, Nothing, A]] {
-      override def optional: Boolean                                         = ev.optional
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = ev.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: ZQuery[R1, Nothing, A]): Step[R0]          = QueryStep(value.map(ev.resolve))
     }
@@ -531,7 +554,8 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZQuery[R1, E, A]] =
     new Schema[R0, ZQuery[R1, E, A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: ZQuery[R1, E, A]): Step[R0]                = QueryStep(value.map(ev.resolve))
     }
@@ -539,7 +563,8 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZQuery[R1, E, A]] =
     new Schema[R0, ZQuery[R1, E, A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
       override def resolve(value: ZQuery[R1, E, A]): Step[R0]                = QueryStep(value.mapBoth(convertError, ev.resolve))
     }
@@ -547,10 +572,11 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R1, ZStream[R1, Nothing, A]] =
     new Schema[R1, ZStream[R1, Nothing, A]] {
-      override def optional: Boolean                                         = false
+      override def nullable: Boolean                                         = false
+      override def canFail: Boolean                                          = ev.canFail
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
         val t = ev.toType_(isInput, isSubscription)
-        if (isSubscription) t else (if (ev.optional) t else t.nonNull).list
+        if (isSubscription) t else (if (ev.nullable || ev.canFail) t else t.nonNull).list
       }
       override def resolve(value: ZStream[R1, Nothing, A]): Step[R1]         = StreamStep(value.map(ev.resolve))
     }
@@ -558,10 +584,11 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZStream[R1, E, A]] =
     new Schema[R0, ZStream[R1, E, A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
         val t = ev.toType_(isInput, isSubscription)
-        if (isSubscription) t else (if (ev.optional) t else t.nonNull).list
+        if (isSubscription) t else (if (ev.nullable || ev.canFail) t else t.nonNull).list
       }
       override def resolve(value: ZStream[R1, E, A]): Step[R0]               = StreamStep(value.map(ev.resolve))
     }
@@ -569,10 +596,11 @@ trait GenericSchema[R] extends SchemaDerivation[R] with TemporalSchema {
     ev: Schema[R2, A]
   ): Schema[R0, ZStream[R1, E, A]] =
     new Schema[R0, ZStream[R1, E, A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
         val t = ev.toType_(isInput, isSubscription)
-        if (isSubscription) t else (if (ev.optional) t else t.nonNull).list
+        if (isSubscription) t else (if (ev.nullable || ev.canFail) t else t.nonNull).list
       }
       override def resolve(value: ZStream[R1, E, A]): Step[R0]               = StreamStep(value.mapBoth(convertError, ev.resolve))
     }
@@ -687,7 +715,11 @@ trait TemporalSchema {
 
 case class FieldAttributes(isInput: Boolean, isSubscription: Boolean)
 
-abstract class PartiallyAppliedFieldBase[V](name: String, description: Option[String], directives: List[Directive]) {
+abstract class PartiallyAppliedFieldBase[V](
+  name: String,
+  description: Option[String],
+  directives: List[Directive]
+) {
   def apply[R, V1](fn: V => V1)(implicit ev: Schema[R, V1], ft: FieldAttributes): (__Field, V => Step[R]) =
     either[R, V1](v => Left(fn(v)))(ev, ft)
 
@@ -701,31 +733,43 @@ abstract class PartiallyAppliedFieldBase[V](name: String, description: Option[St
       description,
       _ => Nil,
       () =>
-        if (ev.optional) ev.toType_(ft.isInput, ft.isSubscription)
+        if (ev.nullable || ev.canFail) ev.toType_(ft.isInput, ft.isSubscription)
         else ev.toType_(ft.isInput, ft.isSubscription).nonNull,
       isDeprecated = Directives.isDeprecated(directives),
       deprecationReason = Directives.deprecationReason(directives),
-      directives = Some(directives.filter(_.name != "deprecated")).filter(_.nonEmpty)
+      directives = Some(
+        directives.filter(_.name != "deprecated")
+      ).filter(_.nonEmpty)
     )
 }
 
-case class PartiallyAppliedField[V](name: String, description: Option[String], directives: List[Directive])
-    extends PartiallyAppliedFieldBase[V](name, description, directives) {
+case class PartiallyAppliedField[V](
+  name: String,
+  description: Option[String],
+  directives: List[Directive]
+) extends PartiallyAppliedFieldBase[V](name, description, directives) {
   def either[R, V1](
     fn: V => Either[V1, Step[R]]
   )(implicit ev: Schema[R, V1], ft: FieldAttributes): (__Field, V => Step[R]) =
     (makeField, (v: V) => fn(v).fold(ev.resolve, identity))
 }
 
-case class PartiallyAppliedFieldLazy[V](name: String, description: Option[String], directives: List[Directive])
-    extends PartiallyAppliedFieldBase[V](name, description, directives) {
+case class PartiallyAppliedFieldLazy[V](
+  name: String,
+  description: Option[String],
+  directives: List[Directive]
+) extends PartiallyAppliedFieldBase[V](name, description, directives) {
   def either[R, V1](
     fn: V => Either[V1, Step[R]]
   )(implicit ev: Schema[R, V1], ft: FieldAttributes): (__Field, V => Step[R]) =
     (makeField, (v: V) => FunctionStep(_ => fn(v).fold(ev.resolve, identity)))
 }
 
-case class PartiallyAppliedFieldWithArgs[V, A](name: String, description: Option[String], directives: List[Directive]) {
+case class PartiallyAppliedFieldWithArgs[V, A](
+  name: String,
+  description: Option[String],
+  directives: List[Directive]
+) {
   def apply[R, V1](fn: V => (A => V1))(implicit ev1: Schema[R, A => V1], fa: FieldAttributes): (__Field, V => Step[R]) =
     (
       Types.makeField(
@@ -733,11 +777,13 @@ case class PartiallyAppliedFieldWithArgs[V, A](name: String, description: Option
         description,
         ev1.arguments,
         () =>
-          if (ev1.optional) ev1.toType_(fa.isInput, fa.isSubscription)
+          if (ev1.nullable || ev1.canFail) ev1.toType_(fa.isInput, fa.isSubscription)
           else ev1.toType_(fa.isInput, fa.isSubscription).nonNull,
         isDeprecated = Directives.isDeprecated(directives),
         deprecationReason = Directives.deprecationReason(directives),
-        directives = Some(directives.filter(_.name != "deprecated")).filter(_.nonEmpty)
+        directives = Some(
+          directives.filter(_.name != "deprecated")
+        ).filter(_.nonEmpty)
       ),
       (v: V) => ev1.resolve(fn(v))
     )
diff --git a/core/src/main/scala/caliban/schema/SchemaUtils.scala b/core/src/main/scala/caliban/schema/SchemaUtils.scala
index de0e571aa..65d40e368 100644
--- a/core/src/main/scala/caliban/schema/SchemaUtils.scala
+++ b/core/src/main/scala/caliban/schema/SchemaUtils.scala
@@ -4,8 +4,15 @@ import caliban.ResponseValue.ObjectValue
 import caliban.Value.{ EnumValue, NullValue, StringValue }
 import caliban.introspection.adt.{ __DeprecatedArgs, __Field, __Type }
 import caliban.schema.Step.MetadataFunctionStep
+import caliban.parsing.adt.Directive
 
 private[schema] object SchemaUtils {
+
+  /**
+   * Directive used to mark a field as semantically non-nullable.
+   */
+  val SemanticNonNull = Directive("semanticNonNull")
+
   private val fakeField =
     Some(
       List(
diff --git a/core/src/test/scala/caliban/TriState.scala b/core/src/test/scala/caliban/TriState.scala
index 2e9646587..f6ae618f0 100644
--- a/core/src/test/scala/caliban/TriState.scala
+++ b/core/src/test/scala/caliban/TriState.scala
@@ -15,7 +15,7 @@ object TriState {
 
   def schemaCustom[R, A](undefined: PureStep)(implicit ev: Schema[R, A]): Schema[R, TriState[A]] =
     new Schema[R, TriState[A]] {
-      override val optional = true
+      override val nullable = true
 
       override def toType(isInput: Boolean, isSubscription: Boolean) = ev.toType_(isInput, isSubscription)
 
diff --git a/core/src/test/scala/caliban/schema/OptionalSpec.scala b/core/src/test/scala/caliban/schema/OptionalSpec.scala
new file mode 100644
index 000000000..f04d2108e
--- /dev/null
+++ b/core/src/test/scala/caliban/schema/OptionalSpec.scala
@@ -0,0 +1,57 @@
+package caliban.schema
+
+import caliban.{ graphQL, RootResolver }
+import zio._
+import zio.test._
+
+object OptionalSpec extends ZIOSpecDefault {
+  import caliban.schema.Schema._
+
+  override def spec = suite("OptionalSpec")(
+    test("Semantic of Schema.optional is kept consistent across various nullable/canFail combinations") {
+      val expected =
+        """schema {
+          |  query: Query
+          |}
+          |
+          |type Query {
+          |  a: String!
+          |  b: String
+          |  c: String
+          |  d: String
+          |}""".stripMargin
+
+      implicit def wrapperSchema[A](implicit ev: Schema[Any, A]): Schema[Any, Wrapper[A]] =
+        new Schema[Any, Wrapper[A]] {
+          @annotation.nowarn
+          override def optional = ev.optional
+          def toType(isInput: Boolean, isSubscription: Boolean) = ev.toType_(isInput, isSubscription)
+          def resolve(value: Wrapper[A]): Step[Any]             =
+            ev.resolve(value.value)
+        }
+
+      implicit def querySchema: Schema[Any, Query] = Schema.gen[Any, Query]
+
+      val resolver = RootResolver(
+        Query(
+          Wrapper("a"),
+          Wrapper(Some("b")),
+          Wrapper(ZIO.succeed("c")),
+          Wrapper(ZIO.succeed(Some("d")))
+        )
+      )
+      val gql      = graphQL(resolver)
+
+      assertTrue(gql.render == expected)
+    }
+  )
+
+  case class Wrapper[A](value: A)
+
+  case class Query(
+    a: Wrapper[String],
+    b: Wrapper[Option[String]],
+    c: Wrapper[Task[String]],
+    d: Wrapper[Task[Option[String]]]
+  )
+}
diff --git a/core/src/test/scala/caliban/schema/SchemaSpec.scala b/core/src/test/scala/caliban/schema/SchemaSpec.scala
index 8fb204034..925c24466 100644
--- a/core/src/test/scala/caliban/schema/SchemaSpec.scala
+++ b/core/src/test/scala/caliban/schema/SchemaSpec.scala
@@ -2,7 +2,7 @@ package caliban.schema
 
 import caliban.Value.StringValue
 import caliban._
-import caliban.introspection.adt.{ __DeprecatedArgs, __Type, __TypeKind }
+import caliban.introspection.adt.{ __DeprecatedArgs, __Field, __Type, __TypeKind }
 import caliban.parsing.adt.Directive
 import caliban.schema.Annotations._
 import caliban.schema.ArgBuilder.auto._
@@ -386,6 +386,7 @@ object SchemaSpec extends ZIOSpecDefault {
     )
 
   case class EffectfulFieldSchema(q: Task[Int], @GQLNonNullable qAnnotated: Task[Int])
+  case class OptionalEffectfulFieldSchema(q: Task[Option[String]], @GQLNonNullable qAnnotated: Task[Option[String]])
   case class InfallibleFieldSchema(q: UIO[Int], @GQLNullable qAnnotated: UIO[Int])
   case class FutureFieldSchema(q: Future[Int])
   case class IDSchema(id: UUID)
diff --git a/core/src/test/scala/caliban/schema/SemanticNonNullSchemaSpec.scala b/core/src/test/scala/caliban/schema/SemanticNonNullSchemaSpec.scala
new file mode 100644
index 000000000..b368ae9c1
--- /dev/null
+++ b/core/src/test/scala/caliban/schema/SemanticNonNullSchemaSpec.scala
@@ -0,0 +1,108 @@
+package caliban.schema
+
+import caliban._
+import caliban.introspection.adt.{ __DeprecatedArgs, __Field, __Type, __TypeKind }
+import caliban.parsing.adt.Directive
+import caliban.schema.Annotations._
+import zio._
+import zio.test.Assertion._
+import zio.test._
+
+object SemanticNonNullSchema extends SchemaDerivation[Any] {
+  override def config = DerivationConfig(enableSemanticNonNull = true)
+}
+
+object SemanticNonNullSchemaSpec extends ZIOSpecDefault {
+  override def spec =
+    suite("SemanticNonNullSchemaSpec")(
+      test("effectful field as semanticNonNull") {
+        val field = effectfulFieldObjectSchema.toType_().fields(__DeprecatedArgs()).toList.flatten.apply(0)
+        assert(field)(
+          hasField[__Field, Option[List[Directive]]](
+            "directives",
+            _.directives,
+            isSome(contains(SchemaUtils.SemanticNonNull))
+          )
+        )
+        assert(field._type)(
+          hasField[__Type, __TypeKind]("kind", _.kind, equalTo(__TypeKind.SCALAR))
+        )
+      },
+      test("effectful field as non-nullable") {
+        val field = effectfulFieldObjectSchema.toType_().fields(__DeprecatedArgs()).toList.flatten.apply(1)
+        assert(field)(
+          hasField[__Field, Option[List[Directive]]](
+            "directives",
+            _.directives.map(_.filter(_ == SchemaUtils.SemanticNonNull)).filter(_.nonEmpty),
+            isNone
+          )
+        )
+        assert(field._type)(
+          hasField[__Type, __TypeKind]("kind", _.kind, equalTo(__TypeKind.NON_NULL))
+        )
+      },
+      test("nullable effectful field") {
+        val field = nullableEffectfulFieldObjectSchema.toType_().fields(__DeprecatedArgs()).toList.flatten.apply(0)
+        assert(field)(
+          hasField[__Field, Option[List[Directive]]](
+            "directives",
+            _.directives.map(_.filter(_ == SchemaUtils.SemanticNonNull)).filter(_.nonEmpty),
+            isNone
+          )
+        )
+        assert(field._type)(
+          hasField[__Type, __TypeKind]("kind", _.kind, equalTo(__TypeKind.SCALAR))
+        )
+      },
+      test("nullable effectful field as non-nullable") {
+        val field = nullableEffectfulFieldObjectSchema.toType_().fields(__DeprecatedArgs()).toList.flatten.apply(1)
+        assert(field)(
+          hasField[__Field, Option[List[Directive]]](
+            "directives",
+            _.directives,
+            isNone
+          )
+        )
+        assert(field._type)(
+          hasField[__Type, __TypeKind]("kind", _.kind, equalTo(__TypeKind.NON_NULL))
+        )
+      },
+      test("infallible effectful field") {
+        val field = infallibleEffectfulFieldSchema.toType_().fields(__DeprecatedArgs()).toList.flatten.apply(0)
+        assert(field)(
+          hasField[__Field, Option[List[Directive]]](
+            "directives",
+            _.directives,
+            isNone
+          )
+        )
+        assert(field._type)(
+          hasField[__Type, __TypeKind]("kind", _.kind, equalTo(__TypeKind.NON_NULL))
+        )
+      },
+      test("infallible effectful field as nullable") {
+        val field = infallibleEffectfulFieldSchema.toType_().fields(__DeprecatedArgs()).toList.flatten.apply(1)
+        assert(field)(
+          hasField[__Field, Option[List[Directive]]](
+            "directives",
+            _.directives,
+            isNone
+          )
+        )
+        assert(field._type)(
+          hasField[__Type, __TypeKind]("kind", _.kind, equalTo(__TypeKind.SCALAR))
+        )
+      }
+    )
+
+  case class EffectfulFieldObject(q: Task[Int], @GQLNonNullable qAnnotated: Task[Int])
+  case class NullableEffectfulFieldObject(q: Task[Option[String]], @GQLNonNullable qAnnotated: Task[Option[String]])
+  case class InfallibleFieldObject(q: UIO[Int], @GQLNullable qAnnotated: UIO[Int])
+
+  implicit val effectfulFieldObjectSchema: Schema[Any, EffectfulFieldObject]                 =
+    SemanticNonNullSchema.gen[Any, EffectfulFieldObject]
+  implicit val nullableEffectfulFieldObjectSchema: Schema[Any, NullableEffectfulFieldObject] =
+    SemanticNonNullSchema.gen[Any, NullableEffectfulFieldObject]
+  implicit val infallibleEffectfulFieldSchema: Schema[Any, InfallibleFieldObject]            =
+    SemanticNonNullSchema.gen[Any, InfallibleFieldObject]
+}
diff --git a/federation/src/main/scala/caliban/federation/FederationSupport.scala b/federation/src/main/scala/caliban/federation/FederationSupport.scala
index 5eef75a02..e4d695f94 100644
--- a/federation/src/main/scala/caliban/federation/FederationSupport.scala
+++ b/federation/src/main/scala/caliban/federation/FederationSupport.scala
@@ -60,7 +60,7 @@ abstract class FederationSupport(
     import genericSchema.auto._
 
     implicit val entitySchema: Schema[R, _Entity] = new Schema[R, _Entity] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type =
         __Type(
           __TypeKind.UNION,
diff --git a/interop/cats/src/main/scala/caliban/interop/cats/CatsInterop.scala b/interop/cats/src/main/scala/caliban/interop/cats/CatsInterop.scala
index b355eef9a..488347988 100644
--- a/interop/cats/src/main/scala/caliban/interop/cats/CatsInterop.scala
+++ b/interop/cats/src/main/scala/caliban/interop/cats/CatsInterop.scala
@@ -190,8 +190,9 @@ object CatsInterop {
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type =
         ev.toType_(isInput, isSubscription)
 
-      override def optional: Boolean =
-        ev.optional
+      override def nullable: Boolean =
+        ev.nullable
+      override def canFail: Boolean  = true
 
       override def resolve(value: F[A]): Step[R] =
         QueryStep(ZQuery.fromZIO(interop.fromEffect(value).map(ev.resolve)))
diff --git a/interop/cats/src/main/scala/caliban/interop/fs2/Fs2Interop.scala b/interop/cats/src/main/scala/caliban/interop/fs2/Fs2Interop.scala
index 181cf247f..c2a3239f2 100644
--- a/interop/cats/src/main/scala/caliban/interop/fs2/Fs2Interop.scala
+++ b/interop/cats/src/main/scala/caliban/interop/fs2/Fs2Interop.scala
@@ -14,7 +14,8 @@ object Fs2Interop {
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type =
         ev.toType_(isInput, isSubscription)
 
-      override def optional: Boolean = ev.optional
+      override def nullable: Boolean = ev.nullable
+      override def canFail: Boolean  = true
 
       override def resolve(value: Stream[RIO[R, *], A]): Step[R] =
         ev.resolve(value.toZStream())
@@ -28,7 +29,8 @@ object Fs2Interop {
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type =
         ev.toType_(isInput, isSubscription)
 
-      override def optional: Boolean = ev.optional
+      override def nullable: Boolean = ev.nullable
+      override def canFail: Boolean  = true
 
       override def resolve(value: Stream[F, A]): Step[R] =
         ev.resolve(value.translate(interop.fromEffectK))
diff --git a/interop/monix/src/main/scala/caliban/interop/monix/MonixInterop.scala b/interop/monix/src/main/scala/caliban/interop/monix/MonixInterop.scala
index e06864d65..c29e38e8e 100644
--- a/interop/monix/src/main/scala/caliban/interop/monix/MonixInterop.scala
+++ b/interop/monix/src/main/scala/caliban/interop/monix/MonixInterop.scala
@@ -60,7 +60,8 @@ object MonixInterop {
   def taskSchema[R, A](implicit ev: Schema[R, A], ev2: ConcurrentEffect[MonixTask]): Schema[R, MonixTask[A]] =
     new Schema[R, MonixTask[A]] {
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = ev.toType_(isInput, isSubscription)
-      override def optional: Boolean                                         = ev.optional
+      override def nullable: Boolean                                         = ev.nullable
+      override def canFail: Boolean                                          = true
       override def resolve(value: MonixTask[A]): Step[R]                     =
         QueryStep(ZQuery.fromZIO(value.to[Task].map(ev.resolve)))
     }
@@ -69,10 +70,10 @@ object MonixInterop {
     queueSize: Int
   )(implicit ev: Schema[R, A], ev2: ConcurrentEffect[MonixTask]): Schema[R, Observable[A]] =
     new Schema[R, Observable[A]] {
-      override def optional: Boolean                                         = true
+      override def nullable: Boolean                                         = true
       override def toType(isInput: Boolean, isSubscription: Boolean): __Type = {
         val t = ev.toType_(isInput, isSubscription)
-        if (isSubscription) t else (if (ev.optional) t else t.nonNull).list
+        if (isSubscription) t else (if (ev.nullable) t else t.nonNull).list
       }
       override def resolve(value: Observable[A]): Step[R]                    =
         StreamStep(