beamr 0.19.1

use std::collections::HashMap;
use std::sync::Arc;

use proptest::prelude::*;

use super::*;
use crate::{
    atom::Atom,
    constant_pool::materialise_literals,
    io::resource::{FD_RESOURCE_WORDS, FdInner, FdResource, write_fd_resource},
    loader::{Instruction, Literal},
    module::{Module, ModuleOrigin},
    process::ExitReason,
    term::{
        boxed::{BoxedTag, Closure, Cons, ProcBin, Tuple, write_closure, write_cons, write_tuple},
        shared_binary::{SharedBinary, write_proc_bin},
        sub_binary::{SUB_BINARY_WORDS, write_sub_binary},
    },
};

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum Snapshot {
    Int(i64),
    Atom(Atom),
    Nil,
    Tuple(Vec<Snapshot>),
    List(Vec<Snapshot>),
    Other,
}

pub(crate) fn module_pin(name: Atom) -> Arc<Module> {
    Arc::new(Module {
        name,
        generation: 0,
        origin: ModuleOrigin::Preloaded,
        exports: HashMap::new(),
        label_index: HashMap::from([(1, 0)]),
        code: vec![Instruction::Label { label: 1 }],
        literals: Vec::new(),
        constant_pool: Default::default(),
        resolved_imports: Vec::new(),
        lambdas: Vec::new(),
        string_table: Vec::new(),
        function_table: Vec::new(),
        line_table: Vec::new(),
        line_info: Vec::new(),
    })
}

pub(crate) fn alloc_tuple(process: &mut Process, elements: &[Term]) -> Term {
    let ptr = alloc(process, 1 + elements.len()).expect("tuple allocation via GC should fit");
    // SAFETY: GC allocation returned `1 + elements.len()` writable words.
    let words = unsafe { std::slice::from_raw_parts_mut(ptr, 1 + elements.len()) };
    write_tuple(words, elements).expect("tuple writer should fit allocated words")
}

pub(crate) fn alloc_cons(process: &mut Process, head: Term, tail: Term) -> Term {
    let ptr = alloc(process, 2).expect("cons allocation via GC should fit");
    // SAFETY: GC allocation returned two writable words.
    let words = unsafe { std::slice::from_raw_parts_mut(ptr, 2) };
    write_cons(words, head, tail).expect("cons writer should fit allocated words")
}

pub(crate) fn alloc_closure(
    process: &mut Process,
    generation: u64,
    unique_id: u64,
    free_vars: &[Term],
) -> Term {
    let words_count = 7 + free_vars.len();
    let ptr = alloc(process, words_count).expect("closure allocation via GC should fit");
    // SAFETY: GC allocation returned `words_count` writable words.
    let words = unsafe { std::slice::from_raw_parts_mut(ptr, words_count) };
    write_closure(words, Atom::OK, 0, 0, generation, unique_id, free_vars)
        .expect("closure writer should fit allocated words")
}

pub(crate) fn alloc_proc_bin(process: &mut Process, shared: &SharedBinary) -> Term {
    let ptr = alloc(process, 3).expect("proc bin allocation via GC should fit");
    // SAFETY: GC allocation returned three writable words.
    let words = unsafe { std::slice::from_raw_parts_mut(ptr, 3) };
    let term = write_proc_bin(words, shared).expect("proc bin writer should fit allocated words");
    process
        .heap_mut()
        .mark_last_young_allocation_maybe_refcounted();
    process.increase_virtual_binary_heap(shared.len());
    term
}

pub(crate) fn alloc_fd_resource(process: &mut Process, inner: Arc<FdInner>) -> Term {
    let ptr = alloc(process, FD_RESOURCE_WORDS).expect("fd resource allocation via GC should fit");
    // SAFETY: GC allocation returned the fixed FdResource word count.
    let words = unsafe { std::slice::from_raw_parts_mut(ptr, FD_RESOURCE_WORDS) };
    let term =
        write_fd_resource(words, inner).expect("fd resource writer should fit allocated words");
    process
        .heap_mut()
        .mark_last_young_allocation_maybe_refcounted();
    term
}

pub(crate) fn snapshot(term: Term) -> Snapshot {
    if let Some(value) = term.as_small_int() {
        return Snapshot::Int(value);
    }
    if let Some(atom) = term.as_atom() {
        return Snapshot::Atom(atom);
    }
    if term.is_nil() {
        return Snapshot::Nil;
    }
    if let Some(tuple) = Tuple::new(term) {
        return Snapshot::Tuple(
            (0..tuple.arity())
                .filter_map(|i| tuple.get(i))
                .map(snapshot)
                .collect(),
        );
    }
    if term.is_list() {
        let mut values = Vec::new();
        let mut tail = term;
        while tail.is_list() {
            let Some(cons) = Cons::new(tail) else {
                return Snapshot::Other;
            };
            values.push(snapshot(cons.head()));
            tail = cons.tail();
        }
        if tail.is_nil() {
            return Snapshot::List(values);
        }
    }
    Snapshot::Other
}

