diff --git a/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/RenderWorkflowInTest.kt b/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/RenderWorkflowInTest.kt
index 5d770e6a3..79b8e3d97 100644
--- a/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/RenderWorkflowInTest.kt
+++ b/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/RenderWorkflowInTest.kt
@@ -6,44 +6,30 @@ import com.squareup.workflow1.internal.ParameterizedTestRunner
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineExceptionHandler
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers.Unconfined
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.consumeAsFlow
-import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.produceIn
+import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.plus
 import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.advanceUntilIdle
-import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
 import okio.ByteString
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
 
-@OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class, WorkflowExperimentalRuntime::class)
+@OptIn(ExperimentalCoroutinesApi::class, WorkflowExperimentalRuntime::class)
 class RenderWorkflowInTest {
 
-  /**
-   * A [TestScope] that will not run until explicitly told to.
-   */
-  private lateinit var pausedTestScope: TestScope
-
-  /**
-   * A [TestScope] that will automatically dispatch enqueued routines.
-   */
-  private lateinit var testScope: TestScope
-
   private val traces: StringBuilder = StringBuilder()
   private val testTracer: WorkflowTracer = object : WorkflowTracer {
     var prefix: String = ""
@@ -58,23 +44,27 @@ class RenderWorkflowInTest {
     }
   }
 
-  private val runtimeOptions: Sequence<Pair<RuntimeConfig, WorkflowTracer?>> = arrayOf(
-    RuntimeConfigOptions.RENDER_PER_ACTION to null,
-    RuntimeConfigOptions.RENDER_PER_ACTION to testTracer,
-    setOf(RENDER_ONLY_WHEN_STATE_CHANGES) to null,
-    setOf(RENDER_ONLY_WHEN_STATE_CHANGES) to testTracer,
-    setOf(CONFLATE_STALE_RENDERINGS) to null,
-    setOf(CONFLATE_STALE_RENDERINGS) to testTracer,
-    setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES) to null,
-    setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES) to testTracer,
-  ).asSequence()
+  private val runtimes = setOf<RuntimeConfig>(
+    RuntimeConfigOptions.RENDER_PER_ACTION,
+    setOf(RENDER_ONLY_WHEN_STATE_CHANGES),
+    setOf(CONFLATE_STALE_RENDERINGS),
+    setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES)
+  )
+
+  private val tracerOptions = setOf<WorkflowTracer?>(
+    null,
+    testTracer
+  )
+
+  private val runtimeOptions: Sequence<Pair<RuntimeConfig, WorkflowTracer?>> = cartesianProduct(
+    runtimes.asSequence(),
+    tracerOptions.asSequence()
+  )
 
   private val runtimeTestRunner = ParameterizedTestRunner<Pair<RuntimeConfig, WorkflowTracer?>>()
 
   private fun setup() {
     traces.clear()
-    pausedTestScope = TestScope()
-    testScope = TestScope(UnconfinedTestDispatcher())
   }
 
   @Test fun initial_rendering_is_calculated_synchronously() {
@@ -82,18 +72,20 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val props = MutableStateFlow("foo")
-      val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
-      // Don't allow the workflow runtime to actually start.
-
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = pausedTestScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      assertEquals("props: foo", renderings.value.rendering)
+      runTest(UnconfinedTestDispatcher()) {
+        val props = MutableStateFlow("foo")
+        val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
+        // Don't allow the workflow runtime to actually start.
+
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        assertEquals("props: foo", renderings.value.rendering)
+      }
     }
   }
 
@@ -102,18 +94,21 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val props = MutableStateFlow("foo")
-      val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
-
-      pausedTestScope.cancel()
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = pausedTestScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      assertEquals("props: foo", renderings.value.rendering)
+      runTest(UnconfinedTestDispatcher()) {
+        val props = MutableStateFlow("foo")
+        val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
+
+        val testScope = TestScope(testScheduler)
+        testScope.cancel()
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        assertEquals("props: foo", renderings.value.rendering)
+      }
     }
   }
 
@@ -123,24 +118,27 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var sideEffectWasRan = false
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect("test") {
-          sideEffectWasRan = true
+      runTest(UnconfinedTestDispatcher()) {
+        var sideEffectWasRan = false
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect("test") {
+            sideEffectWasRan = true
+          }
         }
-      }
 
-      testScope.cancel()
-      renderWorkflowIn(
-        workflow,
-        testScope,
-        MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      testScope.advanceUntilIdle()
-
-      assertFalse(sideEffectWasRan)
+        val testScope = TestScope(testScheduler)
+        testScope.cancel()
+        renderWorkflowIn(
+          workflow,
+          testScope,
+          MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        advanceUntilIdle()
+
+        assertFalse(sideEffectWasRan)
+      }
     }
   }
 
@@ -150,27 +148,30 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var sideEffectWasRan = false
-      val childWorkflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect("test") {
-          sideEffectWasRan = true
+      runTest(UnconfinedTestDispatcher()) {
+        var sideEffectWasRan = false
+        val childWorkflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect("test") {
+            sideEffectWasRan = true
+          }
+        }
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          renderChild(childWorkflow)
         }
-      }
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        renderChild(childWorkflow)
-      }
 
-      testScope.cancel()
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      testScope.advanceUntilIdle()
-
-      assertFalse(sideEffectWasRan)
+        val testScope = TestScope(testScheduler)
+        testScope.cancel()
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        advanceUntilIdle()
+
+        assertFalse(sideEffectWasRan)
+      }
     }
   }
 
@@ -179,65 +180,31 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val props = MutableStateFlow("foo")
-      val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      assertEquals("props: foo", renderings.value.rendering)
-
-      props.value = "bar"
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-
-      assertEquals("props: bar", renderings.value.rendering)
+      runTest(UnconfinedTestDispatcher()) {
+        val props = MutableStateFlow("foo")
+        val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+
+        assertEquals("props: foo", renderings.value.rendering)
+
+        props.value = "bar"
+
+        assertEquals("props: bar", renderings.value.rendering)
+      }
     }
   }
 
