monty 0.0.21

#![expect(
    unsafe_code,
    reason = "Paged arena hands out typed, aliasable views into UnsafeCell entries"
)]

#[cfg(feature = "ref-count-return")]
use std::collections::HashSet;
use std::{
    cell::{Cell, UnsafeCell},
    collections::BTreeMap,
    fmt,
    iter::once,
    marker::PhantomData,
    mem::ManuallyDrop,
    ops::{Deref, DerefMut},
    ptr::{self, NonNull},
    sync::Arc,
};

use monty_types::{ResourceError, ResourceTracker};
use serde::ser::SerializeStruct;

// Re-export items moved to `heap_traits` so that `crate::heap::DropGuard` etc. continue
// to resolve (used by the `defer_drop!` macros and throughout the codebase).
pub(crate) use crate::heap_data::HeapData;
pub(crate) use crate::heap_traits::{ContainsHeap, DropGuard, DropWithContext, HeapItem};
#[cfg(feature = "ref-count-return")]
use crate::types::Type;
use crate::{
    asyncio::{Awaiter, Coroutine, ExternalFuture, ExternalFutureState, GatherFuture, GatherState},
    exception_private::SimpleException,
    heap_data::{CellValue, Closure, FunctionDefaults},
    modules::dataclasses::DataclassField,
    types::{
        BoundMethod, Bytes, BytesIterator, Class, Dataclass, Deque, Dict, DictItemIterator, DictItemsView,
        DictKeyIterator, DictKeysView, DictValueIterator, DictValuesView, ExtFunction, FrozenSet, Instance,
        ItertoolsIter, List, LongInt, Module, NamedTuple, NamedTupleClass, OpenFile, Path, Range, RangeIterator,
        ReMatch, RePattern, Set, SetIterator, Slice, Str, StringIterator, TimeZone, Tuple, TupleIterator,
        callable_iterator::CallableIterator, date, datetime, deque::DequeIterator, list::ListIterator, timedelta,
        timezone,
    },
    value::Value,
};

mod free_list;
mod stable_heap;
use stable_heap::StableHeap;

/// Unique identifier for values stored inside the heap arena.
///
/// The ID does not encode ownership. Local IDs should normally be borrowed or
/// wrapped immediately in `Value::Ref`; owned fields must document and release
/// their reference through `HeapItem` or `DropWithContext` cleanup.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct HeapId(usize);

impl HeapId {
    /// Creates a `HeapId` from a raw index.
    #[inline]
    pub(crate) fn from_index(index: usize) -> Self {
        Self(index)
    }

    /// Returns the raw index value.
    #[inline]
    pub fn index(self) -> usize {
        self.0
    }
}

/// The empty tuple is a singleton which is allocated at startup.
const EMPTY_TUPLE_ID: HeapId = HeapId(0);

/// Color tag used by the trial-deletion cycle collector (Bacon–Rajan, ECOOP 2001).
///
/// Each [`HeapEntry`] carries a color that represents what the collector currently
/// believes about the entry. Outside of a running collection, every reachable
/// entry is either [`Black`](Self::Black) (live, not part of any suspected cycle)
/// or [`Purple`](Self::Purple) (a candidate cycle root discovered by `dec_ref`,
/// awaiting investigation). [`Gray`](Self::Gray) and [`White`](Self::White) are
/// transient states used only during a [`Heap::collect_cycles`] call.
///
/// The encoding fits in a single byte and is serialized as part of every
/// [`HeapEntry`]: a snapshot taken with cycles pending must round-trip through
/// serde so the entries stay enrolled as candidates after restore (otherwise a
/// graph that becomes garbage just before snapshot would leak permanently).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub(crate) enum CcColor {
    /// Live and not currently a cycle candidate. Default state for every newly
    /// allocated entry.
    #[default]
    Black,
    /// Visited by `MarkGray` during a collection cycle. Children's refcounts
    /// have been provisionally decremented; a later `Scan` pass decides whether
    /// to resurrect (back to [`Black`](Self::Black)) or condemn
    /// ([`White`](Self::White)) the entry.
    Gray,
    /// Confirmed unreachable by the current collection: every reference into
    /// the entry comes from another condemned entry. `CollectWhite` will free
    /// it. Only seen mid-collection.
    White,
    /// Candidate cycle root. Set by `dec_ref` whenever a GC-tracked entry's
    /// refcount drops to a non-zero value — the only situation in which a new
    /// reference cycle can become unreachable. The collector seeds its work
    /// from every entry currently flagged Purple.
    Purple,
}

/// This structure allows for reading into the heap more efficiently than repeated calls to `Heap::get` and
/// `Heap::get_mut` by performing the indexing and type lookup once, and then using the borrow checker to
/// safely deference the resulting pointers for short-lived borrows.
///
/// The safety boundary is primarily that `HeapRead` pointers generated by the `HeapReader::read` API must remain valid
/// for their lifetime, see the safety notes in `HeapRead::get` for how that is guaranteed.
pub(crate) struct HeapReader<'a> {
    pub(crate) heap: &'a mut Heap,
    /// Makes the lifetime `'a` invariant.
    phantom: PhantomData<fn(&'a ()) -> &'a ()>,
}

impl HeapReader<'_> {
    /// The ONLY way to get a `HeapReader`. By only providing an API which takes a closure which
    /// must be satisfied for all `'a`, it's impossible to create other `HeapReader` with the
    /// exact same lifetime `'a`.
    ///
    /// To allow other data to be borrowed alongside the `HeapReader`, the closure is given a
    /// `&'a mut D` with the same lifetime as the `HeapReader`, which is forwarded from the
    /// `&mut D` passed to this function. This rebranding lets callers (most notably the
    /// `VM`) hold borrows whose lifetime matches the `HeapReader`'s invariant brand `'a`,
    /// which is what allows the `VM` to be parameterized by a single lifetime.
    pub fn with<R, D: ?Sized>(
        heap: &mut Heap,
        data: &mut D,
        f: impl for<'a> FnOnce(&'a mut HeapReader<'a>, &'a mut D) -> R,
    ) -> R {
        f(
            &mut HeapReader {
                heap: &mut *heap,
                phantom: PhantomData,
            },
            data,
        )
    }
}

impl<'a> HeapReader<'a> {
    /// Resolves a `HeapId` to a stable, branded [`HeapPtr<'a>`] for its entry.
    ///
    /// The returned `HeapPtr` can be used for efficient repeated access to the same entry
    /// without needing to re-index into the paged storage on every access.
    ///
    /// # Panics
    ///
    /// Panics if `id` is out of bounds.
    pub(crate) fn read_ptr(&self, id: HeapId) -> HeapPtr<'a> {
        // SAFETY: [DH] - `HeapPtr` prevents holding reference to freed slots across calls to allocate; it
        // always hands out either live `&HeapData` or `None`, never `&Option<HeapData>`.
        let slot = unsafe { self.heap.entries.slot_at(id) }.expect("HeapReader::read_ptr - id out of bounds");
        HeapPtr {
            inner: NonNull::from(slot),
            brand: PhantomData,
        }
    }

    /// Indexes into the heap.
    ///
    /// Thin wrapper around [`HeapPtr::read`]: resolves `id` to a `HeapPtr` and
    /// delegates the typed match/reader-count logic there. Panics if `id` is out
    /// of bounds or the slot is currently freed.
    pub fn read(&self, id: HeapId) -> HeapReadOutput<'a> {
        self.read_ptr(id).read(self)
    }

    #[expect(clippy::unused_self, reason = "'a lifetime is used to create the safety guarantees")]
    pub fn protect<'t, U: ?Sized>(&mut self, value: &'t U) -> BorrowedHeapRead<'t, 'a, U> {
        BorrowedHeapRead {
            inner: ManuallyDrop::new(HeapRead {
                value: NonNull::from(value),
                readers: NonNull::dangling(),
                borrow: PhantomData,
            }),
            original: PhantomData,
        }
    }

    #[expect(clippy::unused_self, reason = "'a lifetime is used to create the safety guarantees")]
    pub fn protect_mut<'t, U: ?Sized>(&mut self, value: &'t mut U) -> BorrowedHeapReadMut<'t, 'a, U> {
        BorrowedHeapReadMut {
            inner: ManuallyDrop::new(HeapRead {
                value: NonNull::from(value),
                readers: NonNull::dangling(),
                borrow: PhantomData,
            }),
            original: PhantomData,
        }
    }
}

impl ContainsHeap for HeapReader<'_> {
    fn heap(&self) -> &Heap {
        self.heap.heap()
    }
    fn heap_mut(&mut self) -> &mut Heap {
        self.heap.heap_mut()
    }
}

impl Deref for HeapReader<'_> {
    type Target = Heap;

    fn deref(&self) -> &Self::Target {
        self.heap
    }
}

impl DerefMut for HeapReader<'_> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.heap
    }
}

