import Foundation
import NucleicProtocol
import Testing

@testable import NucleicCore

/// Contract tests for the per-session event reducer (docs/MAIN_THREAD_PERFORMANCE_PLAN.md
/// item 2): order preservation, terminal flush, semantic-event immediacy during a
/// streaming burst, coalescing, cancellation with a pending batch, and cadence promotion.
@Suite struct SessionUIProjectorTests {
    func makeEvent(seq: UInt64, kind: AgentEvent.Kind) -> AgentEvent {
        AgentEvent(
            sessionID: SessionID(rawValue: "p-session"),
            seq: seq,
            at: Date(timeIntervalSince1970: 1_750_000_000),
            backend: .claudeCode,
            nativeType: nil,
            kind: kind)
    }

    func delta(_ seq: UInt64) -> AgentEvent {
        makeEvent(seq: seq, kind: .assistantText(TextChunk(messageID: "m", text: "t\(seq)", isPartial: true)))
    }

    func terminal(_ seq: UInt64) -> AgentEvent {
        makeEvent(seq: seq, kind: .runFinished(RunFinished(outcome: .completed, finalText: "done")))
    }

    /// Batches recorded on the main actor, exactly as AppStore would receive them.
    @MainActor final class BatchRecorder {
        private(set) var batches: [[AgentEvent]] = []
        func record(_ batch: [AgentEvent]) { batches.append(batch) }
        var flat: [AgentEvent] { batches.flatMap { $0 } }
    }

    /// Poll until `condition` holds (or ~5s pass); the projector's flushes are async, so
    /// assertions on the recorder need a settle wait that isn't a bare sleep.
    func waitUntil(_ condition: @MainActor @Sendable @escaping () -> Bool) async -> Bool {
        for _ in 0..<500 {
            if await MainActor.run(body: condition) { return true }
            try? await Task.sleep(for: .milliseconds(10))
        }
        return await MainActor.run(body: condition)
    }

    @Test func orderPreservedAndTerminalFlushesPendingDeltas() async {
        let recorder = await BatchRecorder()
        // A window far longer than the test: only immediacy rules can flush.
        let projector = SessionUIProjector(
            cadence: .background, backgroundInterval: .seconds(60)
        ) { batch in recorder.record(batch) }
        let (stream, cont) = AsyncStream.makeStream(of: AgentEvent.self)
        let run = Task { await projector.run(stream) }

        let events = (1...5).map { delta(UInt64($0)) } + [terminal(6)]
        for event in events { cont.yield(event) }

        // The terminal event is semantic: everything pending must land without waiting
        // out the 60s window, in canonical order.
        #expect(await waitUntil { recorder.flat.count == events.count })
        #expect(await MainActor.run { recorder.flat.map(\.seq) } == events.map(\.seq))

        cont.finish()
        await run.value
    }

    @Test func semanticEventLandsImmediatelyDuringBurst() async {
        let recorder = await BatchRecorder()
        let projector = SessionUIProjector(
            cadence: .background, backgroundInterval: .seconds(60)
        ) { batch in recorder.record(batch) }
        let (stream, cont) = AsyncStream.makeStream(of: AgentEvent.self)
        let run = Task { await projector.run(stream) }

        // First delta flushes immediately (quiet spell); the rest sit in the 60s window…
        cont.yield(delta(1))
        #expect(await waitUntil { recorder.flat.count == 1 })
        for seq in 2...20 { cont.yield(delta(UInt64(seq))) }
        // …until a semantic boundary (a user message) cuts the window short.
        cont.yield(makeEvent(seq: 21, kind: .userText(TextChunk(messageID: "u", text: "hi", isPartial: false))))

        #expect(await waitUntil { recorder.flat.count == 21 })
        #expect(await MainActor.run { recorder.flat.map(\.seq) } == (1...21).map { UInt64($0) })

        cont.finish()
        await run.value
    }