-  private val runtimeMatrix: Sequence<Pair<RuntimeConfig, RuntimeConfig>> = arrayOf(
-    Pair(RuntimeConfigOptions.RENDER_PER_ACTION, RuntimeConfigOptions.RENDER_PER_ACTION),
-    Pair(RuntimeConfigOptions.RENDER_PER_ACTION, setOf(RENDER_ONLY_WHEN_STATE_CHANGES)),
-    Pair(RuntimeConfigOptions.RENDER_PER_ACTION, setOf(CONFLATE_STALE_RENDERINGS)),
-    Pair(
-      RuntimeConfigOptions.RENDER_PER_ACTION,
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES)
-    ),
-    Pair(setOf(RENDER_ONLY_WHEN_STATE_CHANGES), RuntimeConfigOptions.RENDER_PER_ACTION),
-    Pair(setOf(RENDER_ONLY_WHEN_STATE_CHANGES), setOf(RENDER_ONLY_WHEN_STATE_CHANGES)),
-    Pair(setOf(RENDER_ONLY_WHEN_STATE_CHANGES), setOf(CONFLATE_STALE_RENDERINGS)),
-    Pair(
-      setOf(RENDER_ONLY_WHEN_STATE_CHANGES),
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES)
-    ),
-    Pair(setOf(CONFLATE_STALE_RENDERINGS), RuntimeConfigOptions.RENDER_PER_ACTION),
-    Pair(setOf(CONFLATE_STALE_RENDERINGS), setOf(RENDER_ONLY_WHEN_STATE_CHANGES)),
-    Pair(setOf(CONFLATE_STALE_RENDERINGS), setOf(CONFLATE_STALE_RENDERINGS)),
-    Pair(
-      setOf(CONFLATE_STALE_RENDERINGS),
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES)
-    ),
-    Pair(
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES),
-      RuntimeConfigOptions.RENDER_PER_ACTION
-    ),
-    Pair(
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES),
-      setOf(RENDER_ONLY_WHEN_STATE_CHANGES)
-    ),
-    Pair(
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES),
-      setOf(CONFLATE_STALE_RENDERINGS)
-    ),
-    Pair(
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES),
-      setOf(CONFLATE_STALE_RENDERINGS, RENDER_ONLY_WHEN_STATE_CHANGES)
-    ),
-  ).asSequence()
+  private val runtimeMatrix: Sequence<Pair<RuntimeConfig, RuntimeConfig>> = cartesianProduct(
+    runtimes.asSequence(),
+    runtimes.asSequence()
+  )
+
   private val runtimeMatrixTestRunner =
     ParameterizedTestRunner<Pair<RuntimeConfig, RuntimeConfig>>()
 
@@ -246,60 +213,59 @@ class RenderWorkflowInTest {
       paramSource = runtimeMatrix,
       before = ::setup,
     ) { (runtimeConfig1, runtimeConfig2) ->
-      val workflow = Workflow.stateful<Unit, String, Nothing, Pair<String, (String) -> Unit>>(
-        initialState = { _, snapshot ->
-          snapshot?.bytes?.parse { it.readUtf8WithLength() } ?: "initial state"
-        },
-        snapshot = { state ->
-          Snapshot.write { it.writeUtf8WithLength(state) }
-        },
-        render = { _, renderState ->
-          Pair(
-            renderState,
-            { newState -> actionSink.send(action("") { state = newState }) }
-          )
-        }
-      )
-      val props = MutableStateFlow(Unit)
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = props,
-        runtimeConfig = runtimeConfig1,
-        workflowTracer = null,
-      ) {}
-
-      // Interact with the workflow to change the state.
-      renderings.value.rendering.let { (state, updateState) ->
-        runtimeMatrixTestRunner.assertEquals("initial state", state)
-        updateState("updated state")
-      }
-
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-
-      val snapshot = renderings.value.let { (rendering, snapshot) ->
-        val (state, updateState) = rendering
-        runtimeMatrixTestRunner.assertEquals("updated state", state)
-        updateState("ignored rendering")
-        return@let snapshot
-      }
-
-      // Create a new scope to launch a second runtime to restore.
-      val restoreScope = TestScope()
-      val restoredRenderings =
-        renderWorkflowIn(
+      runTest(UnconfinedTestDispatcher()) {
+        val workflow = Workflow.stateful<Unit, String, Nothing, Pair<String, (String) -> Unit>>(
+          initialState = { _, snapshot ->
+            snapshot?.bytes?.parse { it.readUtf8WithLength() } ?: "initial state"
+          },
+          snapshot = { state ->
+            Snapshot.write { it.writeUtf8WithLength(state) }
+          },
+          render = { _, renderState ->
+            Pair(
+              renderState,
+              { newState -> actionSink.send(action("") { state = newState }) }
+            )
+          }
+        )
+        val props = MutableStateFlow(Unit)
+        val renderings = renderWorkflowIn(
           workflow = workflow,
-          scope = restoreScope,
+          scope = backgroundScope,
           props = props,
-          initialSnapshot = snapshot,
+          runtimeConfig = runtimeConfig1,
           workflowTracer = null,
-          runtimeConfig = runtimeConfig2
         ) {}
-      runtimeMatrixTestRunner.assertEquals(
-        "updated state",
-        restoredRenderings.value.rendering.first
-      )
+
+        // Interact with the workflow to change the state.
+        renderings.value.rendering.let { (state, updateState) ->
+          runtimeMatrixTestRunner.assertEquals("initial state", state)
+          updateState("updated state")
+        }
+
+        val snapshot = renderings.value.let { (rendering, snapshot) ->
+          val (state, updateState) = rendering
+          runtimeMatrixTestRunner.assertEquals("updated state", state)
+          updateState("ignored rendering")
+          return@let snapshot
+        }
+
+        // Create a new scope to launch a second runtime to restore.
+        val restoreScope = TestScope(testScheduler)
+        val restoredRenderings =
+          renderWorkflowIn(
+            workflow = workflow,
+            scope = restoreScope,
+            props = props,
+            initialSnapshot = snapshot,
+            workflowTracer = null,
+            runtimeConfig = runtimeConfig2
+          ) {}
+        runtimeMatrixTestRunner.assertEquals(
+          "updated state",
+          restoredRenderings.value.rendering.first
+        )
+      }
     }
   }
 
