diff --git a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/AbstractWorkflowNode.kt b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/AbstractWorkflowNode.kt
new file mode 100644
index 000000000..6f5ad0f0c
--- /dev/null
+++ b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/AbstractWorkflowNode.kt
@@ -0,0 +1,143 @@
+package com.squareup.workflow1.internal
+
+import com.squareup.workflow1.ActionApplied
+import com.squareup.workflow1.ActionProcessingResult
+import com.squareup.workflow1.NoopWorkflowInterceptor
+import com.squareup.workflow1.RuntimeConfig
+import com.squareup.workflow1.RuntimeConfigOptions
+import com.squareup.workflow1.TreeSnapshot
+import com.squareup.workflow1.Workflow
+import com.squareup.workflow1.WorkflowIdentifier
+import com.squareup.workflow1.WorkflowInterceptor
+import com.squareup.workflow1.WorkflowInterceptor.WorkflowSession
+import com.squareup.workflow1.WorkflowTracer
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.selects.SelectBuilder
+import kotlin.coroutines.CoroutineContext
+
+internal fun <PropsT, OutputT, RenderingT> createWorkflowNode(
+  id: WorkflowNodeId,
+  workflow: Workflow<PropsT, OutputT, RenderingT>,
+  initialProps: PropsT,
+  snapshot: TreeSnapshot?,
+  baseContext: CoroutineContext,
+  // Providing default value so we don't need to specify in test.
+  runtimeConfig: RuntimeConfig = RuntimeConfigOptions.DEFAULT_CONFIG,
+  workflowTracer: WorkflowTracer? = null,
+  emitAppliedActionToParent: (ActionApplied<OutputT>) -> ActionProcessingResult = { it },
+  parent: WorkflowSession? = null,
+  interceptor: WorkflowInterceptor = NoopWorkflowInterceptor,
+  idCounter: IdCounter? = null
+): AbstractWorkflowNode<PropsT, OutputT, RenderingT> = when (workflow) {
+  // is ComposeWorkflow<*, *, *> -> ComposeWorkflowNode(
+  //   id = id,
+  //   workflow = workflow,
+  //   initialProps = initialProps,
+  //   snapshot = snapshot,
+  //   baseContext = baseContext,
+  //   runtimeConfig = runtimeConfig,
+  //   workflowTracer = workflowTracer,
+  //   emitAppliedActionToParent = emitAppliedActionToParent,
+  //   parent = parent,
+  //   interceptor = interceptor,
+  //   idCounter = idCounter,
+  // )
+
+  else -> StatefulWorkflowNode(
+    id = id,
+    workflow = workflow.asStatefulWorkflow(),
+    initialProps = initialProps,
+    snapshot = snapshot,
+    baseContext = baseContext,
+    runtimeConfig = runtimeConfig,
+    workflowTracer = workflowTracer,
+    emitAppliedActionToParent = emitAppliedActionToParent,
+    parent = parent,
+    interceptor = interceptor,
+    idCounter = idCounter,
+  )
+}
+
+internal abstract class AbstractWorkflowNode<PropsT, OutputT, RenderingT>(
+  val id: WorkflowNodeId,
+  final override val parent: WorkflowSession?,
+  final override val workflowTracer: WorkflowTracer?,
+  final override val runtimeConfig: RuntimeConfig,
+  protected val interceptor: WorkflowInterceptor,
+  protected val emitAppliedActionToParent: (ActionApplied<OutputT>) -> ActionProcessingResult,
+  baseContext: CoroutineContext,
+  idCounter: IdCounter?,
+) : WorkflowSession,
+  CoroutineScope {
+
+  /**
+   * Context that has a job that will live as long as this node.
+   * Also adds a debug name to this coroutine based on its ID.
+   */
+  final override val coroutineContext = baseContext +
+    Job(baseContext[Job]) +
+    CoroutineName(id.toString())
+
+  // WorkflowSession properties
+  final override val identifier: WorkflowIdentifier get() = id.identifier
+  final override val renderKey: String get() = id.name
+  final override val sessionId: Long = idCounter.createId()
+
+  final override fun toString(): String {
+    val parentDescription = parent?.let { "WorkflowInstance(…)" }
+    return "WorkflowInstance(" +
+      "identifier=$identifier, " +
+      "renderKey=$renderKey, " +
+      "instanceId=$sessionId, " +
+      "parent=$parentDescription" +
+      ")"
+  }
+
+  /**
+   * Walk the tree of workflows, rendering each one and using
+   * [RenderContext][com.squareup.workflow1.BaseRenderContext] to give its children a chance to
+   * render themselves and aggregate those child renderings.
+   *
+   * @param workflow The "template" workflow instance used in the current render pass. This isn't
+   * necessarily the same _instance_ every call, but will be the same _type_.
+   */
+  abstract fun render(
+    workflow: Workflow<PropsT, OutputT, RenderingT>,
+    input: PropsT
+  ): RenderingT
+
+  /**
+   * Walk the tree of state machines again, this time gathering snapshots and aggregating them
+   * automatically.
+   */
+  abstract fun snapshot(): TreeSnapshot
+
+  /**
+   * Gets the next [result][ActionProcessingResult] from the state machine. This will be an
+   * [OutputT] or null.
+   *
+   * Walk the tree of state machines, asking each one to wait for its next event. If something happen
+   * that results in an output, that output is returned. Null means something happened that requires
+   * a re-render, e.g. my state changed or a child state changed.
+   *
+   * It is an error to call this method after calling [cancel].
+   *
+   * @return [Boolean] whether or not the queues were empty for this node and its children at the
+   *    time of suspending.
+   */
+  abstract fun onNextAction(selector: SelectBuilder<ActionProcessingResult>): Boolean
+
+  /**
+   * Cancels this state machine host, and any coroutines started as children of it.
+   *
+   * This must be called when the caller will no longer call [onNextAction]. It is an error to call [onNextAction]
+   * after calling this method.
+   */
+  open fun cancel(cause: CancellationException? = null) {
+    coroutineContext.cancel(cause)
+  }
+}
diff --git a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/ComposeWorkflowNode.kt b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/ComposeWorkflowNode.kt
new file mode 100644
index 000000000..692245a8e
--- /dev/null
+++ b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/ComposeWorkflowNode.kt
@@ -0,0 +1,63 @@
+package com.squareup.workflow1.internal
+
+import com.squareup.workflow1.ActionApplied
+import com.squareup.workflow1.ActionProcessingResult
+import com.squareup.workflow1.NoopWorkflowInterceptor
+import com.squareup.workflow1.RuntimeConfig
+import com.squareup.workflow1.RuntimeConfigOptions
+import com.squareup.workflow1.TreeSnapshot
+import com.squareup.workflow1.Workflow
+import com.squareup.workflow1.WorkflowInterceptor
+import com.squareup.workflow1.WorkflowInterceptor.WorkflowSession
+import com.squareup.workflow1.WorkflowTracer
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.selects.SelectBuilder
+import kotlin.coroutines.CoroutineContext
+
+internal class ComposeWorkflowNode<PropsT, OutputT, RenderingT>(
+  id: WorkflowNodeId,
+  // workflow: ComposeWorkflow<PropsT, OutputT, RenderingT>,
+  initialProps: PropsT,
+  snapshot: TreeSnapshot?,
+  baseContext: CoroutineContext,
+  // Providing default value so we don't need to specify in test.
+  runtimeConfig: RuntimeConfig = RuntimeConfigOptions.DEFAULT_CONFIG,
+  workflowTracer: WorkflowTracer? = null,
+  emitAppliedActionToParent: (ActionApplied<OutputT>) -> ActionProcessingResult = { it },
+  parent: WorkflowSession? = null,
+  interceptor: WorkflowInterceptor = NoopWorkflowInterceptor,
+  idCounter: IdCounter? = null
+) : AbstractWorkflowNode<PropsT, OutputT, RenderingT>(
+  id = id,
+  runtimeConfig = runtimeConfig,
+  workflowTracer = workflowTracer,
+  parent = parent,
+  baseContext = baseContext,
+  idCounter = idCounter,
+  interceptor = interceptor,
+  emitAppliedActionToParent = emitAppliedActionToParent,
+) {
+
+  init {
+    interceptor.onSessionStarted(workflowScope = this, session = this)
+  }
+
+  override fun render(
+    workflow: Workflow<PropsT, OutputT, RenderingT>,
+    input: PropsT
+  ): RenderingT {
+    TODO("Not yet implemented")
+  }
+
+  override fun snapshot(): TreeSnapshot {
+    TODO("Not yet implemented")
+  }
+
+  override fun onNextAction(selector: SelectBuilder<ActionProcessingResult>): Boolean {
+    TODO("Not yet implemented")
+  }
+
+  override fun cancel(cause: CancellationException?) {
+    TODO("Not yet implemented")
+  }
+}
diff --git a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowNode.kt b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/StatefulWorkflowNode.kt
similarity index 83%
rename from workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowNode.kt
rename to workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/StatefulWorkflowNode.kt
index 04c680fb2..5db085968 100644
--- a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowNode.kt
+++ b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/StatefulWorkflowNode.kt
@@ -14,7 +14,6 @@ import com.squareup.workflow1.Workflow
 import com.squareup.workflow1.WorkflowAction
 import com.squareup.workflow1.WorkflowExperimentalApi
 import com.squareup.workflow1.WorkflowExperimentalRuntime