    @Test func cancellationDeliversPendingTail() async {
        let recorder = await BatchRecorder()
        let projector = SessionUIProjector(
            cadence: .background, backgroundInterval: .seconds(60)
        ) { batch in recorder.record(batch) }
        let (stream, cont) = AsyncStream.makeStream(of: AgentEvent.self)
        let run = Task { await projector.run(stream) }

        cont.yield(delta(1))
        #expect(await waitUntil { recorder.flat.count == 1 })
        // These sit pending inside the 60s window when the observer is torn down.
        cont.yield(delta(2))
        cont.yield(delta(3))
        // Wait until the pump has actually enqueued both before cancelling — an event
        // still in the stream buffer at cancel is outside the projector's contract.
        var enqueued = false
        for _ in 0..<500 {
            if await projector.pendingCount() == 2 { enqueued = true; break }
            try? await Task.sleep(for: .milliseconds(10))
        }
        #expect(enqueued)
        run.cancel()
        await run.value

        // A coalesce window must never turn into event loss: the tail is delivered.
        #expect(await MainActor.run { recorder.flat.map(\.seq) } == [1, 2, 3])
    }

    @Test func burstCoalescesIntoFewBatches() async {
        let recorder = await BatchRecorder()
        let projector = SessionUIProjector(
            cadence: .open, openInterval: .milliseconds(100)
        ) { batch in recorder.record(batch) }
        let (stream, cont) = AsyncStream.makeStream(of: AgentEvent.self)
        let run = Task { await projector.run(stream) }

        for seq in 1...50 { cont.yield(delta(UInt64(seq))) }
        #expect(await waitUntil { recorder.flat.count == 50 })
        // 50 back-to-back deltas may not commit 50 times — that is the whole point.
        // (First flush is immediate; the rest coalesce into ~1–2 trailing windows.)
        #expect(await MainActor.run { recorder.batches.count } < 10)
        #expect(await MainActor.run { recorder.flat.map(\.seq) } == (1...50).map { UInt64($0) })

        cont.finish()
        await run.value
    }

    @Test func promotionToOpenCutsBackgroundWindowShort() async {
        let recorder = await BatchRecorder()
        let projector = SessionUIProjector(
            cadence: .background, openInterval: .milliseconds(50), backgroundInterval: .seconds(60)
        ) { batch in recorder.record(batch) }
        let (stream, cont) = AsyncStream.makeStream(of: AgentEvent.self)
        let run = Task { await projector.run(stream) }

        cont.yield(delta(1))
        #expect(await waitUntil { recorder.flat.count == 1 })
        // Sits in the 60s background window…
        cont.yield(delta(2))
        // …until the user opens the chat: promotion must not wait out the old window.
        await projector.setCadence(.open)
        #expect(await waitUntil { recorder.flat.count == 2 })

        cont.finish()
        await run.value
    }

    @Test func streamEndFlushesTail() async {
        let recorder = await BatchRecorder()
        let projector = SessionUIProjector(
            cadence: .background, backgroundInterval: .seconds(60)
        ) { batch in recorder.record(batch) }
        let (stream, cont) = AsyncStream.makeStream(of: AgentEvent.self)
        let run = Task { await projector.run(stream) }

        cont.yield(delta(1))
        #expect(await waitUntil { recorder.flat.count == 1 })
        cont.yield(delta(2))
        cont.finish()  // controller shut down mid-window
        await run.value

        #expect(await MainActor.run { recorder.flat.map(\.seq) } == [1, 2])
    }

    @Test func classificationSeparatesStreamingFromSemantic() {
        #expect(!SessionUIProjector.flushesImmediately(
            .assistantText(TextChunk(messageID: "m", text: "t", isPartial: true))))
        #expect(!SessionUIProjector.flushesImmediately(
            .thinking(TextChunk(messageID: "m", text: "t", isPartial: true))))
        #expect(SessionUIProjector.flushesImmediately(
            .runFinished(RunFinished(outcome: .completed, finalText: nil))))
        #expect(SessionUIProjector.flushesImmediately(
            .userText(TextChunk(messageID: "m", text: "t", isPartial: false))))
    }
}