@@ -309,66 +275,64 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      lateinit var sink: Sink<String>
-      var snapped = false
-
-      val workflow = Workflow.stateful<Unit, String, Nothing, String>(
-        initialState = { _, _ -> "unchanging state" },
-        snapshot = {
-          Snapshot.of {
-            snapped = true
-            ByteString.of(1)
+      runTest(UnconfinedTestDispatcher()) {
+        lateinit var sink: Sink<String>
+        var snapped = false
+
+        val workflow = Workflow.stateful<Unit, String, Nothing, String>(
+          initialState = { _, _ -> "unchanging state" },
+          snapshot = {
+            Snapshot.of {
+              snapped = true
+              ByteString.of(1)
+            }
+          },
+          render = { _, renderState ->
+            sink = actionSink.contraMap { action("") { state = it } }
+            renderState
           }
-        },
-        render = { _, renderState ->
-          sink = actionSink.contraMap { action("") { state = it } }
-          renderState
+        )
+        val props = MutableStateFlow(Unit)
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+
+        val emitted = mutableListOf<RenderingAndSnapshot<String>>()
+        val collectionJob = launch {
+          renderings.collect { emitted += it }
         }
-      )
-      val props = MutableStateFlow(Unit)
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      val emitted = mutableListOf<RenderingAndSnapshot<String>>()
-      val scope = CoroutineScope(Unconfined)
-      scope.launch {
-        renderings.collect { emitted += it }
-      }
 
-      if (runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES)) {
-        // we have to change state then or it won't render.
-        sink.send("changing state")
-      } else {
-        sink.send("unchanging state")
-      }
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
+        if (runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES)) {
+          // we have to change state then or it won't render.
+          sink.send("changing state")
+        } else {
+          sink.send("unchanging state")
+        }
 
-      if (runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES)) {
-        // we have to change state then or it won't render.
-        sink.send("changing state, again")
-      } else {
-        sink.send("unchanging state")
+        if (runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES)) {
+          // we have to change state then or it won't render.
+          sink.send("changing state, again")
+        } else {
+          sink.send("unchanging state")
+        }
+        advanceUntilIdle()
+
+        collectionJob.cancel()
+
+        assertFalse(snapped)
+        assertNotSame(
+          emitted[0].snapshot.workflowSnapshot,
+          emitted[1].snapshot.workflowSnapshot
+        )
+        assertNotSame(
+          emitted[1].snapshot.workflowSnapshot,
+          emitted[2].snapshot.workflowSnapshot
+        )
       }
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-
-      scope.cancel()
-
-      assertFalse(snapped)
-      assertNotSame(
-        emitted[0].snapshot.workflowSnapshot,
-        emitted[1].snapshot.workflowSnapshot
-      )
-      assertNotSame(
-        emitted[1].snapshot.workflowSnapshot,
-        emitted[2].snapshot.workflowSnapshot
-      )
     }
   }
 
@@ -377,85 +341,168 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val trigger = Channel<String>()
-      val workflow = Workflow.stateless<Unit, String, Unit> {
-        runningWorker(
-          trigger.consumeAsFlow()
-            .asWorker()
-        ) { action("") { setOutput(it) } }
-      }
-      val receivedOutputs = mutableListOf<String>()
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {
-        receivedOutputs += it
-      }
-      assertTrue(receivedOutputs.isEmpty())
+      runTest(UnconfinedTestDispatcher()) {
+        val trigger = Channel<String>()
+        val workflow = Workflow.stateless<Unit, String, Unit> {
+          runningWorker(
+            trigger.receiveAsFlow()
+              .asWorker()
+          ) { action("") { setOutput(it) } }
+        }
+        val receivedOutputs = mutableListOf<String>()
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {
+          receivedOutputs += it
+        }
+        assertTrue(receivedOutputs.isEmpty())
 
-      trigger.trySend("foo").isSuccess
-      assertEquals(listOf("foo"), receivedOutputs)
+        assertTrue(trigger.trySend("foo").isSuccess)
+        assertEquals(listOf("foo"), receivedOutputs)
 
-      trigger.trySend("bar").isSuccess
-      assertEquals(listOf("foo", "bar"), receivedOutputs)
+        assertTrue(trigger.trySend("bar").isSuccess)
+        assertEquals(listOf("foo", "bar"), receivedOutputs)
+      }
     }
   }
 
+  /**
+   * This is a bit of a tricky test. Everything comes down to how your coroutines are dispatched.
+   * This test confirms that we are setting the value on the StateFlow of the updated rendering
+   * before onOutput is called.
+   *
+   * It uses an [UnconfinedTestDispatcher] for the runtime as would be typical.
+   *
+   * If we were collecting the renderings, that would happen after [onOutput] as it would have
+   * to wait to be dispatched after onOutput was complete.
+   *
+   * See [onOutput_called_after_rendering_emitted_and_collected] for alternate behaviour with
+   * a different dispatcher for the runtime.
+   */
   @Test fun onOutput_called_after_rendering_emitted() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val trigger = Channel<String>()
-      val workflow = Workflow.stateful<String, String, String>(
-        initialState = "initial",
-        render = { renderState ->
-          runningWorker(
-            trigger.consumeAsFlow()
-              .asWorker()
-          ) {
-            action("") {
-              state = it
-              setOutput(it)
+      runTest(UnconfinedTestDispatcher()) {
+        val trigger = Channel<String>()
+        val workflow = Workflow.stateful<String, String, String>(
+          initialState = "initial",
+          render = { renderState ->
+            runningWorker(
+              trigger.receiveAsFlow()
+                .asWorker()
+            ) {
+              action("") {
+                state = it
+                setOutput(it)
+              }
             }
+            renderState
           }
-          renderState
+        )
+
+        val receivedOutputs = mutableListOf<String>()
+        lateinit var renderings: StateFlow<RenderingAndSnapshot<String>>
+        renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) { it: String ->
+          receivedOutputs += it
+          // The value of the updated rendering has already been set by the time onOutput is
+          // called
+          assertEquals(it, renderings.value.rendering)
         }
-      )
-
-      val emittedRenderings = mutableListOf<String>()
-      val receivedOutputs = mutableListOf<String>()
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) { it: String ->
-        receivedOutputs += it
-        assertTrue(emittedRenderings.contains(it))
+        assertTrue(receivedOutputs.isEmpty())
+
+        assertTrue(trigger.trySend("foo").isSuccess)
+        assertEquals(listOf("foo"), receivedOutputs)
       }
-      assertTrue(receivedOutputs.isEmpty())
+    }
+  }
 
