Cool exploration but it doesn't work unless everything is sendable, and AsyncSequence and AsyncStream don't seem to be sendable

AsyncMonitorTask.swift

// UPDATE: See https://github.com/samsonjs/AsyncMonitor if this is your kind of thing!

import Foundation

public final class AsyncMonitorTask: Hashable {
    let task: Task<Void, Never>

    init<Element: Sendable>(
        sequence: sending any AsyncSequence<Element, Never>,
        performing block: @escaping @MainActor (Element) async -> Void
    ) {
        self.task = Task {
            for await element in sequence {
                await block(element)
            }
        }
    }

    deinit {
        task.cancel()
    }

    public func store(in set: inout Set<AsyncMonitorTask>) {
        set.insert(self)
    }
}

// MARK: - Hashable conformance

public extension AsyncMonitorTask {
    static func == (lhs: AsyncMonitorTask, rhs: AsyncMonitorTask) -> Bool {
        lhs.task == rhs.task
    }

    func hash(into hasher: inout Hasher) {
        hasher.combine(task)
    }
}

public extension AsyncSequence where Element: Sendable, Failure == Never {
    func monitor(_ block: @escaping @Sendable (Element) async -> Void) -> AsyncMonitorTask {
        AsyncMonitorTask(sequence: self, performing: block)
    }

    func monitor<Context: AnyObject & Sendable>(
        context: Context,
        _ block: @escaping @Sendable (Context, Element) async -> Void
    ) -> AsyncMonitorTask {
        AsyncMonitorTask(sequence: self) { [weak context] element in
            guard let context else { return }
            await block(context, element)
        }
    }
}

Ok I had a look!

I made a few changes. Used an isolated param to avoid needing Sendable/sending. I think this could be close. However, I did remove the MainActor-ness of the closures and that may or may not be annoying depending on how this is supposed to be used.

public final class AsyncMonitorTask: Hashable {
    let task: Task<Void, Never>

    init<Element: Sendable>(
        isolation: isolated (any Actor)? = #isolation,
        sequence: any AsyncSequence<Element, Never>,
        performing block: @escaping (Element) async -> Void
    ) {
        self.task = Task {
            _ = isolation // use capture trick to inherit isolation

            for await element in sequence {
                await block(element)
            }
        }
    }

    deinit {
        task.cancel()
    }

    public func store(in set: inout Set<AsyncMonitorTask>) {
        set.insert(self)
    }
}

// MARK: - Hashable conformance

public extension AsyncMonitorTask {
    static func == (lhs: AsyncMonitorTask, rhs: AsyncMonitorTask) -> Bool {
        lhs.task == rhs.task
    }

    func hash(into hasher: inout Hasher) {
        hasher.combine(task)
    }
}

public extension AsyncSequence where Element: Sendable, Failure == Never {
    func monitor(_ block: @escaping (Element) async -> Void) -> AsyncMonitorTask {
        AsyncMonitorTask(sequence: self, performing: block)
    }

    func monitor<Context: AnyObject & Sendable>(
        context: Context,
        _ block: @escaping (Context, Element) async -> Void
    ) -> AsyncMonitorTask {
        AsyncMonitorTask(sequence: self) { [weak context] element in
            guard let context else { return }
            await block(context, element)
        }
    }
}

This looks pretty much perfect! Thanks for teaching me about the isolation capture trick too, I didn't know about that one. Losing MainActor-ness here is perfect really, I think it should inherit the caller's isolation and stay there, unlike where this was extracted/expanded from.