//! Integration tests for `smarm-causal` — native causal profiling (RFC 007).
//!
//! Gated on the feature; run with:
//!   cargo test --test causal --features smarm-causal
#![cfg(feature = "smarm-causal")]

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

/// Progress points count, and snapshots expose deltas by name.
#[test]
fn progress_point_counts() {
    smarm::init(smarm::Config::exact(1)).run(|| {
        let before = smarm::causal::progress_snapshot();
        let h = smarm::spawn(|| {
            for _ in 0..100 {
                smarm::progress!("units-a");
            }
            for _ in 0..7 {
                smarm::progress!("units-b");
            }
        });
        h.join().unwrap();
        let after = smarm::causal::progress_snapshot();
        let delta = |name: &str| {
            after.iter().find(|(n, _)| n == name).map(|(_, c)| *c).unwrap()
                - before
                    .iter()
                    .find(|(n, _)| n == name)
                    .map(|(_, c)| *c)
                    .unwrap_or(0)
        };
        assert_eq!(delta("units-a"), 100);
        assert_eq!(delta("units-b"), 7);
    });
}

/// Site guards nest and restore, and site identity is per-actor: observable
/// from inside the actor and gone once the guard drops.
#[test]
fn site_guard_nesting_restores() {
    smarm::init(smarm::Config::exact(1)).run(|| {
        let h = smarm::spawn(|| {
            assert_eq!(smarm::causal::current_site_name(), None);
            {
                let _outer = smarm::causal_site!("outer");
                assert_eq!(
                    smarm::causal::current_site_name().as_deref(),
                    Some("outer")
                );
                {
                    let _inner = smarm::causal_site!("inner");
                    assert_eq!(
                        smarm::causal::current_site_name().as_deref(),
                        Some("inner")
                    );
                }
                assert_eq!(
                    smarm::causal::current_site_name().as_deref(),
                    Some("outer")
                );
            }
            assert_eq!(smarm::causal::current_site_name(), None);
        });
        h.join().unwrap();
    });
}

/// The core Coz transposition: while an experiment targets site S with a
/// nonzero speedup, work inside S grows the global delay ledger and credits
/// itself; an actor *outside* S absorbs the difference (its per-actor ledger
/// catches up to the global one) instead of running delay-free.
#[test]
fn virtual_speedup_ledger() {
    let _s = serial();
    let global_out = Arc::new(AtomicU64::new(0));
    let absorbed_out = Arc::new(AtomicU64::new(0));
    let g_out = global_out.clone();
    let a_out = absorbed_out.clone();

    smarm::init(smarm::Config::exact(2)).run(move || {
        let stop = Arc::new(AtomicBool::new(false));

        // Bystander: never in the target site; must absorb injected delay.
        let stop2 = stop.clone();
        let out2 = a_out.clone();
        let bystander = smarm::spawn(move || {
            while !stop2.load(Ordering::Relaxed) {
                smarm::check!();
                out2.store(
                    smarm::causal::my_absorbed_delay_cycles(),
                    Ordering::Relaxed,
                );
            }
        });

        // Target: spins inside the experiment's site.
        let stop3 = stop.clone();
        let target = smarm::spawn(move || {
            let _g = smarm::causal_site!("hot-site");
            while !stop3.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });

        let base = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("hot-site", 50);
        smarm::sleep(Duration::from_millis(200));
        g_out.store(
            smarm::causal::global_delay_cycles() - base,
            Ordering::Relaxed,
        );
        smarm::causal::end_experiment_for_test();

        stop.store(true, Ordering::Relaxed);
        bystander.join().unwrap();
        target.join().unwrap();
    });

    let global = global_out.load(Ordering::Relaxed);
    let absorbed = absorbed_out.load(Ordering::Relaxed);

    // The target's site work must have grown the ledger…
    assert!(global > 0, "global delay ledger never grew");
    // …and the bystander must have absorbed a meaningful share of it (it can
    // lag by a few check intervals; require half to be robust).
    assert!(
        absorbed >= global / 2,
        "bystander absorbed {absorbed} of {global} global delay cycles"
    );
}