pub enum HeapReadOutput<'a> {
    Str(HeapRead<'a, Str>),
    Bytes(HeapRead<'a, Bytes>),
    List(HeapRead<'a, List>),
    Deque(HeapRead<'a, Deque>),
    Tuple(HeapRead<'a, Tuple>),
    NamedTuple(HeapRead<'a, NamedTuple>),
    NamedTupleClass(HeapRead<'a, NamedTupleClass>),
    Dict(HeapRead<'a, Dict>),
    DictItemsView(HeapRead<'a, DictItemsView>),
    DictKeysView(HeapRead<'a, DictKeysView>),
    DictValuesView(HeapRead<'a, DictValuesView>),
    Set(HeapRead<'a, Set>),
    FrozenSet(HeapRead<'a, FrozenSet>),
    Closure(HeapRead<'a, Closure>),
    FunctionDefaults(HeapRead<'a, FunctionDefaults>),
    ExtFunction(HeapRead<'a, ExtFunction>),
    Cell(HeapRead<'a, CellValue>),
    Range(HeapRead<'a, Range>),
    Slice(HeapRead<'a, Slice>),
    Exception(HeapRead<'a, SimpleException>),
    Dataclass(HeapRead<'a, Dataclass>),
    Class(HeapRead<'a, Class>),
    Instance(HeapRead<'a, Instance>),
    BoundMethod(HeapRead<'a, BoundMethod>),
    DataclassField(HeapRead<'a, DataclassField>),
    ListIterator(HeapRead<'a, ListIterator>),
    DequeIterator(HeapRead<'a, DequeIterator>),
    TupleIterator(HeapRead<'a, TupleIterator>),
    StringIterator(HeapRead<'a, StringIterator>),
    BytesIterator(HeapRead<'a, BytesIterator>),
    RangeIterator(HeapRead<'a, RangeIterator>),
    DictKeyIterator(HeapRead<'a, DictKeyIterator>),
    DictItemIterator(HeapRead<'a, DictItemIterator>),
    DictValueIterator(HeapRead<'a, DictValueIterator>),
    SetIterator(HeapRead<'a, SetIterator>),
    CallableIterator(HeapRead<'a, CallableIterator>),
    Itertools(HeapRead<'a, ItertoolsIter>),
    LongInt(HeapRead<'a, LongInt>),
    Module(HeapRead<'a, Module>),
    Coroutine(HeapRead<'a, Coroutine>),
    GatherFuture(HeapRead<'a, GatherFuture>),
    ExternalFuture(HeapRead<'a, ExternalFuture>),
    Path(HeapRead<'a, Path>),
    OpenFile(HeapRead<'a, OpenFile>),
    RePattern(HeapRead<'a, RePattern>),
    ReMatch(HeapRead<'a, ReMatch>),
    Date(HeapRead<'a, date::Date>),
    DateTime(HeapRead<'a, datetime::DateTime>),
    TimeDelta(HeapRead<'a, timedelta::TimeDelta>),
    TimeZone(HeapRead<'a, timezone::TimeZone>),
}

pub struct HeapRead<'a, T: ?Sized> {
    value: NonNull<T>,
    /// Pointer to the `readers` counter in the owning `HeapValue`.
    ///
    /// Incremented on creation, decremented on drop. This ensures `dec_ref`
    /// cannot free the entry while any `HeapRead` pointing into it exists.
    readers: NonNull<Cell<usize>>,
    /// Makes the lifetime `'a` invariant. In combination with the invariant lifetime
    /// on `HeapReader` and the `HeapReader::with` API, this guarantees that this
    /// `HeapRead` originated from that matching `HeapReader` (there is no way to
    /// construct another `HeapReader` with the same lifetime).
    borrow: PhantomData<fn(&'a T) -> &'a T>,
}

impl<T: ?Sized> Drop for HeapRead<'_, T> {
    fn drop(&mut self) {
        // SAFETY: (DH) the readers pointer is valid for the lifetime of the HeapValue,
        // which is guaranteed by the paged storage (addresses never move) and the
        // reader count itself (dec_ref cannot free an entry with active readers).
        let cell = unsafe { self.readers.as_ref() };
        cell.set(cell.get() - 1);
    }
}

impl<'a, T: ?Sized> HeapRead<'a, T> {
    /// Accesses the value contained in this reference.
    pub fn get<'r>(&self, _: &'r HeapReader<'a>) -> &'r T {
        // SAFETY: (DH)
        //  - The HeapReader has an invariant lifetime 'a which guarantees that this HeapRead
        //    came from the heap borrowed by this HeapReader.
        //  - The address of the `HeapValue` never changes because entries are stored in
        //    paged storage (`HeapEntries`) where each page is never reallocated or moved.
        //  - The HeapRead holds a strong reader reference (via the `readers` counter in
        //    `HeapValue`) which guarantees the entry will never be freed by `dec_ref`
        //    or `collect_cycles` while this `HeapRead` exists. The cycle collector's
        //    `Scan` phase treats `readers > 0` as an external reference and resurrects
        //    the entry to Black instead of condemning it as White.
        //  - The type of the `HeapValue` can never change once allocated. This is
        //    guaranteed by never exposing `&mut HeapData` outside of this module.
        //  - The borrow on `HeapReader` guarantees that there are no mutable borrows on any heap
        //    data while the return value of this function is alive.
        unsafe { self.value.as_ref() }
    }

    /// Mutably accesses the value contained in this reference.
    pub fn get_mut<'r>(&mut self, _: &'r mut HeapReader<'a>) -> &'r mut T {
        // SAFETY: see same constraints as in get() above.
        unsafe { self.value.as_mut() }
    }

    /// Casts this reader to a field of type `U` at some `offset` within the struct.
    ///
    /// Transfers ownership of the reader count from `self` to the returned `HeapRead`.
    ///
    /// # Safety
    ///   - The field of type `U` must ALWAYS exist at `offset` within `T` (i.e. `T` cannot be an enum, union etc)
    unsafe fn cast_as_member_ref<U>(&self, offset: usize) -> BorrowedHeapRead<'_, 'a, U> {
        BorrowedHeapRead {
            // SAFETY: (DH) - caller of this function guarantees the offset & cast is valid
            inner: ManuallyDrop::new(HeapRead {
                // SAFETY: caller guarantees offset points to a valid field of type U within T
                value: unsafe { self.value.byte_add(offset) }.cast(),
                // dangling is fine because this heapread will never be dropped, and it is
                // also not `Clone` so there's no risk of this value ever being used
                readers: NonNull::dangling(),
                borrow: PhantomData,
            }),
            original: PhantomData,
        }
    }

    /// Casts this reader to a field of type `U` at some `offset` within the struct.
    ///
    /// Transfers ownership of the reader count from `self` to the returned `HeapRead`.
    ///
    /// # Safety
    ///   - The field of type `U` must ALWAYS exist at `offset` within `T` (i.e. `T` cannot be an enum, union etc)
    unsafe fn cast_as_member_ref_mut<U>(&mut self, offset: usize) -> BorrowedHeapReadMut<'_, 'a, U> {
        BorrowedHeapReadMut {
            // SAFETY: (DH) - caller of this function guarantees the offset & cast is valid
            inner: ManuallyDrop::new(HeapRead {
                // SAFETY: caller guarantees offset points to a valid field of type U within T
                value: unsafe { self.value.byte_add(offset) }.cast(),
                // dangling is fine because this heapread will never be dropped, and it is
                // also not `Clone` so there's no risk of this value ever being used
                readers: NonNull::dangling(),
                borrow: PhantomData,
            }),
            original: PhantomData,
        }
    }
}

impl<'a, T> HeapRead<'a, Vec<T>> {
    pub fn as_slice(&self, reader: &HeapReader<'a>) -> BorrowedHeapRead<'_, 'a, [T]> {
        BorrowedHeapRead {
            inner: ManuallyDrop::new(HeapRead {
                value: NonNull::from(self.get(reader).as_slice()),
                readers: NonNull::dangling(),
                borrow: PhantomData,
            }),
            original: PhantomData,
        }
    }
}

impl<'a, T: ?Sized> HeapRead<'a, Box<T>> {
    pub fn as_box_value(&self, reader: &HeapReader<'a>) -> BorrowedHeapRead<'_, 'a, T> {
        BorrowedHeapRead {
            inner: ManuallyDrop::new(HeapRead {
                value: NonNull::from(self.get(reader).as_ref()),
                readers: NonNull::dangling(),
                borrow: PhantomData,
            }),
            original: PhantomData,
        }
    }
}

/// Represents the reborrow of a `HeapRead` as a reference to a field of the original type.
pub struct BorrowedHeapRead<'original, 'a, U: ?Sized> {
    // inner is a projected HeapRead which will never be dropped
    inner: ManuallyDrop<HeapRead<'a, U>>,
    original: PhantomData<&'original U>,
}

// NB no DerefMut - would need to have a `BorrowedHeapReadMut`
impl<'a, U: ?Sized> Deref for BorrowedHeapRead<'_, 'a, U> {
    type Target = HeapRead<'a, U>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

/// Unsafe helper for `heap_read_as_field`, do not use. Same safety invariants as `HeapRead::cast_as_member`.
pub(crate) unsafe fn cast_as_member_ref_type_hinted<'r, 'a, T, U>(
    heap_read: &'r HeapRead<'a, T>,
    offset: usize,
    _type_hint: impl for<'s> Fn(&'s HeapRead<'a, T>) -> *const U,
) -> BorrowedHeapRead<'r, 'a, U> {
    // SAFETY: (DH) - caller upholds `cast_as_member` contract
    unsafe { heap_read.cast_as_member_ref(offset) }
}

macro_rules! heap_read_ref_as_field {
    ($heap_read:ident, $ty:ty, $field:tt) => {{
        let offset = std::mem::offset_of!($ty, $field);
        #[expect(unreachable_code)]
        let type_hint = |read: &$crate::heap::HeapRead<'_, $ty>| &raw const read.get(unreachable!()).$field;
        // SAFETY: (DH)
        //  - `std::mem::offset_of!` guarantees there is a field at fixed offset
        //  - `type_hint` guarantees that the field is of type `U` for the safety contract
        #[expect(unsafe_code)]
        unsafe {
            $crate::heap::cast_as_member_ref_type_hinted($heap_read, offset, type_hint)
        }
    }};
}

pub(crate) use heap_read_ref_as_field;

/// Represents the reborrow of a `HeapRead` as a reference to a field of the original type.
pub struct BorrowedHeapReadMut<'original, 'a, U: ?Sized> {
    // inner is a projected HeapRead which will never be dropped
    inner: ManuallyDrop<HeapRead<'a, U>>,
    original: PhantomData<&'original mut U>,
}

impl<'a, U: ?Sized> Deref for BorrowedHeapReadMut<'_, 'a, U> {
    type Target = HeapRead<'a, U>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl<U: ?Sized> DerefMut for BorrowedHeapReadMut<'_, '_, U> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.inner
    }
}

/// Unsafe helper for `heap_read_as_field`, do not use. Same safety invariants as `HeapRead::cast_as_member`.
pub(crate) unsafe fn cast_as_member_ref_mut_type_hinted<'r, 'a, T, U>(
    heap_read: &'r mut HeapRead<'a, T>,
    offset: usize,
    _type_hint: impl for<'s> Fn(&'s HeapRead<'a, T>) -> *const U,
) -> BorrowedHeapReadMut<'r, 'a, U> {
    // SAFETY: (DH) - caller upholds `cast_as_member` contract
    unsafe { heap_read.cast_as_member_ref_mut(offset) }
}

macro_rules! heap_read_ref_as_field_mut {
    ($heap_read:ident, $ty:ty, $field:tt) => {{
        let offset = std::mem::offset_of!($ty, $field);
        #[expect(unreachable_code)]
        let type_hint = |read: &$crate::heap::HeapRead<'_, $ty>| &raw const read.get(unreachable!()).$field;
        // SAFETY: (DH)
        //  - `std::mem::offset_of!` guarantees there is a field at fixed offset
        //  - `type_hint` guarantees that the field is of type `U` for the safety contract
        #[expect(unsafe_code)]
        unsafe {
            $crate::heap::cast_as_member_ref_mut_type_hinted($heap_read, offset, type_hint)
        }
    }};
}

pub(crate) use heap_read_ref_as_field_mut;

/// Stable, branded pointer to a heap slot's `Option<HeapEntry>`.
///
/// `HeapPtr<'a>` carries the same invariant lifetime `'a` as the [`HeapReader<'a>`]
/// that produced it (via [`HeapReader::entry_ptr`]). The brand mechanism is identical
/// to the one on [`HeapRead`]: pointers minted by one reader cannot be dereferenced
/// via a reader from a different `HeapReader::with` scope (or, looking ahead, a
/// different [`Heap`]), so same-heap origin is checked at compile time rather than
/// trusted by convention.
///
/// The pointer addresses the *slot's `Option<HeapEntry>`*, not the inner `HeapEntry`
/// directly. This makes it well-defined across the full lifetime of the slot — even
/// across `dec_ref`-driven frees and subsequent reuse by `allocate` — because the
/// slot's memory location never moves; only its `Some`/`None` state changes.
///
/// This mirrors the semantics of [`Heap::get`]/[`StableHeap::entry`] where the outer
/// `Option` is the live/freed signal.
///
/// Used as a fast handle inside paths that would otherwise re-index into paged storage
/// per access — currently the cycle collector, which converts `HeapId → HeapPtr` once
/// at push-time and uses the pointer directly on every subsequent pop, avoiding the
/// `pages[page_idx][slot_idx]` lookup chain.
#[derive(Copy, Clone)]
#[repr(transparent)]
pub(crate) struct HeapPtr<'a> {
    inner: NonNull<Option<HeapEntry>>,
    /// Makes `'a` invariant. Matches the brand on [`HeapReader<'a>`] / [`HeapRead<'a, T>`]
    /// so a `HeapPtr` cannot be reborrowed under a different reader scope.
    brand: PhantomData<fn(&'a ()) -> &'a ()>,
}

impl<'a> HeapPtr<'a> {
    /// Returns the live [`HeapEntry`] this pointer refers to, panicking if the slot
    /// has been freed.
    ///
    /// All `HeapEntry` fields are interior-mutable — `refcount`/`readers`/`color`
    /// via `Cell` and `data` via `UnsafeCell` — so callers can mutate them through
    /// the returned `&HeapEntry` without ever needing `&mut HeapEntry`. That's
    /// what makes a `&self`-derived `HeapPtr` (with Shared provenance) sound to
    /// dereference: we never derive `&mut` from it, so the SB/TB rules permit
    /// interior mutation via the embedded `Cell`/`UnsafeCell`.
    ///
    /// Use [`Self::try_entry`] for code paths that may legitimately encounter
    /// freed slots (e.g. linear scans over `0..heap.entries.len()`).
    pub fn entry<'r>(self, reader: &'r HeapReader<'a>) -> &'r HeapEntry {
        self.try_entry(reader).expect("HeapPtr::entry: slot has been freed")
    }

    /// Returns the [`HeapEntry`] this pointer refers to, or `None` if the slot is
    /// currently freed.
    ///
    /// Use where a freed slot is part of the expected state (linear scans, root
    /// reseeds, etc.). Mutation paths (cycle collector mark/scan inner loops) should
    /// prefer [`Self::entry`] so that an unexpectedly-freed entry surfaces as a
    /// loud panic rather than a silent skip.
    pub(crate) fn try_entry<'r>(self, _reader: &'r HeapReader<'a>) -> Option<&'r HeapEntry> {
        // SAFETY:
        //  - The invariant `'a` on `_reader` matches this pointer's brand, which is
        //    only settable inside `HeapReader::with`. That guarantees same-heap
        //    origin: a `HeapPtr<'a>` from a different reader scope cannot satisfy
        //    this signature.
        //  - `StableHeap::entry_ptr` only returns pointers to initialized slots, so
        //    the `Option<HeapEntry>` behind the pointer is always a valid place.
        //  - The `&HeapReader` borrow excludes any `&mut HeapReader` op that could
        //    free the slot during the returned reference's lifetime.
        unsafe { self.inner.as_ref() }.as_ref()
    }

    /// Returns the entry's [`HeapData`] payload via the `UnsafeHeapData` interior-
    /// mutability boundary.
    ///
    /// `UnsafeCell::get` produces a `*mut HeapData` with `SharedReadWrite` permission,
    /// so the returned `&HeapData` is sound even though the underlying `HeapPtr` was
    /// minted via `&self`-derived `entry_ptr`. Mutation of the payload still requires
    /// `&mut HeapReader` via the [`HeapRead`] machinery.
    pub fn data<'r>(self, reader: &'r HeapReader<'a>) -> &'r HeapData {
        let entry = self.entry(reader);
        // SAFETY: `UnsafeCell::get` yields a `SharedReadWrite` pointer; the `&HeapReader`
        // borrow on `reader` excludes any concurrent `&mut HeapReader` operation that
        // could mutate the payload during the returned reference's lifetime.
        unsafe { &*entry.data.0.get() }
    }

    /// Returns the typed [`HeapReadOutput`] for this entry, incrementing the
    /// reader count so the produced [`HeapRead<T>`] handles participate in the
    /// reader-count GC safety net (they are decremented on `Drop`).
    ///
    /// All `HeapRead<T>` handle pointers are derived through the `UnsafeHeapData`
    /// `UnsafeCell`, so they retain `SharedReadWrite` permission and remain valid
    /// for both read and mutable access (the latter via `HeapRead::get_mut`,
    /// which requires `&mut HeapReader`).
    pub fn read(self, reader: &HeapReader<'a>) -> HeapReadOutput<'a> {
        /// Computes a `HeapRead` from the raw `UnsafeCell` pointer and a shared reference
        /// to the variant field. The `&T` is only used to compute the field's byte offset
        /// within the `HeapData` enum; the returned `NonNull` is derived from the original
        /// `*mut HeapData` pointer so it inherits the `SharedReadWrite` permission from
        /// the `UnsafeCell`, allowing both reads and writes.
        #[inline]
        fn heap_read<'a, T>(base: *mut HeapData, field: &T, readers: NonNull<Cell<usize>>) -> HeapRead<'a, T> {
            let base_addr = base as usize;
            let field_addr = ptr::from_ref(field) as usize;
            let offset = field_addr - base_addr;
            HeapRead {
                // SAFETY: The pointer is derived from the UnsafeCell's `*mut` via byte
                // offset, preserving the `SharedReadWrite` permission. No reference retag
                // occurs — we only use the `&T` for its address, not to derive the pointer.
                value: unsafe { NonNull::new_unchecked(base.byte_add(offset).cast::<T>()) },
                readers,
                borrow: PhantomData,
            }
        }

        /// Like `heap_read` but for `Box<T>` fields inside `HeapData` variants.
        ///
        /// The `&Box<T>` is only used to locate the box field inside the enum; the
        /// inner pointer is *loaded out of* that field rather than derived by
        /// dereferencing the box. Dereferencing (`boxed.as_ref()`) would create a
        /// shared `&T` whose `SharedReadOnly` provenance makes later
        /// `HeapRead::get_mut` writes UB; the loaded pointer instead carries the
        /// box's own stored (writeable) provenance.
        #[expect(
            clippy::borrowed_box,
            reason = "We intentionally take &Box<T> to signal this is for boxed HeapData variants; &T would lose that context"
        )]
        fn heap_read_boxed<'a, T>(
            base: *mut HeapData,
            boxed: &Box<T>,
            readers: NonNull<Cell<usize>>,
        ) -> HeapRead<'a, T> {
            let base_addr = base as usize;
            let field_addr = ptr::from_ref(boxed) as usize;
            let offset = field_addr - base_addr;
            // SAFETY: `offset` locates the live `Box<T>` field within the enum, and the
            // pointer to it is derived from the `UnsafeCell`'s `*mut` (not from `&Box<T>`),
            // preserving `SharedReadWrite` permission for the load below. `Box<T>` with
            // `T: Sized` is guaranteed to be represented as a single non-null pointer, so
            // loading the field as `*mut T` yields the box's data pointer together with
            // the read/write provenance it was stored with.
            let value = unsafe { NonNull::new_unchecked(base.byte_add(offset).cast::<*mut T>().read()) };
            HeapRead {
                value,
                readers,
                borrow: PhantomData,
            }
        }

        let entry = self.entry(reader);
        // Increment the reader count for this entry. The corresponding decrement
        // happens in `HeapRead::drop`.
        entry.readers.set(entry.readers.get() + 1);
        let readers = NonNull::from(&entry.readers);
        // Get the raw pointer from the UnsafeCell — this has SharedReadWrite permission.
        let base: *mut HeapData = entry.data.0.get();
        // SAFETY: Match on a shared reference (`&*base`) to read the discriminant without
        // creating a Unique retag. The shared retag is compatible with existing
        // SharedReadWrite permissions from prior `read()` calls into the same UnsafeCell.
        // The `heap_read` helper then derives the NonNull from `base` (not from `&T`),
        // so the returned pointer retains full SharedReadWrite permission.
        match unsafe { &*base } {
            HeapData::Str(s) => HeapReadOutput::Str(heap_read(base, s, readers)),
            HeapData::Bytes(bytes) => HeapReadOutput::Bytes(heap_read(base, bytes, readers)),
            HeapData::List(list) => HeapReadOutput::List(heap_read(base, list, readers)),
            HeapData::Deque(deque) => HeapReadOutput::Deque(heap_read(base, deque, readers)),
            HeapData::Tuple(tuple) => HeapReadOutput::Tuple(heap_read(base, tuple, readers)),
            HeapData::NamedTuple(named_tuple) => {
                HeapReadOutput::NamedTuple(heap_read_boxed(base, named_tuple, readers))
            }
            HeapData::NamedTupleClass(class) => HeapReadOutput::NamedTupleClass(heap_read_boxed(base, class, readers)),
            HeapData::Dict(dict) => HeapReadOutput::Dict(heap_read(base, dict, readers)),
            HeapData::DictItemsView(v) => HeapReadOutput::DictItemsView(heap_read(base, v, readers)),
            HeapData::DictKeysView(v) => HeapReadOutput::DictKeysView(heap_read(base, v, readers)),
            HeapData::DictValuesView(v) => HeapReadOutput::DictValuesView(heap_read(base, v, readers)),
            HeapData::Set(set) => HeapReadOutput::Set(heap_read(base, set, readers)),
            HeapData::FrozenSet(frozen_set) => HeapReadOutput::FrozenSet(heap_read(base, frozen_set, readers)),
            HeapData::Closure(closure) => HeapReadOutput::Closure(heap_read(base, closure, readers)),
            HeapData::FunctionDefaults(function_defaults) => {
                HeapReadOutput::FunctionDefaults(heap_read(base, function_defaults, readers))
            }
            HeapData::ExtFunction(name) => HeapReadOutput::ExtFunction(heap_read(base, name, readers)),
            HeapData::Cell(cell_value) => HeapReadOutput::Cell(heap_read(base, cell_value, readers)),
            HeapData::Range(range) => HeapReadOutput::Range(heap_read(base, range, readers)),
            HeapData::Slice(slice) => HeapReadOutput::Slice(heap_read(base, slice, readers)),
            HeapData::Exception(simple_exception) => {
                HeapReadOutput::Exception(heap_read(base, simple_exception, readers))
            }
            HeapData::Dataclass(dataclass) => HeapReadOutput::Dataclass(heap_read_boxed(base, dataclass, readers)),
            HeapData::Class(class) => HeapReadOutput::Class(heap_read_boxed(base, class, readers)),
            HeapData::Instance(instance) => HeapReadOutput::Instance(heap_read_boxed(base, instance, readers)),
            HeapData::BoundMethod(bound_method) => HeapReadOutput::BoundMethod(heap_read(base, bound_method, readers)),
            HeapData::DataclassField(field) => HeapReadOutput::DataclassField(heap_read(base, field, readers)),
            HeapData::ListIterator(iter) => HeapReadOutput::ListIterator(heap_read(base, iter, readers)),
            HeapData::DequeIterator(iter) => HeapReadOutput::DequeIterator(heap_read(base, iter, readers)),
            HeapData::TupleIterator(iter) => HeapReadOutput::TupleIterator(heap_read(base, iter, readers)),
            HeapData::StringIterator(iter) => HeapReadOutput::StringIterator(heap_read(base, iter, readers)),
            HeapData::BytesIterator(iter) => HeapReadOutput::BytesIterator(heap_read(base, iter, readers)),
            HeapData::RangeIterator(iter) => HeapReadOutput::RangeIterator(heap_read(base, iter, readers)),
            HeapData::DictKeyIterator(iter) => HeapReadOutput::DictKeyIterator(heap_read(base, iter, readers)),
            HeapData::DictItemIterator(iter) => HeapReadOutput::DictItemIterator(heap_read(base, iter, readers)),
            HeapData::DictValueIterator(iter) => HeapReadOutput::DictValueIterator(heap_read(base, iter, readers)),
            HeapData::SetIterator(iter) => HeapReadOutput::SetIterator(heap_read(base, iter, readers)),
            HeapData::CallableIterator(c) => HeapReadOutput::CallableIterator(heap_read(base, c, readers)),
            HeapData::Itertools(i) => HeapReadOutput::Itertools(heap_read(base, i, readers)),
            HeapData::LongInt(l) => HeapReadOutput::LongInt(heap_read(base, l, readers)),
            HeapData::Module(module) => HeapReadOutput::Module(heap_read_boxed(base, module, readers)),
            HeapData::Coroutine(coroutine) => HeapReadOutput::Coroutine(heap_read(base, coroutine, readers)),
            HeapData::GatherFuture(gather_future) => {
                HeapReadOutput::GatherFuture(heap_read_boxed(base, gather_future, readers))
            }
            HeapData::ExternalFuture(external_future) => {
                HeapReadOutput::ExternalFuture(heap_read_boxed(base, external_future, readers))
            }
            HeapData::Path(path) => HeapReadOutput::Path(heap_read(base, path, readers)),
            HeapData::OpenFile(file) => HeapReadOutput::OpenFile(heap_read_boxed(base, file, readers)),
            HeapData::RePattern(re_pattern) => HeapReadOutput::RePattern(heap_read_boxed(base, re_pattern, readers)),
            HeapData::ReMatch(re_match) => HeapReadOutput::ReMatch(heap_read_boxed(base, re_match, readers)),
            HeapData::Date(d) => HeapReadOutput::Date(heap_read(base, d, readers)),
            HeapData::DateTime(d) => HeapReadOutput::DateTime(heap_read(base, d, readers)),
            HeapData::TimeDelta(d) => HeapReadOutput::TimeDelta(heap_read(base, d, readers)),
            HeapData::TimeZone(d) => HeapReadOutput::TimeZone(heap_read(base, d, readers)),
        }
    }
}

/// A single entry inside the heap arena, storing refcount and payload.
///
/// Hashing state lives on the per-type structs that benefit from it
/// ([`Str`], [`Bytes`], [`Tuple`], [`NamedTuple`], [`FrozenSet`] (via
/// `SetStorage`), [`Path`]). Cheap-to-hash types ([`Range`], [`Slice`],
/// dates etc.) recompute on demand. Unhashable types ([`List`], [`Dict`],
/// [`Set`]) return `None` from `py_hash` directly. None of these need
/// per-entry metadata.
///
/// The `color` field encodes the entry's state for the trial-deletion cycle
/// collector (see [`CcColor`]). Outside of a running collection, every live
/// entry is either Black (uninteresting) or Purple (a cycle-root candidate
/// queued for investigation); Gray and White are transient states only seen
/// during [`Heap::collect_cycles`]. Cell-typed for symmetry with `refcount`
/// — every write goes through an `&mut Heap` path (`dec_ref` and the
/// collector's `mark_gray`/`scan`/`scan_black`).
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct HeapEntry {
    refcount: Cell<usize>,
    /// Number of active `HeapRead` pointers into this entry's data.
    ///
    /// Incremented when `HeapReader::read` creates a `HeapRead`, decremented when
    /// the `HeapRead` is dropped. `dec_ref` panics if it would free an entry that
    /// still has active readers — this guarantees that `HeapRead` pointers remain
    /// valid for as long as they exist.
    #[serde(skip, default)] // should always be 0 during serde ops
    readers: Cell<usize>,
    /// The payload data
    data: UnsafeHeapData,
    /// Cycle-collector color. See [`CcColor`].
    ///
    /// Round-trips through serde because a snapshot taken between bytecode
    /// instructions can capture entries in the [`Purple`](CcColor::Purple)
    /// pending-collection state; dropping the color on restore would leak
    /// any cycle that became unreachable just before the snapshot.
    #[serde(default)]
    color: Cell<CcColor>,
}

/// This wrapper containing `UnsafeCell` exists to allow for data inside of `HeapValue`
/// to be safely pointed to via the `HeapReader` API.
///
/// The safety invariants are protected by the `Heap` / `HeapReader` API:
///   - It is never possible to alias mutable and immutable borrows into heap values,
///     whether they are the same or different value.
///   - When a mutable borrow of a heap value exists, no other heap value may be
///     borrowed. (See `Heap::get_mut` and `HeapRead::get`, which both require a `&mut`
///     borrow on the heap.)
struct UnsafeHeapData(UnsafeCell<HeapData>);

impl fmt::Debug for UnsafeHeapData {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // SAFETY: (DH) Debug formatting is read-only and never called concurrently
        // with mutation. This matches the safety invariants of the HeapReader API.
        let data = unsafe { &*self.0.get() };
        f.debug_tuple("UnsafeHeapData").field(data).finish()
    }
}

impl serde::Serialize for UnsafeHeapData {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        // SAFETY: when heap data is being serialized, there is no mutable borrow
        // possible on any data contents
        HeapData::serialize(unsafe { &*self.0.get() }, serializer)
    }
}

impl<'de> serde::Deserialize<'de> for UnsafeHeapData {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        Ok(Self(UnsafeCell::new(HeapData::deserialize(deserializer)?)))
    }
}

/// Reference-counted arena that backs all heap-only runtime values.
///
/// Uses a free list to reuse slots from freed values, keeping memory usage
/// constant for long-running loops that repeatedly allocate and free values.
/// When an value is freed via `dec_ref`, its slot ID is added to the free list.
/// New allocations pop from the free list when available, otherwise append.
///
/// Cycle collection uses Bacon–Rajan trial deletion: candidates come from
/// `dec_ref` (every container whose refcount drops to a non-zero value is
/// flagged [`Purple`](CcColor::Purple)), so the VM does not enumerate live
/// roots — refcount math itself proves reachability and values held only on
/// the Rust stack are correctly preserved by their non-zero refcount.
///
/// Owns the [`ResourceTracker`] that enforces memory/time limits and schedules
/// GC; unconfigured limits reduce every check to a single predictable branch.
#[derive(Debug)]
pub(crate) struct Heap {
    /// Paged storage for heap entries with integrated free list.
    entries: StableHeap<HeapEntry>,
    /// Resource tracker for enforcing limits and scheduling GC.
    tracker: ResourceTracker,
    /// Number of entries currently flagged [`Purple`](CcColor::Purple) — i.e.,
    /// suspected cycle roots awaiting collection.
    ///
    /// Used as an early-out: when zero, `collect_cycles` has no candidates
    /// and skips its heap walk entirely. The actual GC *frequency* is still
    /// driven by `allocations_since_gc` against the configured interval, so
    /// programs that produce no cycle candidates pay no collector cost
    /// regardless of how many allocations they perform.
    ///
    /// All `dec_ref` paths that mutate this counter take `&mut self`, so a
    /// plain `usize` is sufficient (no interior mutability needed).
    purple_count: usize,
    /// Number of GC-applicable allocations since the last cycle collection.
    ///
    /// Incremented for every GC-tracked allocation (see [`HeapData::is_gc_tracked`])
    /// and reset to zero at the end of every successful [`collect_cycles`]
    /// call. Combined with [`purple_count`](Self::purple_count) to gate
    /// automatic collections in [`should_gc`](Self::should_gc).
    ///
    /// Uses `Cell` for interior mutability so that `allocate(&self)` can
    /// increment.
    allocations_since_gc: Cell<u32>,
    /// When true, [`should_gc`](Self::should_gc) returns false regardless of
    /// the candidate count, suppressing automatic cycle-collection passes.
    /// Toggled by the `gc.disable()` / `gc.enable()` Python helpers (only
    /// registered under the `test-hooks` feature). Explicit `gc.collect()`
    /// calls still run.
    #[cfg(feature = "test-hooks")]
    gc_disabled: bool,
    /// Cached HeapId for the `datetime.timezone.utc` singleton.
    ///
    /// Lazily allocated on first access to `timezone.utc`. Once created, the refcount
    /// is incremented on each access so the caller can drop their reference normally.
    timezone_utc: Option<HeapId>,
    /// Live external functions indexed by name without owning heap references.
    ///
    /// Uses `BTreeMap` to avoid large residual capacity from spikes of `ExtFunction` allocations.
    ext_function_cache: BTreeMap<Arc<str>, HeapId>,
}

impl serde::Serialize for Heap {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut state = serializer.serialize_struct("Heap", 5)?;
        state.serialize_field("entries", &self.entries)?;
        state.serialize_field("tracker", &self.tracker)?;
        state.serialize_field("purple_count", &self.purple_count)?;
        state.serialize_field("allocations_since_gc", &self.allocations_since_gc.get())?;
        state.serialize_field("timezone_utc", &self.timezone_utc)?;
        state.end()
    }
}

impl<'de> serde::Deserialize<'de> for Heap {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(serde::Deserialize)]
        struct HeapFields {
            entries: StableHeap<HeapEntry>,
            tracker: ResourceTracker,
            #[serde(default)]
            purple_count: usize,
            #[serde(default)]
            allocations_since_gc: u32,
            #[serde(default)]
            timezone_utc: Option<HeapId>,
        }
        let fields = HeapFields::deserialize(deserializer)?;
        let mut entries = fields.entries;
        let mut ext_function_cache = BTreeMap::new();
        for index in 0..entries.len() {
            let id = HeapId::from_index(index);
            if let Some(mut entry) = entries.entry(id)
                && let HeapData::ExtFunction(function) = entry.get_mut().data.0.get_mut()
            {
                ext_function_cache.insert(function.cache_key(), id);
            }
        }
        Ok(Self {
            entries,
            tracker: fields.tracker,
            purple_count: fields.purple_count,
            allocations_since_gc: Cell::new(fields.allocations_since_gc),
            #[cfg(feature = "test-hooks")]
            gc_disabled: false,
            timezone_utc: fields.timezone_utc,
            ext_function_cache,
        })
    }
}

/// Default GC interval — run cycle collection every 100 000 GC-tracked
/// allocations unless the configured resource tracker overrides it.
///
/// The trial-deletion collector additionally short-circuits the trace when
/// `purple_count == 0`, so programs that produce no cycle candidates pay no
/// collector cost regardless of their allocation rate.
///
/// When the `memory-model-checks` feature is enabled, this is reduced to 1 to
/// stress-test GC behavior on every allocation.
const DEFAULT_GC_INTERVAL: usize = if cfg!(feature = "memory-model-checks") {
    1
} else {
    100_000
};

impl Heap {
    /// Creates a new heap with the given resource tracker.
    ///
    /// Use this to create heaps with custom resource limits or GC scheduling.
    pub fn new(capacity: usize, tracker: ResourceTracker) -> Self {
        let this = Self {
            entries: StableHeap::with_capacity(capacity),
            tracker,
            purple_count: 0,
            allocations_since_gc: Cell::new(0),
            #[cfg(feature = "test-hooks")]
            gc_disabled: false,
            timezone_utc: None,
            ext_function_cache: BTreeMap::new(),
        };

        // The empty-tuple singleton starts with refcount = 1 — that single ref *is* the
        // permanent heap-owned reference. `get_empty_tuple` bumps the refcount on each
        // hand-out so callers can `dec_ref` normally; the heap-owned ref keeps the
        // singleton's rc ≥ 1 forever, which is why trial deletion needs no special-case
        // rooting for it (a debug_assert in `dec_ref` enforces the invariant).
        let empty_tuple = HeapData::Tuple(Tuple::default());
        let new_entry = HeapEntry {
            refcount: Cell::new(1),
            readers: Cell::new(0),
            data: UnsafeHeapData(UnsafeCell::new(empty_tuple)),
            color: Cell::new(CcColor::Black),
        };

        let empty_tuple = this.entries.allocate(new_entry);
        debug_assert_eq!(empty_tuple, EMPTY_TUPLE_ID);
        this
    }

    /// Returns a reference to the resource tracker.
    pub fn tracker(&self) -> &ResourceTracker {
        &self.tracker
    }

    /// Returns a mutable reference to the resource tracker.
    pub fn tracker_mut(&mut self) -> &mut ResourceTracker {
        &mut self.tracker
    }

    /// Checks whether a configured time or memory limit has been exceeded.
    ///
    /// Delegates to the resource tracker's `check_time()`, which polls
    /// allocator-backed usage against `max_memory` and elapsed execution time
    /// against `max_duration` (each a no-op when unset).
    ///
    /// Call this inside Rust-side loops (builtins, sort, iterator collection)
    /// that execute within a single bytecode instruction and would otherwise
    /// bypass the VM's per-instruction checkpoint.
    #[inline]
    pub fn check_time(&self) -> Result<(), ResourceError> {
        self.tracker.check_time()
    }

    /// Number of entries in the heap (including freed slots).
    pub fn size(&self) -> usize {
        self.entries.len()
    }

    /// Returns the number of GC-tracked allocations since the last cycle
    /// collection. Reset to zero by [`collect_cycles`](Self::collect_cycles).
    ///
    /// Used by `run_ref_counts` to expose the GC trigger metric to tests:
    /// a value much smaller than the configured `gc_interval` after many
    /// allocations is evidence that collection ran.
    #[cfg(feature = "ref-count-return")]
    pub fn get_allocations_since_gc(&self) -> u32 {
        self.allocations_since_gc.get()
    }

    /// Allocates a new heap entry.
    ///
    /// GC-tracked types bump `allocations_since_gc` so that
    /// [`should_gc`](Self::should_gc) eventually fires; trial deletion's own
    /// candidate enrollment happens later, at `dec_ref` time. Leaf types
    /// (strings, bytes, …) cannot participate in cycles and don't count
    /// against the GC interval.
    pub fn allocate(&self, data: HeapData) -> HeapId {
        if data.is_gc_tracked() {
            self.allocations_since_gc
                .set(self.allocations_since_gc.get().wrapping_add(1));
        }

        let new_entry = HeapEntry {
            refcount: Cell::new(1),
            readers: Cell::new(0),
            data: UnsafeHeapData(UnsafeCell::new(data)),
            color: Cell::new(CcColor::Black),
        };

        self.entries.allocate(new_entry)
    }

    /// Returns the singleton empty tuple.
    ///
    /// In Python, `() is ()` is always `True` because empty tuples are interned.
    /// This method provides the same optimization by returning the same `HeapId`
    /// for all empty tuple allocations.
    ///
    /// The returned `Value` has its reference count incremented, so the caller
    /// owns a reference and must call `dec_ref` when done.
    pub fn get_empty_tuple(&self) -> Value {
        // Return existing singleton with incremented refcount
        self.inc_ref(EMPTY_TUPLE_ID);
        Value::Ref(EMPTY_TUPLE_ID)
    }

    /// Returns the external function for `name`, reusing a live object when possible.
    ///
    /// The cache holds no reference count of its own. Its entry is removed when the
    /// last owning reference is dropped, before the heap slot can be reused.
    pub fn get_ext_function(&mut self, name: &str) -> Value {
        if let Some(id) = self.ext_function_cache.get(name).copied() {
            self.inc_ref(id);
            Value::Ref(id)
        } else {
            let function = ExtFunction::new(name);
            let cache_key = function.cache_key();
            let id = self.allocate(HeapData::ExtFunction(function));
            let previous = self.ext_function_cache.insert(cache_key, id);
            debug_assert!(previous.is_none());
            Value::Ref(id)
        }
    }

    /// Removes `id` from the weak cache without disturbing a duplicate-name winner.
    fn remove_ext_function_cache_entry(cache: &mut BTreeMap<Arc<str>, HeapId>, name: &str, id: HeapId) {
        if cache.get(name) == Some(&id) {
            cache.remove(name);
        }
    }

    /// Returns the cached `datetime.timezone.utc` singleton, lazily creating it on first access.
    ///
    /// The returned `Value::Ref` has its refcount incremented so the caller can drop
    /// it normally. The singleton itself is kept alive by the `timezone_utc` field.
    pub fn get_timezone_utc(&mut self) -> Value {
        if let Some(id) = self.timezone_utc {
            self.inc_ref(id);
            Value::Ref(id)
        } else {
            let tz = TimeZone::utc();
            let id = self.allocate(HeapData::TimeZone(tz));
            // Keep an extra refcount for the singleton cache
            self.inc_ref(id);
            self.timezone_utc = Some(id);
            Value::Ref(id)
        }
    }

    /// Increments the reference count for an existing heap entry.
    ///
    /// # Panics
    /// Panics if the value ID is invalid or the value has already been freed.
    pub fn inc_ref(&self, id: HeapId) {
        let value = self.entries.get(id);
        value.refcount.update(|r| r + 1);
    }

    /// Decrements the reference count and frees the value (plus children) once it hits zero.
    ///
    /// This is the low-level release operation for an owned raw `HeapId`. Ordinary
    /// control flow should instead keep local ownership in `Value::Ref` and use
    /// `defer_drop!` or `DropGuard`. Heap-stored owners declare child references in
    /// `HeapItem::py_dec_ref_ids`; direct calls are appropriate in cleanup for other
    /// structures that own raw IDs.
    ///
    /// Uses an iterative work stack instead of recursion to avoid Rust stack overflow
    /// when freeing deeply nested containers (e.g., a list nested 10,000 levels deep).
    /// This is analogous to CPython's "trashcan" mechanism for safe deallocation.
    ///
    /// Implements the candidate-enrollment side of Bacon–Rajan trial deletion: any
    /// GC-tracked entry whose refcount survives the decrement gets flagged
    /// [`Purple`](CcColor::Purple), so the next [`collect_cycles`](Self::collect_cycles)
    /// can investigate it. Entries that drop to zero are freed immediately on the
    /// existing fast path; if such an entry was Purple, the heap-wide
    /// `purple_count` is rebalanced so it stays in sync with the actual number
    /// of Purple entries.
    ///
    /// # Panics
    /// Panics if the value ID is invalid, the value has already been freed, or
    /// the refcount would reach zero while active `HeapRead` readers exist.
    pub fn dec_ref(&mut self, id: HeapId) {
        HeapReader::with(self, &mut (), |reader, ()| {
            let mut current_id = id;
            // A fresh Vec is deliberate: it costs nothing unless children are
            // actually pushed, whereas pooling it on the Heap was measured
            // (CodSpeed, PR #536) to add take/restore traffic to every call.
            let mut work_stack = Vec::new();
            loop {
                // Using `HeapPtr` avoids the possibility of aliasing with live borrows
                // held by `HeapRead` handles.
                let ptr = reader.read_ptr(current_id);
                let heap_entry = ptr.entry(reader);
                if heap_entry.refcount.get() > 1 {
                    heap_entry.refcount.update(|r| r - 1);

                    let is_gc_tracked = ptr.data(reader).is_gc_tracked();
                    if is_gc_tracked && heap_entry.color.get() != CcColor::Purple {
                        // The refcount survived — a newly unreachable cycle could
                        // now be hiding. Flag it as a candidate for the next `collect_cycles`.
                        heap_entry.color.set(CcColor::Purple);
                        reader.heap.purple_count += 1;
                    }
                } else {
                    debug_assert!(
                        current_id != EMPTY_TUPLE_ID,
                        "Heap::dec_ref: empty-tuple singleton's heap-owned refcount must never reach zero",
                    );
                    assert!(
                        heap_entry.readers.get() == 0,
                        "Heap::dec_ref: cannot free HeapId({}) with {} active reader(s)",
                        current_id.index(),
                        heap_entry.readers.get(),
                    );
                    // If the entry was a pending cycle candidate, decrement
                    // `purple_count` to reflect that it is leaving the heap before
                    // the collector reaches it.
                    if heap_entry.color.get() == CcColor::Purple {
                        reader.heap.purple_count -= 1;
                    }
                    // Remove weak-cache entries before the slot becomes available for reuse.
                    let ext_function_name = match ptr.data(reader) {
                        HeapData::ExtFunction(function) => Some(function.cache_key()),
                        _ => None,
                    };
                    // Clear the cache (only if it points to this exact function, it's possible for
                    // snapshot deserialization to create duplicate functions with the same name)
                    if let Some(name) = ext_function_name {
                        Self::remove_ext_function_cache_entry(&mut reader.heap.ext_function_cache, &name, current_id);
                    }

                    // It is not possible to free from `HeapPtr` because it is created through
                    // a &self borrow on `StableHeap`. At least this repeated lookup is already
                    // on the slow path.
                    let mut value = reader.heap.entries.entry(current_id).expect("already looked up").free();

                    // Collect child IDs and push onto work stack for iterative processing
                    py_dec_ref_ids_for_data(value.data.0.get_mut(), &mut work_stack);
                }

                let Some(next_id) = work_stack.pop() else {
                    break;
                };
                current_id = next_id;
            }
        });
    }

    /// Returns an immutable reference to the heap data stored at the given ID. This can be more efficient
    /// than `.read()` for short-lived borrows that need read-only access (avoids reader bookkeeping).
    ///
    /// # Panics
    /// Panics if the value ID is invalid, the value has already been freed,
    /// or the data is currently borrowed via `with_entry_mut`/`call_attr`.
    #[must_use]
    pub fn get(&self, id: HeapId) -> &HeapData {
        let data = &self.entries.get(id).data;
        // SAFETY: (DH) no mutable references into `HeapData` is possible while the heap is borrowed
        unsafe { &*data.0.get() }
    }

    /// Returns the reference count for the heap entry at the given ID.
    ///
    /// This is primarily used for testing reference counting behavior.
    ///
    /// # Panics
    /// Panics if the value ID is invalid or the value has already been freed.
    #[must_use]
    #[cfg(feature = "ref-count-return")]
    pub fn get_refcount(&self, id: HeapId) -> usize {
        self.entries.get(id).refcount.get()
    }

    /// Returns the number of live (non-freed) values on the heap.
    ///
    /// This is primarily used for testing to verify that all heap entries
    /// are accounted for in reference count tests.
    ///
    /// Excludes the empty tuple singleton since it's an internal optimization
    /// detail that persists even when not explicitly used by user code.
    #[must_use]
    #[cfg(feature = "ref-count-return")]
    pub fn entry_count(&self) -> usize {
        // Skip index 0 which is the empty tuple singleton
        self.entries.iter().skip(1).count()
    }

    /// Returns whether cycle collection should run.
    ///
    /// True when the configured allocation interval has elapsed *and* at
    /// least one [`Purple`](CcColor::Purple) candidate is pending. The
    /// alloc-count check sets the maximum collector frequency the user
    /// asked for; the `purple_count` check is an additional early-out so
    /// programs that produce no cycle candidates pay no collector cost
    /// regardless of their allocation rate.
    ///
    /// Always returns false when [`disable_gc`](Self::disable_gc) has been
    /// called without a matching [`enable_gc`](Self::enable_gc); explicit
    /// [`collect_cycles`](Self::collect_cycles) calls still run regardless.
    #[inline]
    pub fn should_gc(&self) -> bool {
        #[cfg(feature = "test-hooks")]
        if self.gc_disabled {
            return false;
        }
        if self.purple_count == 0 {
            return false;
        }
        let interval = self.tracker.gc_interval().unwrap_or(DEFAULT_GC_INTERVAL);
        (self.allocations_since_gc.get() as usize) >= interval
    }

    /// Suppresses automatic garbage collection until [`enable_gc`](Self::enable_gc)
    /// is called.
    ///
    /// Explicit [`collect_cycles`](Self::collect_cycles) calls still run while
    /// disabled, so a script can build a known amount of garbage and then time
    /// exactly one collection pass.
    #[cfg(feature = "test-hooks")]
    pub fn disable_gc(&mut self) {
        self.gc_disabled = true;
    }

    /// Resumes automatic garbage collection after a prior [`disable_gc`](Self::disable_gc).
    ///
    /// Calling [`enable_gc`](Self::enable_gc) on an already-enabled heap is a no-op.
    #[cfg(feature = "test-hooks")]
    pub fn enable_gc(&mut self) {
        self.gc_disabled = false;
    }

    /// Runs Bacon–Rajan trial-deletion cycle collection.
    ///
    /// Walks every entry currently flagged [`Purple`](CcColor::Purple) (the
    /// candidates accumulated by `dec_ref`) and frees any references that turn
    /// out to live entirely inside an unreachable cycle. Refcount math itself
    /// proves liveness — entries reachable from outside the candidate set
    /// (including those held only on the Rust stack and those with active
    /// `HeapRead` readers) survive automatically because their refcount or
    /// reader count remains non-zero — so no explicit root walk is required.
    ///
    /// Phases:
    ///
    /// 1. **`MarkRoots`** — single linear pass over `entries` that finds
    ///    Purple entries, runs `MarkGray` on each, and collects the resulting
    ///    seed list. Purple entries reached transitively by an earlier seed's
    ///    `MarkGray` turn Gray and are correctly skipped, so each cycle root
    ///    is only seeded once.
    /// 2. **`Scan`** — for each seed, decide whether the subtree is alive
    ///    (`s.refcount > 0 || s.readers > 0`, resurrect to Black) or condemned
    ///    (mark White and recurse).
    /// 3. **`CollectWhite`** — free White entries iteratively. Child
    ///    refcounts were already balanced by `MarkGray`/`ScanBlack`, so this
    ///    phase does **not** call `dec_ref` on children — it only walks them
    ///    to free transitively.
    ///
    /// All four phases iterate via explicit work stacks instead of recursion
    /// (the textbook formulation is recursive); a 10 000-deep nested cycle
    /// must collect without a Rust stack overflow.
    ///
    /// Returns the number of unreachable entries that were freed during the sweep.
    ///
    /// # Caller Responsibility
    /// The caller should check [`should_gc`](Self::should_gc) before calling
    /// this method. With `purple_count == 0` the function returns immediately
    /// without touching the heap.
    pub fn collect_cycles(&mut self) -> usize {
        if self.purple_count == 0 {
            return 0;
        }
        // The mark/scan phases work in terms of `HeapPtr<'a>` to avoid re-indexing
        // paged storage on every entry access; the brand `'a` comes from the
        // surrounding `HeapReader::with` scope.
        HeapReader::with(self, &mut (), |reader, ()| reader.collect_cycles_inner())
    }

    fn collect_white(&mut self, roots: Vec<HeapId>) -> usize {
        let mut work_stack = roots;
        let mut freed = 0;
        while let Some(id) = work_stack.pop() {
            let Some(mut entry) = self.entries.entry(id) else {
                // Already freed via another seed's traversal — ignore.
                continue;
            };
            let heap_entry = entry.get_mut();
            if heap_entry.color.get() != CcColor::White {
                // Either resurrected to Black by `Scan` or never visited
                // (still Black/Gray from somewhere). Don't free.
                continue;
            }
            debug_assert!(
                heap_entry.readers.get() == 0,
                "collect_white: cannot free HeapId({}) with {} active reader(s)",
                id.index(),
                heap_entry.readers.get(),
            );
            let mut value = entry.free();
            // Clear weak entries before freeing their slots, just as `dec_ref`.
            if let HeapData::ExtFunction(function) = value.data.0.get_mut() {
                Self::remove_ext_function_cache_entry(&mut self.ext_function_cache, &function.cache_key(), id);
            }
            freed += 1;
            // Walk children, marking child `Value::Ref`s as `Dereferenced`
            // under `memory-model-checks` so dropping the freed entry's data
            // doesn't trip a Drop-panic on a live `Value::Ref` payload. The
            // pushed child IDs feed the work stack so we recursively walk
            // White grandchildren — we do *not* `dec_ref` these children
            // (`MarkGray`/`ScanBlack` already balanced their refcounts).
            py_dec_ref_ids_for_data(value.data.0.get_mut(), &mut work_stack);
        }
        freed
    }
}

/// Cycle-collection inner phases.
///
/// Implemented on [`HeapReader<'a>`] so the work stacks can carry
/// [`HeapPtr<'a>`] instead of [`HeapId`]: once an entry is reached, every
/// subsequent access is a pointer deref rather than a paged-storage lookup.
/// Entered only via [`Heap::collect_cycles`], which establishes the brand `'a`
/// through [`HeapReader::with`]. [`Heap::collect_white`] is intentionally kept
/// in `HeapId` space because freeing a slot requires consulting the free list,
/// which is keyed by id.
impl<'a> HeapReader<'a> {
    /// Implementation of [`Heap::collect_cycles`]. Phase 1 (discover Purple
    /// roots and `mark_gray` their subtrees), Phase 2 (`scan`), and Phase 3
    /// (`collect_white`) run sequentially. After `scan` the `HeapPtr` work
    /// stack is drained, so no branded pointer escapes into Phase 3.
    ///
    /// Each phase iterates `for_each_child_id` inline rather than staging child
    /// `HeapId`s through a scratch `Vec`. Mutation of entry fields goes through
    /// the `Cell`/`UnsafeCell`-backed interior mutability on `HeapEntry`, so the
    /// `&HeapEntry`/`&HeapData` borrows from [`HeapPtr::entry`]/[`HeapPtr::data`]
    /// can coexist with the `&self.entry_ptr` calls in the closure body.
    fn collect_cycles_inner(&mut self) -> usize {
        let mut roots: Vec<HeapId> = Vec::new();
        let mut work_stack: Vec<HeapPtr<'a>> = Vec::new();

        // 1. Discover roots by finding Purple entries. Mark each root (and its subtree) Gray.
        //    The linear scan may legitimately encounter freed slots, so we use
        //    `try_entry` to skip them rather than panicking.
        for i in 0..self.heap.entries.len() {
            let id = HeapId(i);
            let ptr = self.read_ptr(id);
            let Some(entry) = ptr.try_entry(self) else {
                continue;
            };
            if entry.color.get() != CcColor::Purple {
                continue;
            }
            if entry.readers.get() > 0 {
                // This entry cannot possibly be a root since it has active readers; reset
                // to Black so it won't be a candidate in the next cycle.
                entry.color.set(CcColor::Black);
                continue;
            }
            roots.push(id);
            self.mark_gray(ptr, &mut work_stack);
            debug_assert!(work_stack.is_empty(), "mark_gray must drain its work stack");
        }

        // 2. For each root, scan and resurrect if alive (refcount > 0 or active readers).
        //    Roots were live at Phase 1 and nothing has freed them between then and
        //    now, so `entry_ptr` returning `None` would indicate a bug.
        work_stack.extend(roots.iter().map(|&id| self.read_ptr(id)));
        self.scan(&mut work_stack);
        debug_assert!(work_stack.is_empty(), "scan must drain its work stack");

        // 3. Collect each root's White children as unreachable garbage. This phase
        //    keeps working in HeapId-space because `entry.free()` pushes to the
        //    free list, which is keyed by id.
        let freed = self.heap.collect_white(roots);

        // After this pass no Purple entries remain in the heap; zero the
        // counter so the next `dec_ref` event re-seeds from a clean baseline,
        // and reset the alloc-count gate so the next interval starts now.
        self.heap.purple_count = 0;
        self.heap.allocations_since_gc.set(0);
        freed
    }

    /// `MarkGray` (iterative): paint `s` and its transitive children Gray,
    /// decrementing each child's refcount once per traversal edge.
    ///
    /// After this completes for every root, every Gray entry's refcount equals
    /// the count of *external* references into it (refs originating outside
    /// the candidate subgraph). `Scan` uses that property to decide
    /// alive/condemned.
    fn mark_gray(&mut self, mut ptr: HeapPtr<'a>, work_stack: &mut Vec<HeapPtr<'a>>) {
        ptr.entry(self).color.set(CcColor::Gray);
        loop {
            for_each_child_id(ptr.data(self), |child_id| {
                let child_ptr = self.read_ptr(child_id);
                let entry = child_ptr.entry(self);
                debug_assert!(entry.refcount.get() > 0, "mark_gray: refcount underflow");
                entry.refcount.update(|r| r - 1);
                if entry.color.replace(CcColor::Gray) == CcColor::Gray {
                    // Already marked via another edge, don't push again
                    return;
                }
                work_stack.push(child_ptr);
            });
            let Some(next_ptr) = work_stack.pop() else {
                break;
            };
            ptr = next_ptr;
        }
    }

    /// `Scan` (iterative): each Gray entry is either resurrected via
    /// `ScanBlack` (external reference exists — refcount > 0 or active
    /// `HeapRead` reader) or painted White and its Gray children recursed.
    fn scan(&mut self, work_stack: &mut Vec<HeapPtr<'a>>) {
        let mut black_work_stack: Vec<HeapPtr<'a>> = Vec::new();
        while let Some(ptr) = work_stack.pop() {
            let entry = ptr.entry(self);
            if entry.color.get() != CcColor::Gray {
                // Already processed via another edge
                continue;
            }
            if entry.refcount.get() == 0 && entry.readers.get() == 0 {
                entry.color.set(CcColor::White);
                for_each_child_id(ptr.data(self), |child_id| {
                    let child_ptr = self.read_ptr(child_id);
                    let entry = child_ptr.entry(self);
                    if entry.color.get() != CcColor::Gray {
                        // Already processed via another edge
                        return;
                    }
                    work_stack.push(child_ptr);
                });
            } else {
                // External reference exists (either a refcount we couldn't
                // account for inside the candidate set, or a live `HeapRead`
                // pointing into the entry). Resurrect this entry and its
                // transitive Gray children back to Black via `mark_black`.
                self.mark_black(ptr, &mut black_work_stack);
                debug_assert!(black_work_stack.is_empty());
            }
        }
    }

    /// `ScanBlack` (iterative): resurrect a subtree by re-incrementing
    /// children's refcounts that `MarkGray` previously decremented, restoring
    /// the heap to the state it would have had if no cycle was suspected.
    ///
    /// Children's refcounts are incremented once per traversal edge — even if
    /// the child is already Black — so multi-edge graphs (a child reachable
    /// from two parents in the resurrected subtree) balance the matching
    /// per-edge decrements `MarkGray` performed. Recursion only descends into
    /// non-Black children so each entry is processed at most once.
    fn mark_black(&mut self, mut ptr: HeapPtr<'a>, work_stack: &mut Vec<HeapPtr<'a>>) {
        ptr.entry(self).color.set(CcColor::Black);
        loop {
            for_each_child_id(ptr.data(self), |child_id| {
                let child_ptr = self.read_ptr(child_id);
                let entry = child_ptr.entry(self);
                entry.refcount.update(|r| r + 1);
                if entry.color.replace(CcColor::Black) == CcColor::Black {
                    // Already marked via another edge
                    return;
                }
                work_stack.push(child_ptr);
            });
            let Some(next_ptr) = work_stack.pop() else {
                break;
            };
            ptr = next_ptr;
        }
    }
}

/// Leak detection for reference-counting tests.
///
/// Implemented on [`HeapReader`] because reading an entry's children needs
/// [`HeapPtr::data`]. It borrows the cycle collector's [`for_each_child_id`] to
/// walk edges but is not part of collection: nothing here mutates the heap, and
/// it is reached only from `run_ref_counts`, never from [`Heap::collect_cycles`].
#[cfg(feature = "ref-count-return")]
impl HeapReader<'_> {
    /// Returns live heap entries unreachable from `roots`, with each entry's type
    /// so callers holding `Interns` can name it ([`Type::name`]).
    ///
    /// `run_ref_counts` uses this to prove a test leaked nothing: an entry that
    /// is alive but reachable from no named variable is a missed `drop_with`.
    /// The walk is transitive, so an object owned by another object — a class's
    /// `__annotations__`, a nested list, an instance attribute — is accounted
    /// for by its owner and need not be bound to a name by the test itself.
    ///
    /// Unlike the cycle collector's [`mark_gray`](Self::mark_gray) this touches
    /// no colors or refcounts, so it cannot perturb the state under test.
    pub(crate) fn unreachable_entries(&self, roots: impl IntoIterator<Item = HeapId>) -> Vec<(HeapId, Type)> {
        let mut seen: HashSet<HeapId> = HashSet::new();
        let mut work_stack: Vec<HeapId> = Vec::new();
        for root in roots {
            if seen.insert(root) {
                work_stack.push(root);
            }
        }

        // A root's subtree is live by construction, but `try_entry` keeps the
        // walk total in case a test observes the heap mid-teardown.
        while let Some(id) = work_stack.pop() {
            let ptr = self.read_ptr(id);
            if ptr.try_entry(self).is_none() {
                continue;
            }
            for_each_child_id(ptr.data(self), |child_id| {
                if seen.insert(child_id) {
                    work_stack.push(child_id);
                }
            });
        }

        self.entries
            .iter()
            // The empty tuple singleton is an internal optimisation that is
            // always live and never named, so it is not a leak. See `entry_count`.
            .filter(|(id, _)| *id != EMPTY_TUPLE_ID && !seen.contains(id))
            .map(|(id, _)| (id, self.read_ptr(id).data(self).py_type()))
            .collect()
    }
}

// With `memory-model-checks` enabled, need to manually clean up the heap to avoid the
// bookkeeping causing panics at shutdown.
#[cfg(feature = "memory-model-checks")]
impl Drop for Heap {
    fn drop(&mut self) {
        for id in 0..self.entries.len() {
            if let Some(mut entry) = self.entries.entry(HeapId::from_index(id)) {
                // Mark all `Value::Ref` payloads as `Dereferenced` so they don't panic when dropped
                py_dec_ref_ids_for_data(entry.get_mut().data.0.get_mut(), &mut Vec::new());
                entry.free();
            }
        }
    }
}

/// Walks the GC-relevant children of a `HeapData` value and calls `on_child`
/// for each contained `HeapId`.
///
/// The cycle collector's mark/scan phases use this directly, combining the
/// child-id iteration with a `HeapId → HeapPtr` conversion in the closure
/// body — that's why this is closure-shaped rather than producing a `Vec`.
/// [`py_dec_ref_ids_for_data`] mirrors the same match arms for the
/// freeing/decref paths; the two must stay in sync.
fn for_each_child_id<F: FnMut(HeapId)>(data: &HeapData, mut on_child: F) {
    match data {
        HeapData::List(list) => {
            // Skip iteration if no refs - major GC optimization for lists of primitives
            if !list.contains_refs() {
                return;
            }
            for value in list.as_slice() {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        // MUST report exactly the same ids as `Deque::py_dec_ref_ids` — reporting
        // fewer leaks, reporting more is a use-after-free.
        HeapData::Deque(deque) => {
            if !deque.contains_refs() {
                return;
            }
            for value in deque.iter() {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        HeapData::Tuple(tuple) => {
            // Skip iteration if no refs - GC optimization for tuples of primitives
            if !tuple.contains_refs() {
                return;
            }
            for value in tuple.as_slice() {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        HeapData::NamedTuple(nt) => {
            // Report the owned reference to the class object (factory instances
            // only) before the `contains_refs` early-out — that flag is computed
            // from `items` alone, so an all-primitive instance like `Point(1, 2)`
            // still has a live class edge. MUST report exactly the same ids as
            // `NamedTuple::py_dec_ref_ids`.
            if let Some(class_id) = nt.class_id() {
                on_child(class_id);
            }
            // Skip iteration if no refs - GC optimization for namedtuples of primitives
            if !nt.contains_refs() {
                return;
            }
            for value in nt.as_vec() {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        // A namedtuple *class* owns its default values and its `__module__`.
        // MUST report the same ids as `NamedTupleClass::py_dec_ref_ids`.
        HeapData::NamedTupleClass(class) => {
            if !class.contains_refs() {
                return;
            }
            for value in class.defaults().iter().chain(once(class.module())) {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        HeapData::Dict(dict) => {
            // Report the default_factory (a defaultdict with a heap-ref factory,
            // e.g. a lambda) before the `has_refs` early-out. MUST report exactly
            // the same ids as `Dict::py_dec_ref_ids`: reporting fewer leaks,
            // reporting more is a use-after-free.
            if let Some(Value::Ref(id)) = dict.default_factory() {
                on_child(*id);
            }
            // Skip iteration if no refs - major GC optimization for dicts of primitives
            if !dict.has_refs() {
                return;
            }
            for (k, v) in dict {
                if let Value::Ref(id) = k {
                    on_child(*id);
                }
                if let Value::Ref(id) = v {
                    on_child(*id);
                }
            }
        }
        HeapData::DictKeysView(view) => {
            on_child(view.dict_id());
        }
        HeapData::DictItemsView(view) => {
            on_child(view.dict_id());
        }
        HeapData::DictValuesView(view) => {
            on_child(view.dict_id());
        }
        HeapData::Set(set) => {
            for value in set.storage().iter() {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        HeapData::FrozenSet(frozenset) => {
            for value in frozenset.storage().iter() {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        HeapData::Closure(closure) => {
            // Add captured cells to work list
            for cell_id in &closure.cells {
                on_child(*cell_id);
            }
            // Add default values that are heap references
            for default in &closure.defaults {
                if let Value::Ref(id) = default {
                    on_child(*id);
                }
            }
        }
        HeapData::FunctionDefaults(fd) => {
            // Add default values that are heap references
            for default in &fd.defaults {
                if let Value::Ref(id) = default {
                    on_child(*id);
                }
            }
        }
        HeapData::Cell(cell) => {
            // Cell can contain a reference to another heap value
            if let Value::Ref(id) = &cell.0 {
                on_child(*id);
            }
        }
        HeapData::Dataclass(dc) => {
            // Dataclass attrs are stored in a Dict - iterate through entries
            for (k, v) in dc.attrs() {
                if let Value::Ref(id) = k {
                    on_child(*id);
                }
                if let Value::Ref(id) = v {
                    on_child(*id);
                }
            }
        }
        HeapData::Class(class) => {
            // The class namespace holds method/class-variable values.
            for (k, v) in class.namespace() {
                if let Value::Ref(id) = k {
                    on_child(*id);
                }
                if let Value::Ref(id) = v {
                    on_child(*id);
                }
            }
        }
        HeapData::Instance(instance) => {
            // An instance references its class plus its attribute dict's entries.
            on_child(instance.class());
            for (k, v) in instance.attrs() {
                if let Value::Ref(id) = k {
                    on_child(*id);
                }
                if let Value::Ref(id) = v {
                    on_child(*id);
                }
            }
        }
        HeapData::BoundMethod(bm) => {
            if let Value::Ref(id) = &bm.instance {
                on_child(*id);
            }
            if let Value::Ref(id) = &bm.func {
                on_child(*id);
            }
        }
        HeapData::DataclassField(field) => {
            // A captured default can reach back to the class the field belongs
            // to (`x: object = SomeInstanceOfIt`), closing a cycle.
            if let Value::Ref(id) = field.annotation() {
                on_child(*id);
            }
            if let Some(Value::Ref(id)) = field.default() {
                on_child(*id);
            }
        }
        HeapData::ListIterator(iter) => on_child(iter.list_id()),
        HeapData::DequeIterator(iter) => on_child(iter.deque_id()),
        HeapData::TupleIterator(iter) => on_child(iter.source_id()),
        HeapData::StringIterator(iter) => {
            if let Some(id) = iter.source_id() {
                on_child(id);
            }
        }
        HeapData::BytesIterator(iter) => {
            if let Some(id) = iter.source_id() {
                on_child(id);
            }
        }
        HeapData::RangeIterator(_) => {}
        HeapData::DictKeyIterator(iter) => on_child(iter.source_id()),
        HeapData::DictItemIterator(iter) => on_child(iter.source_id()),
        HeapData::DictValueIterator(iter) => on_child(iter.source_id()),
        HeapData::SetIterator(iter) => on_child(iter.source_id()),
        HeapData::CallableIterator(iter) => iter.for_each_child_id(on_child),
        HeapData::Itertools(iter) => iter.for_each_child_id(on_child),
        HeapData::Module(m) => {
            // Module attrs can contain references to heap values
            if !m.has_refs() {
                return;
            }
            for (k, v) in m.attrs() {
                if let Value::Ref(id) = k {
                    on_child(*id);
                }
                if let Value::Ref(id) = v {
                    on_child(*id);
                }
            }
        }
        HeapData::Coroutine(coro) => {
            // Add namespace values that are heap references
            for value in &coro.namespace {
                if let Value::Ref(id) = value {
                    on_child(*id);
                }
            }
        }
        HeapData::GatherFuture(gather) => {
            // Add inc_ref'd item HeapIds. Both coroutines and external
            // futures are owned by the gather for its entire lifecycle.
            for item in &gather.items {
                on_child(*item);
            }
            // Walk per-state heap refs: in-flight slot results plus this
            // gather's own awaiter (if `GatherSlot`, it owns an inc_ref on
            // the outer gather), or the cached result list once the gather
            // has completed successfully. `Pending` and `Failed` carry no
            // heap refs.
            match &gather.state {
                GatherState::Awaited(awaited) => {
                    if let Awaiter::GatherSlot { gather, .. } = &awaited.awaiter {
                        on_child(*gather);
                    }
                    for result in awaited.results.iter().flatten() {
                        if let Value::Ref(id) = result {
                            on_child(*id);
                        }
                    }
                }
                GatherState::Completed(Value::Ref(id)) => on_child(*id),
                GatherState::Pending | GatherState::Failed(_) | GatherState::Completed(_) => {}
            }
        }
        HeapData::ExternalFuture(fut) => {
            // `Pending { awaiter: Some(GatherSlot { gather, .. }) }` owns an
            // inc_ref on `gather`. `Awaiter::Task` / `None` and the `Failed`
            // state carry no heap refs. `Resolved` owns the cached value.
            match &fut.state {
                ExternalFutureState::Resolved(Value::Ref(id)) => on_child(*id),
                ExternalFutureState::Pending {
                    awaiter: Some(Awaiter::GatherSlot { gather, .. }),
                } => on_child(*gather),
                _ => {}
            }
        }
        HeapData::DateTime(dt) => {
            // Aware datetimes retain a heap reference to the tzinfo object so that
            // `dt.tzinfo is tz` identity is preserved across attribute lookups.
            // GC must follow that reference, otherwise the timezone gets swept
            // while the datetime still points at the freed slot.
            if let Some(tz_id) = dt.tzinfo_ref() {
                on_child(tz_id);
            }
        }
        HeapData::OpenFile(file) => {
            // Kept in sync with `py_dec_ref_ids_for_data`: the file owns one
            // ref on its loaded buffer. (`OpenFile` is not GC-tracked today, so
            // this arm is not reached by the collector, but the two walkers must
            // mirror each other per the contract above.)
            if let Some(buffer_id) = file.buffer_id() {
                on_child(buffer_id);
            }
        }
        HeapData::ReMatch(m) => {
            // Mirror `py_dec_ref_ids_for_data`: a match holds one ref on its
            // shared subject string (`None` for an interned subject).
            if let Value::Ref(id) = m.subject_ref() {
                on_child(*id);
            }
        }
        // Leaf types with no heap references
        _ => {}
    }
}

fn py_dec_ref_ids_for_data(data: &mut HeapData, stack: &mut Vec<HeapId>) {
    match data {
        HeapData::Str(s) => s.py_dec_ref_ids(stack),
        HeapData::Bytes(b) => b.py_dec_ref_ids(stack),
        HeapData::List(l) => l.py_dec_ref_ids(stack),
        HeapData::Deque(d) => d.py_dec_ref_ids(stack),
        HeapData::Tuple(t) => t.py_dec_ref_ids(stack),
        HeapData::NamedTuple(nt) => nt.py_dec_ref_ids(stack),
        HeapData::NamedTupleClass(class) => class.py_dec_ref_ids(stack),
        HeapData::Dict(d) => d.py_dec_ref_ids(stack),
        HeapData::DictKeysView(view) => view.py_dec_ref_ids(stack),
        HeapData::DictItemsView(view) => view.py_dec_ref_ids(stack),
        HeapData::DictValuesView(view) => view.py_dec_ref_ids(stack),
        HeapData::Set(s) => s.py_dec_ref_ids(stack),
        HeapData::FrozenSet(fs) => fs.py_dec_ref_ids(stack),
        HeapData::Closure(closure) => {
            // Decrement ref count for captured cells
            stack.extend(closure.cells.iter().copied());
            // Decrement ref count for default values that are heap references
            for default in &mut closure.defaults {
                default.py_dec_ref_ids(stack);
            }
        }
        HeapData::FunctionDefaults(fd) => {
            // Decrement ref count for default values that are heap references
            for default in &mut fd.defaults {
                default.py_dec_ref_ids(stack);
            }
        }
        HeapData::Cell(cell) => cell.0.py_dec_ref_ids(stack),
        HeapData::Dataclass(dc) => dc.py_dec_ref_ids(stack),
        HeapData::Class(class) => class.py_dec_ref_ids(stack),
        HeapData::Instance(instance) => instance.py_dec_ref_ids(stack),
        HeapData::BoundMethod(bm) => bm.py_dec_ref_ids(stack),
        HeapData::DataclassField(field) => field.py_dec_ref_ids(stack),
        HeapData::ListIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::DequeIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::TupleIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::StringIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::BytesIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::RangeIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::DictKeyIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::DictItemIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::DictValueIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::SetIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::CallableIterator(iter) => iter.py_dec_ref_ids(stack),
        HeapData::Itertools(iter) => iter.py_dec_ref_ids(stack),
        HeapData::Module(m) => m.py_dec_ref_ids(stack),
        HeapData::Coroutine(coro) => {
            // Decrement ref count for namespace values that are heap references
            for value in &mut coro.namespace {
                value.py_dec_ref_ids(stack);
            }
        }
        HeapData::GatherFuture(gather) => {
            // Decrement ref count for owned item HeapIds (coroutines and
            // external futures are both owned by the gather).
            stack.extend(gather.items.iter().copied());
            // Release per-state heap refs: in-flight slot results plus this
            // gather's own awaiter (if `GatherSlot`, it owns an inc_ref on
            // the outer gather), or the cached result list once the gather
            // has completed successfully. `Pending` and `Failed` carry no
            // heap refs.
            match &mut gather.state {
                GatherState::Awaited(awaited) => {
                    if let Awaiter::GatherSlot { gather, .. } = &awaited.awaiter {
                        stack.push(*gather);
                    }
                    for result in awaited.results.iter_mut().flatten() {
                        result.py_dec_ref_ids(stack);
                    }
                }
                GatherState::Completed(value) => value.py_dec_ref_ids(stack),
                GatherState::Pending | GatherState::Failed(_) => {}
            }
        }
        HeapData::ExternalFuture(fut) => match &mut fut.state {
            ExternalFutureState::Resolved(value) => value.py_dec_ref_ids(stack),
            ExternalFutureState::Pending {
                awaiter: Some(Awaiter::GatherSlot { gather, .. }),
            } => stack.push(*gather),
            ExternalFutureState::Pending {
                awaiter: None | Some(Awaiter::Task(_)),
            }
            | ExternalFutureState::Failed(_) => {}
        },
        HeapData::DateTime(dt) => {
            // Mirror `for_each_child_id`: when an aware datetime is freed we must
            // also drop the retained tzinfo reference so its refcount is balanced.
            if let Some(tz_id) = dt.tzinfo_ref() {
                stack.push(tz_id);
            }
        }
        HeapData::OpenFile(f) => {
            // Kept in sync with `for_each_child_id`: release the file's owned
            // ref on its loaded buffer when the file is freed (e.g. read but
            // never `close()`d).
            f.py_dec_ref_ids(stack);
        }
        // Release the shared subject reference (mirrors `for_each_child_id`).
        HeapData::ReMatch(m) => m.py_dec_ref_ids(stack),
        // other types have no nested heap references
        _ => {}
    }
}

/// Compile-fail soundness tests for [`HeapReader`].
///
/// Gated behind `--cfg heap_reader_compile_fail_tests` so they are only compiled
/// when the integration test harness runs `cargo check` with the appropriate flags.
#[cfg(heap_reader_compile_fail_tests)]
#[path = "../../tests/heap_reader_compile_fail_cases/cases.rs"]
mod heap_reader_compile_fail_cases;

/// Cycle-collector unit tests.
///
/// These live inside `heap.rs` (rather than under `crates/monty/tests/`)
/// because they need to manipulate `Heap` state directly — building a cycle
/// without a VM, peeking at `purple_count`, and rooting an entry only via a
/// Rust local binding. The integration-test surface only exposes
/// Python-driven execution and cannot construct any of those scenarios.
///
/// In particular, the [`cstack_only_cycle_survives_collection`] test
/// validates the central correctness property of trial deletion: a heap
/// entry referenced *only* from the Rust C stack survives a cycle
/// collection because its non-zero refcount is itself proof of liveness.
/// That behavior was previously a known soundness gap of the explicit-roots
/// mark–sweep collector.
#[cfg(test)]
mod tests {
    use monty_types::ResourceTracker;

    use super::*;
    use crate::{
        types::{List, callable_iterator::CallableIterator},
        value::Value,
    };

    /// Returns whether a heap entry is still allocated at `id`.
    fn is_alive(heap: &Heap, id: HeapId) -> bool {
        heap.entries.iter().any(|(other, _)| other == id)
    }

    /// Allocates a self-referencing one-element list and returns its id.
    ///
    /// The list's items become `[Value::Ref(id)]` and its refcount is bumped
    /// to 2 to reflect both the caller's ref and the new self-reference.
    fn alloc_self_cycle(heap: &Heap) -> HeapId {
        let id = heap.allocate(HeapData::List(List::new(vec![])));
        let entry = heap
            .entries
            .iter()
            .find(|(other, _)| *other == id)
            .map(|(_, e)| e)
            .expect("entry just allocated");
        // SAFETY: no other borrow into this entry's data exists during the test.
        let data = unsafe { &mut *entry.data.0.get() };
        match data {
            HeapData::List(list) => {
                list.set_contains_refs();
                list.as_vec_mut().push(Value::Ref(id));
            }
            _ => unreachable!(),
        }
        // The new self-pointer counts as one more reference into the entry.
        heap.inc_ref(id);
        id
    }

    /// Allocates a two-element cycle where one direction has multiplicity 3:
    /// `P → [A, A, A]` and `A → [P]`. Returns `(p_id, a_id)`.
    ///
    /// Final refcounts: `P.rc = 2` (alloc + one edge from A), `A.rc = 4`
    /// (alloc + three edges from P). The caller "owns" one of each — both
    /// can be dropped via `dec_ref` to isolate the cycle.
    ///
    /// Exercises the duplicate-edges-within-one-element shape that
    /// `mark_gray`/`mark_black` must handle correctly: each edge from P
    /// to A is one independent decrement (or increment), but the work
    /// stack must not grow with edge multiplicity — otherwise the outer
    /// pop processes A's children multiple times and over-counts.
    fn alloc_dup_child_cycle(heap: &Heap) -> (HeapId, HeapId) {
        let p_id = heap.allocate(HeapData::List(List::new(vec![])));
        let a_id = heap.allocate(HeapData::List(List::new(vec![])));

        let push_refs = |target: HeapId, refs: &[HeapId]| {
            let entry = heap
                .entries
                .iter()
                .find(|(other, _)| *other == target)
                .map(|(_, e)| e)
                .expect("entry just allocated");
            // SAFETY: no other borrow into this entry's data exists during the test.
            let data = unsafe { &mut *entry.data.0.get() };
            match data {
                HeapData::List(list) => {
                    list.set_contains_refs();
                    for r in refs {
                        list.as_vec_mut().push(Value::Ref(*r));
                    }
                }
                _ => unreachable!(),
            }
            for r in refs {
                heap.inc_ref(*r);
            }
        };

        push_refs(p_id, &[a_id, a_id, a_id]);
        push_refs(a_id, &[p_id]);
        (p_id, a_id)
    }

    #[test]
    fn cstack_only_cycle_survives_collection() {
        let mut heap = Heap::new(16, ResourceTracker::default());
        let id = alloc_self_cycle(&heap);

        // Simulate a Rust-side local `Value::Ref` binding by bumping the
        // refcount one extra time. Then `dec_ref` it back down to 2 — that
        // dec_ref is what enrolls the entry as a Purple candidate, mimicking
        // exactly the situation under the old GC where the local binding
        // wasn't published in any explicit root set.
        heap.inc_ref(id); // rc = 3
        heap.dec_ref(id); // rc = 2, flagged Purple
        assert_eq!(heap.purple_count, 1);

        // Cycle collection must not free the entry: the local "C-stack" ref
        // contributes one of its two surviving refcount units, so trial
        // deletion sees rc > 0 after MarkGray and resurrects the subtree.
        heap.collect_cycles();
        assert_eq!(heap.purple_count, 0);
        assert!(is_alive(&heap, id), "C-stack-rooted cycle was freed");
        assert!(matches!(heap.get(id), HeapData::List(_)));

        // Drop the simulated Rust local. Now the cycle is genuinely isolated
        // (rc 1 = self-pointer only). The next collection must reclaim it.
        heap.dec_ref(id); // rc = 1, re-flagged Purple
        assert_eq!(heap.purple_count, 1);
        heap.collect_cycles();
        assert_eq!(heap.purple_count, 0);
        assert!(!is_alive(&heap, id), "isolated cycle should have been freed");
    }

    #[test]
    fn heap_read_rooted_cycle_survives_collection() {
        let mut heap = Heap::new(16, ResourceTracker::default());
        let id = alloc_self_cycle(&heap);

        // Bump `readers` manually to mimic a live `HeapRead` pointing into
        // the entry. The borrow checker prevents holding a real `HeapRead`
        // across `collect_cycles` (which requires `&mut Heap`), so we
        // splice the same counter that `HeapRead::Drop` decrements.
        let readers_before = heap.entries.get(id).readers.get();
        heap.entries.get(id).readers.set(readers_before + 1);

        // Drive the entry into Purple via dec_ref: rc 2 → 1. Without the
        // `readers > 0` special-case in `Scan`, the resulting cycle would
        // be condemned to White and freed.
        heap.dec_ref(id); // rc = 1, flagged Purple
        assert_eq!(heap.purple_count, 1);

        heap.collect_cycles();
        assert!(
            is_alive(&heap, id),
            "entry with active HeapRead reader was freed by collect_cycles"
        );

        // Restore the simulated reader so `Heap::drop` can clean up
        // without tripping the `dec_ref` active-readers assertion.
        heap.entries.get(id).readers.set(readers_before);
        // The entry is leaked here on purpose (rc = 1 from the self-ref,
        // no external root remains, but the collector ran already and the
        // color is Black — the next dec_ref would try to recurse into the
        // self-pointer after freeing the entry). `Heap::drop` walks every
        // slot and tears them down regardless of refcount, so leaking
        // here is safe for the duration of the test.
    }

    /// Regression test for a soundness gap in [`Heap::dec_ref`]'s trial-deletion
    /// candidate-enrollment path.
    ///
    /// The surviving-refcount branch calls `heap_entry.data.0.get_mut()` to
    /// check `is_gc_tracked`, even though [`HeapData::is_gc_tracked`] only
    /// needs `&self`. That `UnsafeCell::get_mut()` is a fresh `&mut HeapData`
    /// (Unique) retag of the same allocation the live `HeapRead` is already
    /// pointing into via a `SharedReadWrite` raw pointer, so it invalidates
    /// the prior pointer under Stacked / Tree Borrows.
    ///
    /// Pure `cargo test` cannot observe this — the pointer arithmetic still
    /// reads valid bytes — but `cargo +nightly miri test` flags the access.
    /// The branch is reachable from normal Monty code (e.g. `list.remove`
    /// on a self-referential list holds a `HeapRead<List>`, clones the
    /// matching element, and the deferred drop of that clone calls
    /// `dec_ref` on the same `HeapId` while the list reader is still live).
    #[test]
    fn dec_ref_must_not_invalidate_live_heap_read() {
        let mut heap = Heap::new(16, ResourceTracker::default());
        let id = heap.allocate(HeapData::List(List::new(vec![])));
        // Bump refcount so `dec_ref` enters the non-freeing branch where
        // the offending `data.0.get_mut()` lives. `List` is GC-tracked, so
        // `is_gc_tracked` returns true and the branch is fully exercised.
        heap.inc_ref(id);

        HeapReader::with(&mut heap, &mut (), |heap, ()| {
            let HeapReadOutput::List(list) = heap.read(id) else {
                unreachable!()
            };
            // Holding `list` does NOT borrow the heap — only `list.get(heap)`
            // does. That is what lets the borrow checker accept this
            // sequence, while the underlying raw-pointer aliasing is
            // nevertheless violated by the next call.
            heap.dec_ref(id);
            // Read through the now-invalidated `SharedReadWrite` raw pointer.
            // Miri's aliasing model fails here.
            let _ = list.get(heap).as_slice().len();
        });
    }

    #[test]
    fn isolated_simple_cycle_is_collected() {
        // Sanity check: a self-reference cycle with no external rooting
        // gets collected on the next `collect_cycles` call.
        let mut heap = Heap::new(16, ResourceTracker::default());
        let id = alloc_self_cycle(&heap);
        // After alloc_self_cycle: rc = 2 (allocate's 1 + self-ref's 1).
        // Drop the caller's reference. rc 2 → 1, marks Purple.
        heap.dec_ref(id);
        assert_eq!(heap.purple_count, 1);
        heap.collect_cycles();
        assert!(!is_alive(&heap, id));
        assert_eq!(heap.purple_count, 0);
    }

    #[test]
    fn empty_tuple_singleton_survives_collection() {
        // The empty-tuple singleton is no longer rooted explicitly by the
        // collector. Its refcount stays ≥ 1 forever (initial heap-owned
        // ref), which is what keeps it alive — verify the collector does
        // not accidentally free it even after spurious Purple flagging.
        let mut heap = Heap::new(16, ResourceTracker::default());
        // Fake a dec_ref event that would mark the empty tuple Purple.
        heap.inc_ref(EMPTY_TUPLE_ID);
        heap.dec_ref(EMPTY_TUPLE_ID);
        heap.collect_cycles();
        assert!(
            is_alive(&heap, EMPTY_TUPLE_ID),
            "empty tuple singleton must survive collection"
        );
    }

    #[test]
    fn pending_purple_cycle_round_trips_through_serde() {
        // A snapshot can be taken between any two bytecode instructions, so
        // entries flagged Purple by `dec_ref` but not yet visited by the
        // collector must survive serde round-trips. Otherwise a cycle that
        // becomes garbage just before snapshot would leak permanently after
        // restore (the post-restore VM would never re-touch it).
        let mut heap = Heap::new(16, ResourceTracker::default());
        let id = alloc_self_cycle(&heap);
        // Drop the caller's external ref so the entry is genuinely
        // unreachable except via its self-pointer. dec_ref flags Purple.
        heap.dec_ref(id); // rc 2 → 1
        assert_eq!(heap.purple_count, 1);
        assert_eq!(heap.entries.get(id).color.get(), CcColor::Purple);

        // Round-trip through postcard.
        let bytes = postcard::to_allocvec(&heap).expect("serialize");
        let mut restored: Heap = postcard::from_bytes(&bytes).expect("deserialize");

        // `purple_count` and the per-entry color must round-trip.
        assert_eq!(restored.purple_count, 1);
        assert_eq!(restored.entries.get(id).color.get(), CcColor::Purple);

        // Run the collector on the restored heap; the cycle is unreachable
        // and must be reclaimed.
        restored.collect_cycles();
        assert!(!is_alive(&restored, id));
        assert_eq!(restored.purple_count, 0);
    }

    #[test]
    fn isolated_cycle_with_duplicate_child_refs_is_collected() {
        // Regression: an unreachable cycle where one element references its
        // sibling multiple times within its child list. The mark phase must
        // (a) decrement the sibling's refcount once per edge, and (b) push
        // the sibling onto the work stack at most once, even when many
        // edges from the same parent target it.
        //
        // Without (b), the outer pop walks the sibling's children once per
        // duplicate edge, over-decrementing the *sibling's* children's
        // refcounts. In debug builds this trips `mark_gray`'s
        // refcount-underflow `debug_assert`; in release it wraps the
        // refcount to `usize::MAX` and `scan` resurrects the cycle —
        // either way, the cycle is not collected.
        let mut heap = Heap::new(16, ResourceTracker::default());
        let (p_id, a_id) = alloc_dup_child_cycle(&heap);
        // After construction: P.rc = 2, A.rc = 4.

        // Drop the caller's refs; the cycle is now genuinely unreachable.
        heap.dec_ref(p_id); // P.rc 2 → 1, flagged Purple
        heap.dec_ref(a_id); // A.rc 4 → 3, flagged Purple
        assert_eq!(heap.purple_count, 2);

        heap.collect_cycles();

        assert!(!is_alive(&heap, p_id), "P should be collected");
        assert!(!is_alive(&heap, a_id), "A should be collected");
        assert_eq!(heap.purple_count, 0);
    }

    #[test]
    fn cstack_rooted_cycle_with_duplicate_child_refs_collects_after_pin_dropped() {
        // Regression: when `scan` resurrects a Purple cycle via `mark_black`,
        // the per-edge refcount *increment* must mirror `mark_gray`'s
        // per-edge decrement. Duplicate edges from one element to its
        // sibling must not over-increment via repeated outer pops of the
        // sibling.
        //
        // We pin P externally (extra inc_ref) so `scan` is forced to take
        // the resurrect path through `mark_black`. After collection the
        // refcounts must be exactly the pre-collection values — verified by
        // dropping the pin and confirming the next `collect_cycles` reclaims
        // the cycle. If `mark_black` over-incremented during resurrection,
        // dropping the pin leaves the refcounts artificially high and the
        // cycle leaks.
        let mut heap = Heap::new(16, ResourceTracker::default());
        let (p_id, a_id) = alloc_dup_child_cycle(&heap);
        // After construction: P.rc = 2, A.rc = 4.

        // Pin P externally. After the inc_ref + dec_ref pair, P is Purple
        // but its refcount still includes one unit not accounted for by
        // any in-cycle edge — `scan` must resurrect via `mark_black`.
        heap.inc_ref(p_id); // P.rc = 3 (alloc + from A + pin)
        heap.dec_ref(p_id); // P.rc = 2, flagged Purple
        heap.dec_ref(a_id); // A.rc = 3, flagged Purple

        heap.collect_cycles();

        // First pass: cycle survives because of P's external pin.
        assert!(is_alive(&heap, p_id), "P should survive (external pin)");
        assert!(is_alive(&heap, a_id), "A should survive (reachable via P)");

        // Drop the pin. If `mark_black` correctly restored refcounts during
        // resurrection, the cycle is now isolated and the next collection
        // reclaims it. Over-incrementing in `mark_black` leaves P's
        // refcount artificially high so `scan` resurrects it again.
        heap.dec_ref(p_id);
        heap.collect_cycles();

        assert!(!is_alive(&heap, p_id), "P should be collected after pin dropped");
        assert!(!is_alive(&heap, a_id), "A should be collected after pin dropped");
    }

    /// The GC must see BOTH refs a `callable_iterator` owns, with correct
    /// multiplicity: under-tracing here would let a live object be collected.
    #[test]
    fn callable_iterator_traces_callable_and_sentinel() {
        let heap = Heap::new(16, ResourceTracker::default());
        let c = heap.allocate(HeapData::List(List::new(vec![])));
        let s = heap.allocate(HeapData::List(List::new(vec![])));

        heap.inc_ref(c);
        heap.inc_ref(s);
        let iter = heap.allocate(HeapData::CallableIterator(CallableIterator::new(
            Value::Ref(c),
            Value::Ref(s),
        )));

        let mut traced = vec![];
        for_each_child_id(heap.get(iter), |id| traced.push(id));
        assert_eq!(traced, vec![c, s], "callable and sentinel are both traced");

        // Multiplicity: when callable IS sentinel, two counted refs point at one
        // object, so the id must be reported twice or trial deletion
        // under-decrements and frees a live object.
        heap.inc_ref(c);
        heap.inc_ref(c);
        let shared = heap.allocate(HeapData::CallableIterator(CallableIterator::new(
            Value::Ref(c),
            Value::Ref(c),
        )));

        let mut dup = vec![];
        for_each_child_id(heap.get(shared), |id| dup.push(id));
        assert_eq!(dup, vec![c, c], "a shared callable/sentinel is traced twice");
    }

    /// End-to-end: a cycle through the callable must be collected and the
    /// non-cycle sentinel released — exercising the mark phase
    /// (`for_each_child_id`) and the free phase (`py_dec_ref_ids`) together.
    #[test]
    fn callable_iterator_cycle_is_collected() {
        let mut heap = Heap::new(16, ResourceTracker::default());
        let sentinel = heap.allocate(HeapData::List(List::new(vec![])));
        let list = heap.allocate(HeapData::List(List::new(vec![])));
        heap.inc_ref(list);
        heap.inc_ref(sentinel);
        let iter = heap.allocate(HeapData::CallableIterator(CallableIterator::new(
            Value::Ref(list),
            Value::Ref(sentinel),
        )));

        // Close the cycle: the callable list references the iterator back.
        heap.inc_ref(iter);
        HeapReader::with(&mut heap, &mut (), |reader, ()| {
            let HeapReadOutput::List(mut l) = reader.read(list) else {
                unreachable!("just allocated a list")
            };
            let l = l.get_mut(reader);
            l.set_contains_refs();
            l.as_vec_mut().push(Value::Ref(iter));
        });
        // list.rc = 2, iter.rc = 2, sentinel.rc = 2.

        // Drop the allocation refs: {iter, list} is now an unreachable cycle and
        // the sentinel is held only by the iterator.
        heap.dec_ref(list);
        heap.dec_ref(iter);
        heap.dec_ref(sentinel);

        heap.collect_cycles();

        assert!(!is_alive(&heap, iter), "callable_iterator in a cycle must be collected");
        assert!(!is_alive(&heap, list), "the cycle's other node must be collected");
        assert!(
            !is_alive(&heap, sentinel),
            "sentinel held only by the freed iterator must be released"
        );
    }
}