-      val scope = CoroutineScope(Unconfined)
-      scope.launch {
-        renderings.collect { rendering: RenderingAndSnapshot<String> ->
-          emittedRenderings += rendering.rendering
+  /**
+   * A different form of [onOutput_called_after_rendering_emitted]. Here we launch the workflow
+   * runtime on its own [TestScope] with a [StandardTestDispatcher], which will be paused until
+   * told to advance.
+   *
+   * We *collect* emitted renderings on the [UnconfinedTestDispatcher] of the [runTest].
+   * The point here is that when the runtime sets the value on the StateFlow - a non-suspending
+   * operation - and then it calls [onOutput] - a suspending operation - the [onOutput] handler
+   * will not be immediately dispatched (it is waiting for dispatch from the scheduler), but the
+   * collector of the renderings [StateFlow] will be dispatched and update the 'emitted' renderings.
+   * Then when we let the runtime's scheduler go ahead, it will have already been populated.
+   */
+  @Test fun onOutput_called_after_rendering_emitted_and_collected() {
+    runtimeTestRunner.runParametrizedTest(
+      paramSource = runtimeOptions,
+      before = ::setup,
+    ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
+      runTest(UnconfinedTestDispatcher()) {
+        val trigger = Channel<String>()
+        val workflow = Workflow.stateful<String, String, String>(
+          initialState = "initial",
+          render = { renderState ->
+            runningWorker(
+              trigger.receiveAsFlow()
+                .asWorker()
+            ) {
+              action("") {
+                state = it
+                setOutput(it)
+              }
+            }
+            renderState
+          }
+        )
+
+        val runtimeTestDispatcher = StandardTestDispatcher()
+        val testScope = TestScope(runtimeTestDispatcher)
+        val emittedRenderings = mutableListOf<String>()
+        val receivedOutputs = mutableListOf<String>()
+        val renderings: StateFlow<RenderingAndSnapshot<String>> = renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) { it: String ->
+          // The list collecting the renderings already contains it by the time onOutput is fired.
+          assertTrue(emittedRenderings.contains(it))
+          receivedOutputs += it
+        }
+        assertTrue(receivedOutputs.isEmpty())
+
+        val collectionJob = launch {
+          renderings.collect {
+            emittedRenderings += it.rendering
+          }
+        }
+
+        launch {
+          trigger.send("foo")
         }
-      }
 
-      trigger.trySend("foo").isSuccess
+        testScope.advanceUntilIdle()
 
-      trigger.trySend("bar").isSuccess
+        assertEquals(listOf("foo"), receivedOutputs)
 
-      scope.cancel()
+        collectionJob.cancel()
+      }
     }
   }
 
-  @Test fun tracer_includes_expected_sections() {
+  @Test fun tracer_includes_expected_sections() = runTest(UnconfinedTestDispatcher()) {
+    // Only test default so we only have one 'golden value' to assert against.
+    // We are only testing the tracer correctness here, which should be agnostic of runtime.
+    // We include 'tracers' in the other test to test against unexpected side effects.
     val runtimeConfig = RuntimeConfigOptions.DEFAULT_CONFIG
     val workflowTracer = testTracer
     setup()
@@ -464,7 +511,7 @@ class RenderWorkflowInTest {
       initialState = "initial",
       render = { renderState ->
         runningWorker(
-          trigger.consumeAsFlow()
+          trigger.receiveAsFlow()
             .asWorker()
         ) {
           action("") {
@@ -480,30 +527,25 @@ class RenderWorkflowInTest {
     val receivedOutputs = mutableListOf<String>()
     val renderings = renderWorkflowIn(
       workflow = workflow,
-      scope = testScope,
+      scope = backgroundScope,
       props = MutableStateFlow(Unit),
       runtimeConfig = runtimeConfig,
       workflowTracer = workflowTracer,
-    ) { it: String ->
-      receivedOutputs += it
-      assertTrue(emittedRenderings.contains(it))
-    }
+      onOutput = {}
+    )
     assertTrue(receivedOutputs.isEmpty())
 
-    val scope = CoroutineScope(Unconfined)
-    scope.launch {
+    val collectionJob = launch {
       renderings.collect { rendering: RenderingAndSnapshot<String> ->
         emittedRenderings += rendering.rendering
       }
     }
 
-    trigger.trySend("foo").isSuccess
-
-    trigger.trySend("bar").isSuccess
-
-    scope.cancel()
+    assertTrue(trigger.trySend("foo").isSuccess)
 
     assertEquals(EXPECTED_TRACE, traces.toString().trim())
+
+    collectionJob.cancel()
   }
 
   @Test fun onOutput_is_not_called_when_no_output_emitted() {
@@ -511,30 +553,28 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val workflow = Workflow.stateless<Int, String, Int> { props -> props }
-      var onOutputCalls = 0
-      val props = MutableStateFlow(0)
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) { onOutputCalls++ }
-      assertEquals(0, renderings.value.rendering)
-      assertEquals(0, onOutputCalls)
-
-      props.value = 1
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-      assertEquals(1, renderings.value.rendering)
-      assertEquals(0, onOutputCalls)
-
-      props.value = 2
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-      assertEquals(2, renderings.value.rendering)
-      assertEquals(0, onOutputCalls)
+      runTest(UnconfinedTestDispatcher()) {
+        val workflow = Workflow.stateless<Int, String, Int> { props -> props }
+        var onOutputCalls = 0
+        val props = MutableStateFlow(0)
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) { onOutputCalls++ }
+        assertEquals(0, renderings.value.rendering)
+        assertEquals(0, onOutputCalls)
+
+        props.value = 1
+        assertEquals(1, renderings.value.rendering)
+        assertEquals(0, onOutputCalls)
+
+        props.value = 2
+        assertEquals(2, renderings.value.rendering)
+        assertEquals(0, onOutputCalls)
+      }
     }
   }
 
@@ -548,19 +588,21 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        throw ExpectedException()
-      }
-      assertFailsWith<ExpectedException> {
-        renderWorkflowIn(
-          workflow = workflow,
-          scope = testScope,
-          props = MutableStateFlow(Unit),
-          runtimeConfig = runtimeConfig,
-          workflowTracer = workflowTracer,
-        ) {}
+      runTest(UnconfinedTestDispatcher()) {
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          throw ExpectedException()
+        }
+        assertFailsWith<ExpectedException> {
+          renderWorkflowIn(
+            workflow = workflow,
+            scope = backgroundScope,
+            props = MutableStateFlow(Unit),
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+        }
+        assertTrue(backgroundScope.isActive)
       }
-      assertTrue(testScope.isActive)
     }
   }
 
@@ -570,24 +612,26 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var sideEffectWasRan = false
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect("test") {
-          sideEffectWasRan = true
+      runTest(UnconfinedTestDispatcher()) {
+        var sideEffectWasRan = false
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect("test") {
+            sideEffectWasRan = true
+          }
+          throw ExpectedException()
         }
-        throw ExpectedException()
-      }
 
-      assertFailsWith<ExpectedException> {
-        renderWorkflowIn(
-          workflow = workflow,
-          scope = testScope,
-          props = MutableStateFlow(Unit),
-          runtimeConfig = runtimeConfig,
-          workflowTracer = workflowTracer,
-        ) {}
+        assertFailsWith<ExpectedException> {
+          renderWorkflowIn(
+            workflow = workflow,
+            scope = backgroundScope,
+            props = MutableStateFlow(Unit),
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+        }
+        assertFalse(sideEffectWasRan)
       }
-      assertFalse(sideEffectWasRan)
     }
   }
 