/// The forgiveness rule must not leak: delay is forgiven on resume from a
/// *park* (Coz's blocked-thread rule), but an actor that merely yields on
/// timeslice expiry stays runnable and must PAY — its work rate has to drop
/// while an experiment targets someone else. (Regression: forgiving on every
/// resume made all bystanders delay-immune, so experiments moved nothing.)
#[test]
fn runnable_bystander_pays_delay() {
    let _s = serial();
    let baseline = Arc::new(AtomicU64::new(0));
    let during = Arc::new(AtomicU64::new(0));
    let b_out = baseline.clone();
    let d_out = during.clone();

    smarm::init(smarm::Config::exact(1)).run(move || {
        let stop = Arc::new(AtomicBool::new(false));

        // Target: in-site spinner; grows the ledger while the experiment runs.
        let stop_t = stop.clone();
        let target = smarm::spawn(move || {
            let _g = smarm::causal_site!("pay-site");
            while !stop_t.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });

        // Bystander: pure check!-loop — yields on slice expiry, never parks.
        let stop_b = stop.clone();
        let iters = Arc::new(AtomicU64::new(0));
        let iters2 = iters.clone();
        let absorbed = Arc::new(AtomicU64::new(0));
        let absorbed2 = absorbed.clone();
        let bystander = smarm::spawn(move || {
            while !stop_b.load(Ordering::Relaxed) {
                iters2.fetch_add(1, Ordering::Relaxed);
                smarm::check!();
                absorbed2.store(
                    smarm::causal::my_absorbed_delay_cycles(),
                    Ordering::Relaxed,
                );
            }
        });

        let window = |out: &AtomicU64| {
            let i0 = iters.load(Ordering::Relaxed);
            let t = std::time::Instant::now();
            smarm::sleep(Duration::from_millis(150));
            let rate =
                (iters.load(Ordering::Relaxed) - i0) as f64 / t.elapsed().as_secs_f64();
            out.store(rate as u64, Ordering::Relaxed);
        };

        window(&b_out);
        let d0 = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("pay-site", 50);
        window(&d_out);
        smarm::causal::end_experiment_for_test();
        let injected = smarm::causal::global_delay_cycles() - d0;

        stop.store(true, Ordering::Relaxed);
        target.join().unwrap();
        bystander.join().unwrap();

        // The never-parking bystander's ledger can only advance by actually
        // spinning (forgiveness requires a real park), so this is no longer
        // vacuous: most of the injected delay must have been paid for real.
        let a = absorbed.load(Ordering::Relaxed);
        assert!(injected > 0, "experiment injected nothing");
        assert!(
            a >= injected / 2,
            "bystander spun off {a} of {injected} injected cycles"
        );
    });

    let b = baseline.load(Ordering::Relaxed) as f64;
    let d = during.load(Ordering::Relaxed) as f64;
    // Time-shared single scheduler: target is on-CPU ~50% of wall, so a 50%
    // virtual speedup injects ~25% of wall — expected slowdown ×0.8 (measured
    // exactly that live); on real parallelism it's ×0.67. Either way <0.9.
    assert!(b > 0.0, "bystander never ran");
    assert!(
        d < b * 0.9,
        "runnable bystander did not pay: baseline {b:.0}/s, during experiment {d:.0}/s"
    );
}

/// A zero-percent experiment must inject nothing: the null experiment is the
/// baseline Coz relies on, and it doubles as the overhead sanity check.
#[test]
fn zero_speedup_injects_nothing() {
    let _s = serial();
    smarm::init(smarm::Config::exact(2)).run(|| {
        let stop = Arc::new(AtomicBool::new(false));
        let stop2 = stop.clone();
        let worker = smarm::spawn(move || {
            let _g = smarm::causal_site!("zero-site");
            while !stop2.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });

        let before = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("zero-site", 0);
        smarm::sleep(Duration::from_millis(50));
        smarm::causal::end_experiment_for_test();
        let after = smarm::causal::global_delay_cycles();

        stop.store(true, Ordering::Relaxed);
        worker.join().unwrap();
        assert_eq!(after, before, "0% speedup must not grow the delay ledger");
    });
}

/// `impact_pct` computes the normalized throughput impact of a (site,
/// speedup) cell against that site's own 0% baseline: injected virtual delay
/// is removed from the divisor (Coz's normalization), so a bottleneck site
/// shows positive impact and a fully-overlapped one shows ~zero.
#[test]
fn impact_from_synthetic_results() {
    use smarm::causal::{impact_pct, tsc_hz, ExperimentResult};
    let hz = tsc_hz();
    let secs = |s: f64| Duration::from_secs_f64(s);
    let cycles = |s: f64| (s * hz) as u64;
    let results = vec![
        // Baseline: 1000 units in 1s, nothing injected.
        ExperimentResult {
            site: "bottleneck".into(),
            speedup_pct: 0,
            duration: secs(1.0),
            deltas: vec![("units".into(), 1000)],
            injected_cycles: 0,
            ..Default::default()
        },
        // Bottleneck at 50%: raw count unchanged, 0.5s injected
        // -> rate 1000/0.5 = 2x baseline -> +100%.
        ExperimentResult {
            site: "bottleneck".into(),
            speedup_pct: 50,
            duration: secs(1.0),
            deltas: vec![("units".into(), 1000)],
            injected_cycles: cycles(0.5),
            ..Default::default()
        },
        ExperimentResult {
            site: "overlapped".into(),
            speedup_pct: 0,
            duration: secs(1.0),
            deltas: vec![("units".into(), 1000)],
            injected_cycles: 0,
            ..Default::default()
        },
        // Overlapped at 50%: everyone else (incl. the real bottleneck)
        // slowed, count drops with the divisor -> ~0% impact.
        ExperimentResult {
            site: "overlapped".into(),
            speedup_pct: 50,
            duration: secs(1.0),
            deltas: vec![("units".into(), 500)],
            injected_cycles: cycles(0.5),
            ..Default::default()
        },
    ];
    let bn = impact_pct(&results, "bottleneck", 50, "units").unwrap();
    assert!((bn - 100.0).abs() < 5.0, "bottleneck impact: {bn}");
    let ov = impact_pct(&results, "overlapped", 50, "units").unwrap();
    assert!(ov.abs() < 5.0, "overlapped impact: {ov}");
    // Missing cells yield None, not garbage.
    assert!(impact_pct(&results, "nosuch", 50, "units").is_none());
    assert!(impact_pct(&results, "bottleneck", 50, "nosuch").is_none());
}

