[kyo-kernel][kyo-core] tests for nested-computation resumption and fatal-error handling

PendingTest: map/flatMap/andThen must apply their continuation when the resumption over a nested
value is deferred at a denied safepoint, so the inner effect stays handled.

IOTaskTest: a computation that throws a fatal error on a scheduler fiber must still run its
finalizers, so a promise a finalizer completes is not left pending.

IOPromiseTest: a completion callback that throws must not abort the flush and leave the promise's
other waiters unnotified.

Author: fwbrasil

kyo-core/shared/src/test/scala/kyo/scheduler/IOPromiseTest.scala

@@ -1248,4 +1248,24 @@ class IOPromiseTest extends kyo.test.Test[Any]:
         }
     }
 
+    "flush" - {
+        "a fatal error in one completion callback still runs the others" in {
+            // When a promise completes, its registered callbacks run in a single flush loop. A callback that throws
+            // must not abort that loop and leave the promise's other waiters unnotified. InterruptedException is a
+            // portable throwable that scala.util.control.NonFatal classifies as fatal; placing it between two
+            // counting callbacks means an aborted flush would skip one of them regardless of order.
+            var ran = 0
+            val p   = new IOPromise[Nothing, Int]()
+            p.onComplete(_ => ran += 1)
+            p.onComplete(_ => throw new InterruptedException("fatal error"))
+            p.onComplete(_ => ran += 1)
+            try p.completeDiscard(Result.succeed(1))
+            catch case _: Throwable => () // contain any throwable that escapes the flush so the assertion below runs
+            assert(
+                ran == 2,
+                s"a fatal callback aborted the flush and orphaned the other waiters; only $ran of 2 non-fatal callbacks ran"
+            )
+        }
+    }
+
 end IOPromiseTest

kyo-core/shared/src/test/scala/kyo/scheduler/IOTaskTest.scala

@@ -109,4 +109,30 @@ class IOTaskTest extends kyo.test.Test[Any]:
 
     }
 
+    "fatal error in a guarded body" - {
+        // A computation on a scheduler fiber throws a fatal error. IOTask completes the fiber's own promise with a
+        // Panic, but must still run the computation's finalizers; here a finalizer completes `probe`, standing in
+        // for a promise awaited elsewhere. If finalizers are skipped on the fatal path, `probe` is never completed
+        // and the timed await below expires. LinkageError is used because scala.util.control.NonFatal (which the
+        // scheduler gates on) classifies it as fatal. JVM-only: it relies on worker-thread semantics (one worker
+        // taking the fatal while the timeout fires on another), which the single-worker Native and single-threaded
+        // JS runtimes do not provide.
+        "runs the ensure finalizer even though the fatal aborts the fiber".onlyJvm in {
+            for
+                probe <- Promise.init[Unit, Any]
+                _ <- Fiber.initUnscoped {
+                    Sync.ensure { probe.completeDiscard(Result.succeed(())) } {
+                        Sync.defer[Unit, Any](throw new LinkageError("fatal error"))
+                    }
+                }
+                // Await only `probe`, never the fatal fiber's own result: awaiting a fatally-aborted fiber re-raises
+                // the fatal here. Bounded so a missing completion fails fast rather than blocking indefinitely.
+                finished <- Abort.run[Any](Async.timeout(5.seconds)(probe.get))
+            yield assert(
+                finished.isSuccess,
+                "the Sync.ensure finalizer did not run when the guarded body threw a fatal error; the awaited promise was never completed"
+            )
+        }
+    }
+
 end IOTaskTest

kyo-kernel/shared/src/test/scala/kyo/kernel/PendingTest.scala

@@ -351,6 +351,23 @@ class PendingTest extends kyo.test.Test[Any]:
                 ContextEffect.handle(Tag[TestEffect3], value)(v)
         end TestEffect3
 
+        // A fresh interceptor per call (it is stateful) that denies entry into the second pending computation: the
+        // resumption over a nested `A < S` value, where map/flatMap/andThen must re-apply their continuation.
+        def denyNestedResumption: Safepoint.Interceptor =
+            new Safepoint.Interceptor:
+                var seenNested                                                 = 0
+                def addFinalizer(f: Maybe[Result.Error[Any]] => Unit): Unit    = ()
+                def removeFinalizer(f: Maybe[Result.Error[Any]] => Unit): Unit = ()
+                def enter(frame: Frame, value: Any): Boolean =
+                    value match
+                        case _: kyo.kernel.internal.Kyo[?, ?] =>
+                            seenNested += 1
+                            seenNested != 2
+                        case _ => true
+
+        def nestedScenario: Int < TestEffect2 < TestEffect1 =
+            Kyo.lift(TestEffect1(10)).map(_.map(s => Kyo.lift(TestEffect2(s))))
+
         "basic nesting operations" in {
             val nested: String < TestEffect1 < Any = Kyo.lift(TestEffect1(42))
             assert(TestEffect1.run(nested.flatten).eval == "Effect1:42")
@@ -370,6 +387,40 @@ class PendingTest extends kyo.test.Test[Any]:
             assert(result2.eval == "Effect1:10".length + 10)
         }
 
+        // When the safepoint denies entry (as it does under fiber preemption), map/flatMap/andThen defer the
+        // resumption over a nested `A < S` value; each must still apply its continuation so the inner effect stays
+        // handled and does not leak past its handler. `denyNestedResumption` forces that deferral deterministically.
+        // One leaf per operation.
+
+        "multiple effects with the nested map deferred at a denied safepoint" in {
+            assert(nestedScenario.map(_.handle(TestEffect2.run)).handle(TestEffect1.run).eval == "Effect1:10".length + 10)
+            val deferred =
+                Safepoint.immediate(denyNestedResumption) {
+                    nestedScenario.map(_.handle(TestEffect2.run)).handle(TestEffect1.run)
+                }
+            assert(deferred.eval == "Effect1:10".length + 10)
+        }
+
+        "multiple effects with the nested flatMap deferred at a denied safepoint" in {
+            assert(nestedScenario.flatMap(_.handle(TestEffect2.run)).handle(TestEffect1.run).eval == "Effect1:10".length + 10)
+            val deferred =
+                Safepoint.immediate(denyNestedResumption) {
+                    nestedScenario.flatMap(_.handle(TestEffect2.run)).handle(TestEffect1.run)
+                }
+            assert(deferred.eval == "Effect1:10".length + 10)
+        }
+
+        "discarding a nested computation with andThen deferred at a denied safepoint" in {
+            // andThen discards the inner `Int < TestEffect2`; the deferred resumption must drop the whole nested
+            // value rather than re-evaluate it at the wrong nesting.
+            assert(nestedScenario.andThen(99).handle(TestEffect1.run).eval == 99)
+            val deferred =
+                Safepoint.immediate(denyNestedResumption) {
+                    nestedScenario.andThen(99).handle(TestEffect1.run)
+                }
+            assert(deferred.eval == 99)
+        }
+
         "map" in {
             val nested: String < TestEffect1 < Any = Kyo.lift(TestEffect1(50))