//! RFC 005 wake-slot tests: correctness with the slot on, push-policy
//! discrimination (actor vs scheduler context, spawns), displacement, the
//! default-off contract, and per-run counter reset.
//!
//! The slot is same-thread plain ops behind the existing queue-op contract,
//! so there is nothing new to model-check (RFC 005 §Summary); these tests
//! pin the *policy* — who lands in the slot, who never does — which is
//! observable through the `slot_hits` / `slot_displacements` counters.

use smarm::channel::channel;
use smarm::runtime::{init, Config};
use smarm::spawn;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

/// Ping-pong over channels: the slot's home pattern. Returns total roundtrips.
fn ping_pong(pairs: usize, roundtrips: u64) -> impl FnOnce() + Send + 'static {
    move || {
        let handles: Vec<_> = (0..pairs)
            .map(|_| {
                spawn(move || {
                    let (tx_ab, rx_ab) = channel::<u64>();
                    let (tx_ba, rx_ba) = channel::<u64>();
                    let echo = spawn(move || {
                        for _ in 0..roundtrips {
                            let v = rx_ab.recv().expect("echo recv");
                            tx_ba.send(v + 1).expect("echo send");
                        }
                    });
                    for i in 0..roundtrips {
                        tx_ab.send(i).expect("ping send");
                        assert_eq!(rx_ba.recv().expect("ping recv"), i + 1);
                    }
                    let _ = echo.join();
                })
            })
            .collect();
        for h in handles {
            h.join().expect("pair");
        }
    }
}

// ---------------------------------------------------------------------------
// Correctness: messaging workloads complete with the slot on, 1 and N threads
// ---------------------------------------------------------------------------

#[test]
fn ping_pong_correct_with_slot_on_single_thread() {
    let rt = init(Config::exact(1).wake_slot(true));
    rt.run(ping_pong(4, 200));
}

#[test]
fn ping_pong_correct_with_slot_on_multi_thread() {
    let rt = init(Config::exact(4).wake_slot(true));
    rt.run(ping_pong(8, 200));
}

// ---------------------------------------------------------------------------
// Push policy: actor-context wakes hit the slot; the default is off
// ---------------------------------------------------------------------------

#[test]
fn messaging_workload_exercises_the_slot() {
    let rt = init(Config::exact(1).wake_slot(true));
    rt.run(ping_pong(2, 100));
    let stats = rt.stats();
    assert!(
        stats.slot_hits() > 0,
        "send-wakes are actor-context unparks — the slot must see hits, got 0"
    );
}

#[test]
fn slot_off_means_zero_slot_traffic() {
    // Off explicitly…
    let rt = init(Config::exact(1).wake_slot(false));
    rt.run(ping_pong(2, 100));
    assert_eq!(rt.stats().slot_hits(), 0);
    assert_eq!(rt.stats().slot_displacements(), 0);

    // …and off by default (RFC 005: default off until the shootout accepts).
    let rt = init(Config::exact(1));
    rt.run(ping_pong(2, 100));
    assert_eq!(rt.stats().slot_hits(), 0, "wake_slot must default to OFF");
}

// ---------------------------------------------------------------------------
// Push policy: spawns and join-wakes (finalize = scheduler context) bypass
// ---------------------------------------------------------------------------

#[test]
fn spawn_join_workload_bypasses_the_slot() {
    let rt = init(Config::exact(1).wake_slot(true));
    rt.run(|| {
        for _ in 0..20 {
            let handles: Vec<_> = (0..50).map(|_| spawn(|| {})).collect();
            for h in handles {
                h.join().expect("trivial actor");
            }
        }
    });
    let stats = rt.stats();
    assert_eq!(
        stats.slot_hits(),
        0,
        "spawns go shared by policy and joiner wakes fire from finalize \
         (scheduler context) — a pure spawn/join workload must never touch \
         the slot"
    );
    assert_eq!(stats.slot_displacements(), 0);
}

// ---------------------------------------------------------------------------
// Displacement: newest wake takes the slot, occupant goes shared — and runs
// ---------------------------------------------------------------------------

#[test]
fn displaced_occupant_reaches_the_shared_queue_and_runs() {
    // Single thread, strict FIFO (rq-mutex default): rx1 and rx2 park on
    // their recvs before the sender runs; the sender's back-to-back sends
    // then produce two actor-context wakes — the second displaces the first.
    let rt = init(Config::exact(1).wake_slot(true));
    let ran = Arc::new(AtomicU64::new(0));
    let (r1, r2) = (ran.clone(), ran.clone());
    rt.run(move || {
        let (tx1, rx1) = channel::<u32>();
        let (tx2, rx2) = channel::<u32>();
        let h1 = spawn(move || {
            assert_eq!(rx1.recv().expect("rx1"), 1);
            r1.fetch_add(1, Ordering::Relaxed);
        });
        let h2 = spawn(move || {
            assert_eq!(rx2.recv().expect("rx2"), 2);
            r2.fetch_add(1, Ordering::Relaxed);
        });
        let sender = spawn(move || {
            tx1.send(1).expect("send 1"); // rx1 → slot
            tx2.send(2).expect("send 2"); // rx2 → slot, rx1 displaced → shared
        });
        sender.join().expect("sender");
        h1.join().expect("h1");
        h2.join().expect("h2");
    });
    assert_eq!(ran.load(Ordering::Relaxed), 2, "both receivers must run");
    let stats = rt.stats();
    assert!(
        stats.slot_displacements() >= 1,
        "back-to-back wakes of parked receivers must displace at least once"
    );
    assert!(stats.slot_hits() >= 1, "the displacing wake is slot-popped");
}

// ---------------------------------------------------------------------------
// Counters reset at the start of each run() on a reused Runtime
// ---------------------------------------------------------------------------

#[test]
fn slot_counters_reset_per_run() {
    let rt = init(Config::exact(1).wake_slot(true));
    rt.run(ping_pong(2, 100));
    let first = rt.stats().slot_hits();
    assert!(first > 0);

    // A slot-bypassing run on the same handle must read 0, not `first`.
    rt.run(|| {
        let h = spawn(|| {});
        h.join().expect("trivial");
    });
    assert_eq!(
        rt.stats().slot_hits(),
        0,
        "counters are reset at run() start; the second run had no slot traffic"
    );
}