/// End-to-end controller pass over a toy workload: the report must contain a
/// human summary naming every (site × speedup) cell and a Coz-compatible
/// section with experiment / throughput-point lines.
#[test]
fn controller_produces_report() {
    let _s = serial();
    smarm::init(smarm::Config::exact(2)).run(|| {
        let stop = Arc::new(AtomicBool::new(false));
        let stop2 = stop.clone();
        let worker = smarm::spawn(move || {
            while !stop2.load(Ordering::Relaxed) {
                {
                    let _g = smarm::causal_site!("stage-x");
                    for _ in 0..50 {
                        smarm::check!();
                    }
                }
                smarm::progress!("items");
            }
        });

        let results = smarm::causal::run_experiments(&smarm::causal::ExperimentPlan {
            speedups_pct: vec![0, 50],
            experiment: Duration::from_millis(60),
            cooldown: Duration::from_millis(20),
        });
        stop.store(true, Ordering::Relaxed);
        worker.join().unwrap();

        assert!(!results.is_empty());
        let summary = smarm::causal::render_summary(&results);
        assert!(summary.contains("stage-x"), "summary: {summary}");
        assert!(summary.contains("items"), "summary: {summary}");
        assert!(
            summary.contains("impacts are lower bounds"),
            "fidelity footer missing: {summary}"
        );

        let coz = smarm::causal::render_coz(&results);
        assert!(coz.contains("experiment\tselected=stage-x"), "coz: {coz}");
        assert!(coz.contains("throughput-point\tname=items"), "coz: {coz}");
    });
}

/// A target-site entry that never hits a cold causal check must still be
/// attributed: the guard boundaries flush the pending interval on exit and
/// re-arm the clock on entry. Without the exit flush, in-site time between
/// the last check and guard drop is silently discarded — measured live at
/// ~6-7% of all target time (eff 0.93), under-reporting every impact
/// (+83.5% where theory says +100%).
#[test]
fn site_boundaries_flush_tail() {
    let _s = serial();
    fn tsc() -> u64 {
        // SAFETY: rdtsc has no preconditions on x86_64.
        unsafe { core::arch::x86_64::_rdtsc() }
    }

    let injected = Arc::new(AtomicU64::new(0));
    let spent = Arc::new(AtomicU64::new(0));
    let inj_out = injected.clone();
    let spent_out = spent.clone();

    smarm::init(smarm::Config::exact(1)).run(move || {
        // Experiment is live before the actor ever enters the site.
        smarm::causal::begin_experiment_for_test("tail-site", 50);
        let base = smarm::causal::global_delay_cycles();

        let a = smarm::spawn(move || {
            let t0 = tsc();
            {
                let _g = smarm::causal_site!("tail-site");
                // Burn ~20M cycles with no allocations and no `check!()`:
                // the amortised cold check can never fire, so only the
                // guard-boundary flush can attribute this time.
                let start = tsc();
                while tsc().wrapping_sub(start) < 20_000_000 {
                    core::hint::spin_loop();
                }
            }
            spent_out.store(tsc().wrapping_sub(t0), Ordering::Relaxed);
        });
        a.join().unwrap();

        smarm::causal::end_experiment_for_test();
        inj_out.store(
            smarm::causal::global_delay_cycles() - base,
            Ordering::Relaxed,
        );
    });

    let injected = injected.load(Ordering::Relaxed);
    let spent = spent.load(Ordering::Relaxed);
    // At 50% the flush should attribute ≈ half the in-guard time; allow a
    // generous band for guard overhead and clock skew.
    assert!(
        injected >= spent * 40 / 100 && injected <= spent * 60 / 100,
        "boundary flush attributed {injected} of {spent} spent cycles (want ~50%)"
    );
}