pub(crate) fn assert_no_reachable_pointer_into_young(process: &Process) {
    for term in process.x_regs() {
        assert_no_term_pointer_into_young(process, *term);
    }
    for term in process.mailbox().scan_iter() {
        assert_no_term_pointer_into_young(process, *term);
    }
}

pub(crate) fn assert_no_term_pointer_into_young(process: &Process, term: Term) {
    let mut stack = vec![term];
    while let Some(current) = stack.pop() {
        if let Some(ptr) = current.heap_ptr() {
            assert!(!process.heap().young_contains(ptr));
            if current.is_list() {
                let cons = Cons::new(current).expect("valid cons");
                stack.push(cons.head());
                stack.push(cons.tail());
            } else if let Some(tuple) = Tuple::new(current) {
                for index in 0..tuple.arity() {
                    if let Some(element) = tuple.get(index) {
                        stack.push(element);
                    }
                }
            } else if let Some(closure) = Closure::new(current) {
                for index in 0..closure.num_free() {
                    if let Some(free_var) = closure.free_var(index) {
                        stack.push(free_var);
                    }
                }
            } else if let Some(sub_binary) = crate::term::boxed::SubBinary::new(current) {
                stack.push(sub_binary.parent());
            }
        }
    }
}

#[test]
fn constant_pool_terms_remain_external_roots_across_gc() {
    let literals = vec![Literal::Tuple(vec![
        Literal::Integer(1),
        Literal::Integer(2),
    ])];
    let pool = materialise_literals(&literals, None).expect("constant pool");
    let literal = pool.get(0).expect("literal root");
    let literal_ptr = literal.heap_ptr();
    let mut process = Process::new(1, 32);
    process.set_x_reg(0, literal);

    collect_minor(&mut process).expect("minor GC succeeds");
    assert_eq!(process.x_reg(0).raw(), literal.raw());
    assert_eq!(process.x_reg(0).heap_ptr(), literal_ptr);

    collect_major(&mut process).expect("major GC succeeds");
    assert_eq!(process.x_reg(0).raw(), literal.raw());
    assert_eq!(process.x_reg(0).heap_ptr(), literal_ptr);
    assert_eq!(
        snapshot(process.x_reg(0)),
        Snapshot::Tuple(vec![Snapshot::Int(1), Snapshot::Int(2)])
    );
}

#[test]
fn constant_pool_terms_nested_in_process_heap_are_not_copied() {
    let literals = vec![Literal::List(
        vec![Literal::Integer(7)],
        Box::new(Literal::Nil),
    )];
    let pool = materialise_literals(&literals, None).expect("constant pool");
    let literal = pool.get(0).expect("literal root");
    let literal_ptr = literal.heap_ptr();
    let mut process = Process::new(1, 32);
    let tuple = alloc_tuple(&mut process, &[literal]);
    process.set_x_reg(0, tuple);

    collect_minor(&mut process).expect("minor GC succeeds");
    let tuple = Tuple::new(process.x_reg(0)).expect("tuple after minor");
    assert_eq!(tuple.get(0).expect("literal field").raw(), literal.raw());
    assert_eq!(tuple.get(0).expect("literal field").heap_ptr(), literal_ptr);

    collect_major(&mut process).expect("major GC succeeds");
    let tuple = Tuple::new(process.x_reg(0)).expect("tuple after major");
    assert_eq!(tuple.get(0).expect("literal field").raw(), literal.raw());
    assert_eq!(tuple.get(0).expect("literal field").heap_ptr(), literal_ptr);
}

#[test]
fn gc_process_isolation_does_not_touch_other_process() {
    let mut process_a = Process::new(1, 8);
    let mut process_b = Process::new(2, 8);
    let b_term = alloc_tuple(&mut process_b, &[Term::small_int(99)]);
    process_b.set_x_reg(0, b_term);
    let b_young_used = process_b.heap().young_used();
    let b_old_used = process_b.heap().old_used();
    let b_root = process_b.x_reg(0);

    let a_term = alloc_tuple(&mut process_a, &[Term::small_int(1)]);
    process_a.set_x_reg(0, a_term);
    collect_minor(&mut process_a).expect("minor GC succeeds");

    assert_eq!(process_b.heap().young_used(), b_young_used);
    assert_eq!(process_b.heap().old_used(), b_old_used);
    assert_eq!(process_b.x_reg(0).raw(), b_root.raw());
    assert_eq!(
        snapshot(process_b.x_reg(0)),
        Snapshot::Tuple(vec![Snapshot::Int(99)])
    );
}

