diff --git a/.scala-steward.conf b/.scala-steward.conf
index fae8c166..f6bf87c9 100644
--- a/.scala-steward.conf
+++ b/.scala-steward.conf
@@ -1,3 +1,3 @@
 updates.pin = [
-  { groupId = "org.scala-lang", artifactId = "scala3-library", version = "3.3." }
+  { groupId = "org.scala-lang", artifactId = "scala3-library", version = "3.9." }
 ]
diff --git a/build.sbt b/build.sbt
index cfef6bce..d814b5ee 100644
--- a/build.sbt
+++ b/build.sbt
@@ -7,8 +7,8 @@ commonSmlBuildSettings
 ossPublishSettings
 
 organization := "com.softwaremill.ox"
-scalaVersion := "3.3.8"
-scalacOptions ++= Seq("-Yfuture-lazy-vals", "-java-output-version", "21")
+scalaVersion := "3.9.0"
+scalacOptions ++= Seq("-java-output-version", "21")
 Test / scalacOptions += "-Wconf:msg=unused value of type org.scalatest.Assertion:s"
 Test / scalacOptions += "-Wconf:msg=unused value of type org.scalatest.compatible.Assertion:s"
 mimaPreviousArtifacts := Set.empty // we only use MiMa for `core` for now, using enableMimaSettings
diff --git a/core/src/main/scala/ox/Chunk.scala b/core/src/main/scala/ox/Chunk.scala
index f46524b0..22bbe232 100644
--- a/core/src/main/scala/ox/Chunk.scala
+++ b/core/src/main/scala/ox/Chunk.scala
@@ -2,6 +2,7 @@ package ox
 
 import java.nio.charset.Charset
 import java.nio.charset.StandardCharsets
+import scala.annotation.unused
 import scala.reflect.ClassTag
 
 /** An immutable finite indexed sequence of elements, backed by multiple IArrays. Optimized for efficient concatenation by maintaining a
@@ -184,10 +185,11 @@ case object Empty extends Chunk[Nothing]:
 object Chunk:
   def empty[A]: Chunk[A] = Empty
 
-  def fromArray[A: ClassTag](array: Array[A]): Chunk[A] =
+  // keeping unused parameter for binary compatibility - deleting it triggered MiMa
+  def fromArray[A](array: Array[A])(using @unused ct: ClassTag[A]): Chunk[A] =
     if array.isEmpty then Empty else NonEmptyChunk(List(IArray.unsafeFromArray(array)))
 
-  def fromIArray[A: ClassTag](array: IArray[A]): Chunk[A] =
+  def fromIArray[A](array: IArray[A])(using @unused ct: ClassTag[A]): Chunk[A] =
     if array.isEmpty then Empty else NonEmptyChunk(List(array))
 
   /** Creates a chunk from elements */
diff --git a/core/src/main/scala/ox/ErrorMode.scala b/core/src/main/scala/ox/ErrorMode.scala
index 5b912693..fe599f39 100644
--- a/core/src/main/scala/ox/ErrorMode.scala
+++ b/core/src/main/scala/ox/ErrorMode.scala
@@ -1,5 +1,6 @@
 package ox
 
+import scala.annotation.unused
 import scala.reflect.ClassTag
 
 /** Describes the representation of application errors. Such errors have type `E` and are reported in context `F`.
@@ -32,12 +33,12 @@ trait ErrorMode[E, F[_]]:
   /** Adds a suppressed exception to the value being represented by `error`. This is only called if `isError(error)` returns `true`. By
     * default, the suppressed exception is discarded and the original value is returned.
     */
-  def addSuppressedException[T](error: F[T], e: Throwable): F[T] = error
+  def addSuppressedException[T](error: F[T], @unused e: Throwable): F[T] = error
 
   /** Adds a suppressed application error to the value being represented by `error`. This is only called if `isError(error)` returns `true`.
     * By default, the suppressed application error is discarded and the original value is returned.
     */
