Restore unconfined dispatcher usage

This ensures all values that can be seen will be seen by recording to the channel in the same stackframe as the upstream emission.

src/commonMain/kotlin/app/cash/turbine/flow.kt

@@ -20,6 +20,7 @@ import kotlin.time.Duration.Companion.seconds
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
+import kotlinx.coroutines.Dispatchers.Unconfined
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.coroutineScope
@@ -110,7 +111,7 @@ public fun <T> Flow<T>.testIn(scope: CoroutineScope): ReceiveTurbine<T> {
 private fun <T> Flow<T>.collectTurbineIn(scope: CoroutineScope): Turbine<T> {
   lateinit var channel: Channel<T>
 
-  val job = scope.launch(start = UNDISPATCHED) {
+  val job = scope.launch(Unconfined, start = UNDISPATCHED) {
     channel = collectIntoChannel(this)
   }
 

src/commonTest/kotlin/app/cash/turbine/FlowTest.kt

@@ -29,6 +29,7 @@ import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.RENDEZVOUS
 import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.flow
@@ -471,6 +472,19 @@ class FlowTest {
       }
   }
 
+  @Test fun valuesDoNotConflate() = runTest {
+    val flow = MutableStateFlow(0)
+    flow.test {
+      flow.value = 1
+      flow.value = 2
+      flow.value = 3
+      assertEquals(0, awaitItem())
+      assertEquals(1, awaitItem())
+      assertEquals(2, awaitItem())
+      assertEquals(3, awaitItem())
+    }
+  }
+
   @Test fun assertNullValuesWithExpectMostRecentItem() = runTest {
     flowOf(1, 2, null).test {
       assertEquals(null, expectMostRecentItem())