import Foundation
import Testing

@testable import NucleicCore

/// The generic hang detector: it must alert when a process goes silent with idle CPU, stay quiet
/// while the process is active (or busy-but-CPU-hot), and — once it has alerted — report a
/// self-resolution when the process recovers or exits. Driven by a fake handle whose activity clock
/// and CPU sampler are controlled directly, so the tiny timing windows here are deterministic.
@Suite struct ProcessStallMonitorTests {
    @Test func productionDefaultsAllowLongQuietPhases() {
        let config = ProcessStallConfig.default
        #expect(config.sampleIntervalNanos == 5 * 1_000_000_000)
        #expect(config.idleThresholdNanos == 300 * 1_000_000_000)
        #expect(config.ioByteThreshold == 64 * 1024)
        #expect(config.lowCPUSustainNanos == 60 * 1_000_000_000)
    }

    /// A `ProcessHandle` whose `lastActivityNanos` and exit are driven by the test.
    final class FakeHandle: ProcessHandle, @unchecked Sendable {
        let stdoutLines: AsyncThrowingStream<Data, Error>
        let stderrLines: AsyncThrowingStream<Data, Error>
        var processID: Int32 { 0 }

        private let lock = NSLock()
        private var activityNanos = DispatchTime.now().uptimeNanoseconds
        private var exited = false
        private var waiters: [CheckedContinuation<Int32, Never>] = []

        init() {
            stdoutLines = AsyncThrowingStream { _ in }
            stderrLines = AsyncThrowingStream { _ in }
        }

        var lastActivityNanos: UInt64 { lock.withLock { activityNanos } }

        /// Pretend the last output was `nanosAgo` in the past (to exceed the idle threshold).
        func setSilent(forNanos nanosAgo: UInt64) {
            let t = DispatchTime.now().uptimeNanoseconds &- nanosAgo
            lock.withLock { activityNanos = t }
        }

        /// Mark output as just-now (active).
        func bump() {
            let t = DispatchTime.now().uptimeNanoseconds
            lock.withLock { activityNanos = t }
        }

        func writeLine(_ data: Data) throws {}
        func closeStdin() {}
        func sendSignal(_ sig: Int32) {}

        func wait() async -> Int32 {
            // Cancellation-aware so the monitor's exit-watcher task completes when the monitor is
            // cancelled, even though this fake process never exits on its own.
            await withTaskCancellationHandler {
                await withCheckedContinuation { (c: CheckedContinuation<Int32, Never>) in
                    let already = lock.withLock { () -> Bool in
                        if exited { return true }
                        waiters.append(c)
                        return false
                    }
                    if already { c.resume(returning: 0) }
                }
            } onCancel: {
                markExited()
            }
        }

        func markExited() {
            let toWake: [CheckedContinuation<Int32, Never>] = lock.withLock {
                exited = true
                let w = waiters
                waiters.removeAll()
                return w
            }
            for c in toWake { c.resume(returning: 0) }
        }
    }

    /// One-shot signal a callback can fire and a test polls for with a timeout.
    final class Latch: @unchecked Sendable {
        private let lock = NSLock()
        private var fired = false
        func fire() { lock.withLock { fired = true } }
        var didFire: Bool { lock.withLock { fired } }
    }

    /// Poll (rather than park on a continuation) so a never-fired latch simply times out — no leaked
    /// task can wedge the test the way a task-group awaiting an un-resumable child would.
    private func fires(within nanos: UInt64, _ latch: Latch) async -> Bool {
        let deadline = DispatchTime.now().uptimeNanoseconds &+ nanos
        while DispatchTime.now().uptimeNanoseconds < deadline {
            if latch.didFire { return true }
            try? await Task.sleep(nanoseconds: 5_000_000)
        }
        return latch.didFire
    }

    /// Fast config for the tests: sample every 20ms, consider ≥10ms of silence idle, require ~40ms
    /// (2 samples) of sustained near-zero CPU before alerting.
    private var fastConfig: ProcessStallConfig {
        ProcessStallConfig(
            sampleIntervalNanos: 20_000_000, idleThresholdNanos: 10_000_000,
            cpuFractionThreshold: 0.02, lowCPUSustainNanos: 40_000_000)
    }