#[test]
fn gc_triggered_allocation_reclaims_empty_nursery_without_growth() {
    let mut process = Process::new(1, 233);
    let _ptr = process.heap_mut().alloc(233).expect("fill nursery");

    let ptr = alloc(&mut process, 1).expect("GC allocation should collect then allocate");

    assert!(process.heap().young_contains(ptr));
    assert_eq!(process.heap().capacity(), 233);
    assert_eq!(process.heap().young_used(), 1);
}

#[test]
fn ensure_space_grows_with_fibonacci_policy_when_needed() {
    let mut process = Process::new(1, 233);

    ensure_space(&mut process, 300, 0).expect("growth below max succeeds");

    assert_eq!(process.heap().capacity(), 377);
    assert!(process.heap().available() >= 300);
}

#[test]
fn ensure_space_reports_heap_full_when_growth_exceeds_max() {
    let mut process = Process::new(1, 8);
    process.heap_mut().set_max_capacity(8);

    let error = ensure_space(&mut process, 9, 0).expect_err("growth above max fails");

    assert!(matches!(error, GcError::HeapFull(_)));
    assert_eq!(process.heap().capacity(), 8);
}

#[test]
fn mixed_x_y_and_mailbox_roots_survive_minor_gc() {
    let mut process = Process::new(1, 32);
    let x_term = alloc_tuple(&mut process, &[Term::small_int(5)]);
    let y_term = alloc_tuple(&mut process, &[Term::small_int(6)]);
    let mail_term = alloc_tuple(&mut process, &[Term::small_int(7)]);
    process.set_x_reg(5, x_term);
    process
        .stack_mut()
        .push_frame(Atom::OK, 0, module_pin(Atom::OK), 3)
        .expect("frame fits");
    process.stack_mut().set_y_reg(2, y_term).expect("Y2 exists");
    process.mailbox_mut().push_owned_for_test(mail_term);
    let expected = [snapshot(x_term), snapshot(y_term), snapshot(mail_term)];

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(snapshot(process.x_reg(5)), expected[0]);
    assert_eq!(
        snapshot(process.stack().y_reg(2).expect("Y2 exists")),
        expected[1]
    );
    assert_eq!(
        snapshot(process.mailbox().front_for_test().expect("mailbox root")),
        expected[2]
    );
    assert_no_reachable_pointer_into_young(&process);
}

#[test]
fn closure_metadata_and_free_vars_survive_minor_gc() {
    let mut process = Process::new(1, 32);
    let captured = alloc_tuple(&mut process, &[Term::small_int(42)]);
    let closure = alloc_closure(&mut process, 7, 0x1234_5678, &[captured]);
    process.set_x_reg(0, closure);

    collect_minor(&mut process).expect("minor GC succeeds");

    let closure = Closure::new(process.x_reg(0)).expect("closure root");
    assert_eq!(closure.module(), Some(Atom::OK));
    assert_eq!(closure.function_index(), 0);
    assert_eq!(closure.arity(), 0);
    assert_eq!(closure.num_free(), 1);
    assert_eq!(closure.generation(), 7);
    assert_eq!(closure.unique_id(), 0x1234_5678);
    let free_var = closure.free_var(0).expect("captured term");
    assert_eq!(snapshot(free_var), Snapshot::Tuple(vec![Snapshot::Int(42)]));
    assert_no_term_pointer_into_young(&process, process.x_reg(0));
}

#[test]
fn sub_binary_traces_proc_bin_parent_across_minor_gc() {
    let data: Vec<u8> = (0..64).collect();
    let shared = SharedBinary::new(data);
    let mut process = Process::new(1, 32);
    let parent = alloc_proc_bin(&mut process, &shared);
    let ptr = process
        .heap_mut()
        .alloc(SUB_BINARY_WORDS)
        .expect("sub binary allocation fits");
    let sub_binary = write_sub_binary(
        crate::interpreter::opcodes::core::heap_slice(ptr, SUB_BINARY_WORDS),
        parent,
        10,
        12,
    )
    .expect("sub binary writer fits");
    process.set_x_reg(0, sub_binary);
    assert_eq!(shared.ref_count(), 2);

    collect_minor(&mut process).expect("minor GC succeeds");

    let sub_binary = crate::term::boxed::SubBinary::new(process.x_reg(0)).expect("sub binary root");
    assert_eq!(sub_binary.as_bytes(), &shared.as_bytes()[10..22]);
    let parent = sub_binary.parent();
    let parent_ptr = parent.heap_ptr().expect("rewritten parent pointer");
    assert!(!process.heap().young_contains(parent_ptr));
    assert_no_term_pointer_into_young(&process, process.x_reg(0));
}