@@ -597,35 +641,37 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var sideEffectWasRan = false
-      var cancellationException: Throwable? = null
-      val childWorkflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect("test") {
-          sideEffectWasRan = true
-          suspendCancellableCoroutine { continuation ->
-            continuation.invokeOnCancellation { cause -> cancellationException = cause }
+      runTest(UnconfinedTestDispatcher()) {
+        var sideEffectWasRan = false
+        var cancellationException: Throwable? = null
+        val childWorkflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect("test") {
+            sideEffectWasRan = true
+            suspendCancellableCoroutine { continuation ->
+              continuation.invokeOnCancellation { cause -> cancellationException = cause }
+            }
           }
         }
-      }
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        renderChild(childWorkflow)
-        throw ExpectedException()
-      }
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          renderChild(childWorkflow)
+          throw ExpectedException()
+        }
 
-      assertFailsWith<ExpectedException> {
-        renderWorkflowIn(
-          workflow = workflow,
-          scope = testScope,
-          props = MutableStateFlow(Unit),
-          runtimeConfig = runtimeConfig,
-          workflowTracer = workflowTracer,
-        ) {}
+        assertFailsWith<ExpectedException> {
+          renderWorkflowIn(
+            workflow = workflow,
+            scope = backgroundScope,
+            props = MutableStateFlow(Unit),
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+        }
+        assertTrue(sideEffectWasRan)
+        assertNotNull(cancellationException)
+        val realCause = generateSequence(cancellationException) { it.cause }
+          .firstOrNull { it !is CancellationException }
+        assertTrue(realCause is ExpectedException)
       }
-      assertTrue(sideEffectWasRan)
-      assertNotNull(cancellationException)
-      val realCause = generateSequence(cancellationException) { it.cause }
-        .firstOrNull { it !is CancellationException }
-      assertTrue(realCause is ExpectedException)
     }
   }
 
@@ -635,27 +681,29 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var sideEffectWasRan = false
-      val childWorkflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect("test") {
-          sideEffectWasRan = true
+      runTest(UnconfinedTestDispatcher()) {
+        var sideEffectWasRan = false
+        val childWorkflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect("test") {
+            sideEffectWasRan = true
+          }
+          throw ExpectedException()
+        }
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          renderChild(childWorkflow)
         }
-        throw ExpectedException()
-      }
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        renderChild(childWorkflow)
-      }
 
-      assertFailsWith<ExpectedException> {
-        renderWorkflowIn(
-          workflow = workflow,
-          scope = testScope,
-          props = MutableStateFlow(Unit),
-          runtimeConfig = runtimeConfig,
-          workflowTracer = workflowTracer,
-        ) {}
+        assertFailsWith<ExpectedException> {
+          renderWorkflowIn(
+            workflow = workflow,
+            scope = backgroundScope,
+            props = MutableStateFlow(Unit),
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+        }
+        assertFalse(sideEffectWasRan)
       }
-      assertFalse(sideEffectWasRan)
     }
   }
 
@@ -664,31 +712,34 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val trigger = CompletableDeferred<Unit>()
-      // Throws an exception when trigger is completed.
-      val workflow = Workflow.stateful<Unit, Boolean, Nothing, Unit>(
-        initialState = { false },
-        render = { _, throwNow ->
-          runningWorker(Worker.from { trigger.await() }) { action("") { state = true } }
-          if (throwNow) {
-            throw ExpectedException()
+      runTest(UnconfinedTestDispatcher()) {
+        val trigger = CompletableDeferred<Unit>()
+        // Throws an exception when trigger is completed.
+        val workflow = Workflow.stateful<Unit, Boolean, Nothing, Unit>(
+          initialState = { false },
+          render = { _, throwNow ->
+            runningWorker(Worker.from { trigger.await() }) { action("") { state = true } }
+            if (throwNow) {
+              throw ExpectedException()
+            }
           }
-        }
-      )
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      assertTrue(testScope.isActive)
-
-      trigger.complete(Unit)
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-      assertFalse(testScope.isActive)
+        )
+        val testScope = TestScope(testScheduler)
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+
+        assertTrue(testScope.isActive)
+
+        trigger.complete(Unit)
+        advanceUntilIdle()
+
+        assertFalse(testScope.isActive)
+      }
     }
   }
 
@@ -697,29 +748,32 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val trigger = CompletableDeferred<Unit>()
-      // Throws an exception when trigger is completed.
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningWorker(Worker.from { trigger.await() }) {
-          action("") {
-            throw ExpectedException()
+      runTest(UnconfinedTestDispatcher()) {
+        val trigger = CompletableDeferred<Unit>()
+        // Throws an exception when trigger is completed.
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningWorker(Worker.from { trigger.await() }) {
+            action("") {
+              throw ExpectedException()
+            }
           }
         }
+        val testScope = TestScope(testScheduler)
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+
+        assertTrue(testScope.isActive)
+
+        trigger.complete(Unit)
+        advanceUntilIdle()
+
+        assertFalse(testScope.isActive)
       }
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      assertTrue(testScope.isActive)
-
-      trigger.complete(Unit)
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-      assertFalse(testScope.isActive)
     }
   }
 
@@ -728,27 +782,34 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var cancellationException: Throwable? = null
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect(key = "test1") {
-          suspendCancellableCoroutine { continuation ->
-            continuation.invokeOnCancellation { cause -> cancellationException = cause }
+      runTest(UnconfinedTestDispatcher()) {
+        var cancellationException: Throwable? = null
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect(key = "test1") {
+            suspendCancellableCoroutine { continuation ->
+              continuation.invokeOnCancellation { cause -> cancellationException = cause }
+            }
           }
         }
+        val testScope = TestScope(testScheduler)
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        assertNull(cancellationException)
+        assertTrue(testScope.isActive)
+        advanceUntilIdle()
+
+        testScope.cancel()
+
+        advanceUntilIdle()
+
+        assertTrue(cancellationException is CancellationException)
+        assertNull(cancellationException!!.cause)
       }
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      assertNull(cancellationException)
-      assertTrue(testScope.isActive)
-
-      testScope.cancel()
-      assertTrue(cancellationException is CancellationException)
-      assertNull(cancellationException!!.cause)
     }
   }
 