    @Test func alertsWhenSilentAndIdleCPU() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)  // 1s of silence — well past the 10ms threshold
        let stalled = Latch()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig, cpuSampler: nil,
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 1_000_000_000, stalled))
    }

    @Test func staysQuietWhileBusyButCPUHot() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)  // silent…
        let stalled = Latch()
        // …but CPU keeps climbing far faster than the 2% threshold, so it's clearly working.
        let cpu = ManagedAtomicNanos()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig,
            cpuSampler: { cpu.advance(by: 1_000_000_000) },  // ~a full core each sample
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 300_000_000, stalled) == false)
    }

    @Test func staysQuietWhileMovingIOAtIdleCPU() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)  // silent…
        let stalled = Latch()
        // …CPU flat (a network/registry push burns almost none), but disk I/O keeps climbing well past
        // the byte threshold each sample, so it's plainly moving data — the `cctl images push` case.
        let cpu = ManagedAtomicNanos()
        let io = ManagedAtomicNanos()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig,
            cpuSampler: { cpu.advance(by: 0) },  // no CPU progress
            ioSampler: { io.advance(by: 4 * 1024 * 1024) },  // 4 MiB/sample of I/O
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 300_000_000, stalled) == false)
    }

    @Test func alertsWhenSilentAndBothResourcesQuiet() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)
        let stalled = Latch()
        // Both samplers present but flat (no CPU, no I/O) and output silent → a genuine wedge.
        let cpu = ManagedAtomicNanos()
        let io = ManagedAtomicNanos()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig,
            cpuSampler: { cpu.advance(by: 0) }, ioSampler: { io.advance(by: 0) },
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 1_000_000_000, stalled))
    }

    @Test func bufferingTailDiscountsOutputIdleButHonorsResources() async {
        // A terminal `tail` withholds output, so `outputIdleCounts: false`. Even with output *fresh*
        // (never silent), a flat-CPU/flat-I/O tree must still be judged wedged on the resource signals.
        let handle = FakeHandle()  // fresh activity clock — not silent
        let stalled = Latch()
        let cpu = ManagedAtomicNanos()
        let io = ManagedAtomicNanos()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig,
            cpuSampler: { cpu.advance(by: 0) }, ioSampler: { io.advance(by: 0) },
            outputIdleCounts: false,
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 1_000_000_000, stalled))
    }

    @Test func bufferingTailStillSparedWhileMovingIO() async {
        // The real win: a `| tail -6`-wrapped push. Output is silent (buffered by tail), output-idle
        // is discounted, and disk I/O is flowing → not a hang.
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)
        let stalled = Latch()
        let cpu = ManagedAtomicNanos()
        let io = ManagedAtomicNanos()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig,
            cpuSampler: { cpu.advance(by: 0) },
            ioSampler: { io.advance(by: 4 * 1024 * 1024) },
            outputIdleCounts: false,
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 300_000_000, stalled) == false)
    }

    @Test func reportsExitAfterAlerting() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)
        let stalled = Latch()
        let exitedResolved = Latch()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig, cpuSampler: nil,
            onStall: { _ in stalled.fire() },
            onResolve: { resolution in if resolution == .exited { exitedResolved.fire() } })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 1_000_000_000, stalled))
        handle.markExited()
        #expect(await fires(within: 1_000_000_000, exitedResolved))
    }

    @Test func reportsResumeWhenOutputReturns() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 5_000_000_000)  // 5s silent — well past the 300ms idle threshold
        let stalled = Latch()
        let resumed = Latch()
        // The idle threshold (300ms) must exceed the sample interval (20ms) so a single fresh chunk
        // clears the idle window for the next sample and the monitor sees recovery.
        let config = ProcessStallConfig(
            sampleIntervalNanos: 20_000_000, idleThresholdNanos: 300_000_000,
            cpuFractionThreshold: 0.02, lowCPUSustainNanos: 40_000_000)
        let monitor = ProcessStallMonitor(
            handle: handle, config: config, cpuSampler: nil,
            onStall: { _ in stalled.fire() },
            onResolve: { resolution in if resolution == .resumed { resumed.fire() } })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 2_000_000_000, stalled))
        handle.bump()  // output is flowing again
        #expect(await fires(within: 2_000_000_000, resumed))
    }

    /// The verdict `host_exec`'s wall-clock ceiling and a bounded monitor's age both consult before
    /// reaping anything: it must read false while the tree is working, and only flip once the process
    /// is confirmed wedged. A ceiling that fires on `isStalled == false` would be the old
    /// kill-a-slow-build bug.
    @Test func isStalledReportsTheLiveVerdict() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)
        let stalled = Latch()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig, cpuSampler: nil,
            onStall: { _ in stalled.fire() }, onResolve: { _ in })
        #expect(monitor.isStalled == false)  // nothing sampled yet
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 1_000_000_000, stalled))
        #expect(monitor.isStalled)
    }

    /// A busy-but-silent tree is NOT stalled, however long it has been running — the case that used to
    /// get killed at the timeout. CPU keeps climbing, so the ceiling must never see a stall verdict.
    @Test func isStalledStaysFalseWhileTheTreeBurnsCPU() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 1_000_000_000)
        let cpu = ManagedAtomicNanos()
        let monitor = ProcessStallMonitor(
            handle: handle, config: fastConfig,
            cpuSampler: { cpu.advance(by: 1_000_000_000) },
            onStall: { _ in }, onResolve: { _ in })
        monitor.start()
        defer { monitor.cancel() }
        // Well past several sample intervals and the sustain window.
        try? await Task.sleep(nanoseconds: 300_000_000)
        #expect(monitor.isStalled == false)
    }

    /// Recovery clears the verdict, so a command that goes quiet and then resumes work is spared by a
    /// ceiling that happens to land in between.
    @Test func isStalledClearsWhenOutputReturns() async {
        let handle = FakeHandle()
        handle.setSilent(forNanos: 5_000_000_000)
        let stalled = Latch()
        let resumed = Latch()
        let config = ProcessStallConfig(
            sampleIntervalNanos: 20_000_000, idleThresholdNanos: 300_000_000,
            cpuFractionThreshold: 0.02, lowCPUSustainNanos: 40_000_000)
        let monitor = ProcessStallMonitor(
            handle: handle, config: config, cpuSampler: nil,
            onStall: { _ in stalled.fire() },
            onResolve: { resolution in if resolution == .resumed { resumed.fire() } })
        monitor.start()
        defer { monitor.cancel() }
        #expect(await fires(within: 2_000_000_000, stalled))
        #expect(monitor.isStalled)
        handle.bump()
        #expect(await fires(within: 2_000_000_000, resumed))
        #expect(monitor.isStalled == false)
    }

    /// Minimal monotonic counter for the CPU sampler.
    final class ManagedAtomicNanos: @unchecked Sendable {
        private let lock = NSLock()
        private var value: UInt64 = 0
        func advance(by delta: UInt64) -> UInt64 { lock.withLock { value &+= delta; return value } }
    }
}
