A ReactiveX observable like ReplaySubject but replaying only items that have never been observed?

Question

I need an Rx.NET observable that accumulates items while there is no active subscribers, and emits the whole accumulated sequence (and any future items) to new subscribes, as soon as there is any.

It is different from ReplaySubject in that it doesn't keep the items which have been once replayed to any subscriber. Thus, once a queued item has been observed by the current subscribers, it gets removed from the queue and won't be seen by any new future subscribers.

Can something like that be composed using standard Rx.NET operators?

I need it to tackle races conditions in the following scenario. There is a looping async workflow RunWorkflowAsync which needs to perform ResetAsync task when it observes a specific ResetRequestedEvent message.

Here's the whole thing as a .NET 6 console app:

using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading.Channels;

try
{
    await TestAsync();
}
catch (Exception ex)
{
    Console.WriteLine(ex.Message);
}

async Task TestAsync()
{
    var resetRequestsSubject = new Subject<ResetRequestedEvent>();

    using var cts = new CancellationTokenSource(20000);
    await Task.WhenAll(
        SimulateResetRequests(cts.Token), 
        RunWorkflowAsync(resetRequestsSubject, cts.Token));

    // simulate emmiting reset requests
    async Task SimulateResetRequests(CancellationToken cancelToken)
    {
        async Task Raise(int n, int delay)
        {
            var ev = new ResetRequestedEvent(n);
            Console.WriteLine($"{ev} issued");
            resetRequestsSubject!.OnNext(ev);
            await Task.Delay(delay, cancelToken);
        }

        await Raise(1, 50);
        await Raise(2, 50);
        await Raise(3, 50);
        await Raise(4, 1000);
        await Raise(5, 5000);
        await Raise(6, 4000);
        await Raise(7, 3000);
        resetRequestsSubject.OnCompleted();
    }

    // simulate the reset task
    async Task ResetAsync(CancellationToken cancelToken)
    {
        await Task.Delay(1000, cancelToken);
        Console.WriteLine("Reset done");
    }

    // simulate the work task
    async Task DoWorkAsync(CancellationToken cancelToken)
    {
        await Task.Delay(2000, cancelToken);
        Console.WriteLine("Work done");
    }


    // do reset, then work in a loop until cancelled
    async Task RunWorkflowAsync(IObservable<ResetRequestedEvent> resetRequests, CancellationToken externalCancelToken)
    {
        // from this point, make sure reset requests don't go unobserved
        var accumulatedResetRequests = resetRequests.Accumulate(externalCancelToken);
        using var auto1 = accumulatedResetRequests.Connect();

        while (true)
        {
            externalCancelToken.ThrowIfCancellationRequested(); // stops the whole workflow

            using var internalCts = CancellationTokenSource.CreateLinkedTokenSource(externalCancelToken);
            var internalCancelToken = internalCts.Token;

            // signal cancellation upon the most recent reset request
            using var auto2 = accumulatedResetRequests
                .Do(ev => Console.WriteLine($"{ev} seen"))
                .Throttle(TimeSpan.FromMilliseconds(100))
                .Do(ev => Console.WriteLine($"{ev} acted upon"))
                .Subscribe(_ => internalCts.Cancel());

            try
            {
                // start with a reset
                await ResetAsync(internalCancelToken);
                
                // do work until another reset is requested
                while (true)
                {
                    await DoWorkAsync(internalCancelToken);
                }
            }
            catch (OperationCanceledException)
            {
            }
        }
    }
}

record ResetRequestedEvent(int Number);

public static class RxExt
{
    class CumulativeObservable<T> : IConnectableObservable<T>
    {
        readonly IObservable<T> _source;
        readonly Channel<T> _channel;
        readonly CancellationToken _cancelToken;

        public CumulativeObservable(IObservable<T> source, CancellationToken cancellationToken)
        {
            _source = source;
            _channel = Channel.CreateUnbounded<T>();
            _cancelToken = cancellationToken;
        }

        public IDisposable Connect() =>
            _source.Subscribe(
                onNext: item => _channel.Writer.TryWrite(item),
                onError: ex => _channel.Writer.Complete(ex),
                onCompleted: () => _channel.Writer.Complete());

        public IDisposable Subscribe(IObserver<T> observer) =>
            _channel.Reader.ReadAllAsync(_cancelToken).ToObservable().Subscribe(observer);
    }
    public static IConnectableObservable<T> Accumulate<T>(
        this IObservable<T> @this, 
        CancellationToken cancellationToken) => 
        new CumulativeObservable<T>(@this, cancellationToken);
}

The idea is to stop all pending tasks inside RunWorkflowAsync and perform ResetAsync when ResetRequestedEvent message comes along.

I realize there's more than one way to cook an egg (and implement RunWorkflowAsync), but I like this approach as I don't need to think about thread safety when I use and recycle internalCts cancellation token source (to stop all pending task before another iteration).

Above, CumulativeObservable does what I want, but it's a very naive implementation which only supports one concurrent observable (unlike, say, ReplaySubject) and lacks any safety checks.

I'd prefer a composition that can be built using standard operators.