-import com.squareup.workflow1.WorkflowIdentifier
 import com.squareup.workflow1.WorkflowInterceptor
 import com.squareup.workflow1.WorkflowInterceptor.WorkflowSession
 import com.squareup.workflow1.WorkflowTracer
@@ -50,32 +49,32 @@ import kotlin.reflect.KType
  * structured concurrency).
  */
 @OptIn(WorkflowExperimentalApi::class, WorkflowExperimentalRuntime::class)
-internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
-  val id: WorkflowNodeId,
+internal class StatefulWorkflowNode<PropsT, StateT, OutputT, RenderingT>(
+  id: WorkflowNodeId,
   workflow: StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>,
   initialProps: PropsT,
   snapshot: TreeSnapshot?,
   baseContext: CoroutineContext,
   // Providing default value so we don't need to specify in test.
-  override val runtimeConfig: RuntimeConfig = RuntimeConfigOptions.DEFAULT_CONFIG,
-  override val workflowTracer: WorkflowTracer? = null,
-  private val emitAppliedActionToParent: (ActionApplied<OutputT>) -> ActionProcessingResult =
-    { it },
-  override val parent: WorkflowSession? = null,
-  private val interceptor: WorkflowInterceptor = NoopWorkflowInterceptor,
+  runtimeConfig: RuntimeConfig = RuntimeConfigOptions.DEFAULT_CONFIG,
+  workflowTracer: WorkflowTracer? = null,
+  emitAppliedActionToParent: (ActionApplied<OutputT>) -> ActionProcessingResult = { it },
+  parent: WorkflowSession? = null,
+  interceptor: WorkflowInterceptor = NoopWorkflowInterceptor,
   idCounter: IdCounter? = null
-) : CoroutineScope, SideEffectRunner, RememberStore, WorkflowSession {
+) : AbstractWorkflowNode<PropsT, OutputT, RenderingT>(
+  id = id,
+  runtimeConfig = runtimeConfig,
+  workflowTracer = workflowTracer,
+  parent = parent,
+  baseContext = baseContext,
+  idCounter = idCounter,
+  interceptor = interceptor,
+  emitAppliedActionToParent = emitAppliedActionToParent,
+),
+  SideEffectRunner,
+  RememberStore {
 
-  /**
-   * Context that has a job that will live as long as this node.
-   * Also adds a debug name to this coroutine based on its ID.
-   */
-  override val coroutineContext = baseContext + Job(baseContext[Job]) + CoroutineName(id.toString())
-
-  // WorkflowInstance properties
-  override val identifier: WorkflowIdentifier get() = id.identifier
-  override val renderKey: String get() = id.name
-  override val sessionId: Long = idCounter.createId()
   private var cachedWorkflowInstance: StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>
   private var interceptedWorkflowInstance: StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>
 
@@ -116,45 +115,37 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
     state = interceptedWorkflowInstance.initialState(initialProps, snapshot?.workflowSnapshot, this)
   }
 