#[test]
fn proc_bin_survives_minor_gc_as_leaf_with_shared_bytes() {
    let shared = SharedBinary::new(vec![0xCD; 100 * 1024]);
    let mut process = Process::new(1, 32);
    let proc_bin = alloc_proc_bin(&mut process, &shared);
    process.set_x_reg(0, proc_bin);
    assert_eq!(shared.ref_count(), 2);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(shared.ref_count(), 2);
    assert_eq!(process.virtual_binary_heap(), 100 * 1024);
    let proc_bin = ProcBin::new(process.x_reg(0)).expect("proc bin root");
    assert_eq!(proc_bin.len(), 100 * 1024);
    assert_eq!(proc_bin.as_bytes(), shared.as_bytes());
    assert_no_term_pointer_into_young(&process, process.x_reg(0));
}

#[test]
fn unreachable_young_proc_bin_releases_shared_bytes_after_minor_gc() {
    let shared = SharedBinary::new(vec![0xAA; 64 * 1024]);
    let mut process = Process::new(1, 32);
    let _unreachable = alloc_proc_bin(&mut process, &shared);
    assert_eq!(shared.ref_count(), 2);
    assert_eq!(process.virtual_binary_heap(), 64 * 1024);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(shared.ref_count(), 1);
    assert_eq!(process.virtual_binary_heap(), 0);
    assert_eq!(process.heap().young_used(), 0);
}

#[test]
fn shared_proc_bin_bytes_survive_when_parent_reference_remains_after_minor_gc() {
    let shared = SharedBinary::new(vec![0xBB; 32 * 1024]);
    let observer = shared.clone();
    let mut process = Process::new(1, 32);
    let _unreachable = alloc_proc_bin(&mut process, &shared);
    assert_eq!(shared.ref_count(), 3);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(shared.ref_count(), 2);
    assert_eq!(observer.as_bytes(), shared.as_bytes());
    assert_eq!(process.virtual_binary_heap(), 0);
}

#[test]
fn major_gc_releases_unreachable_proc_bins_from_young_and_old_sources() {
    let old_shared = SharedBinary::new(vec![0xCC; 8 * 1024]);
    let young_shared = SharedBinary::new(vec![0xDD; 16 * 1024]);
    let mut process = Process::new(1, 64);
    let old_proc_bin = alloc_proc_bin(&mut process, &old_shared);
    process.set_x_reg(0, old_proc_bin);
    collect_minor(&mut process).expect("promote old proc bin");
    process.set_x_reg(0, Term::NIL);
    let _young_proc_bin = alloc_proc_bin(&mut process, &young_shared);
    assert_eq!(old_shared.ref_count(), 2);
    assert_eq!(young_shared.ref_count(), 2);
    assert_eq!(process.virtual_binary_heap(), 24 * 1024);

    collect_major(&mut process).expect("major GC succeeds");

    assert_eq!(old_shared.ref_count(), 1);
    assert_eq!(young_shared.ref_count(), 1);
    assert_eq!(process.virtual_binary_heap(), 0);
    assert_eq!(process.heap().young_used(), 0);
    assert_eq!(process.heap().old_used(), 0);
}

#[test]
fn surviving_proc_bin_has_stable_ref_count_after_major_gc() {
    let shared = SharedBinary::new(vec![0xEE; 12 * 1024]);
    let mut process = Process::new(1, 32);
    let proc_bin = alloc_proc_bin(&mut process, &shared);
    process.set_x_reg(0, proc_bin);
    assert_eq!(shared.ref_count(), 2);

    collect_major(&mut process).expect("major GC succeeds");

    assert_eq!(shared.ref_count(), 2);
    assert_eq!(process.virtual_binary_heap(), 12 * 1024);
    let proc_bin = ProcBin::new(process.x_reg(0)).expect("proc bin root");
    assert_eq!(proc_bin.as_bytes(), shared.as_bytes());
}