// ---------------------------------------------------------------------------
// RFC 007 timer virtual time: injected delay dilates virtual time for
// everyone, so a pending timer's *effective* deadline is its raw deadline
// plus whatever delay was injected while it was armed. Without this, a
// sleep or receive-timeout fires "early" in virtual terms and timeout-driven
// behaviour speeds up relative to the dilated world.
//
// These tests (and every ledger-delta test above) share the process-global
// delay ledger, so they serialize on `serial()`.
// ---------------------------------------------------------------------------

fn serial() -> std::sync::MutexGuard<'static, ()> {
    static M: std::sync::Mutex<()> = std::sync::Mutex::new(());
    M.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// A pending timer must not fire at its raw deadline if delay was injected
/// after it was armed; it fires once wall time covers raw + injected.
#[test]
fn timer_deadline_shifts_with_injected_delay() {
    let _s = serial();
    use smarm::pid::Pid;
    use smarm::timer::Timers;
    use std::time::Instant;

    let hz = smarm::causal::tsc_hz();
    let mut t = Timers::new();
    let now = Instant::now();
    t.insert_sleep(now + Duration::from_millis(50), Pid::new(0, 0), 1);

    // 100ms of debt lands while the timer is pending.
    smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64);

    // Raw deadline passed, effective deadline not: nothing fires, entry kept.
    assert!(t.pop_due(now + Duration::from_millis(60)).is_empty());
    assert!(!t.is_empty(), "shifted entry must be re-queued, not dropped");

    // Past raw + injected (with margin) it must fire. Chase in case a
    // parallel test injected more debt meanwhile.
    let mut fire_at = now + Duration::from_millis(170);
    let mut fired = false;
    for _ in 0..40 {
        if !t.pop_due(fire_at).is_empty() {
            fired = true;
            break;
        }
        fire_at += Duration::from_millis(50);
    }
    assert!(fired, "timer never fired after covering injected delay");
}

/// With no debt accrued since insertion, behaviour is byte-identical to
/// before: due entries fire at their raw deadline.
#[test]
fn timer_zero_debt_pops_at_raw_deadline() {
    let _s = serial();
    use smarm::pid::Pid;
    use smarm::timer::Timers;
    use std::time::Instant;

    let mut t = Timers::new();
    let now = Instant::now();
    t.insert_sleep(now - Duration::from_millis(1), Pid::new(0, 0), 1);
    assert_eq!(t.pop_due(now).len(), 1);
    assert!(t.is_empty());
}

/// A wall-anchored timer (RFC 007 controller fix) fires at its raw deadline
/// regardless of injected delay, while a virtual sibling in the same heap
/// still shifts. This is what keeps the causal controller's experiment
/// windows a fixed wall length even under heavy injection.
#[test]
fn wall_timer_ignores_injected_delay() {
    let _s = serial();
    use smarm::pid::Pid;
    use smarm::timer::Timers;
    use std::time::Instant;

    let hz = smarm::causal::tsc_hz();
    let mut t = Timers::new();
    let now = Instant::now();
    t.insert_sleep_wall(now + Duration::from_millis(50), Pid::new(0, 0), 1);
    t.insert_sleep(now + Duration::from_millis(50), Pid::new(1, 0), 1);

    // 100ms of debt lands while both are pending.
    smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64);

    // Just past the raw deadline: the wall entry fires, the virtual one is
    // re-queued at its shifted deadline.
    let due = t.pop_due(now + Duration::from_millis(60));
    assert_eq!(due.len(), 1, "exactly the wall entry must fire at raw deadline");
    assert_eq!(due[0].pid, Pid::new(0, 0));
    assert!(!t.is_empty(), "virtual sibling must remain queued, shifted");
}

/// Re-queueing a shifted entry must preserve its identity: a `send_after`
/// cancelled *after* being shifted past its raw deadline must still cancel
/// (return true) and must never deliver.
#[test]
fn send_after_cancel_survives_reheap() {
    let _s = serial();
    use smarm::pid::Pid;
    use smarm::timer::Timers;
    use std::sync::atomic::AtomicBool;
    use std::time::Instant;

    let hz = smarm::causal::tsc_hz();
    let mut t = Timers::new();
    let now = Instant::now();
    let delivered = Arc::new(AtomicBool::new(false));
    let d2 = delivered.clone();
    let id = t.insert_send(
        now + Duration::from_millis(20),
        Pid::new(0, 0),
        Box::new(move || d2.store(true, Ordering::Relaxed)),
    );

    smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64);

    // Raw deadline passed: shifted, not fired.
    assert!(t.pop_due(now + Duration::from_millis(30)).is_empty());
    // Cancellation still works on the re-queued entry.
    assert!(t.cancel(id), "cancel lost track of a re-queued send_after");
    // Even far past the effective deadline nothing is delivered.
    for e in t.pop_due(now + Duration::from_secs(3600)) {
        if let smarm::timer::Reason::Send { fire } = e.reason {
            fire();
        }
    }
    assert!(
        !delivered.load(Ordering::Relaxed),
        "cancelled send_after delivered after re-heap"
    );
}