-  override fun toString(): String {
-    val parentDescription = parent?.let { "WorkflowInstance(…)" }
-    return "WorkflowInstance(" +
-      "identifier=$identifier, " +
-      "renderKey=$renderKey, " +
-      "instanceId=$sessionId, " +
-      "parent=$parentDescription" +
-      ")"
-  }
-
   /**
    * Walk the tree of workflows, rendering each one and using
    * [RenderContext][com.squareup.workflow1.BaseRenderContext] to give its children a chance to
    * render themselves and aggregate those child renderings.
    */
   @Suppress("UNCHECKED_CAST")
-  fun render(
-    workflow: StatefulWorkflow<PropsT, *, OutputT, RenderingT>,
+  override fun render(
+    workflow: Workflow<PropsT, OutputT, RenderingT>,
     input: PropsT
-  ): RenderingT =
-    renderWithStateType(workflow as StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>, input)
+  ): RenderingT = renderWithStateType(
+    workflow = workflow.asStatefulWorkflow() as StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>,
+    props = input
+  )
 
   /**
    * Walk the tree of state machines again, this time gathering snapshots and aggregating them
    * automatically.
    */
-  fun snapshot(workflow: StatefulWorkflow<*, *, *, *>): TreeSnapshot {
-    @Suppress("UNCHECKED_CAST")
-    val typedWorkflow = workflow as StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>
-    maybeUpdateCachedWorkflowInstance(typedWorkflow)
-    return interceptor.onSnapshotStateWithChildren({
-      val childSnapshots = subtreeManager.createChildSnapshots()
-      val rootSnapshot = interceptedWorkflowInstance.snapshotState(state)
-      TreeSnapshot(
-        workflowSnapshot = rootSnapshot,
-        // Create the snapshots eagerly since subtreeManager is mutable.
-        childTreeSnapshots = { childSnapshots }
-      )
-    }, this)
+  override fun snapshot(): TreeSnapshot {
+    return interceptor.onSnapshotStateWithChildren(
+      proceed = {
+        val childSnapshots = subtreeManager.createChildSnapshots()
+        val rootSnapshot = interceptedWorkflowInstance.snapshotState(state)
+        TreeSnapshot(
+          workflowSnapshot = rootSnapshot,
+          // Create the snapshots eagerly since subtreeManager is mutable.
+          childTreeSnapshots = { childSnapshots }
+        )
+      },
+      session = this
+    )
   }
 
   override fun runningSideEffect(
@@ -212,7 +203,7 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
    *    time of suspending.
    */
   @OptIn(ExperimentalCoroutinesApi::class, DelicateCoroutinesApi::class)
-  fun onNextAction(selector: SelectBuilder<ActionProcessingResult>): Boolean {
+  override fun onNextAction(selector: SelectBuilder<ActionProcessingResult>): Boolean {
     // Listen for any child workflow updates.
     var empty = subtreeManager.onNextChildAction(selector)
 
@@ -230,11 +221,11 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
   /**
    * Cancels this state machine host, and any coroutines started as children of it.
    *
-   * This must be called when the caller will no longer call [onNextAction]. It is an error to call [onNextAction]
-   * after calling this method.
+   * This must be called when the caller will no longer call [onNextAction]. It is an error to call
+   * [onNextAction] after calling this method.
    */
-  fun cancel(cause: CancellationException? = null) {
-    coroutineContext.cancel(cause)
+  override fun cancel(cause: CancellationException?) {
+    super.cancel(cause)
     lastRendering = NullableInitBox()
   }
 
@@ -314,7 +305,6 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
    * Applies [action] to this workflow's [state] and then passes the resulting [ActionApplied]
    * via [emitAppliedActionToParent] to the parent, with additional information as to whether or
    * not this action has changed the current node's state.
-   *
    */
   private fun applyAction(
     action: WorkflowAction<PropsT, StateT, OutputT>,
diff --git a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/SubtreeManager.kt b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/SubtreeManager.kt
index 09fb7608a..7fc6dbd7c 100644
--- a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/SubtreeManager.kt
+++ b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/SubtreeManager.kt
@@ -19,12 +19,12 @@ import kotlin.coroutines.CoroutineContext
  * Responsible for tracking child workflows, starting them and tearing them down when necessary.
  * Also manages restoring children from snapshots.
  *
- * Child workflows are stored in [WorkflowChildNode]s, which associate the child's [WorkflowNode]
+ * Child workflows are stored in [WorkflowChildNode]s, which associate the child's [StatefulWorkflowNode]
  * with its output handler.
  *
  * ## Rendering
  *
- * This class implements [RealRenderContext.Renderer], and [WorkflowNode] will pass its instance
+ * This class implements [RealRenderContext.Renderer], and [StatefulWorkflowNode] will pass its instance
  * of this class to the [RealRenderContext] on each render pass to render children. That means that
  * when a workflow renders a child, this class does the actual work.
  *
@@ -54,7 +54,7 @@ import kotlin.coroutines.CoroutineContext
  *      active:  [bar]
  *      staging: [foo, baz]
  *      ```
- *   4. When the workflow's render method returns, the [WorkflowNode] calls
+ *   4. When the workflow's render method returns, the [StatefulWorkflowNode] calls
  *      [commitRenderedChildren], which:
  *        1. Tears down all the children remaining in the active list
  *           ```
@@ -142,11 +142,11 @@ internal class SubtreeManager<PropsT, StateT, OutputT>(
         )
       }
     stagedChild.setHandler(handler)
-    return stagedChild.render(child.asStatefulWorkflow(), props)
+    return stagedChild.render(child, props)
   }
 
   /**
-   * Uses [selector] to invoke [WorkflowNode.onNextAction] for every running child workflow this instance
+   * Uses [selector] to invoke [StatefulWorkflowNode.onNextAction] for every running child workflow this instance
    * is managing.
    *
    * @return [Boolean] whether or not the children action queues are empty.
@@ -164,8 +164,7 @@ internal class SubtreeManager<PropsT, StateT, OutputT>(
   fun createChildSnapshots(): Map<WorkflowNodeId, TreeSnapshot> {
     val snapshots = mutableMapOf<WorkflowNodeId, TreeSnapshot>()
     children.forEachActive { child ->
-      val childWorkflow = child.workflow.asStatefulWorkflow()
-      snapshots[child.id] = child.workflowNode.snapshot(childWorkflow)
+      snapshots[child.id] = child.workflowNode.snapshot()
     }
     return snapshots
   }
@@ -190,9 +189,9 @@ internal class SubtreeManager<PropsT, StateT, OutputT>(
 
     val childTreeSnapshots = snapshotCache?.get(id)
 
-    val workflowNode = WorkflowNode(
+    val workflowNode = createWorkflowNode(
       id = id,
-      workflow = child.asStatefulWorkflow(),
+      workflow = child,
       initialProps = initialProps,
       snapshot = childTreeSnapshots,
       baseContext = contextForChildren,
diff --git a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowChildNode.kt b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowChildNode.kt
index ea2d46876..1843e90af 100644
--- a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowChildNode.kt
+++ b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowChildNode.kt
@@ -1,6 +1,5 @@
 package com.squareup.workflow1.internal
 
-import com.squareup.workflow1.StatefulWorkflow
 import com.squareup.workflow1.Workflow
 import com.squareup.workflow1.WorkflowAction
 import com.squareup.workflow1.WorkflowTracer
@@ -10,7 +9,7 @@ import com.squareup.workflow1.trace
 /**
  * Representation of a child workflow that has been rendered by another workflow.
  *
- * Associates the child's [WorkflowNode] (which includes the key passed to `renderChild`) with the
+ * Associates the child's [StatefulWorkflowNode] (which includes the key passed to `renderChild`) with the
  * output handler function that was passed to `renderChild`.
  */
 internal class WorkflowChildNode<
@@ -22,11 +21,11 @@ internal class WorkflowChildNode<
   >(
   val workflow: Workflow<*, ChildOutputT, *>,
   private var handler: (ChildOutputT) -> WorkflowAction<ParentPropsT, ParentStateT, ParentOutputT>,
-  val workflowNode: WorkflowNode<ChildPropsT, *, ChildOutputT, *>
+  val workflowNode: AbstractWorkflowNode<ChildPropsT, ChildOutputT, *>
 ) : InlineListNode<WorkflowChildNode<*, *, *, *, *>> {
   override var nextListNode: WorkflowChildNode<*, *, *, *, *>? = null
 
-  /** The [WorkflowNode]'s [WorkflowNodeId]. */
+  /** The [StatefulWorkflowNode]'s [WorkflowNodeId]. */
   val id get() = workflowNode.id
 
   /**
@@ -48,15 +47,15 @@ internal class WorkflowChildNode<
   }
 
   /**
-   * Wrapper around [WorkflowNode.render] that allows calling it with erased types.
+   * Wrapper around [StatefulWorkflowNode.render] that allows calling it with erased types.
    */
   fun <R> render(
-    workflow: StatefulWorkflow<*, *, *, *>,
+    workflow: Workflow<*, *, *>,
     props: Any?
   ): R {
     @Suppress("UNCHECKED_CAST")
     return workflowNode.render(
-      workflow as StatefulWorkflow<ChildPropsT, *, ChildOutputT, Nothing>,
+      workflow as Workflow<ChildPropsT, ChildOutputT, Nothing>,
       props as ChildPropsT
     ) as R
   }
diff --git a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowRunner.kt b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowRunner.kt
index 9eb66bb1b..39a35acc0 100644
--- a/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowRunner.kt
+++ b/workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowRunner.kt
@@ -50,7 +50,7 @@ internal class WorkflowRunner<PropsT, OutputT, RenderingT>(
   private val propsChannel = props.dropWhile { it == currentProps }
     .produceIn(scope)
 
-  private val rootNode = WorkflowNode(
+  private val rootNode = createWorkflowNode(
     id = workflow.id(),
     workflow = workflow,
     initialProps = currentProps,
@@ -71,7 +71,7 @@ internal class WorkflowRunner<PropsT, OutputT, RenderingT>(
   fun nextRendering(): RenderingAndSnapshot<RenderingT> {
     return interceptor.onRenderAndSnapshot(currentProps, { props ->
       val rendering = rootNode.render(workflow, props)
-      val snapshot = rootNode.snapshot(workflow)
+      val snapshot = rootNode.snapshot()
       RenderingAndSnapshot(rendering, snapshot)
     }, rootNode)
   }
diff --git a/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/WorkflowNodeTest.kt b/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/StatefulWorkflowNodeTest.kt
similarity index 94%
rename from workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/WorkflowNodeTest.kt
rename to workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/StatefulWorkflowNodeTest.kt
index 616630284..769e8beb7 100644
--- a/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/WorkflowNodeTest.kt
+++ b/workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/StatefulWorkflowNodeTest.kt
@@ -55,7 +55,7 @@ import kotlin.test.assertTrue
 import kotlin.test.fail
 
 @Suppress("UNCHECKED_CAST")
-internal class WorkflowNodeTest {
+internal class StatefulWorkflowNodeTest {
 
   abstract class StringWorkflow : StatefulWorkflow<String, String, String, String>() {
     override fun snapshotState(state: String): Snapshot = fail("not expected")
@@ -108,7 +108,7 @@ internal class WorkflowNodeTest {
       oldAndNewProps += old to new
       return@PropsRenderingWorkflow state
     }
-    val node = WorkflowNode(workflow.id(), workflow, "old", null, context)
+    val node = StatefulWorkflowNode(workflow.id(), workflow, "old", null, context)
 
     node.render(workflow, "new")
 
@@ -121,7 +121,7 @@ internal class WorkflowNodeTest {
       oldAndNewProps += old to new
       return@PropsRenderingWorkflow state
     }
-    val node = WorkflowNode(workflow.id(), workflow, "old", null, context)
+    val node = StatefulWorkflowNode(workflow.id(), workflow, "old", null, context)
 
     node.render(workflow, "old")
 
@@ -132,7 +132,7 @@ internal class WorkflowNodeTest {
     val workflow = PropsRenderingWorkflow { old, new, _ ->
       "$old->$new"
     }
-    val node = WorkflowNode(workflow.id(), workflow, "foo", null, context)
+    val node = StatefulWorkflowNode(workflow.id(), workflow, "foo", null, context)
 
     val rendering = node.render(workflow, "foo2")
 
@@ -173,7 +173,7 @@ internal class WorkflowNodeTest {
         return context.eventHandler("") { event -> setOutput(event) }
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       "",
@@ -215,7 +215,7 @@ internal class WorkflowNodeTest {
         return context.eventHandler("") { event -> setOutput(event) }
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       "",
@@ -267,7 +267,7 @@ internal class WorkflowNodeTest {
         return ""
       }
     }
-    val node = WorkflowNode(workflow.id(), workflow, "", null, context)
+    val node = StatefulWorkflowNode(workflow.id(), workflow, "", null, context)
 
     node.render(workflow, "")
     sink.send(action("") { setOutput("event") })
@@ -284,7 +284,7 @@ internal class WorkflowNodeTest {
       }
       assertFalse(started)
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -305,7 +305,7 @@ internal class WorkflowNodeTest {
         contextFromWorker = coroutineContext
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -327,7 +327,7 @@ internal class WorkflowNodeTest {
         actionSink.send(action("") { setOutput("result") })
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -359,7 +359,7 @@ internal class WorkflowNodeTest {
         }
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = true,
@@ -388,7 +388,7 @@ internal class WorkflowNodeTest {
         }
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -417,7 +417,7 @@ internal class WorkflowNodeTest {
         }
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = 0,
@@ -445,7 +445,7 @@ internal class WorkflowNodeTest {
         seenProps += props
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = 0,
@@ -469,7 +469,7 @@ internal class WorkflowNodeTest {
       runningSideEffect("same") { fail() }
       runningSideEffect("same") { fail() }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -500,7 +500,7 @@ internal class WorkflowNodeTest {
       if (props == 2) runningSideEffect("three", recordingSideEffect(events3))
     }
       .asStatefulWorkflow()
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       initialProps = 0,
@@ -536,7 +536,7 @@ internal class WorkflowNodeTest {
       runningSideEffect("one") { started1 = true }
       runningSideEffect("two") { started2 = true }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -566,7 +566,7 @@ internal class WorkflowNodeTest {
         }
       }
     )
-    val originalNode = WorkflowNode(
+    val originalNode = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       initialProps = "initial props",
@@ -575,10 +575,10 @@ internal class WorkflowNodeTest {
     )
 
     assertEquals("initial props", originalNode.render(workflow, "foo"))
-    val snapshot = originalNode.snapshot(workflow)
+    val snapshot = originalNode.snapshot()
     assertNotEquals(0, snapshot.toByteString().size)
 
-    val restoredNode = WorkflowNode(
+    val restoredNode = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       // These props should be ignored, since snapshot is non-null.
@@ -595,7 +595,7 @@ internal class WorkflowNodeTest {
       render = { _, state -> state },
       snapshot = { Snapshot.of("restored") }
     )
-    val originalNode = WorkflowNode(
+    val originalNode = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       initialProps = "initial props",
@@ -604,10 +604,10 @@ internal class WorkflowNodeTest {
     )
 
     assertEquals("initial props", originalNode.render(workflow, "foo"))
-    val snapshot = originalNode.snapshot(workflow)
+    val snapshot = originalNode.snapshot()
     assertNotEquals(0, snapshot.toByteString().size)
 
-    val restoredNode = WorkflowNode(
+    val restoredNode = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       // These props should be ignored, since snapshot is non-null.
@@ -652,7 +652,7 @@ internal class WorkflowNodeTest {
       }
     )
 
-    val originalNode = WorkflowNode(
+    val originalNode = StatefulWorkflowNode(
       parentWorkflow.id(),
       parentWorkflow,
       initialProps = "initial props",
@@ -661,10 +661,10 @@ internal class WorkflowNodeTest {
     )
 
     assertEquals("initial props|child props", originalNode.render(parentWorkflow, "foo"))
-    val snapshot = originalNode.snapshot(parentWorkflow)
+    val snapshot = originalNode.snapshot()
     assertNotEquals(0, snapshot.toByteString().size)
 
-    val restoredNode = WorkflowNode(
+    val restoredNode = StatefulWorkflowNode(
       parentWorkflow.id(),
       parentWorkflow,
       // These props should be ignored, since snapshot is non-null.
@@ -695,13 +695,13 @@ internal class WorkflowNodeTest {
         }
       }
     )
-    val node = WorkflowNode(workflow.id(), workflow, Unit, null, Unconfined)
+    val node = StatefulWorkflowNode(workflow.id(), workflow, Unit, null, Unconfined)
 
     assertEquals(0, snapshotCalls)
     assertEquals(0, snapshotWrites)
     assertEquals(0, restoreCalls)
 
-    val snapshot = node.snapshot(workflow)
+    val snapshot = node.snapshot()
 
     assertEquals(1, snapshotCalls)
     assertEquals(0, snapshotWrites)
@@ -713,7 +713,7 @@ internal class WorkflowNodeTest {
     assertEquals(1, snapshotWrites)
     assertEquals(0, restoreCalls)
 
-    WorkflowNode(workflow.id(), workflow, Unit, snapshot, Unconfined)
+    StatefulWorkflowNode(workflow.id(), workflow, Unit, snapshot, Unconfined)
 
     assertEquals(1, snapshotCalls)
     assertEquals(1, snapshotWrites)
@@ -732,7 +732,7 @@ internal class WorkflowNodeTest {
       render = { _, state -> state },
       snapshot = { state -> Snapshot.write { it.writeUtf8WithLength(state) } }
     )
-    val originalNode = WorkflowNode(
+    val originalNode = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       initialProps = "initial props",
@@ -741,10 +741,10 @@ internal class WorkflowNodeTest {
     )
 
     assertEquals("initial props", originalNode.render(workflow, "foo"))
-    val snapshot = originalNode.snapshot(workflow)
+    val snapshot = originalNode.snapshot()
     assertNotEquals(0, snapshot.toByteString().size)
 
-    val restoredNode = WorkflowNode(
+    val restoredNode = StatefulWorkflowNode(
       workflow.id(),
       workflow,
       initialProps = "new props",
@@ -756,7 +756,7 @@ internal class WorkflowNodeTest {
 
   @Test fun toString_formats_as_WorkflowInstance_without_parent() {
     val workflow = Workflow.rendering<Nothing, Unit>(Unit)
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -774,7 +774,7 @@ internal class WorkflowNodeTest {
 
   @Test fun toString_formats_as_WorkflowInstance_with_parent() {
     val workflow = Workflow.rendering<Nothing, Unit>(Unit)
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -807,7 +807,7 @@ internal class WorkflowNodeTest {
       }
     }
     val workflow = Workflow.rendering<Nothing, Unit>(Unit)
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -848,7 +848,7 @@ internal class WorkflowNodeTest {
       }
     }
     val workflow = Workflow.rendering<Nothing, Unit>(Unit)
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -895,7 +895,7 @@ internal class WorkflowNodeTest {
       initialState = { props -> "state($props)" },
       render = { _, _ -> fail() }
     )
-    WorkflowNode(
+    StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = "props",
@@ -941,7 +941,7 @@ internal class WorkflowNodeTest {
       onPropsChanged = { old, new, state -> "onPropsChanged($old, $new, $state)" },
       render = { _, state -> state }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = "old",
@@ -987,7 +987,7 @@ internal class WorkflowNodeTest {
       initialState = { "state" },
       render = { props, state -> "render($props, $state)" }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = "props",
@@ -1029,7 +1029,7 @@ internal class WorkflowNodeTest {
       render = { _, state -> state },
       snapshot = { state -> Snapshot.of("snapshot($state)") }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = "old",
@@ -1038,7 +1038,7 @@ internal class WorkflowNodeTest {
       baseContext = Unconfined,
       parent = TestSession(42)
     )
-    val snapshot = node.snapshot(workflow)
+    val snapshot = node.snapshot()
 
     assertEquals("state", interceptedState)
     assertEquals(Snapshot.of("snapshot(state)"), interceptedSnapshot)
@@ -1070,7 +1070,7 @@ internal class WorkflowNodeTest {
       render = { _, state -> state },
       snapshot = { null }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = workflow.id(key = "foo"),
       workflow = workflow.asStatefulWorkflow(),
       initialProps = "old",
@@ -1079,7 +1079,7 @@ internal class WorkflowNodeTest {
       baseContext = Unconfined,
       parent = TestSession(42)
     )
-    val snapshot = node.snapshot(workflow)
+    val snapshot = node.snapshot()
 
     assertEquals("state", interceptedState)
     assertNull(interceptedSnapshot)
@@ -1111,7 +1111,7 @@ internal class WorkflowNodeTest {
         "root(${renderChild(leafWorkflow, props)})"
       }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       id = rootWorkflow.id(key = "foo"),
       workflow = rootWorkflow.asStatefulWorkflow(),
       initialProps = "props",
@@ -1131,7 +1131,7 @@ internal class WorkflowNodeTest {
       val sink = eventHandler("eventHandler") { fail("Expected handler to fail.") }
       sink()
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1159,7 +1159,7 @@ internal class WorkflowNodeTest {
     val workflow = Workflow.stateless {
       actionSink.send(TestAction())
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1186,7 +1186,7 @@ internal class WorkflowNodeTest {
         }
       }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1211,7 +1211,7 @@ internal class WorkflowNodeTest {
     val workflow = Workflow.stateless<Unit, String, Sink<String>> {
       actionSink.contraMap { action("") { setOutput(it) } }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1238,7 +1238,7 @@ internal class WorkflowNodeTest {
     val workflow = Workflow.stateless<Unit, String?, Sink<String?>> {
       actionSink.contraMap { action("") { setOutput(null) } }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1269,7 +1269,7 @@ internal class WorkflowNodeTest {
         return@stateful renderState
       }
     )
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1292,7 +1292,7 @@ internal class WorkflowNodeTest {
         actionSink.send(action("") { setOutput("child:hello") })
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1318,7 +1318,7 @@ internal class WorkflowNodeTest {
         actionSink.send(action("") { setOutput(null) })
       }
     }
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       workflow.asStatefulWorkflow(),
       initialProps = Unit,
@@ -1343,7 +1343,7 @@ internal class WorkflowNodeTest {
     }
     val stateful = workflow.asStatefulWorkflow()
 
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       stateful,
       initialProps = "",
@@ -1367,7 +1367,7 @@ internal class WorkflowNodeTest {
       remember(key, typeOf<String>(), input) { value }
     }
     val stateful = workflow.asStatefulWorkflow()
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       stateful,
       initialProps = "",
@@ -1393,7 +1393,7 @@ internal class WorkflowNodeTest {
       remember(key, returnType) { value }
     }
     val stateful = workflow.asStatefulWorkflow()
-    val node = WorkflowNode(
+    val node = StatefulWorkflowNode(
       workflow.id(),
       stateful,
       initialProps = "" to ("" as Any),