#[test]
fn fd_resource_survives_minor_and_major_gc_with_stable_ref_count() {
    let inner = Arc::new(FdInner::new(-1, 77));
    let mut process = Process::new(77, 32);
    let resource = alloc_fd_resource(&mut process, Arc::clone(&inner));
    process.set_x_reg(0, resource);
    assert_eq!(Arc::strong_count(&inner), 2);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(Arc::strong_count(&inner), 2);
    let resource = FdResource::new(process.x_reg(0)).expect("fd resource after minor");
    assert_eq!(resource.fd(), -1);
    assert_eq!(resource.owner_pid(), 77);
    assert_no_term_pointer_into_young(&process, process.x_reg(0));

    collect_major(&mut process).expect("major GC succeeds");

    assert_eq!(Arc::strong_count(&inner), 2);
    let resource = FdResource::new(process.x_reg(0)).expect("fd resource after major");
    assert_eq!(resource.fd(), -1);
    assert_eq!(resource.owner_pid(), 77);
    release_all_refcounted_resources(&mut process);
    assert_eq!(Arc::strong_count(&inner), 1);
}

#[test]
fn unreachable_fd_resource_releases_heap_arc_after_minor_gc() {
    let inner = Arc::new(FdInner::new(-1, 77));
    let mut process = Process::new(77, 32);
    let _unreachable = alloc_fd_resource(&mut process, Arc::clone(&inner));
    assert_eq!(Arc::strong_count(&inner), 2);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(Arc::strong_count(&inner), 1);
    assert_eq!(process.heap().young_used(), 0);
}

#[test]
fn virtual_binary_pressure_triggers_minor_gc_even_with_nursery_space() {
    let shared = SharedBinary::new(vec![0xEF; 4096]);
    let mut process = Process::new(1, 32);
    let _unreachable = alloc_proc_bin(&mut process, &shared);
    let used_before = process.heap().young_used();
    assert!(process.heap().available() > 1);
    assert_eq!(shared.ref_count(), 2);

    let _ptr = alloc(&mut process, 1).expect("allocation under binary pressure succeeds");

    assert_eq!(process.heap().young_used(), 1);
    assert!(used_before > process.heap().young_used());
    assert_eq!(shared.ref_count(), 1);
    assert_eq!(process.virtual_binary_heap(), 0);
}

#[test]
fn exception_roots_are_rewritten_after_minor_gc() {
    let mut process = Process::new(1, 32);
    let reason = alloc_tuple(&mut process, &[Term::small_int(8)]);
    let stacktrace = alloc_tuple(&mut process, &[Term::small_int(9)]);
    let expected_reason = snapshot(reason);
    let expected_stacktrace = snapshot(stacktrace);
    process.set_current_exception(Some(crate::process::Exception {
        class: Term::atom(Atom::ERROR),
        reason,
        stacktrace,
    }));

    collect_minor(&mut process).expect("minor GC succeeds");

    let exception = process
        .current_exception()
        .expect("exception should remain installed");
    assert_eq!(snapshot(exception.reason), expected_reason);
    assert_eq!(snapshot(exception.stacktrace), expected_stacktrace);
    assert_no_term_pointer_into_young(&process, exception.reason);
    assert_no_term_pointer_into_young(&process, exception.stacktrace);
}

#[test]
fn dictionary_heap_terms_survive_minor_gc() {
    let mut process = Process::new(1, 32);
    let key = alloc_tuple(&mut process, &[Term::small_int(10)]);
    let value = alloc_tuple(&mut process, &[Term::small_int(11)]);
    let expected_key = snapshot(key);
    let expected_value = snapshot(value);

    process.dict_put(key, value);

    collect_minor(&mut process).expect("minor GC succeeds");

    let entries = process.dict_get_all();
    assert_eq!(entries.len(), 1);
    let (key, value) = entries[0];
    assert_eq!(snapshot(key), expected_key);
    assert_eq!(snapshot(value), expected_value);
    assert_no_term_pointer_into_young(&process, key);
    assert_no_term_pointer_into_young(&process, value);
}

#[test]
fn dictionary_immediate_terms_survive_minor_gc_unchanged() {
    let mut process = Process::new(1, 32);
    let key = Term::atom(Atom::OK);
    let value = Term::small_int(12);

    process.dict_put(key, value);
    let _garbage = alloc_tuple(&mut process, &[Term::small_int(13)]);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(process.dict_get_all(), &[(key, value)]);
    assert_eq!(process.dict_get(key), value);
}

#[test]
fn unreachable_young_terms_are_reclaimed() {
    let mut process = Process::new(1, 32);
    let reachable = alloc_tuple(&mut process, &[Term::small_int(1)]);
    let _unreachable = alloc_tuple(&mut process, &[Term::small_int(2)]);
    process.set_x_reg(0, reachable);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(process.heap().young_used(), 0);
    assert_eq!(process.heap().old_used(), 2);
    assert_eq!(
        snapshot(process.x_reg(0)),
        Snapshot::Tuple(vec![Snapshot::Int(1)])
    );
}