/// End-to-end through the runtime: an actor mid-`sleep` pays delay injected
/// while it is parked by sleeping *longer* — which is exactly what makes the
/// park-gated resume-credit fast-forward correct rather than forgiving.
#[test]
fn sleeping_actor_pays_injected_delay() {
    let _s = serial();
    use std::time::Instant;

    let slept = Arc::new(AtomicU64::new(0));
    let s_out = slept.clone();

    smarm::init(smarm::Config::exact(1)).run(move || {
        let hz = smarm::causal::tsc_hz();
        let s2 = s_out.clone();
        let sleeper = smarm::spawn(move || {
            let t0 = Instant::now();
            smarm::sleep(Duration::from_millis(50));
            s2.store(t0.elapsed().as_millis() as u64, Ordering::Relaxed);
        });

        // Let the sleeper arm its timer, then inject 150ms of debt from a
        // plain OS thread — no slot, so nothing spin-absorbs it here and the
        // only path by which it can slow the sleeper is the timer shift.
        smarm::sleep(Duration::from_millis(10));
        std::thread::spawn(move || {
            smarm::causal::inject_delay_cycles_for_test((hz * 0.150) as u64);
        })
        .join()
        .unwrap();

        sleeper.join().unwrap();
    });

    let ms = slept.load(Ordering::Relaxed);
    // Raw sleep 50ms + ~150ms debt injected mid-sleep => ~200ms effective.
    // Loose lower bound; the pre-fix behaviour is ~50ms.
    assert!(
        ms >= 150,
        "sleeper paid nothing: slept {ms}ms, expected >= 150ms (raw 50 + injected 150)"
    );
}

/// RFC 007 user-facing opt-out: a wall-anchored `send_after` entry fires at
/// its raw deadline while a virtual sibling armed at the same instant is
/// shifted by injected delay — the `Send`-reason mirror of
/// `wall_timer_ignores_injected_delay`.
#[test]
fn wall_send_after_ignores_injected_delay() {
    let _s = serial();
    use smarm::pid::Pid;
    use smarm::timer::Timers;
    use std::time::Instant;

    let hz = smarm::causal::tsc_hz();
    let mut t = Timers::new();
    let now = Instant::now();
    let wall_fired = Arc::new(AtomicBool::new(false));
    let virt_fired = Arc::new(AtomicBool::new(false));
    let (w2, v2) = (wall_fired.clone(), virt_fired.clone());
    let _wid = t.insert_send_wall(
        now + Duration::from_millis(50),
        Pid::new(0, 0),
        Box::new(move || w2.store(true, Ordering::Relaxed)),
    );
    let _vid = t.insert_send(
        now + Duration::from_millis(50),
        Pid::new(1, 0),
        Box::new(move || v2.store(true, Ordering::Relaxed)),
    );

    // 100ms of debt lands while both are pending.
    smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64);

    // Just past the raw deadline: only the wall send pops; run its thunk.
    let due = t.pop_due(now + Duration::from_millis(60));
    assert_eq!(due.len(), 1, "exactly the wall send must fire at raw deadline");
    for e in due {
        if let smarm::timer::Reason::Send { fire } = e.reason {
            fire();
        }
    }
    assert!(wall_fired.load(Ordering::Relaxed), "wall send must deliver");
    assert!(!virt_fired.load(Ordering::Relaxed), "virtual send must not");
    assert!(!t.is_empty(), "virtual sibling must remain queued, shifted");
}

/// `cancel` is anchor-agnostic: a wall-anchored `send_after` cancels exactly
/// like a virtual one and never delivers, even with delay debt outstanding.
#[test]
fn wall_send_after_cancels() {
    let _s = serial();
    use smarm::pid::Pid;
    use smarm::timer::Timers;
    use std::time::Instant;

    let hz = smarm::causal::tsc_hz();
    let mut t = Timers::new();
    let now = Instant::now();
    let delivered = Arc::new(AtomicBool::new(false));
    let d2 = delivered.clone();
    let id = t.insert_send_wall(
        now + Duration::from_millis(20),
        Pid::new(0, 0),
        Box::new(move || d2.store(true, Ordering::Relaxed)),
    );

    smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64);

    assert!(t.cancel(id), "cancel must find the wall-anchored send");
    for e in t.pop_due(now + Duration::from_secs(3600)) {
        if let smarm::timer::Reason::Send { fire } = e.reason {
            fire();
        }
    }
    assert!(
        !delivered.load(Ordering::Relaxed),
        "cancelled wall send_after delivered"
    );
}