-  def addSuppressedError[T](error: F[T], e: E): F[T] = error
+  def addSuppressedError[T](error: F[T], @unused e: E): F[T] = error
 end ErrorMode
 
 /** An error mode which doesn't allow reporting application errors.
diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala
index f057a9ee..f9611c00 100644
--- a/core/src/main/scala/ox/channels/SourceOps.scala
+++ b/core/src/main/scala/ox/channels/SourceOps.scala
@@ -2,8 +2,6 @@ package ox.channels
 
 import ox.*
 
-import java.util
-
 trait SourceOps[+T]:
   outer: Source[T] =>
 
diff --git a/core/src/main/scala/ox/channels/select.scala b/core/src/main/scala/ox/channels/select.scala
index e63d2582..c5229d73 100644
--- a/core/src/main/scala/ox/channels/select.scala
+++ b/core/src/main/scala/ox/channels/select.scala
@@ -344,7 +344,7 @@ def selectOrClosedWithin[TV, T1](
   selectOrClosedWithin(timeout, timeoutValue)(source1.receiveClause) match
     case source1.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TV, T1, T2](
@@ -355,7 +355,7 @@ def selectOrClosedWithin[TV, T1, T2](
     case source1.Received(v) => v
     case source2.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TimeoutValue, T1, T2, T3](
@@ -367,7 +367,7 @@ def selectOrClosedWithin[TimeoutValue, T1, T2, T3](
     case source2.Received(v) => v
     case source3.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TV, T1, T2, T3, T4](
@@ -385,7 +385,7 @@ def selectOrClosedWithin[TV, T1, T2, T3, T4](
     case source3.Received(v) => v
     case source4.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TV, T1, T2, T3, T4, T5](
@@ -411,7 +411,7 @@ def selectOrClosedWithin[TV, T1, T2, T3, T4, T5](
     case source4.Received(v) => v
     case source5.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** Select exactly one source from which to receive a value, within the given timeout. Sources should not repeat.
   *
diff --git a/core/src/main/scala/ox/computeIntensive.scala b/core/src/main/scala/ox/computeIntensive.scala
index 356496dc..ad8cfaed 100644
--- a/core/src/main/scala/ox/computeIntensive.scala
+++ b/core/src/main/scala/ox/computeIntensive.scala
@@ -1,5 +1,6 @@
 package ox
 
+import scala.compiletime.uninitialized
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.ExecutorService
@@ -11,7 +12,7 @@ import java.util.concurrent.atomic.AtomicInteger
 // setting & computing the executor
 //
 
-private var customComputeExecutor: ExecutorService = _
+private var customComputeExecutor: ExecutorService = uninitialized
 
 /** Sets the executor used to run computations passed to [[computeIntensive]]. Should be called once, at the start of the application,
   * before any [[computeIntensive]] calls; the executor's lifecycle (shutdown) is then the responsibility of the caller.
diff --git a/core/src/main/scala/ox/either.scala b/core/src/main/scala/ox/either.scala
index 5301b209..8a003442 100644
--- a/core/src/main/scala/ox/either.scala
+++ b/core/src/main/scala/ox/either.scala
@@ -58,7 +58,7 @@ object either:
           t match
             case Left(e)  => break(Left(e))
             case Right(a) => a
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case _: boundary.Label[Either[Nothing, Nothing]] =>
           error("The enclosing `either` call uses a different error type.\nIf it's explicitly typed, is the error type correct?")
         case _ => error("`.ok()` can only be used within an `either` call.\nIs it present?")
       }
@@ -77,7 +77,7 @@ object either:
       summonFrom {
         case given boundary.Label[Either[E, Nothing]] =>
           break(t.asInstanceOf[Either[E, Nothing]])
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case _: boundary.Label[Either[Nothing, Nothing]] =>
           error("The enclosing `either` call uses a different error type.\nIf it's explicitly typed, is the error type correct?")
         case _ => error("`.ok()` can only be used within an `either` call.\nIs it present?")
       }
@@ -91,7 +91,7 @@ object either:
           t match
             case None    => break(Left(()))
             case Some(a) => a
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case _: boundary.Label[Either[Nothing, Nothing]] =>
           error(
             "The enclosing `either` call uses a different error type.\nIf it's explicitly typed, is the error type correct?\nNote that for options, the error type must contain a `Unit`."
           )
@@ -110,8 +110,8 @@ object either:
     /** Unwrap the value of the `Option`, short-circuiting the computation to the enclosing [[either]] on guaranteed `None`. */
     transparent inline def ok(): A =
       summonFrom {
-        case given boundary.Label[Either[Unit, Nothing]]    => break(Left(()))
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case given boundary.Label[Either[Unit, Nothing]] => break(Left(()))
+        case _: boundary.Label[Either[Nothing, Nothing]] =>
           error(
             "The enclosing `either` call uses a different error type.\nIf it's explicitly typed, is the error type correct?\nNote that for options, the error type must contain a `Unit`."
           )
@@ -133,8 +133,8 @@ object either:
     /** Fail the computation, short-circuiting to the enclosing [[either]] block. */
     transparent inline def fail(): Nothing =
       summonFrom {
-        case given boundary.Label[Either[E, Nothing]]       => break(Left(e))
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case given boundary.Label[Either[E, Nothing]]    => break(Left(e))
+        case _: boundary.Label[Either[Nothing, Nothing]] =>
           error("The enclosing `either` call uses a different error type.\nIf it's explicitly typed, is the error type correct?")
       }
   end extension
diff --git a/core/src/main/scala/ox/flow/FlowCompanionOps.scala b/core/src/main/scala/ox/flow/FlowCompanionOps.scala
index 6d8da21f..43fc3559 100644
--- a/core/src/main/scala/ox/flow/FlowCompanionOps.scala
+++ b/core/src/main/scala/ox/flow/FlowCompanionOps.scala
@@ -161,7 +161,7 @@ trait FlowCompanionOps:
         case None        => false
 
   /** Create a flow which sleeps for the given `timeout` and then completes as done. */
-  def timeout[T](timeout: FiniteDuration): Flow[T] = usingEmitInline: emit =>
+  def timeout[T](timeout: FiniteDuration): Flow[T] = usingEmitInline: _ =>
     sleep(timeout)
 
   /** Creates a flow which concatenates the given `flows` in order. First elements from the first flow are emitted, then from the second
diff --git a/core/src/main/scala/ox/flow/FlowOps.scala b/core/src/main/scala/ox/flow/FlowOps.scala
index 23de8a7e..f81223e3 100644
--- a/core/src/main/scala/ox/flow/FlowOps.scala
+++ b/core/src/main/scala/ox/flow/FlowOps.scala
@@ -2,7 +2,6 @@ package ox.flow
 
 import ox.CancellableFork
 import ox.Fork
-import ox.Ox
 import ox.OxUnsupervised
 import ox.channels.BufferCapacity
 import ox.channels.Channel
@@ -623,17 +622,14 @@ class FlowOps[+T]:
     *   An upper bound on the number of child flows that run in parallel.
     */
   def flattenPar[U](parallelism: Int)(using T <:< Flow[U])(using BufferCapacity): Flow[U] = Flow.usingEmitInline: emit =>
-    case class Nested(child: Flow[U])
-    case object ChildDone
-
     unsupervised:
       val childOutputChannel = BufferCapacity.newChannel[U]
-      val childDoneChannel = Channel.unlimited[ChildDone.type]
+      val childDoneChannel = Channel.unlimited[Unit]
 
       // When an error occurs in the parent, propagating it also to `childOutputChannel`, from which we always
       // `select` in the main loop. That way, even if max parallelism is reached, errors in the parent will
       // be discovered without delay.
-      val parentChannel = outer.map(Nested(_)).onError(childOutputChannel.error(_).discard).runToChannel()
+      val parentChannel = outer.map(t => t: Flow[U]).onError(childOutputChannel.error(_).discard).runToChannel()
 
       var runningChannelCount = 1 // parent is running
       var parentDone = false
@@ -641,12 +637,13 @@ class FlowOps[+T]:
       while runningChannelCount > 0 do
         assert(runningChannelCount <= parallelism + 1)
 
-        val pool: List[Source[Nested] | Source[U] | Source[ChildDone.type]] =
+        val received =
           // +1, because of the parent channel.
-          if runningChannelCount == parallelism + 1 || parentDone then List(childOutputChannel, childDoneChannel)
-          else List(childOutputChannel, childDoneChannel, parentChannel)
+          if runningChannelCount == parallelism + 1 || parentDone
+          then selectOrClosed(childOutputChannel.receiveClause, childDoneChannel.receiveClause)
+          else selectOrClosed(childOutputChannel.receiveClause, childDoneChannel.receiveClause, parentChannel.receiveClause)
 
-        selectOrClosed(pool) match
+        received match
           // Only `parentChannel` might be done, child completion is signalled via `childDoneChannel`.
           case ChannelClosed.Done =>
             parentDone = isSourceDone(parentChannel)
@@ -656,16 +653,16 @@ class FlowOps[+T]:
 
           case e: ChannelClosed.Error => throw e.toThrowable
 
-          case ChildDone => runningChannelCount -= 1
+          case childDoneChannel.Received(_) => runningChannelCount -= 1
 
-          case Nested(t) =>
+          case parentChannel.Received(child) =>
             forkUnsupervised:
-              t.onDone(childDoneChannel.send(ChildDone)).runPipeToSink(childOutputChannel, propagateDone = false)
+              child.onDone(childDoneChannel.send(())).runPipeToSink(childOutputChannel, propagateDone = false)
             .discard
 
             runningChannelCount += 1
 
-          case u: U @unchecked => emit(u)
+          case childOutputChannel.Received(u) => emit(u)
         end match
       end while
   end flattenPar
@@ -918,7 +915,7 @@ class FlowOps[+T]:
     require(elements > 0, "elements must be > 0")
     require(per.toMillis > 0, "per time must be >= 1 ms")
     val emitEveryMillis = (per.toMillis / elements).millis
-    tap(t => sleep(emitEveryMillis))
+    tap(_ => sleep(emitEveryMillis))
   end throttle
 
   /** If this flow has no elements then elements from an `alternative` flow are emitted by the returned flow. If this flow is failed then
@@ -1249,7 +1246,7 @@ class FlowOps[+T]:
 
   /** Discard all elements emitted by this flow. The returned flow completes only when this flow completes (successfully or with an error).
     */
-  def drain(): Flow[Nothing] = Flow.usingEmitInline: emit =>
+  def drain(): Flow[Nothing] = Flow.usingEmitInline: _ =>
     last.run(FlowEmit.fromInline(_ => ()))
 
   /** Always runs `f` after the flow completes, whether it's because all elements are emitted, or when there's an error. */
diff --git a/core/src/main/scala/ox/flow/FlowReactiveOps.scala b/core/src/main/scala/ox/flow/FlowReactiveOps.scala
index d3d45be7..58a2c53d 100644
--- a/core/src/main/scala/ox/flow/FlowReactiveOps.scala
+++ b/core/src/main/scala/ox/flow/FlowReactiveOps.scala
@@ -8,7 +8,6 @@ import ox.channels.Sink
 import ox.channels.forkPropagate
 import ox.channels.selectOrClosed
 import ox.discard
-import ox.fork
 import ox.forkDiscard
 import ox.inScopeRunner
 import ox.tapException
diff --git a/core/src/main/scala/ox/flow/FlowTextOps.scala b/core/src/main/scala/ox/flow/FlowTextOps.scala
index dbb39761..0312b221 100644
--- a/core/src/main/scala/ox/flow/FlowTextOps.scala
+++ b/core/src/main/scala/ox/flow/FlowTextOps.scala
@@ -147,7 +147,7 @@ trait FlowTextOps[+T]:
       if str != null then output.apply(str)
       (newBuf, State.Pull)
 
-    def processByteOrderMark(bytes: T, buffer: Chunk[Byte], output: FlowEmit[String]): (Chunk[Byte], State) =
+    def processByteOrderMark(bytes: T, buffer: Chunk[Byte]): (Chunk[Byte], State) =
       // A common case, worth checking in advance
       if buffer == null && bytes.length >= bomSize && !bytes.startsWith(bomUtf8) then (bytes, State.Pull)
       else
@@ -168,7 +168,7 @@ trait FlowTextOps[+T]:
       last.run(
         FlowEmit.fromInline: t =>
           val (newBuffer, newState) = state match
-            case State.ProcessBOM => processByteOrderMark(t, buffer, emit)
+            case State.ProcessBOM => processByteOrderMark(t, buffer)
             case State.Pull       => doPull(t, buffer, emit)
 
           buffer = newBuffer
diff --git a/core/src/main/scala/ox/flow/internal/groupByImpl.scala b/core/src/main/scala/ox/flow/internal/groupByImpl.scala
index 3bd3ff14..98506a3b 100644
--- a/core/src/main/scala/ox/flow/internal/groupByImpl.scala
+++ b/core/src/main/scala/ox/flow/internal/groupByImpl.scala
@@ -15,6 +15,10 @@ import ox.unsupervised
 
 import scala.annotation.tailrec
 
+private enum ChildOutput[+U, +V]:
+  case Value(u: U)
+  case Done(v: V)
+
 private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predicate: T => V)(childFlowTransform: V => Flow[T] => Flow[U])(
     using BufferCapacity
 ): Flow[U] =
@@ -44,17 +48,15 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
     def shouldReceiveFromParentChannel: Boolean = parentDone || pendingFromParent.isDefined
   end GroupByState
 
-  case class ChildDone(v: V)
-
   // Running a pending child flow, after another has completed as done
-  def runChild_ifPending(state: GroupByState, childOutput: Sink[U | ChildDone])(using OxUnsupervised): GroupByState =
+  def runChild_ifPending(state: GroupByState, childOutput: Sink[ChildOutput[U, V]])(using OxUnsupervised): GroupByState =
     state.pendingFromParent match
       case Some((t, v, counter)) => sendToChild_orRunChild_orBuffer(state.copy(pendingFromParent = None), childOutput, t, v, counter)
       case None                  => state
 
   def sendToChild_orRunChild_orBuffer(
       state: GroupByState,
-      childOutput: Sink[U | ChildDone],
+      childOutput: Sink[ChildOutput[U, V]],
       t: T,
       v: V,
       counter: Long
@@ -67,14 +69,15 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
 
       case None if s.children.size < parallelism =>
         // Starting a new child flow, running in the background; the child flow receives values via a channel,
-        // and feeds its output to `childOutput`. Done signals are propagated as values, errors are propagated
+        // and feeds its output to `childOutput`. Elements and done signals are propagated as values, errors are propagated
         // as channel errors.
         val childChannel = BufferCapacity.newChannel[T]
         s = s.withChildAdded(v, childChannel)
 
         forkUnsupervised:
           childFlowTransform(v)(Flow.fromSource(childChannel))
-            .onDone(childOutput.sendOrClosed(ChildDone(v)).discard)
+            .map(ChildOutput.Value(_))
+            .onDone(childOutput.sendOrClosed(ChildOutput.Done(v)).discard)
             // When the child flow is done, making sure that the source channel becomes closed as well
             // otherwise, we'd be risking a deadlock, if there are `childChannel.send`-s pending, and the
             // buffer is full; if the channel is already closed, this is a no-op.
@@ -90,7 +93,7 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
         s = s.withPendingFromParent(t, v, counter)
 
         // Completing as done the child flow which didn't receive an element for the longest time. After
-        // the flow completes, it will send `ChildDone` to `childOutput`.
+        // the flow completes, it will send `ChildOutput.Done` to `childOutput`.
         s = s.withoutLongestInactiveChild.pipe { (vOpt, s2) =>
           vOpt.foreach(s.children(_).done())
           s2
@@ -107,13 +110,12 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
       // clauses. The elements from this channel are then emitted by the returned flow.
       // Completion of children (when the parent is complete, or the parallelism limit is reached) is signalled on
       // this channel as well.
-      val childOutput = BufferCapacity.newChannel[U | ChildDone]
+      val childOutput = BufferCapacity.newChannel[ChildOutput[U, V]]
 
       // Parent channel, from which we receive as long as it's not done, and only when a child flow isn't pending
       // creation (see below). As the receive is conditional, the errors that occur on this channel are also
       // propagated to `childOutput`, which is always the first (priority) clause in the main `select`.
-      case class FromParent(v: T)
-      val parentChannel = parent.map(FromParent(_)).onError(childOutput.errorOrClosed(_).discard).runToChannel()
+      val parentChannel = parent.onError(childOutput.errorOrClosed(_).discard).runToChannel()
 
       var state = GroupByState(parentDone = false, None, Map.empty, 0L, WeightedHeap())
 
@@ -122,16 +124,19 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
         assert(state.children.size <= parallelism) // invariant
 
         // We do not receive from the parent when it's done, or when there's already a pending child flow to create
-        // (but can't be created because of `parallelism` limit); we always receive from child output & child done
-        // signals. In case of parent's error, the error is also propagated above to `childOutput`, so that it's
-        // quickly received. Receiving from child output has priority over child done signals, to receive all child
-        // values before marking a child as done.
-        val pool =
+        // (but can't be created because of `parallelism` limit); we always receive from `childOutput`. In case of
+        // parent's error, the error is also propagated above to `childOutput`, so that it's quickly received.
+        // Child values and done signals share `childOutput`, so all of a child's values are received before its
+        // `Done`; with a separate done channel, `select` might pick the done signal first and drop values.
+        val received =
           if state.shouldReceiveFromParentChannel
-          then List(childOutput)
-          else List(childOutput, parentChannel)
+          then childOutput.receiveOrClosed()
+          else
+            selectOrClosed(childOutput.receiveClause, parentChannel.receiveClause) match
+              case childOutput.Received(c)                     => c
+              case r: (parentChannel.Received | ChannelClosed) => r
 
-        selectOrClosed(pool) match
+        received match
           case ChannelClosed.Done =>
             // Only the parent can be done; child completion is signalled via a value in `childOutput`.
             state = state.withParentDone(isSourceDone(parentChannel))
@@ -147,11 +152,11 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
 
           case e: ChannelClosed.Error => throw e.toThrowable
 
-          case FromParent(t) =>
+          case parentChannel.Received(t) =>
             state = state.withFromParentCounterIncremented
             state = sendToChild_orRunChild_orBuffer(state, childOutput, t, predicate(t), state.fromParentCounter)
 
-          case ChildDone(v) =>
+          case ChildOutput.Done(v) =>
             state = state.withChildRemoved(v)
 
             // Children should only be done because their `childChannel` was completed as done by
@@ -166,7 +171,7 @@ private[flow] def groupByImpl[T, V, U](parent: Flow[T], parallelism: Int, predic
 
             state = runChild_ifPending(state, childOutput)
 
-          case u: U @unchecked => emit(u) // forwarding from `childOutput`
+          case ChildOutput.Value(u) => emit(u)
         end match
       end while
 end groupByImpl
diff --git a/core/src/main/scala/ox/fork.scala b/core/src/main/scala/ox/fork.scala
index 226e6250..bb5dfff5 100644
--- a/core/src/main/scala/ox/fork.scala
+++ b/core/src/main/scala/ox/fork.scala
@@ -182,7 +182,7 @@ def forkCancellable[T](f: => T)(using OxUnsupervised): CancellableFork[T] =
     catch
       // if this thread was interrupted, any context is already captured as part of the thread started in the nested scope
       // hence, ignoring this exception, so that it's not logged by the uncaught exception handler
-      case e: InterruptedException =>
+      case _: InterruptedException =>
 
   new ForkUsingResult(result) with CancellableFork[T]:
     override def cancel(): Either[Throwable, T] =
diff --git a/core/src/main/scala/ox/local.scala b/core/src/main/scala/ox/local.scala
index 19aefdf8..f59eae9e 100644
--- a/core/src/main/scala/ox/local.scala
+++ b/core/src/main/scala/ox/local.scala
@@ -42,6 +42,6 @@ object ForkLocal:
   * stored in a map bound to the concurrency scope, and the binding can only be updated by starting a new concurrency scope (same as with
   * the `ScopedValue`-based implementation). That way locals never escape their scope.
   */
-private[ox] class ForkLocalMap(storage: Map[ForkLocal[_], Any]):
+private[ox] class ForkLocalMap(storage: Map[ForkLocal[?], Any]):
   def get[T](key: ForkLocal[T]): Option[T] = storage.get(key).asInstanceOf[Option[T]]
   def set[T](key: ForkLocal[T], value: T): ForkLocalMap = new ForkLocalMap(storage.updated(key, value))
diff --git a/core/src/main/scala/ox/oxThreadFactory.scala b/core/src/main/scala/ox/oxThreadFactory.scala
index 68f20fc0..fbbd30a6 100644
--- a/core/src/main/scala/ox/oxThreadFactory.scala
+++ b/core/src/main/scala/ox/oxThreadFactory.scala
@@ -1,8 +1,9 @@
 package ox
 
+import scala.compiletime.uninitialized
 import java.util.concurrent.ThreadFactory
 
-private var customThreadFactory: ThreadFactory = _
+private var customThreadFactory: ThreadFactory = uninitialized
 
 /** @see [[oxThreadFactory]] */
 def setOxThreadFactory(tf: ThreadFactory): Unit =
diff --git a/core/src/main/scala/ox/resilience/CircuitBreakerStateMachine.scala b/core/src/main/scala/ox/resilience/CircuitBreakerStateMachine.scala
index 0b49bab0..26448876 100644
--- a/core/src/main/scala/ox/resilience/CircuitBreakerStateMachine.scala
+++ b/core/src/main/scala/ox/resilience/CircuitBreakerStateMachine.scala
@@ -120,26 +120,24 @@ private[resilience] object CircuitBreakerResults:
     private val results = new collection.mutable.ArrayDeque[CircuitBreakerResult](windowSize + 1)
     private var slowCalls = 0
     private var failedCalls = 0
-    private var successCalls = 0
 
     private def clearResults(): Unit =
       results.clear()
       slowCalls = 0
       failedCalls = 0
-      successCalls = 0
 
     def onStateChange(oldState: CircuitBreakerState, newState: CircuitBreakerState): Unit =
       if !oldState.isSameState(newState) then clearResults()
 
     def updateResults(result: CircuitBreakerResult): Unit =
       result match
-        case CircuitBreakerResult.Success => successCalls += 1
+        case CircuitBreakerResult.Success => ()
         case CircuitBreakerResult.Failure => failedCalls += 1
         case CircuitBreakerResult.Slow    => slowCalls += 1
       val resultingQueue = results.addOne(result)
       if resultingQueue.length > windowSize then
         resultingQueue.removeHeadOption(false) match
-          case Some(CircuitBreakerResult.Success) => successCalls -= 1
+          case Some(CircuitBreakerResult.Success) => ()
           case Some(CircuitBreakerResult.Failure) => failedCalls -= 1
           case Some(CircuitBreakerResult.Slow)    => slowCalls -= 1
           case None                               => ()
@@ -164,20 +162,18 @@ private[resilience] object CircuitBreakerResults:
     private val results = collection.mutable.ArrayDeque[(Long, CircuitBreakerResult)]()
     private var slowCalls = 0
     private var failedCalls = 0
-    private var successCalls = 0
 
     private def clearResults(): Unit =
       results.clear()
       slowCalls = 0
       failedCalls = 0
-      successCalls = 0
 
     def calculateMetrics(lastAcquisitionResult: Option[AcquireResult], timestamp: Long): Metrics =
       // filter all entries that happened outside sliding window
       val removed = results.removeHeadWhile((time, _) => timestamp > time + windowDuration.toMillis)
       removed.foreach { (_, result) =>
         result match
-          case CircuitBreakerResult.Success => successCalls -= 1
+          case CircuitBreakerResult.Success => ()
           case CircuitBreakerResult.Failure => failedCalls -= 1
           case CircuitBreakerResult.Slow    => slowCalls -= 1
       }
@@ -195,7 +191,7 @@ private[resilience] object CircuitBreakerResults:
 
     def updateResults(result: CircuitBreakerResult): Unit =
       result match
-        case CircuitBreakerResult.Success => successCalls += 1
+        case CircuitBreakerResult.Success => ()
         case CircuitBreakerResult.Failure => failedCalls += 1
         case CircuitBreakerResult.Slow    => slowCalls += 1
       results.addOne((System.currentTimeMillis(), result))
diff --git a/core/src/main/scala/ox/resource.scala b/core/src/main/scala/ox/resource.scala
index 82f36c8f..4337034e 100644
--- a/core/src/main/scala/ox/resource.scala
+++ b/core/src/main/scala/ox/resource.scala
@@ -11,9 +11,8 @@ import scala.util.NotGiven
 )
 opaque type NoEnclosingConcurrencyScope = Unit
 
-/** Scala 3.9+ erases a result of type `NoEnclosingConcurrencyScope` to `void`, earlier versions to `BoxedUnit`. Overriding this method
-  * makes the compiler emit both signatures, so code compiled with 3.9+ links against Ox (https://github.com/scala/scala3/issues/24653).
-  */
+// Scala 3.9+ erases NoEnclosingConcurrencyScope results to void, older versions to BoxedUnit;
+// kept for binary compatibility (https://github.com/scala/scala3/issues/24653)
 private[ox] trait NoEnclosingConcurrencyScopeVoidErasure:
   def noEnclosingConcurrencyScope(using NotGiven[OxUnsupervised]): Unit
 
@@ -21,6 +20,13 @@ object NoEnclosingConcurrencyScope extends NoEnclosingConcurrencyScopeVoidErasur
   // in the companion, so that it's found via the implicit scope of the type, without any imports
   given noEnclosingConcurrencyScope(using NotGiven[OxUnsupervised]): NoEnclosingConcurrencyScope = ()
 
+  // the BoxedUnit variant (the pre-3.9 erasure of the given above, which is now void), for binary compatibility with code
+  // compiled against Ox built with Scala 3.3 (https://github.com/scala/scala3/issues/24653)
+  @targetName("noEnclosingConcurrencyScope")
+  private[ox] def noEnclosingConcurrencyScopeBoxedUnit(using NotGiven[OxUnsupervised]): scala.runtime.BoxedUnit =
+    scala.runtime.BoxedUnit.UNIT
+end NoEnclosingConcurrencyScope
+
 /** Starts a new resource scope: within the given code block `f`, resources can be registered using [[useInScope]] and
   * [[releaseAfterScope]]. They are released, in reverse registration order, once `f` completes (either successfully or with an exception).
   * Releasing is [[uninterruptible]]. A resource scope is not a concurrency scope: no forks can be started and no [[ForkLocal]] values can
diff --git a/core/src/test/scala/ox/CollectParTest.scala b/core/src/test/scala/ox/CollectParTest.scala
index 521425aa..ba2811f3 100644
--- a/core/src/test/scala/ox/CollectParTest.scala
+++ b/core/src/test/scala/ox/CollectParTest.scala
@@ -13,7 +13,7 @@ class CollectParTest extends AnyFlatSpec with Matchers:
   "collectPar" should "output the same type as input" in {
     val input = List(1, 2, 3)
     val result = input.collectPar(1)(x => x)
-    result shouldBe a[List[_]]
+    result shouldBe a[List[?]]
   }
 
   it should "run computations in parallel" in {
diff --git a/core/src/test/scala/ox/FilterParTest.scala b/core/src/test/scala/ox/FilterParTest.scala
index 5186b62e..dcad3f9b 100644
--- a/core/src/test/scala/ox/FilterParTest.scala
+++ b/core/src/test/scala/ox/FilterParTest.scala
@@ -14,7 +14,7 @@ class FilterParTest extends AnyFlatSpec with Matchers:
   "filterPar" should "output the same type as input" in {
     val input = List(1, 2, 3)
     val result = input.filterPar(1)(_ => true)
-    result shouldBe a[List[_]]
+    result shouldBe a[List[?]]
   }
 
   it should "run computations in parallel" in {
diff --git a/core/src/test/scala/ox/MapParTest.scala b/core/src/test/scala/ox/MapParTest.scala
index cd17c43d..9c279917 100644
--- a/core/src/test/scala/ox/MapParTest.scala
+++ b/core/src/test/scala/ox/MapParTest.scala
@@ -13,7 +13,7 @@ class MapParTest extends AnyFlatSpec with Matchers:
   "mapPar" should "output the same type as input" in {
     val input = List(1, 2, 3)
     val result = input.mapPar(1)(identity)
-    result shouldBe a[List[_]]
+    result shouldBe a[List[?]]
   }
 
   it should "run computations in parallel" in {
diff --git a/core/src/test/scala/ox/flow/FlowOpsCollectTest.scala b/core/src/test/scala/ox/flow/FlowOpsCollectTest.scala
index 2b5d71e4..9da9e892 100644
--- a/core/src/test/scala/ox/flow/FlowOpsCollectTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsCollectTest.scala
@@ -8,7 +8,7 @@ class FlowOpsCollectTest extends AnyFlatSpec with Matchers:
   behavior of "collect"
 
   it should "collect over a source" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
 
     val s = c.collect:
       case i if i % 2 == 0 => i * 10
diff --git a/core/src/test/scala/ox/flow/FlowOpsDebounceByTest.scala b/core/src/test/scala/ox/flow/FlowOpsDebounceByTest.scala
index e6f6695d..d84232a2 100644
--- a/core/src/test/scala/ox/flow/FlowOpsDebounceByTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsDebounceByTest.scala
@@ -13,7 +13,7 @@ class FlowOpsDebounceByTest extends AnyFlatSpec with Matchers:
     s.runToList() shouldBe List.empty
 
   it should "not debounce if applied on a flow containing only distinct f(value)" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val s = c.debounceBy(_ * 2)
     s.runToList() shouldBe (1 to 10)
 
diff --git a/core/src/test/scala/ox/flow/FlowOpsDebounceTest.scala b/core/src/test/scala/ox/flow/FlowOpsDebounceTest.scala
index b80939c6..04fb96ca 100644
--- a/core/src/test/scala/ox/flow/FlowOpsDebounceTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsDebounceTest.scala
@@ -13,7 +13,7 @@ class FlowOpsDebounceTest extends AnyFlatSpec with Matchers:
     s.runToList() shouldBe List.empty
 
   it should "not debounce if applied on a flow containing only distinct values" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val s = c.debounce
     s.runToList() shouldBe (1 to 10)
 
diff --git a/core/src/test/scala/ox/flow/FlowOpsFilterTest.scala b/core/src/test/scala/ox/flow/FlowOpsFilterTest.scala
index a76d2bd7..e6ad6c35 100644
--- a/core/src/test/scala/ox/flow/FlowOpsFilterTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsFilterTest.scala
@@ -13,17 +13,17 @@ class FlowOpsFilterTest extends AnyFlatSpec with Matchers:
     s.runToList() shouldBe List.empty
 
   it should "filter out everything if no element meets 'f'" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val s = c.filter(_ => false)
     s.runToList() shouldBe List.empty
 
   it should "not filter anything if all the elements meet 'f'" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val s = c.filter(_ => true)
     s.runToList() shouldBe (1 to 10)
 
   it should "filter out elements that don't meet 'f'" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val s = c.filter(_ % 2 == 0)
     s.runToList() shouldBe (2 to 10 by 2)
 
diff --git a/core/src/test/scala/ox/flow/FlowOpsSampleTest.scala b/core/src/test/scala/ox/flow/FlowOpsSampleTest.scala
index d711c55b..5f03ad40 100644
--- a/core/src/test/scala/ox/flow/FlowOpsSampleTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsSampleTest.scala
@@ -13,18 +13,18 @@ class FlowOpsSampleTest extends AnyFlatSpec with Matchers:
     s.runToList() shouldBe List.empty
 
   it should "not sample anything when 'n == 0'" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val s = c.sample(0)
     s.runToList() shouldBe List.empty
 
   it should "sample every element of the flow when 'n == 1'" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val n = 1
     val s = c.sample(n)
     s.runToList() shouldBe (n to 10 by n)
 
   it should "sample every nth element of the flow" in:
-    val c = Flow.fromValues(1 to 10: _*)
+    val c = Flow.fromValues(1 to 10*)
     val n = 3
     val s = c.sample(n)
     s.runToList() shouldBe (n to 10 by n)
diff --git a/core/src/test/scala/ox/flow/FlowOpsScanTest.scala b/core/src/test/scala/ox/flow/FlowOpsScanTest.scala
index 50a9e6ce..5f11629f 100644
--- a/core/src/test/scala/ox/flow/FlowOpsScanTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsScanTest.scala
@@ -13,12 +13,12 @@ class FlowOpsScanTest extends AnyFlatSpec with Matchers:
     scannedFlow.runToList() shouldBe List(0)
 
   it should "scan a flow of summed Int" in:
-    val flow = Flow.fromValues(1 to 10: _*)
+    val flow = Flow.fromValues(1 to 10*)
     val scannedFlow = flow.scan(0)((acc, el) => acc + el)
     scannedFlow.runToList() shouldBe List(0, 1, 3, 6, 10, 15, 21, 28, 36, 45, 55)
 
   it should "scan a flow of multiplied Int" in:
-    val flow = Flow.fromValues(1 to 10: _*)
+    val flow = Flow.fromValues(1 to 10*)
     val scannedFlow = flow.scan(1)((acc, el) => acc * el)
     scannedFlow.runToList() shouldBe List(1, 1, 2, 6, 24, 120, 720, 5040, 40320, 362880, 3628800)
 
diff --git a/core/src/test/scala/ox/flow/FlowOpsZipWithIndexTest.scala b/core/src/test/scala/ox/flow/FlowOpsZipWithIndexTest.scala
index 079f40fd..ec70d580 100644
--- a/core/src/test/scala/ox/flow/FlowOpsZipWithIndexTest.scala
+++ b/core/src/test/scala/ox/flow/FlowOpsZipWithIndexTest.scala
@@ -13,7 +13,7 @@ class FlowOpsZipWithIndexTest extends AnyFlatSpec with Matchers with Eventually:
     s.runToList() shouldBe List.empty
 
   it should "zip flow with index" in:
-    val c = Flow.fromValues(1 to 5: _*)
+    val c = Flow.fromValues(1 to 5*)
     val s = c.zipWithIndex
     s.runToList() shouldBe List((1, 0), (2, 1), (3, 2), (4, 3), (5, 4))
 
diff --git a/doc/integrations/kafka.md b/doc/integrations/kafka.md
index 8272f987..1b54af42 100644
--- a/doc/integrations/kafka.md
+++ b/doc/integrations/kafka.md
@@ -80,7 +80,7 @@ import ox.kafka.ReceivedMessage
 
 case class SendPacket[K, V](
   send: List[ProducerRecord[K, V]], 
-  commit: List[ReceivedMessage[_, _]])
+  commit: List[ReceivedMessage[?, ?]])
 ```
 
 The `send` list contains the messages to be sent (each message is a Kafka `ProducerRecord`). The `commit` list contains
diff --git a/doc/structured-concurrency/fork-join.md b/doc/structured-concurrency/fork-join.md
index 21c713ae..c6e335b9 100644
--- a/doc/structured-concurrency/fork-join.md
+++ b/doc/structured-concurrency/fork-join.md
@@ -80,7 +80,7 @@ are cancelled (using interruption). Once all forks complete, the exception is pr
 the `supervised` method invocation:
 
 ```scala mdoc:compile-only
-import ox.{fork, forkUser, Ox, sleep, supervised}
+import ox.{fork, forkUser, sleep, supervised}
 import scala.concurrent.duration.*
 
 supervised {
diff --git a/kafka/src/main/scala/ox/kafka/KafkaStage.scala b/kafka/src/main/scala/ox/kafka/KafkaStage.scala
index 17a2e688..1ce2592b 100644
--- a/kafka/src/main/scala/ox/kafka/KafkaStage.scala
+++ b/kafka/src/main/scala/ox/kafka/KafkaStage.scala
@@ -78,7 +78,7 @@ object KafkaStage:
         // packets which are fully sent, and should be committed
         // using an unlimited buffer so that the I/O thread doesn't get blocked in producer.send callbacks; backpressure is provided
         // by creating a buffered channel in `flow.runToChannel()` below
-        val toCommit = Channel.unlimited[SendPacket[_, _]]
+        val toCommit = Channel.unlimited[SendPacket[?, ?]]
 
         // used to reorder values received from `metadata` using the assigned sequence numbers
         val sendInSequence = SendInSequence(emit)
@@ -174,7 +174,7 @@ object KafkaStage:
       producer: KafkaProducer[K, V],
       packet: SendPacket[K, V],
       sendInSequence: SendInSequence[RecordMetadata],
-      toCommit: Sink[SendPacket[_, _]],
+      toCommit: Sink[SendPacket[?, ?]],
       exceptions: Sink[Exception],
       metadata: Sink[(Long, RecordMetadata)],
       commitOffsets: Boolean
@@ -201,7 +201,7 @@ end KafkaStage
 private class SendInSequence[T](emit: FlowEmit[T]):
   private var sequenceNoNext = 0L
   private var sequenceNoToSendNext = 0L
-  private val toSend = mutable.SortedSet[(Long, T)]()(Ordering.by(_._1))
+  private val toSend = mutable.SortedSet[(Long, T)]()(using Ordering.by(_._1))
 
   def nextSequenceNo: Long =
     val n = sequenceNoNext
diff --git a/kafka/src/main/scala/ox/kafka/kafkaOffsetCommit.scala b/kafka/src/main/scala/ox/kafka/kafkaOffsetCommit.scala
index 075781a6..1c25abb5 100644
--- a/kafka/src/main/scala/ox/kafka/kafkaOffsetCommit.scala
+++ b/kafka/src/main/scala/ox/kafka/kafkaOffsetCommit.scala
@@ -11,7 +11,7 @@ import ox.flow.Flow
 
 /** Common trait for packets that contain messages to commit. */
 trait HasCommit:
-  def commit: List[ReceivedMessage[_, _]]
+  def commit: List[ReceivedMessage[?, ?]]
 
 private[kafka] def doCommit[K, V](consumer: ActorRef[KafkaConsumerWrapper[K, V]], packets: Source[HasCommit]): Unit =
   val commitInterval = 1.second
@@ -39,7 +39,7 @@ private[kafka] def doCommit[K, V](consumer: ActorRef[KafkaConsumerWrapper[K, V]]
   commitAll()
 end doCommit
 
-case class SendPacket[K, V](send: List[ProducerRecord[K, V]], commit: List[ReceivedMessage[_, _]]) extends HasCommit
+case class SendPacket[K, V](send: List[ProducerRecord[K, V]], commit: List[ReceivedMessage[?, ?]]) extends HasCommit
 
 object SendPacket:
   def apply[K, V](send: ProducerRecord[K, V], commit: ReceivedMessage[?, ?]): SendPacket[K, V] =
@@ -49,7 +49,7 @@ object SendPacket:
     SendPacket(send, List(commit))
 
 /** A packet containing only commit messages (consumer records) to be committed. */
-case class CommitPacket(commit: List[ReceivedMessage[_, _]]) extends HasCommit
+case class CommitPacket(commit: List[ReceivedMessage[?, ?]]) extends HasCommit
 
 object CommitPacket:
   def apply(commit: ReceivedMessage[?, ?]): CommitPacket =
diff --git a/kafka/src/test/scala/ox/kafka/KafkaTest.scala b/kafka/src/test/scala/ox/kafka/KafkaTest.scala
index b6adbeaf..16f338a7 100644
--- a/kafka/src/test/scala/ox/kafka/KafkaTest.scala
+++ b/kafka/src/test/scala/ox/kafka/KafkaTest.scala
@@ -1,5 +1,6 @@
 package ox.kafka
 
+import scala.compiletime.uninitialized
 import io.github.embeddedkafka.EmbeddedKafka
 import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}
 import org.apache.kafka.common.serialization.{Deserializer, StringDeserializer}
@@ -14,7 +15,7 @@ import scala.concurrent.duration.*
 import ox.flow.Flow
 
 class KafkaTest extends AnyFlatSpec with Matchers with EmbeddedKafka with BeforeAndAfterAll:
-  private var bootstrapServer: String = _
+  private var bootstrapServer: String = uninitialized
 
   override def beforeAll(): Unit =
     bootstrapServer = s"localhost:${EmbeddedKafka.start().config.kafkaPort}"