#[test]
fn boolean_list_cons_is_not_walked_as_a_refcounted_binary() {
    // C1 regression. `visit_young_boxed_objects` reconstructs an allocation's
    // type from its first word, but cons cells are headerless: word[0] is the
    // list head. The atom `false` encodes to the raw word 0x19, which is exactly
    // `BoxedTag::ProcBin`, so a `[false | _]` cons is reported as a live ProcBin
    // and then handed to `release_proc_bin_arc`, which reads out-of-bounds word
    // two and `Arc::from_raw`s it -- arbitrary-memory free on the next minor GC.
    // Booleans lists are pervasive, so this is reachable from ordinary code.
    let mut process = Process::new(1, 64);
    let cons = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    process.set_x_reg(0, cons);

    let mut reported = Vec::new();
    process
        .heap()
        .visit_young_boxed_objects(|_ptr, tag, _words| reported.push(tag));

    assert!(
        !reported.contains(&BoxedTag::ProcBin),
        "a headerless [false | _] cons was misclassified as a refcounted ProcBin \
         by the GC release walk (reported tags: {reported:?})"
    );
}

// Accounting-sanity walls (AION-ENCODE-GC-DEFECT walls lane 2). The C1
// collision wall above pins WHICH allocations the release walks may visit;
// these walls pin what the walks DO: every ProcBin Arc released exactly once,
// and `virtual_binary_heap` pacing landing at the EXACT expected value after
// each of the three release-walk entry points — young walk (minor), full walk
// (terminate), compacted-sources walk (major) — on adversarial heaps where
// live ProcBins with known byte totals sit adjacent to `[false | _]` conses
// in both regions. Silent pacing corruption (the production ~25 GB residency
// face) shows up here as an inexact value, not a crash.

#[test]
fn minor_release_walk_decrements_pacing_by_exactly_the_unreachable_proc_bin_bytes() {
    let mut process = Process::new(1, 256);
    let live_shared = SharedBinary::new(vec![0xE2; 100]);
    let dead_shared = SharedBinary::new(vec![0x80; 77]);

    let guard_before = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let live_bin = alloc_proc_bin(&mut process, &live_shared);
    let guard_between = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let _dead_bin = alloc_proc_bin(&mut process, &dead_shared);
    let guard_after = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    process.set_x_reg(0, live_bin);
    process.set_x_reg(1, guard_before);
    process.set_x_reg(2, guard_after);
    let _ = guard_between; // unreachable, like the dead ProcBin

    assert_eq!(process.virtual_binary_heap(), 177);
    assert_eq!(live_shared.ref_count(), 2);
    assert_eq!(dead_shared.ref_count(), 2);

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(
        process.virtual_binary_heap(),
        100,
        "pacing must drop by exactly the unreachable ProcBin's 77 bytes"
    );
    assert_eq!(
        live_shared.ref_count(),
        2,
        "live ProcBin keeps exactly one heap-owned Arc across promotion"
    );
    assert_eq!(
        dead_shared.ref_count(),
        1,
        "dead ProcBin's heap-owned Arc is released exactly once"
    );
    let promoted = ProcBin::new(process.x_reg(0)).expect("promoted ProcBin");
    assert_eq!(promoted.as_bytes(), &[0xE2; 100][..]);
}

#[test]
fn terminate_release_walk_releases_each_proc_bin_arc_exactly_once_and_zeroes_pacing() {
    let mut process = Process::new(1, 256);
    let old_shared = SharedBinary::new(vec![0x94; 64]);
    let young_shared = SharedBinary::new(vec![0xAF; 33]);

    // Old-region ProcBin flanked by [false | _] conses: allocate in the
    // nursery, root everything, promote via one minor collection.
    let guard_before = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let old_bin = alloc_proc_bin(&mut process, &old_shared);
    let guard_after = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    process.set_x_reg(0, guard_before);
    process.set_x_reg(1, old_bin);
    process.set_x_reg(2, guard_after);
    collect_minor(&mut process).expect("minor GC succeeds");
    assert_eq!(old_shared.ref_count(), 2);

    // Young-region ProcBin with its own adjacent [false | _] cons, live at
    // termination — the full walk covers both regions.
    let young_guard = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let young_bin = alloc_proc_bin(&mut process, &young_shared);
    process.set_x_reg(3, young_guard);
    process.set_x_reg(4, young_bin);
    assert_eq!(process.virtual_binary_heap(), 97);
    assert_eq!(young_shared.ref_count(), 2);

    process.terminate(ExitReason::Normal);

    assert_eq!(process.virtual_binary_heap(), 0);
    assert_eq!(
        old_shared.ref_count(),
        1,
        "old-region Arc released exactly once at termination"
    );
    assert_eq!(
        young_shared.ref_count(),
        1,
        "young-region Arc released exactly once at termination"
    );
}