// ---------------------------------------------------------------------------
// Ledger audit (RFC 007 deficit hunt) — measure-only counters that decompose
// where injected delay is born, paid, forgiven, and where would-be
// attribution is silently lost. Behaviour under test is accounting only:
// none of these tests assert a change in injection or absorption itself.
// ---------------------------------------------------------------------------

/// A never-parking bystander's spin absorption lands in the audit's
/// `spin_absorbed_cycles` — the "paid for real" bucket.
#[test]
fn audit_counts_spin_absorption() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        let stop = Arc::new(AtomicBool::new(false));
        let stop_t = stop.clone();
        let target = smarm::spawn(move || {
            let _g = smarm::causal_site!("audit-spin-site");
            while !stop_t.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });
        let stop_b = stop.clone();
        let bystander = smarm::spawn(move || {
            while !stop_b.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });
        let c0 = smarm::causal::ledger_counters();
        let g0 = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("audit-spin-site", 50);
        smarm::sleep(Duration::from_millis(150));
        smarm::causal::end_experiment_for_test();
        let injected = smarm::causal::global_delay_cycles() - g0;
        let c1 = smarm::causal::ledger_counters();
        stop.store(true, Ordering::Relaxed);
        target.join().unwrap();
        bystander.join().unwrap();

        let spun = c1.spin_absorbed_cycles - c0.spin_absorbed_cycles;
        assert!(injected > 0, "experiment injected nothing");
        assert!(
            spun >= injected / 2,
            "audit missed spin absorption: {spun} of {injected} injected"
        );
        // Payers are bounded by the debt that exists; 2x headroom covers the
        // main actor's incidental checkpoint payments.
        assert!(
            spun <= injected.saturating_mul(2),
            "audit over-counted spin: {spun} of {injected} injected"
        );
    });
}

/// A wake after a real park waives the sleeper's accrued debt; the waiver is
/// tallied in `park_forgiven_cycles` (Coz's blocked-thread rule, now visible).
#[test]
fn audit_counts_park_forgiveness() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        let stop = Arc::new(AtomicBool::new(false));
        let stop_t = stop.clone();
        let target = smarm::spawn(move || {
            let _g = smarm::causal_site!("audit-forgive-site");
            while !stop_t.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });
        // Parks across the whole injection window; owes everything at wake.
        let sleeper = smarm::spawn(|| smarm::sleep(Duration::from_millis(120)));

        let c0 = smarm::causal::ledger_counters();
        let g0 = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("audit-forgive-site", 50);
        smarm::sleep(Duration::from_millis(60));
        smarm::causal::end_experiment_for_test();
        let injected = smarm::causal::global_delay_cycles() - g0;
        sleeper.join().unwrap();
        let c1 = smarm::causal::ledger_counters();
        stop.store(true, Ordering::Relaxed);
        target.join().unwrap();

        let forgiven = c1.park_forgiven_cycles - c0.park_forgiven_cycles;
        assert!(injected > 0, "experiment injected nothing");
        assert!(
            forgiven >= injected / 2,
            "audit missed park forgiveness: {forgiven} of {injected} injected"
        );
    });
}

/// Parking while *inside* the target site drops the sample tail
/// `[last sample -> park]` — nothing flushes it, and `on_resume` re-arms the
/// clock. The audit counts that loss without changing it.
#[test]
fn audit_counts_park_drop_inside_target_site() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        smarm::causal::begin_experiment_for_test("audit-park-drop-site", 50);
        let c0 = smarm::causal::ledger_counters();
        let h = smarm::spawn(|| {
            let _g = smarm::causal_site!("audit-park-drop-site");
            smarm::check!(); // arm an in-site sample interval
            smarm::sleep(Duration::from_millis(20)); // real park, in-guard
        });
        h.join().unwrap();
        let c1 = smarm::causal::ledger_counters();
        smarm::causal::end_experiment_for_test();

        assert!(
            c1.drop_park_n > c0.drop_park_n,
            "in-site park recorded no drop event"
        );
        assert!(
            c1.drop_park_cycles > c0.drop_park_cycles,
            "in-site park drop carried no cycles"
        );
    });
}

