nucleic/Sources/NucleicCore/SessionUIProjector.swift

import Foundation
import NucleicProtocol

/// Per-session reduction of the canonical event stream into batched UI commits
/// (docs/MAIN_THREAD_PERFORMANCE_PLAN.md item 2).
///
/// One projector runs per live session, on its own actor executor — not the main actor.
/// It consumes the `SessionController.subscribe()` stream, appends every event to a
/// pending batch in arrival order, and hands the batch to the main actor:
///
/// - **immediately** when an event is a semantic boundary (approval requested/resolved,
///   user text, error, run/session status transition, terminal event, note, quota push).
///   The pending streaming deltas ride along in front of it, in order, so a terminal
///   event can never outrun the text it concludes.
/// - **on a cadence** for ordinary streaming traffic (assistant text, thinking, tool
///   lifecycle and input deltas, usage): ~11 commits/s for the open chat, ~3/s for a
///   background session. The first event after a quiet spell flushes with no added
///   latency; only events landing inside a live window coalesce.
///
/// The projector never drops, reorders, or rewrites events — canonical transcript
/// durability, seq order, and sync delivery are untouched; only the *cadence* of
/// main-actor commits changes. On cancellation (re-observe, session teardown) the
/// pending tail is still delivered; after a session delete the main-actor side finds
/// no controller and drops it — that's the delete's decision, not a coalescing loss.
public actor SessionUIProjector {
    /// How urgently coalescable events reach the main actor.
    public enum Cadence: Sendable {
        /// The chat on screen: ~11 batched commits/s keeps text visibly streaming.
        case open
        /// Sidebar-only session: a few commits/s is plenty for summary rows.
        case background
    }

    private let openInterval: Duration
    private let backgroundInterval: Duration
    private var cadence: Cadence
    private let deliver: @MainActor @Sendable ([AgentEvent]) async -> Void

    /// Events awaiting the next flush, in arrival order.
    private var pending: [AgentEvent] = []
    /// A semantic event is pending — the next flush skips the coalesce window.
    private var urgent = false
    /// When the last flush happened; nil until the first (which is always immediate).
    private var lastFlush: ContinuousClock.Instant?
    /// Parks the drain loop while no events are pending.
    private var wake: CheckedContinuation<Void, Never>?
    /// The live coalesce-window sleep, cancellable by an urgent enqueue.
    private var coalesceSleep: Task<Void, Never>?

    /// `openInterval`/`backgroundInterval` are injectable for tests; production uses the
    /// defaults (plan targets: 10–12 commits/s open, 2–4 background).
    public init(
        cadence: Cadence,
        openInterval: Duration = .milliseconds(90),
        backgroundInterval: Duration = .milliseconds(300),
        deliver: @escaping @MainActor @Sendable ([AgentEvent]) async -> Void
    ) {
        self.cadence = cadence
        self.openInterval = openInterval
        self.backgroundInterval = backgroundInterval
        self.deliver = deliver
    }

    /// Retarget the flush cadence (the open chat changed). Cuts any in-flight coalesce
    /// window short so a session promoted to `.open` starts committing at UI cadence
    /// now, not after one last background-length window.
    public func setCadence(_ new: Cadence) {
        guard cadence != new else { return }
        cadence = new
        coalesceSleep?.cancel()
    }

    /// Consume the stream until it ends (session shutdown) or the surrounding task is
    /// cancelled (re-observe, teardown), then deliver any pending tail.
    public func run(_ stream: AsyncStream<AgentEvent>) async {
        await withTaskGroup(of: Void.self) { group in
            group.addTask { await self.pump(stream) }
            group.addTask { await self.drainLoop() }
            // The pump ends when the stream terminates; the drain loop only ends by
            // cancellation — either way, wind the other one down too.
            await group.next()
            group.cancelAll()
        }
        // Events that arrived before the end still reach the UI — a coalesce window
        // must never turn into event loss.
        await flushNow()
    }

    // MARK: - Stream side

    private func pump(_ stream: AsyncStream<AgentEvent>) async {
        for await event in stream { enqueue(event) }
    }

    private func enqueue(_ event: AgentEvent) {
        pending.append(event)
        if Self.flushesImmediately(event.kind) {
            urgent = true
            coalesceSleep?.cancel()
        }
        wake?.resume()
        wake = nil
    }

    /// Semantic boundaries the user must see with no coalescing latency. Everything
    /// else is streaming-progress traffic whose only observable effect is a display
    /// refresh, which the next flush covers.
    static func flushesImmediately(_ kind: AgentEvent.Kind) -> Bool {
        switch kind {
        case .assistantText, .thinking, .toolCallStarted, .toolCallInputDelta,
            .toolCallCompleted, .toolResult, .fileChange, .usage, .raw:
            return false
        case .sessionStarted, .userText, .approvalRequested, .approvalResolved,
            .rateLimit, .codexUsage, .turnCompleted, .runFinished, .error, .note:
            return true
        }
    }

    // MARK: - Flush side

    private var flushInterval: Duration {
        switch cadence {
        case .open: openInterval
        case .background: backgroundInterval
        }
    }

    private func drainLoop() async {
        while !Task.isCancelled {
            await parkUntilEvent()
            if Task.isCancelled { return }
            // Coalesce: the window is measured from the last flush, so the first event
            // after a quiet spell flushes immediately; an urgent event skips the wait.
            if !urgent, let last = lastFlush {
                let target = last + flushInterval
                if ContinuousClock.now < target { await coalesce(until: target) }
            }
            if Task.isCancelled { return }
            await flushNow()
        }
    }

    /// Sleep until `target`, waking early when an urgent event lands (its enqueue
    /// cancels the sleep) or the drain loop is cancelled.
    private func coalesce(until target: ContinuousClock.Instant) async {
        let sleep = Task { () -> Void in
            try? await Task.sleep(until: target, clock: .continuous)
        }
        coalesceSleep = sleep
        await withTaskCancellationHandler {
            _ = await sleep.value
        } onCancel: {
            sleep.cancel()
        }
        coalesceSleep = nil
    }

    private func flushNow() async {
        guard !pending.isEmpty else { return }
        let batch = pending
        pending = []
        urgent = false
        lastFlush = .now
        await deliver(batch)
    }

    /// Park until at least one event is pending. Cancellation-safe: a cancel while
    /// parked resumes the continuation instead of stranding the drain loop.
    private func parkUntilEvent() async {
        await withTaskCancellationHandler {
            await withCheckedContinuation { (cont: CheckedContinuation<Void, Never>) in
                if !pending.isEmpty || Task.isCancelled {
                    cont.resume()
                } else {
                    wake = cont
                }
            }
        } onCancel: {
            Task { await self.releaseWake() }
        }
    }

    private func releaseWake() {
        wake?.resume()
        wake = nil
    }

    /// Test hook: how many events sit in the pending batch right now.
    func pendingCount() -> Int { pending.count }
}