[kyo-reactive-streams] fix flaky hang in publisher-interruption test

PublisherToSubscriberTest "publisher's interuption should end all subscribed
parties" intermittently timed out (a 2m hang) on CI. latchPub only proves
publisher.subscribe was called for each subscriber; onSubscribe is delivered
asynchronously by the publisher's channel-consumer fiber. Interrupting the
publisher before a subscriber is established orphans it: it stays Uninitialized,
its emit-loop await blocks on a promise that only onSubscribe (or its own scope
teardown) can complete, and its run fiber hangs until the leaf timeout. A prior
change removed the settle sleep that had masked this, exposing the race.

Fix by ordering the interruption strictly after establishment: StreamSubscriber
now completes a dedicated `subscribed` promise when it leaves Uninitialized, and
exposes `awaitSubscribed`; the test awaits all four subscribers before
interrupting. The test then exercises real propagation to established parties
(no restored sleep, no manual subscriber interrupts) and no longer races setup.
Validated: the interruption leaf passes 15/15 loops on the JVM and on JS/Native.

Author: fwbrasil

kyo-reactive-streams/shared/src/main/scala/kyo/interop/flow/StreamSubscriber.scala

@@ -23,6 +23,11 @@ final private[kyo] class StreamSubscriber[V](
         UpstreamState.Uninitialized -> Maybe.empty[Fiber.Promise.Unsafe[Unit, Any]]
     )
 
+    // Completed the first time the subscriber leaves Uninitialized (onSubscribe delivered, i.e. the
+    // subscription is established). Lets a caller await establishment without touching the consumption
+    // `await` above, whose promise slot is owned by the emit loop.
+    private val subscribed = Fiber.Promise.Unsafe.init[Unit, Any]()
+
     private inline def throwIfNull[A](b: A): Unit = if isNull(b) then throw new NullPointerException()
 
     override def onSubscribe(subscription: Subscription): Unit =
@@ -33,6 +38,7 @@ final private[kyo] class StreamSubscriber[V](
                 case (UpstreamState.Uninitialized, maybePromise) =>
                     val nextState = UpstreamState.WaitForRequest(subscription, Chunk.empty, 0) -> Absent
                     if state.compareAndSet(curState, nextState) then
+                        subscribed.completeUnitDiscard()
                         maybePromise.foreach(_.completeUnitDiscard())
                     else
                         handleSubscribe()
@@ -156,6 +162,12 @@ final private[kyo] class StreamSubscriber[V](
         Sync.defer(handleAwait())
     end await
 
+    // Resolves once onSubscribe has been delivered (the subscription is established). Used to order a
+    // downstream action (e.g. a publisher interruption) strictly after every subscriber is subscribed,
+    // so it cannot race subscription setup and orphan a still-Uninitialized subscriber.
+    private[interop] def awaitSubscribed(using Frame): Unit < Async =
+        subscribed.safe.use(_ => ())
+
     private[interop] def request(using Frame): Long < Sync =
         @tailrec def handleRequest(): Long < Sync =
             val curState = state.get()

kyo-reactive-streams/shared/src/test/scala/kyo/interop/flow/PublisherToSubscriberTest.scala

@@ -195,6 +195,15 @@ abstract private class PublisherToSubscriberTest extends kyo.test.Test[Any]:
                         .andThen(latchPub.release).andThen(Async.never)
                 )))
                 _ <- latchPub.await
+                // latchPub only proves publisher.subscribe was called for each subscriber; onSubscribe
+                // is delivered asynchronously by the publisher's channel-consumer fiber. Interrupting
+                // before a subscriber is established would orphan it (it never receives onSubscribe or a
+                // terminal signal), and its run fiber would hang. Wait until every subscriber is actually
+                // subscribed, so this tests propagation to established parties rather than racing setup.
+                _ <- subscriber1.awaitSubscribed
+                _ <- subscriber2.awaitSubscribed
+                _ <- subscriber3.awaitSubscribed
+                _ <- subscriber4.awaitSubscribed
                 _ <- publisherFiber.interrupt.unit
                 // Interrupting the publisher fiber closes its scope, which must propagate
                 // cancellation to every subscriber; their run fibers then complete on their own.