#[test]
fn major_release_walk_accounts_compacted_source_proc_bins_exactly() {
    let mut process = Process::new(1, 256);
    let old_live = SharedBinary::new(vec![0x11; 80]);
    let old_dead = SharedBinary::new(vec![0x22; 55]);
    let young_live = SharedBinary::new(vec![0x33; 40]);
    let young_dead = SharedBinary::new(vec![0x44; 21]);

    // Old region: two ProcBins interleaved with [false | _] conses, promoted
    // by a minor collection while both are still rooted.
    let guard_1 = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let old_live_bin = alloc_proc_bin(&mut process, &old_live);
    let guard_2 = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let old_dead_bin = alloc_proc_bin(&mut process, &old_dead);
    let guard_3 = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    process.set_x_reg(0, guard_1);
    process.set_x_reg(1, old_live_bin);
    process.set_x_reg(2, guard_2);
    process.set_x_reg(3, old_dead_bin);
    process.set_x_reg(4, guard_3);
    collect_minor(&mut process).expect("minor GC succeeds");
    process.set_x_reg(3, Term::NIL); // old_dead becomes unreachable in old space

    // Young region: live and dead ProcBins behind a [false | _] cons.
    let young_guard = alloc_cons(&mut process, Term::atom(Atom::FALSE), Term::NIL);
    let young_live_bin = alloc_proc_bin(&mut process, &young_live);
    let _young_dead_bin = alloc_proc_bin(&mut process, &young_dead);
    process.set_x_reg(5, young_guard);
    process.set_x_reg(6, young_live_bin);

    assert_eq!(process.virtual_binary_heap(), 196);

    collect_major(&mut process).expect("major GC succeeds");

    assert_eq!(
        process.virtual_binary_heap(),
        120,
        "pacing must drop by exactly the two unreachable ProcBins' 76 bytes"
    );
    assert_eq!(old_live.ref_count(), 2);
    assert_eq!(young_live.ref_count(), 2);
    assert_eq!(old_dead.ref_count(), 1);
    assert_eq!(young_dead.ref_count(), 1);
    let compacted_old = ProcBin::new(process.x_reg(1)).expect("compacted old ProcBin");
    assert_eq!(compacted_old.as_bytes(), &[0x11; 80][..]);
    let compacted_young = ProcBin::new(process.x_reg(6)).expect("compacted young ProcBin");
    assert_eq!(compacted_young.as_bytes(), &[0x33; 40][..]);
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(128))]
    #[test]
    fn gc_property_random_acyclic_terms_survive(seed in 0u64..10_000) {
        let mut process = Process::new(1, 377);
        let mut terms = vec![Term::small_int(seed as i64), Term::atom(Atom::OK), Term::NIL];
        for index in 0..24usize {
            let left = terms[(seed as usize + index) % terms.len()];
            let right = terms[(seed as usize + index * 7 + 1) % terms.len()];
            let next = if index % 3 == 0 {
                alloc_cons(&mut process, left, right)
            } else {
                alloc_tuple(&mut process, &[left, right, Term::small_int(index as i64)])
            };
            terms.push(next);
        }
        process.set_x_reg(0, terms[terms.len() - 1]);
        process
            .stack_mut()
            .push_frame(Atom::OK, 0, module_pin(Atom::OK), 2)
            .expect("frame fits");
        process.stack_mut().set_y_reg(0, terms[terms.len() / 2]).expect("Y0 exists");
        process.mailbox_mut().push_owned_for_test(terms[terms.len() / 3]);
        let expected_x = snapshot(process.x_reg(0));
        let expected_y = snapshot(process.stack().y_reg(0).expect("Y0 exists"));
        let expected_mail = snapshot(process.mailbox().front_for_test().expect("mailbox root"));

        if seed % 2 == 0 {
            collect_minor(&mut process).expect("minor GC succeeds");
        } else {
            collect_major(&mut process).expect("major GC succeeds");
        }

        prop_assert_eq!(snapshot(process.x_reg(0)), expected_x);
        prop_assert_eq!(snapshot(process.stack().y_reg(0).expect("Y0 exists")), expected_y);
        prop_assert_eq!(snapshot(process.mailbox().front_for_test().expect("mailbox root")), expected_mail);
    }
}