/// An explicit yield inside the target site loses its tail the same way; it
/// lands in the separate yield bucket so park- and yield-losses stay apart.
#[test]
fn audit_counts_yield_drop_inside_target_site() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        smarm::causal::begin_experiment_for_test("audit-yield-drop-site", 50);
        let c0 = smarm::causal::ledger_counters();
        let h = smarm::spawn(|| {
            let _g = smarm::causal_site!("audit-yield-drop-site");
            smarm::check!(); // arm an in-site sample interval
            smarm::yield_now();
            smarm::check!();
        });
        h.join().unwrap();
        let c1 = smarm::causal::ledger_counters();
        smarm::causal::end_experiment_for_test();

        assert!(
            c1.drop_yield_n > c0.drop_yield_n,
            "in-site yield recorded no drop event"
        );
        assert!(
            c1.drop_yield_cycles > c0.drop_yield_cycles,
            "in-site yield drop carried no cycles"
        );
    });
}

/// A single in-site interval past MAX_SAMPLE_CYCLES is discarded by the
/// TSC-weirdness clamp; the audit counts the discard instead of losing it
/// silently.
#[test]
fn audit_counts_overmax_discard() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        smarm::causal::begin_experiment_for_test("audit-overmax-site", 50);
        let c0 = smarm::causal::ledger_counters();
        let h = smarm::spawn(|| {
            let _g = smarm::causal_site!("audit-overmax-site");
            // Stay in-site past MAX_SAMPLE_CYCLES (~33ms at 3GHz) without a
            // single checkpoint: no allocs, no check!. The guard-drop flush
            // then sees one huge interval and must discard it.
            let t0 = std::time::Instant::now();
            while t0.elapsed() < Duration::from_millis(60) {
                std::hint::spin_loop();
            }
        });
        h.join().unwrap();
        let c1 = smarm::causal::ledger_counters();
        smarm::causal::end_experiment_for_test();

        assert!(
            c1.discard_overmax_n > c0.discard_overmax_n,
            "overmax interval recorded no discard event"
        );
        assert!(
            c1.discard_overmax_cycles > c0.discard_overmax_cycles,
            "overmax discard carried no cycles"
        );
    });
}

/// Debt left outstanding when a window ends is paid during the NEXT active
/// experiment — including a 0% baseline cell, whose bystander-chase branch is
/// live at pct=0. Synthesized deterministically: grow the ledger with no
/// experiment active (nobody pays), then watch a 0% window absorb it.
#[test]
fn audit_zero_pct_window_absorbs_leftover_debt() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        let hz = smarm::causal::tsc_hz();
        let stop = Arc::new(AtomicBool::new(false));
        let stop_b = stop.clone();
        let bystander = smarm::spawn(move || {
            while !stop_b.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });

        // No experiment active: checks early-return, debt sits unpaid.
        let debt = (hz * 0.020) as u64; // ~20ms
        smarm::causal::inject_delay_cycles_for_test(debt);
        smarm::sleep(Duration::from_millis(30));

        let c0 = smarm::causal::ledger_counters();
        let g0 = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("audit-zero-pct-site", 0);
        smarm::sleep(Duration::from_millis(60));
        smarm::causal::end_experiment_for_test();
        let c1 = smarm::causal::ledger_counters();
        stop.store(true, Ordering::Relaxed);
        bystander.join().unwrap();

        assert_eq!(
            smarm::causal::global_delay_cycles(),
            g0,
            "a 0% window must inject nothing"
        );
        let spun = c1.spin_absorbed_cycles - c0.spin_absorbed_cycles;
        assert!(
            spun >= debt / 2,
            "leftover debt not absorbed in the 0% window: {spun} of {debt}"
        );
    });
}

/// The audit render lists one line per cell with the ledger columns.
#[test]
fn ledger_audit_renders_per_cell() {
    use smarm::causal::{render_ledger_audit, ExperimentResult};
    let results = vec![ExperimentResult {
        site: "audit-render-site".into(),
        speedup_pct: 50,
        duration: Duration::from_secs(1),
        deltas: vec![("units".into(), 10)],
        injected_cycles: 1_000_000,
        spin_absorbed_cycles: 900_000,
        park_forgiven_cycles: 50_000,
        drop_park_cycles: 30_000,
        drop_park_n: 2,
        drop_yield_cycles: 10_000,
        drop_yield_n: 1,
        discard_overmax_cycles: 5_000,
        discard_overmax_n: 1,
        discard_unarmed_n: 0,
        offcpu_in_site_cycles: 20_000,
        offcpu_in_site_n: 3,
    }];
    let s = render_ledger_audit(&results);
    assert!(s.contains("ledger audit"), "missing header: {s}");
    assert!(s.contains("audit-render-site"), "missing site line: {s}");
    assert!(s.contains("absorbed"), "missing absorbed column: {s}");
    assert!(s.contains("forgiven"), "missing forgiven column: {s}");
    assert!(s.contains("offcpu"), "missing offcpu column: {s}");
}

