Solution: Flow with history

The problem is that the withHistory function always returns the same list reference; so, when its values are changed, all the previously emitted lists are also changed. The simplest solution is to use a read-only list instead of a mutable list.

fun <T> Flow<T>.withHistory(): Flow<List<T>> = flow {
    var history = listOf<T>()
    emit(history)
    collect {
        history += it
        emit(history)
    }
}

This function can also be implemented using scan:

fun <T> Flow<T>.withHistory(): Flow<List<T>> =
    scan(emptyList()) { history, value ->
        history + value
    }

Example solution in playground

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
import org.junit.Test
import kotlin.test.assertEquals

fun <T> Flow<T>.withHistory(): Flow<List<T>> = flow {
    var history = listOf<T>()
    emit(history)
    collect {
        history += it
        emit(history)
    }
}

suspend fun main() {
    flowOf(1, 2, 3)
        .withHistory()
        .toList()
        .let(::println)
    // [[1, 2, 3], [1, 2, 3], [1, 2, 3], [1, 2, 3]]
}

class FlowHistoryTests {
    @Test
    fun `should emit empty for empty`() = runTest {
        assertEquals(listOf(listOf()), flowOf<String>().withHistory().toList())
    }

    @Test
    fun `should emit history`() = runTest {
        val flow = flowOf(1, 2, 3).withHistory()
        assertEquals(listOf(listOf(), listOf(1), listOf(1, 2), listOf(1, 2, 3)), flow.toList())
    }

    @Test
    fun `should emit elements as they appear`() = runTest {
        val flow = flow {
            delay(100)
            emit(1)
            delay(1000)
            emit(2)
            delay(10000)
            emit(3)
        }.withHistory()
            .withVirtualTime(this)
        assertEquals(
            listOf(
                ValueAndTime(listOf(), 0),
                ValueAndTime(listOf(1), 100),
                ValueAndTime(listOf(1, 2), 1100),
                ValueAndTime(listOf(1, 2, 3), 11100),
            ), flow.toList()
        )
    }
}

fun <T> Flow<T>.withVirtualTime(testScope: TestScope): Flow<ValueAndTime<T>> =
    map { ValueAndTime(it, testScope.currentTime) }

data class ValueAndTime<T>(val value: T, val timeMillis: Long)