/// T-iOS-15 · Broadcast adapter for a single-consumer `AsyncStream`.
///
/// `SessionEngine.events` has exactly ONE consumer (engine contract), but the
/// wiring needs three: `TerminalViewModel` (output/connection/exit),
/// `GateViewModel` (gate/digest) and `SessionActivityBridge` (adopted /
/// pending-⚠ / last-session persistence). Both VMs were designed for this —
/// their `events` parameter is injected separately from the engine precisely
/// so a fan-out branch can be passed instead (their type docs say so).
///
/// Shape: `branchCount` child streams are created up front (immutable `let`s
/// — no late subscription, so no missed-element semantics to define); ONE pump
/// task consumes the source and yields to every branch in order. AsyncStream's
/// default unbounded buffering means a slow branch never drops or blocks the
/// others. Source finish — which `SessionEngine.close()` guarantees — finishes
/// every branch; `cancel()` is the teardown belt-and-braces for rebuilds.
final class EventFanOut<Element: Sendable>: Sendable {
    let branches: [AsyncStream<Element>]
    private let continuations: [AsyncStream<Element>.Continuation]
    private let pumpTask: Task<Void, Never>

    init(source: AsyncStream<Element>, branchCount: Int) {
        var branches: [AsyncStream<Element>] = []
        var continuations: [AsyncStream<Element>.Continuation] = []
        for _ in 0..<branchCount {
            let (stream, continuation) = AsyncStream<Element>.makeStream()
            branches.append(stream)
            continuations.append(continuation)
        }
        self.branches = branches
        self.continuations = continuations
        let sinks = continuations
        pumpTask = Task {
            for await element in source {
                for sink in sinks {
                    sink.yield(element)
                }
            }
            for sink in sinks {
                sink.finish()
            }
        }
    }

    /// Stop pumping and finish every branch immediately (idempotent — finish
    /// on a finished continuation is a no-op). Used when a suspended terminal
    /// stack is torn down and rebuilt.
    func cancel() {
        pumpTask.cancel()
        for sink in continuations {
            sink.finish()
        }
    }
}