/// A runnable (yield) deschedule inside the target site opens an off-CPU
/// gap; the resume closing it lands in the same experiment window, so the
/// audit counts it. RFC 007: this is the located @50 deficit — runnable
/// queue-wait is real wall time in-site that on-CPU attribution correctly
/// skips; the bucket exists so the books close instead of leaving a
/// silent ~7% residual.
#[test]
fn audit_counts_offcpu_runnable_gap_in_target_site() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        smarm::causal::begin_experiment_for_test("audit-offcpu-site", 50);
        let c0 = smarm::causal::ledger_counters();
        let h = smarm::spawn(|| {
            let _g = smarm::causal_site!("audit-offcpu-site");
            smarm::check!(); // arm
            smarm::yield_now(); // runnable gap: deschedule -> resume
            smarm::check!();
        });
        h.join().unwrap();
        let c1 = smarm::causal::ledger_counters();
        smarm::causal::end_experiment_for_test();

        assert!(
            c1.offcpu_in_site_n > c0.offcpu_in_site_n,
            "in-site runnable gap recorded no offcpu event"
        );
        assert!(
            c1.offcpu_in_site_cycles > c0.offcpu_in_site_cycles,
            "in-site runnable gap carried no cycles"
        );
    });
}

/// An off-CPU gap that straddles `end()` — even into a later window with
/// an identical site+pct word (live in the attrib probe's 50,50 schedule)
/// — is dropped, not counted: the stash carries the experiment *epoch*,
/// so a straddling resume can never leak a cooldown across windows.
#[test]
fn audit_offcpu_gap_across_windows_not_counted() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        smarm::causal::begin_experiment_for_test("audit-offcpu-straddle", 50);
        let c0 = smarm::causal::ledger_counters();
        let h = smarm::spawn(|| {
            let _g = smarm::causal_site!("audit-offcpu-straddle");
            smarm::check!();
            smarm::yield_now(); // stash carries window A's epoch
            smarm::check!();
        });
        // FIFO on exact(1): let h run to its yield, then close window A and
        // open a same-word window B before h resumes.
        smarm::yield_now();
        smarm::causal::end_experiment_for_test();
        smarm::causal::begin_experiment_for_test("audit-offcpu-straddle", 50);
        h.join().unwrap();
        let c1 = smarm::causal::ledger_counters();
        smarm::causal::end_experiment_for_test();

        assert_eq!(
            c1.offcpu_in_site_n, c0.offcpu_in_site_n,
            "gap straddling windows must not be counted"
        );
    });
}

/// Slot reuse under actor churn must not book the process's accumulated
/// delay history as park forgiveness. A fresh incarnation is born current
/// (`causal_delay = GLOBAL_DELAY` at install), so churning short-lived
/// actors while no experiment is live forgives exactly nothing: the global
/// ledger is frozen (checks gate on the experiment word) and a newborn
/// starts even with it. Found live (urus close-mode load, ~95k conn
/// spawns/s): delay=0 + parked=true at install forgave the whole backlog
/// once per spawn — millions of phantom ms per 700ms window, even in 0%
/// cells, while ka-mode books stayed plausible.
#[test]
fn actor_churn_between_experiments_forgives_nothing() {
    let _s = serial();
    smarm::init(smarm::Config::exact(1)).run(|| {
        // Seed a nonzero global backlog with a real experiment.
        let stop = Arc::new(AtomicBool::new(false));
        let stop_t = stop.clone();
        let target = smarm::spawn(move || {
            let _g = smarm::causal_site!("churn-seed-site");
            while !stop_t.load(Ordering::Relaxed) {
                smarm::check!();
            }
        });
        let g0 = smarm::causal::global_delay_cycles();
        smarm::causal::begin_experiment_for_test("churn-seed-site", 50);
        smarm::sleep(Duration::from_millis(60));
        smarm::causal::end_experiment_for_test();
        let backlog = smarm::causal::global_delay_cycles() - g0;
        assert!(backlog > 0, "seed experiment injected nothing");
        stop.store(true, Ordering::Relaxed);
        target.join().unwrap();

        // Every live actor's wake (the sleep, the joins) is behind us and
        // the ledger is frozen; snapshot, then churn reused slots with
        // trivial actors. The first resume alone is the trigger — the actor
        // body never parks.
        let c0 = smarm::causal::ledger_counters();
        for _ in 0..200 {
            smarm::spawn(|| {}).join().unwrap();
        }
        let c1 = smarm::causal::ledger_counters();

        let forgiven = c1.park_forgiven_cycles - c0.park_forgiven_cycles;
        assert_eq!(
            forgiven, 0,
            "spawn churn booked {forgiven} phantom forgiveness cycles \
             against a frozen {backlog}-cycle backlog"
        );
    });
}