@@ -757,34 +818,41 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val trigger = CompletableDeferred<Unit>()
-      var renderCount = 0
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        renderCount++
-        runningWorker(Worker.from { trigger.await() }) {
-          action("") {
-            testScope.cancel()
+      runTest(UnconfinedTestDispatcher()) {
+        val testScope = TestScope(testScheduler)
+        val trigger = CompletableDeferred<Unit>()
+        var renderCount = 0
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          renderCount++
+          runningWorker(Worker.from { trigger.await() }) {
+            action("") {
+              testScope.cancel()
+            }
           }
         }
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        advanceUntilIdle()
+
+        assertTrue(testScope.isActive)
+        assertTrue(renderCount == 1)
+
+        trigger.complete(Unit)
+
+        advanceUntilIdle()
+
+        assertFalse(testScope.isActive)
+        assertEquals(
+          1,
+          renderCount,
+          "Should not render after CoroutineScope is canceled."
+        )
       }
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      assertTrue(testScope.isActive)
-      assertTrue(renderCount == 1)
-
-      trigger.complete(Unit)
-      testScope.advanceUntilIdle()
-      assertFalse(testScope.isActive)
-      assertEquals(
-        1,
-        renderCount,
-        "Should not render after CoroutineScope is canceled."
-      )
     }
   }
 
@@ -793,27 +861,32 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var cancellationException: Throwable? = null
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect(key = "failing") {
-          suspendCancellableCoroutine { continuation ->
-            continuation.invokeOnCancellation { cause -> cancellationException = cause }
+      runTest(UnconfinedTestDispatcher()) {
+        var cancellationException: Throwable? = null
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {
+          runningSideEffect(key = "failing") {
+            suspendCancellableCoroutine { continuation ->
+              continuation.invokeOnCancellation { cause -> cancellationException = cause }
+            }
           }
         }
+        val testScope = TestScope(testScheduler)
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+        advanceUntilIdle()
+        assertNull(cancellationException)
+        assertTrue(testScope.isActive)
+
+        testScope.cancel(CancellationException("fail!", ExpectedException()))
+        advanceUntilIdle()
+        assertTrue(cancellationException is CancellationException)
+        assertTrue(cancellationException!!.cause is ExpectedException)
       }
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      assertNull(cancellationException)
-      assertTrue(testScope.isActive)
-
-      testScope.cancel(CancellationException("fail!", ExpectedException()))
-      assertTrue(cancellationException is CancellationException)
-      assertTrue(cancellationException!!.cause is ExpectedException)
     }
   }
 
@@ -822,57 +895,27 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {}
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      // Collect in separate scope so we actually test that the parent scope is failed when it's
-      // different from the collecting scope.
-      val collectScope = TestScope(UnconfinedTestDispatcher())
-      collectScope.launch {
-        renderings.collect { throw ExpectedException() }
-      }
-      assertTrue(testScope.isActive)
-      assertFalse(collectScope.isActive)
-    }
-  }
+      runTest(UnconfinedTestDispatcher()) {
+        val workflow = Workflow.stateless<Unit, Nothing, Unit> {}
+        val testScope = TestScope(testScheduler)
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
 
-  @Test fun error_from_renderings_collector_cancels_runtime() {
-    runtimeTestRunner.runParametrizedTest(
-      paramSource = runtimeOptions,
-      before = ::setup,
-    ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      var cancellationException: Throwable? = null
-      val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-        runningSideEffect(key = "test") {
-          suspendCancellableCoroutine { continuation ->
-            continuation.invokeOnCancellation { cause ->
-              cancellationException = cause
-            }
-          }
+        // Collect in separate scope so we actually test that the parent scope is failed when it's
+        // different from the collecting scope.
+        val collectScope = TestScope(testScheduler)
+        collectScope.launch {
+          renderings.collect { throw ExpectedException() }
         }
+        advanceUntilIdle()
+        assertTrue(testScope.isActive)
+        assertFalse(collectScope.isActive)
       }
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = pausedTestScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      pausedTestScope.launch {
-        renderings.collect { throw ExpectedException() }
-      }
-      assertNull(cancellationException)
-
-      pausedTestScope.advanceUntilIdle()
-      assertTrue(cancellationException is CancellationException)
-      assertTrue(cancellationException!!.cause is ExpectedException)
     }
   }
 
@@ -881,164 +924,126 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val trigger = CompletableDeferred<Unit>()
-      // Emits a Unit when trigger is completed.
-      val workflow = Workflow.stateless<Unit, Unit, Unit> {
-        runningWorker(Worker.from { trigger.await() }) { action("") { setOutput(Unit) } }
-      }
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = pausedTestScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {
-        throw ExpectedException()
-      }
-      assertTrue(pausedTestScope.isActive)
-
-      trigger.complete(Unit)
-      assertTrue(pausedTestScope.isActive)
+      runTest(UnconfinedTestDispatcher()) {
+        val trigger = CompletableDeferred<Unit>()
+        // Emits a Unit when trigger is completed.
+        val workflow = Workflow.stateless<Unit, Unit, Unit> {
+          runningWorker(Worker.from { trigger.await() }) { action("") { setOutput(Unit) } }
+        }
+        val testScope = TestScope(testScheduler)
+        renderWorkflowIn(
+          workflow = workflow,
+          scope = testScope,
+          props = MutableStateFlow(Unit),
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {
+          throw ExpectedException()
+        }
+        advanceUntilIdle()
+        assertTrue(testScope.isActive)
 
-      pausedTestScope.advanceUntilIdle()
-      pausedTestScope.runCurrent()
-      assertFalse(pausedTestScope.isActive)
+        trigger.complete(Unit)
+        advanceUntilIdle()
+        assertFalse(testScope.isActive)
+      }
     }
   }
 