// ---------------------------------------------------------------------------
// THE DISCRIMINATING PAIR — lane #112.
//
// ⭐ These two tests are the point of the collection observable, and neither is
// worth anything without the other. A counter that incremented on BOTH events
// would pass ARM A perfectly while being exactly as useless as the
// `total_capacity()` proxy it replaces. ARM B is what pins the confusion.
//
// The defect they descend from: an AR-1 positive control asserted
// `total_capacity()` grew, under a message claiming it witnessed collection
// pressure. It witnesses a RESIZE. A pre-fix replica failed with the heap never
// resizing, so a correctly-failing control was graded as an unpressed cell.
// See docs/design/beamr/briefs/COLLECTION-OBSERVABLE-NOTE.md.
// ---------------------------------------------------------------------------

#[test]
fn a_fresh_process_has_collected_nothing() {
    let process = Process::new(1, 64);
    assert_eq!(process.gc_attempts(), 0);
    assert_eq!(process.gc_completions(), 0);
}

/// ARM A — the observable MOVES on a collection.
#[test]
fn arm_a_collection_counter_moves_on_a_collection() {
    let mut process = Process::new(1, 64);
    let live = alloc_tuple(&mut process, &[Term::small_int(1)]);
    process.set_x_reg(0, live);

    let before_attempts = process.gc_attempts();
    let before_completions = process.gc_completions();

    collect_minor(&mut process).expect("minor GC succeeds");

    assert_eq!(
        process.gc_attempts(),
        before_attempts + 1,
        "a collection was entered, so the attempt counter must move"
    );
    assert_eq!(
        process.gc_completions(),
        before_completions + 1,
        "the collection succeeded, so the completion counter must move"
    );

    // A major collection is counted by the same rule, on the same process, so
    // the counters accumulate rather than latch at one.
    collect_major(&mut process).expect("major GC succeeds");
    assert_eq!(process.gc_attempts(), before_attempts + 2);
    assert_eq!(process.gc_completions(), before_completions + 2);
}

/// ARM B — ⭐ the observable does NOT move on a resize.
///
/// This is the arm that exists because of the finding. It asserts the exact
/// false positive the old guard produced, as a negative, forever: the heap grows
/// — `total_capacity()` moves, which is precisely what the old control read as
/// "a collection happened" — while no collection occurs and the counters stay
/// flat.
#[test]
fn arm_b_collection_counter_does_not_move_on_a_pure_resize() {
    let mut process = Process::new(1, 64);

    let capacity_before = process.heap().total_capacity();
    let attempts_before = process.gc_attempts();
    let completions_before = process.gc_completions();

    // A resize with no collection anywhere near it.
    process.heap_mut().grow_to_next_capacity();

    let capacity_after = process.heap().total_capacity();

    // The precondition of this test is that a resize ACTUALLY HAPPENED. Without
    // this the test would pass trivially on a heap that never grew, which is the
    // asleep-instrument shape the whole lane is about.
    assert!(
        capacity_after > capacity_before,
        "ARM B needs a real resize to be meaningful, got {capacity_before} -> {capacity_after}"
    );

    assert_eq!(
        process.gc_attempts(),
        attempts_before,
        "the heap RESIZED and no collection ran -- the attempt counter must NOT move. \
         If this fires, the observable has been wired to growth and is a proxy again"
    );
    assert_eq!(
        process.gc_completions(),
        completions_before,
        "the heap RESIZED and no collection ran -- the completion counter must NOT move"
    );
}

/// The two observables are independent in BOTH directions.
///
/// ARM B shows capacity can move while the counters do not. This shows the
/// converse: a collection can be counted with capacity completely unchanged, so
/// no future reader can rebuild the equivalence from either side.
#[test]
fn a_collection_can_happen_with_capacity_unchanged() {
    let mut process = Process::new(1, 64);
    let live = alloc_tuple(&mut process, &[Term::small_int(1)]);
    process.set_x_reg(0, live);

    let capacity_before = process.heap().total_capacity();
    collect_minor(&mut process).expect("minor GC succeeds");
    let capacity_after = process.heap().total_capacity();

    assert_eq!(
        capacity_after, capacity_before,
        "this cell is only evidence while the collection does not resize the heap"
    );
    assert_eq!(
        process.gc_completions(),
        1,
        "a collection ran and completed, and the observable saw it even though \
         capacity never moved -- the case the old resize proxy scored as 'nothing happened'"
    );
}