Thanks, it's really great writeup 👍.

By the way, instead of CompletableDeferred, why don't you use suspendCancellableCoroutine by which you can also support cancelling registered callback if Flow is by chance terminated/cancelled/?

Let's say, this is how it looks if implemented with suspendCancellableCoroutine:

object WhileSubscribedOrRetained : SharingStarted {

    private val handler = Handler(Looper.getMainLooper())

    override fun command(subscriptionCount: StateFlow<Int>): Flow<SharingCommand> =
        subscriptionCount
            .transformLatest { count ->
                if (count > 0) {
                    emit(SharingCommand.START)
                } else {
                    awaitChoregrapherFramePostFrontOfQueue()
                    emit(SharingCommand.STOP)
                }
            }
            .dropWhile { it != SharingCommand.START }
            .distinctUntilChanged()

    override fun toString(): String = "WhileSubscribedOrRetained"

    @OptIn(InternalCoroutinesApi::class)
    private suspend fun awaitChoregrapherFramePostFrontOfQueue() {
        suspendCancellableCoroutine<Unit> { cont ->
            val frameCallback = Choreographer.FrameCallback {
                handler.postAtFrontOfQueue {
                    handler.post {
                        if (!cont.isCompleted) {
                            cont.resume(Unit)
                        }
                    }
                }
            }

            Choreographer.getInstance().postFrameCallback(frameCallback)

            cont.invokeOnCancellation {
                Choreographer.getInstance().removeFrameCallback(frameCallback)
            }
        }
    }
}