-  @Test fun output_is_emitted_before_next_render_pass() {
+  // https://github.com/square/workflow-kotlin/issues/224
+  @Test fun exceptions_from_Snapshots_do_not_fail_runtime() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val outputTrigger = CompletableDeferred<String>()
-      // A workflow whose state and rendering is the last output that it emitted.
-      val workflow = Workflow.stateful<Unit, String, String, String>(
-        initialState = { "{no output}" },
-        render = { _, renderState ->
-          runningWorker(Worker.from { outputTrigger.await() }) { output ->
-            action("") {
-              setOutput(output)
-              state = output
+      runTest(UnconfinedTestDispatcher()) {
+        val workflow = Workflow.stateful<Int, Unit, Nothing, Unit>(
+          snapshot = {
+            Snapshot.of {
+              throw ExpectedException()
             }
-          }
-          return@stateful renderState
+          },
+          initialState = { _, _ -> },
+          render = { _, _ -> }
+        )
+        val props = MutableStateFlow(0)
+        val uncaughtExceptions = mutableListOf<Throwable>()
+        val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
+          uncaughtExceptions += throwable
         }
-      )
-      val events = mutableListOf<String>()
-
-      renderWorkflowIn(
-        workflow = workflow,
-        scope = pausedTestScope,
-        props = MutableStateFlow(Unit),
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-        onOutput = { events += "output($it)" }
-      )
-        .onEach { events += "rendering(${it.rendering})" }
-        .launchIn(pausedTestScope)
-      pausedTestScope.advanceUntilIdle()
-      pausedTestScope.runCurrent()
-      assertEquals(listOf("rendering({no output})"), events)
-
-      outputTrigger.complete("output")
-      pausedTestScope.advanceUntilIdle()
-      pausedTestScope.runCurrent()
-      assertEquals(
-        listOf(
-          "rendering({no output})",
-          "output(output)",
-          "rendering(output)",
-        ),
-        events
-      )
+        val mutex = Mutex(locked = true)
+        backgroundScope.launch(exceptionHandler) {
+          val snapshot = renderWorkflowIn(
+            workflow = workflow,
+            scope = this,
+            props = props,
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+            .value
+            .snapshot
+
+          assertFailsWith<ExpectedException> { snapshot.toByteString() }
+          assertTrue(uncaughtExceptions.isEmpty())
+
+          props.value += 1
+          assertFailsWith<ExpectedException> { snapshot.toByteString() }
+          mutex.unlock()
+        }
+        // wait for snapshotting.
+        mutex.lock()
+      }
     }
   }
 
   // https://github.com/square/workflow-kotlin/issues/224
-  @Test fun exceptions_from_Snapshots_do_not_fail_runtime() {
+  @Test fun exceptions_from_renderings_equals_methods_do_not_fail_runtime() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      val workflow = Workflow.stateful<Int, Unit, Nothing, Unit>(
-        snapshot = {
-          Snapshot.of {
+      runTest(UnconfinedTestDispatcher()) {
+        @Suppress("EqualsOrHashCode", "unused")
+        class FailRendering(val value: Int) {
+          override fun equals(other: Any?): Boolean {
             throw ExpectedException()
           }
-        },
-        initialState = { _, _ -> },
-        render = { _, _ -> }
-      )
-      val props = MutableStateFlow(0)
-      val uncaughtExceptions = mutableListOf<Throwable>()
-      val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
-        uncaughtExceptions += throwable
-      }
-      val snapshot = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope + exceptionHandler,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-        .value
-        .snapshot
-
-      assertFailsWith<ExpectedException> { snapshot.toByteString() }
-      assertTrue(uncaughtExceptions.isEmpty())
-
-      props.value += 1
-      assertFailsWith<ExpectedException> { snapshot.toByteString() }
-    }
-  }
+        }
 
-  // https://github.com/square/workflow-kotlin/issues/224
-  @Test fun exceptions_from_renderings_equals_methods_do_not_fail_runtime() {
-    runtimeTestRunner.runParametrizedTest(
-      paramSource = runtimeOptions,
-      before = ::setup,
-    ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      @Suppress("EqualsOrHashCode", "unused")
-      class FailRendering(val value: Int) {
-        override fun equals(other: Any?): Boolean {
-          throw ExpectedException()
+        val workflow = Workflow.stateless<Int, Nothing, FailRendering> { props ->
+          FailRendering(props)
         }
-      }
+        val props = MutableStateFlow(0)
+        val uncaughtExceptions = mutableListOf<Throwable>()
+        val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
+          uncaughtExceptions += throwable
+        }
+        val mutex = Mutex(locked = true)
+        backgroundScope.launch(exceptionHandler) {
+          val ras = renderWorkflowIn(
+            workflow = workflow,
+            scope = this,
+            props = props,
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+          val renderings = ras.map { it.rendering }
+
+          @Suppress("UnusedEquals")
+          assertFailsWith<ExpectedException> {
+            renderings.collect {
+              it.equals(Unit)
+            }
+          }
+          assertTrue(uncaughtExceptions.isEmpty())
 
-      val workflow = Workflow.stateless<Int, Nothing, FailRendering> { props ->
-        FailRendering(props)
-      }
-      val props = MutableStateFlow(0)
-      val uncaughtExceptions = mutableListOf<Throwable>()
-      val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
-        uncaughtExceptions += throwable
-      }
-      val ras = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope + exceptionHandler,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      val renderings = ras.map { it.rendering }
-        .produceIn(testScope)
-
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-
-      @Suppress("UnusedEquals")
-      assertFailsWith<ExpectedException> {
-        renderings.tryReceive()
-          .getOrNull()!!
-          .equals(Unit)
+          // Trigger another render pass.
+          props.value += 1
+          advanceUntilIdle()
+          mutex.unlock()
+        }
+        mutex.lock()
       }
-      assertTrue(uncaughtExceptions.isEmpty())
-
-      // Trigger another render pass.
-      props.value += 1
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
     }
   }
 
@@ -1048,140 +1053,142 @@ class RenderWorkflowInTest {
       paramSource = runtimeOptions,
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      @Suppress("EqualsOrHashCode")
-      data class FailRendering(val value: Int) {
-        override fun hashCode(): Int {
-          throw ExpectedException()
+      runTest(UnconfinedTestDispatcher()) {
+        @Suppress("EqualsOrHashCode")
+        data class FailRendering(val value: Int) {
+          override fun hashCode(): Int {
+            throw ExpectedException()
+          }
         }
-      }
-
-      val workflow = Workflow.stateless<Int, Nothing, FailRendering> { props ->
-        FailRendering(props)
-      }
-      val props = MutableStateFlow(0)
-      val uncaughtExceptions = mutableListOf<Throwable>()
-      val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
-        uncaughtExceptions += throwable
-      }
-      val ras = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope + exceptionHandler,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-      val renderings = ras.map { it.rendering }
-        .produceIn(testScope)
-
-      assertFailsWith<ExpectedException> {
-        renderings.tryReceive()
-          .getOrNull()
-          .hashCode()
-      }
-      assertTrue(uncaughtExceptions.isEmpty())
 
-      props.value += 1
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
+        val workflow = Workflow.stateless<Int, Nothing, FailRendering> { props ->
+          FailRendering(props)
+        }
+        val props = MutableStateFlow(0)
+        val uncaughtExceptions = mutableListOf<Throwable>()
+        val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
+          uncaughtExceptions += throwable
+        }
+        val mutex = Mutex(locked = true)
+        backgroundScope.launch(exceptionHandler) {
+          val ras = renderWorkflowIn(
+            workflow = workflow,
+            scope = this,
+            props = props,
+            runtimeConfig = runtimeConfig,
+            workflowTracer = workflowTracer,
+          ) {}
+          val renderings = ras.map { it.rendering }
+
+          assertFailsWith<ExpectedException> {
+            renderings.collect {
+              it.hashCode()
+            }
+          }
+          assertTrue(uncaughtExceptions.isEmpty())
 
-      assertFailsWith<ExpectedException> {
-        renderings.tryReceive()
-          .getOrNull()
-          .hashCode()
+          // Trigger another render pass.
+          props.value += 1
+          advanceUntilIdle()
+          mutex.unlock()
+        }
+        mutex.lock()
       }
     }
   }
 
   @Test fun for_render_on_state_change_only_we_do_not_render_if_state_not_changed() {
     runtimeTestRunner.runParametrizedTest(
-      paramSource = arrayOf(
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES) to null,
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES) to testTracer,
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES, CONFLATE_STALE_RENDERINGS) to null,
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES, CONFLATE_STALE_RENDERINGS) to testTracer,
-      ).asSequence(),
+      paramSource = runtimeOptions.filter {
+        it.first.contains(RENDER_ONLY_WHEN_STATE_CHANGES)
+      },
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      check(runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES))
-      lateinit var sink: Sink<String>
-
-      val workflow = Workflow.stateful<Unit, String, Nothing, String>(
-        initialState = { "unchanging state" },
-        render = { _, renderState ->
-          sink = actionSink.contraMap { action("") { state = it } }
-          renderState
+      runTest(UnconfinedTestDispatcher()) {
+        check(runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES))
+        lateinit var sink: Sink<String>
+
+        val workflow = Workflow.stateful<Unit, String, Nothing, String>(
+          initialState = { "unchanging state" },
+          render = { _, renderState ->
+            sink = actionSink.contraMap { action("") { state = it } }
+            renderState
+          }
+        )
+        val props = MutableStateFlow(Unit)
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+
+        val emitted = mutableListOf<RenderingAndSnapshot<String>>()
+        val collectionJob = launch {
+          renderings.collect { emitted += it }
         }
-      )
-      val props = MutableStateFlow(Unit)
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      val emitted = mutableListOf<RenderingAndSnapshot<String>>()
-      val scope = CoroutineScope(Unconfined)
-      scope.launch {
-        renderings.collect { emitted += it }
-      }
 
-      sink.send("unchanging state")
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-      scope.cancel()
+        sink.send("unchanging state")
+        advanceUntilIdle()
+        collectionJob.cancel()
 
-      assertEquals(1, emitted.size)
+        assertEquals(1, emitted.size)
+      }
     }
   }
 
   @Test fun for_render_on_state_change_only_we_render_if_state_changed() {
     runtimeTestRunner.runParametrizedTest(
-      paramSource = arrayOf(
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES) to null,
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES) to testTracer,
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES, CONFLATE_STALE_RENDERINGS) to null,
-        setOf(RENDER_ONLY_WHEN_STATE_CHANGES, CONFLATE_STALE_RENDERINGS) to testTracer,
-      ).asSequence(),
+      paramSource = runtimeOptions.filter {
+        it.first.contains(RENDER_ONLY_WHEN_STATE_CHANGES)
+      },
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      check(runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES))
-      lateinit var sink: Sink<String>
-
-      val workflow = Workflow.stateful<Unit, String, Nothing, String>(
-        initialState = { "unchanging state" },
-        render = { _, renderState ->
-          sink = actionSink.contraMap { action("") { state = it } }
-          renderState
+      runTest(UnconfinedTestDispatcher()) {
+        check(runtimeConfig.contains(RENDER_ONLY_WHEN_STATE_CHANGES))
+        lateinit var sink: Sink<String>
+
+        val workflow = Workflow.stateful<Unit, String, Nothing, String>(
+          initialState = { "unchanging state" },
+          render = { _, renderState ->
+            sink = actionSink.contraMap { action("") { state = it } }
+            renderState
+          }
+        )
+        val props = MutableStateFlow(Unit)
+        val renderings = renderWorkflowIn(
+          workflow = workflow,
+          scope = backgroundScope,
+          props = props,
+          runtimeConfig = runtimeConfig,
+          workflowTracer = workflowTracer,
+        ) {}
+
+        val emitted = mutableListOf<RenderingAndSnapshot<String>>()
+        val collectionJob = launch {
+          renderings.collect { emitted += it }
         }
-      )
-      val props = MutableStateFlow(Unit)
-      val renderings = renderWorkflowIn(
-        workflow = workflow,
-        scope = testScope,
-        props = props,
-        runtimeConfig = runtimeConfig,
-        workflowTracer = workflowTracer,
-      ) {}
-
-      val emitted = mutableListOf<RenderingAndSnapshot<String>>()
-      val scope = CoroutineScope(Unconfined)
-      scope.launch {
-        renderings.collect { emitted += it }
-      }
 
-      sink.send("changing state")
-      testScope.advanceUntilIdle()
-      testScope.runCurrent()
-      scope.cancel()
+        advanceUntilIdle()
+        sink.send("changing state")
+        advanceUntilIdle()
+        assertEquals(2, emitted.size)
 
-      assertEquals(2, emitted.size)
+        collectionJob.cancel()
+      }
     }
   }
 
   private class ExpectedException : RuntimeException()
 
+  private fun <T1, T2> cartesianProduct(
+    set1: Sequence<T1>,
+    set2: Sequence<T2>
+  ): Sequence<Pair<T1, T2>> {
+    return set1.flatMap { set1Item -> set2.map { set2Item -> set1Item to set2Item } }
+  }
+
   companion object {
     internal val EXPECTED_TRACE: String = """
 StartingCreateWorkerWorkflow