cljrs-gc 0.2.28

//! Process-wide managed-memory governor.
//!
//! See `docs/managed-memory-governor-plan.md`.  Every isolate keeps its private
//! heap and collects it independently; the governor owns only counters,
//! pressure policy, and thread-safe [`IsolateControl`] handles.
//!
//! **Status: Phase 2 (chunked credit, observe-only).**  Each isolate
//! allocates from local credit that the governor grants in chunks of
//! [`MemoryConfig::credit_chunk`]; granted credit counts as committed memory.
//! The governor derives a process [`PressureLevel`] from committed bytes and
//! requests collection from isolates at `Yellow`/`Red`.  It never rejects a
//! grant: a grant that leaves committed bytes above the hard limit is recorded
//! as a *would-reject* event (see [`MemorySnapshot::over_limit_events`]).
//!
//! The hot path ([`IsolateAccount::charge`]) touches only thread-local `Cell`s.
//! It contacts the governor only when local credit runs out.  A refill is the
//! one point where an isolate publishes its used bytes, so per-isolate
//! metrics lag by less than one chunk.
//! Committed bytes are exact: they change only when credit is granted or
//! returned.

use std::cell::{Cell, OnceCell};
use std::fmt;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, PoisonError, Weak};
use std::time::Duration;

const KIB: usize = 1024;
const MIB: usize = 1024 * 1024;

/// Default isolate credit chunk.
pub const DEFAULT_CREDIT_CHUNK: usize = 64 * KIB;
/// Default number of unused credit chunks an isolate keeps after a
/// collection at `Green` pressure.
pub const DEFAULT_RETAINED_CREDIT_CHUNKS: usize = 2;
/// Granularity of a large credit request (one that exceeds a chunk).
pub const ACCOUNTING_UNIT: usize = 4 * KIB;
/// Floor for the post-collection headroom below which the governor does not
/// request collection at `Yellow`.
pub const MIN_COLLECTION_HEADROOM: usize = MIB;
/// Floor for the default critical reserve.
pub const MIN_DEFAULT_CRITICAL_RESERVE: usize = 4 * MIB;
/// Ceiling for the default critical reserve.
pub const MAX_DEFAULT_CRITICAL_RESERVE: usize = 256 * MIB;
/// Ceiling for the default per-channel queue limit.
pub const DEFAULT_QUEUE_LIMIT: usize = 64 * MIB;

/// Process-wide soft limit, in MiB.
pub const SOFT_LIMIT_ENV: &str = "CLJRS_MEMORY_SOFT_LIMIT_MB";
/// Process-wide hard limit, in MiB.
pub const HARD_LIMIT_ENV: &str = "CLJRS_MEMORY_HARD_LIMIT_MB";
/// Critical reserve for runtime control operations, in MiB.
pub const CRITICAL_RESERVE_ENV: &str = "CLJRS_MEMORY_CRITICAL_RESERVE_MB";
/// Isolate credit chunk, in KiB.
pub const CREDIT_ENV: &str = "CLJRS_MEMORY_CREDIT_KB";
/// Default byte limit for one isolate channel, in MiB.
pub const QUEUE_LIMIT_ENV: &str = "CLJRS_ISOLATE_QUEUE_LIMIT_MB";
/// Deprecated alias for [`SOFT_LIMIT_ENV`].
pub const DEPRECATED_SOFT_LIMIT_ENV: &str = "CLJRS_GC_SOFT_LIMIT_MB";
/// Deprecated alias for [`HARD_LIMIT_ENV`].
pub const DEPRECATED_HARD_LIMIT_ENV: &str = "CLJRS_GC_HARD_LIMIT_MB";

// ── Memory classes and pressure ──────────────────────────────────────────────

/// Owner and lifetime rule for a governed allocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum MemoryClass {
    /// Objects in an isolate's private tracing heap.
    GcHeap,
    /// Bump-region chunks (active and retired).
    Region,
    /// Shared immutable values (`ByteBlob`, `SharedAtom` storage, …).
    SharedValue,
    /// Serialized messages queued on isolate channels.
    MessageQueue,
    /// Program-lifetime static allocations.
    Static,
    /// JIT code and code caches.
    Code,
    /// Other runtime-owned data.
    Runtime,
}

const CLASS_COUNT: usize = 7;

impl MemoryClass {
    /// Every memory class, in reporting order.
    pub const ALL: [MemoryClass; CLASS_COUNT] = [
        MemoryClass::GcHeap,
        MemoryClass::Region,
        MemoryClass::SharedValue,
        MemoryClass::MessageQueue,
        MemoryClass::Static,
        MemoryClass::Code,
        MemoryClass::Runtime,
    ];

    /// Static, allocation-free name used in diagnostics and errors.
    pub const fn name(self) -> &'static str {
        match self {
            MemoryClass::GcHeap => "gc-heap",
            MemoryClass::Region => "region",
            MemoryClass::SharedValue => "shared-value",
            MemoryClass::MessageQueue => "message-queue",
            MemoryClass::Static => "static",
            MemoryClass::Code => "code",
            MemoryClass::Runtime => "runtime",
        }
    }

    const fn index(self) -> usize {
        self as usize
    }
}

/// Process memory pressure, derived from committed bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[repr(u8)]
pub enum PressureLevel {
    /// Below the soft limit.
    Green = 0,
    /// At or above the soft limit.
    Yellow = 1,
    /// Above the hard limit (strict mode would reject new application credit).
    Red = 2,
}

impl PressureLevel {
    fn from_u8(v: u8) -> Self {
        match v {
            0 => PressureLevel::Green,
            1 => PressureLevel::Yellow,
            _ => PressureLevel::Red,
        }
    }

    /// Classify `committed` bytes against the given limits.
    pub fn classify(committed: usize, soft_limit: usize, hard_limit: usize) -> Self {
        if committed > hard_limit {
            PressureLevel::Red
        } else if committed >= soft_limit {
            PressureLevel::Yellow
        } else {
            PressureLevel::Green
        }
    }
}

impl fmt::Display for PressureLevel {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            PressureLevel::Green => "Green",
            PressureLevel::Yellow => "Yellow",
            PressureLevel::Red => "Red",
        })
    }
}

// ── Configuration ────────────────────────────────────────────────────────────

/// Process-wide managed-memory limits.  Read once per process.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryConfig {
    /// Committed bytes at which pressure becomes `Yellow`.
    pub soft_limit: usize,
    /// Committed bytes above which pressure becomes `Red`.
    pub hard_limit: usize,
    /// Capacity reserved for collection, shutdown, and error reporting.
    pub critical_reserve: usize,
    /// Normal isolate credit chunk.
    pub credit_chunk: usize,
    /// Unused credit chunks an isolate keeps after a collection at `Green`.
    /// `Yellow` keeps at most one and `Red` keeps none.
    pub retained_credit_chunks: usize,
    /// Default byte limit for one isolate channel.
    pub queue_limit: usize,
}

/// A rejected [`MemoryConfig`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MemoryConfigError {
    /// The hard limit is zero.
    ZeroHardLimit,
    /// The soft limit is greater than the hard limit.
    SoftAboveHard {
        soft_limit: usize,
        hard_limit: usize,
    },
    /// The critical reserve is greater than the hard limit.
    ReserveAboveHard {
        critical_reserve: usize,
        hard_limit: usize,
    },
    /// The credit chunk is zero.
    ZeroCreditChunk,
    /// The queue limit is greater than the application budget.
    QueueAboveBudget {
        queue_limit: usize,
        hard_limit: usize,
    },
    /// An environment variable is not a non-negative integer.
    InvalidNumber { var: &'static str, value: String },
}

impl fmt::Display for MemoryConfigError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            MemoryConfigError::ZeroHardLimit => write!(f, "the hard memory limit must be non-zero"),
            MemoryConfigError::SoftAboveHard {
                soft_limit,
                hard_limit,
            } => write!(
                f,
                "the soft memory limit ({soft_limit} bytes) is greater than the hard limit ({hard_limit} bytes)"
            ),
            MemoryConfigError::ReserveAboveHard {
                critical_reserve,
                hard_limit,
            } => write!(
                f,
                "the critical reserve ({critical_reserve} bytes) is greater than the hard limit ({hard_limit} bytes)"
            ),
            MemoryConfigError::ZeroCreditChunk => write!(f, "the credit chunk must be non-zero"),
            MemoryConfigError::QueueAboveBudget {
                queue_limit,
                hard_limit,
            } => write!(
                f,
                "the isolate queue limit ({queue_limit} bytes) is greater than the hard limit ({hard_limit} bytes)"
            ),
            MemoryConfigError::InvalidNumber { var, value } => {
                write!(f, "{var}={value:?} is not a non-negative integer")
            }
        }
    }
}

impl std::error::Error for MemoryConfigError {}

impl MemoryConfig {
    /// Build a config from explicit limits; the remaining fields take their
    /// defaults for `hard_limit`.
    pub fn with_limits(soft_limit: usize, hard_limit: usize) -> Result<Self, MemoryConfigError> {
        let config = Self {
            soft_limit,
            hard_limit,
            critical_reserve: default_critical_reserve(hard_limit),
            credit_chunk: DEFAULT_CREDIT_CHUNK,
            retained_credit_chunks: DEFAULT_RETAINED_CREDIT_CHUNKS,
            queue_limit: DEFAULT_QUEUE_LIMIT.min(hard_limit),
        };
        config.validate()?;
        Ok(config)
    }

    /// Build a config from optional soft and hard limits.
    ///
    /// A missing hard limit uses [`default_hard_limit`], raised to the soft
    /// limit when the soft limit is larger.  A missing soft limit is 75% of
    /// the hard limit.
    pub fn from_optional_limits(
        soft_limit: Option<usize>,
        hard_limit: Option<usize>,
    ) -> Result<Self, MemoryConfigError> {
        let hard = match (soft_limit, hard_limit) {
            (_, Some(hard)) => hard,
            (Some(soft), None) => default_hard_limit().max(soft),
            (None, None) => default_hard_limit(),
        };
        let soft = soft_limit.unwrap_or_else(|| default_soft_limit(hard));
        Self::with_limits(soft, hard)
    }

    /// Platform defaults: container limit or a fraction of physical memory.
    pub fn platform_default() -> Self {
        let hard = default_hard_limit().max(1);
        Self::with_limits(default_soft_limit(hard), hard)
            .expect("platform default memory config is valid")
    }

    /// Check the invariants the plan requires of a config.
    pub fn validate(&self) -> Result<(), MemoryConfigError> {
        if self.hard_limit == 0 {
            return Err(MemoryConfigError::ZeroHardLimit);
        }
        if self.soft_limit > self.hard_limit {
            return Err(MemoryConfigError::SoftAboveHard {
                soft_limit: self.soft_limit,
                hard_limit: self.hard_limit,
            });
        }
        if self.critical_reserve > self.hard_limit {
            return Err(MemoryConfigError::ReserveAboveHard {
                critical_reserve: self.critical_reserve,
                hard_limit: self.hard_limit,
            });
        }
        if self.credit_chunk == 0 {
            return Err(MemoryConfigError::ZeroCreditChunk);
        }
        if self.queue_limit > self.hard_limit {
            return Err(MemoryConfigError::QueueAboveBudget {
                queue_limit: self.queue_limit,
                hard_limit: self.hard_limit,
            });
        }
        Ok(())
    }

    /// Read the process config from the environment.
    ///
    /// The deprecated `CLJRS_GC_*_LIMIT_MB` names set process-wide values; a
    /// new variable wins over its old alias.  Each old variable in use
    /// produces one warning on stderr.
    pub fn from_env() -> Result<Self, MemoryConfigError> {
        Self::from_lookup(|var| std::env::var(var).ok(), |msg| eprintln!("{msg}"))
    }

    /// [`Self::from_env`] over an arbitrary variable lookup (for tests).
    pub fn from_lookup(
        lookup: impl Fn(&str) -> Option<String>,
        mut warn: impl FnMut(String),
    ) -> Result<Self, MemoryConfigError> {
        let mut limit = |new: &'static str, old: &'static str| -> Result<Option<usize>, _> {
            let new_val = parse_env(new, lookup(new), MIB)?;
            if let Some(old_val) = lookup(old) {
                if new_val.is_some() {
                    warn(format!(
                        "[memory] warning: {old} is deprecated and ignored because {new} is set"
                    ));
                    return Ok(new_val);
                }
                warn(format!(
                    "[memory] warning: {old} is deprecated; use {new} (it sets a process-wide limit)"
                ));
                return parse_env(old, Some(old_val), MIB);
            }
            Ok(new_val)
        };
        let soft = limit(SOFT_LIMIT_ENV, DEPRECATED_SOFT_LIMIT_ENV)?;
        let hard = limit(HARD_LIMIT_ENV, DEPRECATED_HARD_LIMIT_ENV)?;
        let mut config = Self::from_optional_limits(soft, hard)?;
        if let Some(reserve) = parse_env(CRITICAL_RESERVE_ENV, lookup(CRITICAL_RESERVE_ENV), MIB)? {
            config.critical_reserve = reserve;
        }
        if let Some(chunk) = parse_env(CREDIT_ENV, lookup(CREDIT_ENV), KIB)? {
            config.credit_chunk = chunk;
        }
        if let Some(queue) = parse_env(QUEUE_LIMIT_ENV, lookup(QUEUE_LIMIT_ENV), MIB)? {
            config.queue_limit = queue;
        }
        config.validate()?;
        Ok(config)
    }
}

fn parse_env(
    var: &'static str,
    value: Option<String>,
    unit: usize,
) -> Result<Option<usize>, MemoryConfigError> {
    match value {
        None => Ok(None),
        Some(raw) => raw
            .trim()
            .parse::<usize>()
            .map(|n| Some(n.saturating_mul(unit)))
            .map_err(|_| MemoryConfigError::InvalidNumber { var, value: raw }),
    }
}

/// Default soft limit: 75% of the hard limit.
pub fn default_soft_limit(hard_limit: usize) -> usize {
    // Round up, so a nonzero hard limit never yields a zero soft limit
    // (which would hold the process at `Yellow` permanently).
    hard_limit - hard_limit / 4
}

/// Default critical reserve: the larger of 4 MiB or 1% of the hard limit,
/// capped at [`MAX_DEFAULT_CRITICAL_RESERVE`] and at the hard limit itself.
pub fn default_critical_reserve(hard_limit: usize) -> usize {
    (hard_limit / 100)
        .clamp(MIN_DEFAULT_CRITICAL_RESERVE, MAX_DEFAULT_CRITICAL_RESERVE)
        .min(hard_limit)
}

/// Default process hard limit.
///
/// Uses the cgroup memory limit when one applies (Linux), otherwise half of
/// physical memory.  `wasm32` uses a fixed 256 MiB.
pub fn default_hard_limit() -> usize {
    #[cfg(target_arch = "wasm32")]
    {
        256 * MIB
    }
    #[cfg(not(target_arch = "wasm32"))]
    {
        let physical = system_memory::total() as usize;
        let half = (physical / 2).max(256 * MIB);
        match container_memory_limit() {
            Some(limit) if limit < physical || physical == 0 => limit,
            _ => half,
        }
    }
}

/// The cgroup (v2, then v1) memory limit of this process, if any.
#[cfg(not(target_arch = "wasm32"))]
pub fn container_memory_limit() -> Option<usize> {
    #[cfg(target_os = "linux")]
    {
        const PATHS: [&str; 2] = [
            "/sys/fs/cgroup/memory.max",
            "/sys/fs/cgroup/memory/memory.limit_in_bytes",
        ];
        PATHS.iter().find_map(|path| {
            let raw = std::fs::read_to_string(path).ok()?;
            let raw = raw.trim();
            if raw == "max" {
                return None;
            }
            // cgroup v1 reports "unlimited" as a value near i64::MAX.
            raw.parse::<u64>()
                .ok()
                .filter(|&n| n > 0 && n < (1u64 << 60))
                .map(|n| n as usize)
        })
    }
    #[cfg(not(target_os = "linux"))]
    {
        None
    }
}

// ── Isolate control ──────────────────────────────────────────────────────────

/// Process-unique isolate identifier.
pub type IsolateId = u64;

/// Thread-safe control data for one registered isolate.
///
/// The governor holds weak references to these in its registry; it can
/// request collection or a credit recall, but it cannot trace or sweep
/// another isolate's heap.
pub struct IsolateControl {
    id: IsolateId,
    name: Mutex<Arc<str>>,
    gc_requested: AtomicBool,
    /// The governor asks the isolate to return its unused credit.
    recall_requested: AtomicBool,
    used_bytes: AtomicUsize,
    peak_used_bytes: AtomicUsize,
    free_credit_bytes: AtomicUsize,
    /// Number of completed collections; the collection epoch.
    last_collection_epoch: AtomicU64,
    /// Epoch at which the governor last requested collection (`u64::MAX`: never).
    requested_epoch: AtomicU64,
    /// Used bytes below which the governor does not request collection at
    /// `Yellow` (set from the last collection's result).
    collection_target: AtomicUsize,
    collection_requests: AtomicU64,
    credit_refills: AtomicU64,
    zero_yield_collections: AtomicU64,
    last_bytes_before: AtomicUsize,
    last_bytes_after: AtomicUsize,
    last_duration_nanos: AtomicU64,
}

impl IsolateControl {
    fn new(id: IsolateId, name: Arc<str>) -> Self {
        Self {
            id,
            name: Mutex::new(name),
            gc_requested: AtomicBool::new(false),
            recall_requested: AtomicBool::new(false),
            used_bytes: AtomicUsize::new(0),
            peak_used_bytes: AtomicUsize::new(0),
            free_credit_bytes: AtomicUsize::new(0),
            last_collection_epoch: AtomicU64::new(0),
            requested_epoch: AtomicU64::new(u64::MAX),
            collection_target: AtomicUsize::new(0),
            collection_requests: AtomicU64::new(0),
            credit_refills: AtomicU64::new(0),
            zero_yield_collections: AtomicU64::new(0),
            last_bytes_before: AtomicUsize::new(0),
            last_bytes_after: AtomicUsize::new(0),
            last_duration_nanos: AtomicU64::new(0),
        }
    }

    /// Process-unique id.
    pub fn id(&self) -> IsolateId {
        self.id
    }

    /// Diagnostic name.
    pub fn name(&self) -> Arc<str> {
        self.name
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }

    fn set_name(&self, name: Arc<str>) {
        *self.name.lock().unwrap_or_else(PoisonError::into_inner) = name;
    }

    /// Ask the isolate to collect at its next safepoint.
    pub fn request_collection(&self) {
        self.gc_requested.store(true, Ordering::Release);
    }

    /// Whether a collection request is pending.
    pub fn collection_requested(&self) -> bool {
        self.gc_requested.load(Ordering::Acquire)
    }

    /// Check and clear the pending collection request.
    pub fn take_collection_request(&self) -> bool {
        self.gc_requested.swap(false, Ordering::AcqRel)
    }

    /// Ask the isolate to return its unused credit at its next runtime
    /// service poll ([`IsolateAccount::poll`]).
    pub fn request_recall(&self) {
        self.recall_requested.store(true, Ordering::Release);
    }

    /// Whether a credit recall is pending.
    pub fn recall_requested(&self) -> bool {
        self.recall_requested.load(Ordering::Acquire)
    }

    /// Last published used bytes.
    pub fn used_bytes(&self) -> usize {
        self.used_bytes.load(Ordering::Relaxed)
    }

    /// Last published unused credit.
    pub fn free_credit_bytes(&self) -> usize {
        self.free_credit_bytes.load(Ordering::Relaxed)
    }

    /// Used bytes below which the governor does not request collection at
    /// `Yellow`.
    pub fn collection_target(&self) -> usize {
        self.collection_target.load(Ordering::Relaxed)
    }

    /// Number of completed collections.
    pub fn collection_epoch(&self) -> u64 {
        self.last_collection_epoch.load(Ordering::Relaxed)
    }

    /// Governor-initiated collection requests for this isolate.
    pub fn collection_requests(&self) -> u64 {
        self.collection_requests.load(Ordering::Relaxed)
    }

    /// Credit refills this isolate has requested.
    pub fn credit_refills(&self) -> u64 {
        self.credit_refills.load(Ordering::Relaxed)
    }

    /// Request collection on the governor's behalf, at most once per
    /// collection epoch.  Below `Red`, also skip the request while the heap
    /// is under its post-collection target.  Returns whether a request was
    /// made.
    fn governor_request(&self, level: PressureLevel) -> bool {
        let epoch = self.collection_epoch();
        let prev = self.requested_epoch.load(Ordering::Relaxed);
        if prev == epoch {
            return false;
        }
        if level < PressureLevel::Red && self.used_bytes() < self.collection_target() {
            return false;
        }
        // `evaluate` can run on several refilling threads at once; only the
        // thread that claims this epoch counts the request.
        if self
            .requested_epoch
            .compare_exchange(prev, epoch, Ordering::Relaxed, Ordering::Relaxed)
            .is_err()
        {
            return false;
        }
        self.collection_requests.fetch_add(1, Ordering::Relaxed);
        self.request_collection();
        true
    }

    fn snapshot(&self) -> IsolateSnapshot {
        IsolateSnapshot {
            id: self.id,
            name: self.name(),
            used_bytes: self.used_bytes(),
            free_credit_bytes: self.free_credit_bytes(),
            peak_used_bytes: self.peak_used_bytes.load(Ordering::Relaxed),
            collection_target: self.collection_target(),
            collections: self.collection_epoch(),
            collection_requests: self.collection_requests(),
            credit_refills: self.credit_refills(),
            zero_yield_collections: self.zero_yield_collections.load(Ordering::Relaxed),
            last_bytes_before: self.last_bytes_before.load(Ordering::Relaxed),
            last_bytes_after: self.last_bytes_after.load(Ordering::Relaxed),
            last_duration: Duration::from_nanos(self.last_duration_nanos.load(Ordering::Relaxed)),
        }
    }
}

impl fmt::Debug for IsolateControl {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("IsolateControl")
            .field("id", &self.id)
            .field("name", &self.name())
            .field("used_bytes", &self.used_bytes())
            .field("free_credit_bytes", &self.free_credit_bytes())
            .finish()
    }
}

/// What one collection did, reported to the isolate's account.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CollectionReport {
    /// Heap bytes before collection.
    pub bytes_before: usize,
    /// Heap bytes after collection.
    pub bytes_after: usize,
    /// Bytes freed by sweep.  They become free credit, and the account
    /// returns the credit above its retained limit to the governor.
    pub bytes_returned: usize,
    /// Mark plus sweep time.
    pub duration: Duration,
}

// ── Isolate account ──────────────────────────────────────────────────────────

/// One isolate's single-threaded view of the governor.
///
/// The account holds `used` bytes (live objects and retained garbage) and
/// `credit` (committed capacity it has not used yet).  Their sum is exactly
/// what the governor has committed for this isolate.
///
/// The counters are `Cell`s, so the account is `!Sync`: only one thread
/// uses it at a time.  It is `Send`, so nothing stops moving it to another
/// thread; the runtime keeps it in a thread-local for the isolate's thread.
///
/// Created by [`ProcessMemoryGovernor::register_isolate`].  Dropping the
/// account returns its used bytes and free credit and unregisters it.
pub struct IsolateAccount {
    governor: &'static ProcessMemoryGovernor,
    control: Arc<IsolateControl>,
    used: Cell<usize>,
    credit: Cell<usize>,
    /// Free credit last added to the governor's free-credit counter.
    published_credit: Cell<usize>,
    /// Allocations not yet added to [`crate::stats::GC_STATS`].
    pending_allocs: Cell<u64>,
    pending_alloc_bytes: Cell<u64>,
}

impl IsolateAccount {
    /// Shared control handle.
    pub fn control(&self) -> &Arc<IsolateControl> {
        &self.control
    }

    /// Exact local used bytes (may be ahead of the published value).
    pub fn used_bytes(&self) -> usize {
        self.used.get()
    }

    /// Exact local unused credit.
    pub fn free_credit_bytes(&self) -> usize {
        self.credit.get()
    }

    /// Bytes the governor has committed for this isolate: used plus free
    /// credit.
    pub fn committed_bytes(&self) -> usize {
        self.used.get() + self.credit.get()
    }

    /// Account `bytes` of new heap allocation.  Thread-local unless local
    /// credit is insufficient, in which case the account refills from the
    /// governor.
    #[inline]
    pub fn charge(&self, bytes: usize) {
        self.used.set(self.used.get() + bytes);
        self.pending_allocs.set(self.pending_allocs.get() + 1);
        self.pending_alloc_bytes
            .set(self.pending_alloc_bytes.get() + bytes as u64);
        match self.credit.get().checked_sub(bytes) {
            Some(rest) => self.credit.set(rest),
            None => self.refill(bytes),
        }
    }

    /// Slow path of [`Self::charge`]: obtain credit for the part of `bytes`
    /// that local credit does not cover.  A shortfall up to one chunk gets a
    /// normal chunk; a larger one gets its own size, rounded up to
    /// [`ACCOUNTING_UNIT`].
    #[cold]
    #[inline(never)]
    fn refill(&self, bytes: usize) {
        let credit = self.credit.get();
        let shortfall = bytes - credit;
        let chunk = self.governor.credit_chunk();
        let grant = if shortfall > chunk {
            shortfall.div_ceil(ACCOUNTING_UNIT) * ACCOUNTING_UNIT
        } else {
            chunk
        };
        self.governor.grant(grant);
        self.control.credit_refills.fetch_add(1, Ordering::Relaxed);
        self.credit.set(credit + grant - bytes);
        if self.control.recall_requested() {
            self.service_recall();
        }
        self.sync();
        self.governor.evaluate(Some(&self.control), true);
    }

    /// Return `bytes` freed by sweep.  They become free credit until the
    /// collection report trims it.
    #[inline]
    pub fn release(&self, bytes: usize) {
        let used = self.used.get();
        debug_assert!(
            bytes <= used,
            "isolate {} released {bytes} bytes with only {used} charged",
            self.control.id
        );
        let bytes = bytes.min(used);
        self.used.set(used - bytes);
        self.credit.set(self.credit.get() + bytes);
    }

    /// Publish the local counters and re-evaluate pressure.  Normal
    /// operation publishes at each refill and collection; this is for
    /// metrics polls.
    pub fn publish(&self) {
        self.sync();
        self.governor.evaluate(Some(&self.control), false);
    }

    /// Runtime service poll: return credit if the governor recalled it,
    /// and report whether a collection is requested.  Called from the
    /// safepoint check, so it must stay cheap when nothing is pending.
    #[inline]
    pub fn poll(&self) -> bool {
        if self.control.recall_requested() {
            self.service_recall();
            self.sync();
            self.governor.evaluate(None, false);
        }
        self.control.collection_requested()
    }

    fn service_recall(&self) {
        self.control
            .recall_requested
            .store(false, Ordering::Release);
        self.trim_credit();
    }

    /// Return the free credit above the retained limit for the current
    /// pressure level.
    fn trim_credit(&self) {
        let limit = self.governor.retained_credit(self.governor.pressure());
        let credit = self.credit.get();
        if credit > limit {
            self.credit.set(limit);
            self.governor.return_credit(credit - limit);
        }
    }

    /// Copy the local counters to the control handle and the governor's
    /// free-credit counter, and flush pending allocation statistics.
    fn sync(&self) {
        let used = self.used.get();
        let credit = self.credit.get();
        self.control.used_bytes.store(used, Ordering::Relaxed);
        self.control
            .peak_used_bytes
            .fetch_max(used, Ordering::Relaxed);
        self.control
            .free_credit_bytes
            .store(credit, Ordering::Relaxed);
        let prev = self.published_credit.replace(credit);
        if credit >= prev {
            self.governor.add_free_credit(credit - prev);
        } else {
            self.governor.sub_free_credit(prev - credit);
        }
        self.flush_alloc_stats();
    }

    fn flush_alloc_stats(&self) {
        let allocs = self.pending_allocs.replace(0);
        let bytes = self.pending_alloc_bytes.replace(0);
        if allocs > 0 {
            crate::stats::GC_STATS.record_gc_allocs(allocs, bytes);
        }
    }

    /// Record a completed collection: bump the epoch, set the next
    /// collection target, return excess free credit, and publish.
    ///
    /// The target, which gates governor requests at `Yellow`, is the
    /// surviving bytes plus the larger of half the surviving bytes or
    /// [`ProcessMemoryGovernor::min_headroom`].  After a collection that
    /// frees nothing, the headroom is the larger of the surviving bytes or
    /// that minimum.  The heap's own collection trigger is its `GcConfig`.
    pub fn record_collection(&self, report: CollectionReport) {
        let c = &self.control;
        c.last_collection_epoch.fetch_add(1, Ordering::Relaxed);
        c.last_bytes_before
            .store(report.bytes_before, Ordering::Relaxed);
        c.last_bytes_after
            .store(report.bytes_after, Ordering::Relaxed);
        c.last_duration_nanos.store(
            report.duration.as_nanos().min(u64::MAX as u128) as u64,
            Ordering::Relaxed,
        );
        let min = self.governor.min_headroom();
        let after = report.bytes_after;
        let headroom = if report.bytes_returned == 0 {
            c.zero_yield_collections.fetch_add(1, Ordering::Relaxed);
            after.max(min)
        } else {
            (after / 2).max(min)
        };
        c.collection_target
            .store(after.saturating_add(headroom), Ordering::Relaxed);
        self.service_recall();
        self.sync();
        self.governor.evaluate(Some(c), false);
    }
}

impl Drop for IsolateAccount {
    fn drop(&mut self) {
        // The isolate heap is not torn down at thread exit yet, so this
        // stops counting bytes that remain allocated until process exit.
        let committed = self.used.replace(0) + self.credit.replace(0);
        let published = self.published_credit.replace(0);
        self.control.used_bytes.store(0, Ordering::Relaxed);
        self.control.free_credit_bytes.store(0, Ordering::Relaxed);
        self.flush_alloc_stats();
        self.governor.sub_free_credit(published);
        self.governor.return_credit(committed);
        self.governor.unregister(&self.control);
        self.governor.evaluate(None, false);
    }
}

// ── Charges for independent-lifetime allocations ─────────────────────────────

/// RAII charge for an allocation with its own lifetime.  Dropping it returns
/// the bytes to the governor exactly once.
#[must_use = "dropping a MemoryCharge returns its bytes immediately"]
pub struct MemoryCharge {
    governor: &'static ProcessMemoryGovernor,
    class: MemoryClass,
    bytes: usize,
}

impl MemoryCharge {
    /// Charged class.
    pub fn class(&self) -> MemoryClass {
        self.class
    }

    /// Charged bytes.
    pub fn bytes(&self) -> usize {
        self.bytes
    }
}

impl Drop for MemoryCharge {
    fn drop(&mut self) {
        self.governor.sub(self.class, self.bytes);
        self.governor.evaluate(None, false);
    }
}

impl fmt::Debug for MemoryCharge {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MemoryCharge")
            .field("class", &self.class)
            .field("bytes", &self.bytes)
            .finish()
    }
}

/// A reservation the governor refused.  Fixed-size; building it never
/// allocates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MemoryLimitExceeded {
    /// Class of the refused reservation.
    pub class: MemoryClass,
    /// Requested bytes.
    pub requested: usize,
    /// Committed bytes at the time of the request.
    pub committed: usize,
    /// Hard limit at the time of the request.
    pub hard_limit: usize,
}

impl fmt::Display for MemoryLimitExceeded {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "managed memory limit exceeded: {} bytes of {} requested with {} of {} bytes committed",
            self.requested,
            self.class.name(),
            self.committed,
            self.hard_limit
        )
    }
}

impl std::error::Error for MemoryLimitExceeded {}

// ── Process governor ─────────────────────────────────────────────────────────

/// Process-wide managed-memory governor.  See the module docs.
pub struct ProcessMemoryGovernor {
    config: Mutex<Option<MemoryConfig>>,
    configured: AtomicBool,
    soft_limit: AtomicUsize,
    hard_limit: AtomicUsize,
    credit_chunk: AtomicUsize,
    retained_credit_chunks: AtomicUsize,
    /// Total committed bytes; the sum of `class_bytes`.
    committed: AtomicUsize,
    class_bytes: [AtomicUsize; CLASS_COUNT],
    /// Unused isolate credit, as last published by each isolate.
    free_credit: AtomicUsize,
    peak_committed: AtomicUsize,
    peak_used: AtomicUsize,
    pressure: AtomicU8,
    pressure_transitions: AtomicU64,
    collection_requests: AtomicU64,
    credit_refills: AtomicU64,
    credit_returned_bytes: AtomicU64,
    credit_recalls: AtomicU64,
    over_limit_events: AtomicU64,
    over_limit_bytes: AtomicU64,
    registry: Mutex<Vec<Weak<IsolateControl>>>,
    next_isolate_id: AtomicU64,
    isolates_registered: AtomicU64,
}

impl ProcessMemoryGovernor {
    /// A governor that reads [`MemoryConfig::from_env`] on first use.
    pub const fn new() -> Self {
        Self {
            config: Mutex::new(None),
            configured: AtomicBool::new(false),
            soft_limit: AtomicUsize::new(usize::MAX),
            hard_limit: AtomicUsize::new(usize::MAX),
            credit_chunk: AtomicUsize::new(DEFAULT_CREDIT_CHUNK),
            retained_credit_chunks: AtomicUsize::new(DEFAULT_RETAINED_CREDIT_CHUNKS),
            committed: AtomicUsize::new(0),
            class_bytes: [const { AtomicUsize::new(0) }; CLASS_COUNT],
            free_credit: AtomicUsize::new(0),
            peak_committed: AtomicUsize::new(0),
            peak_used: AtomicUsize::new(0),
            pressure: AtomicU8::new(PressureLevel::Green as u8),
            pressure_transitions: AtomicU64::new(0),
            collection_requests: AtomicU64::new(0),
            credit_refills: AtomicU64::new(0),
            credit_returned_bytes: AtomicU64::new(0),
            credit_recalls: AtomicU64::new(0),
            over_limit_events: AtomicU64::new(0),
            over_limit_bytes: AtomicU64::new(0),
            registry: Mutex::new(Vec::new()),
            next_isolate_id: AtomicU64::new(1),
            isolates_registered: AtomicU64::new(0),
        }
    }

    /// A governor with an explicit config.
    pub fn with_config(config: MemoryConfig) -> Result<Self, MemoryConfigError> {
        let g = Self::new();
        g.configure(config)?;
        Ok(g)
    }

    /// Replace the process limits and re-evaluate pressure.
    pub fn configure(&self, config: MemoryConfig) -> Result<(), MemoryConfigError> {
        config.validate()?;
        let mut slot = self.config.lock().unwrap_or_else(PoisonError::into_inner);
        self.install(&config);
        *slot = Some(config);
        drop(slot);
        self.evaluate(None, false);
        Ok(())
    }

    fn install(&self, config: &MemoryConfig) {
        self.soft_limit.store(config.soft_limit, Ordering::Relaxed);
        self.hard_limit.store(config.hard_limit, Ordering::Relaxed);
        self.credit_chunk
            .store(config.credit_chunk, Ordering::Relaxed);
        self.retained_credit_chunks
            .store(config.retained_credit_chunks, Ordering::Relaxed);
        self.configured.store(true, Ordering::Release);
    }

    fn ensure_configured(&self) {
        if self.configured.load(Ordering::Acquire) {
            return;
        }
        let mut slot = self.config.lock().unwrap_or_else(PoisonError::into_inner);
        if slot.is_some() {
            return;
        }
        let config = MemoryConfig::from_env().unwrap_or_else(|e| {
            eprintln!("[memory] warning: invalid memory configuration ({e}); using defaults");
            MemoryConfig::platform_default()
        });
        self.install(&config);
        *slot = Some(config);
    }

    /// The active config.
    pub fn config(&self) -> MemoryConfig {
        self.ensure_configured();
        self.config
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
            .expect("configured")
    }

    fn credit_chunk(&self) -> usize {
        self.ensure_configured();
        self.credit_chunk.load(Ordering::Relaxed)
    }

    /// Smallest post-collection headroom: the larger of 16 credit chunks or
    /// [`MIN_COLLECTION_HEADROOM`].
    pub fn min_headroom(&self) -> usize {
        self.credit_chunk()
            .saturating_mul(16)
            .max(MIN_COLLECTION_HEADROOM)
    }

    /// Free credit an isolate keeps after a collection or recall at `level`.
    fn retained_credit(&self, level: PressureLevel) -> usize {
        let chunks = self.retained_credit_chunks.load(Ordering::Relaxed);
        let chunks = match level {
            PressureLevel::Green => chunks,
            PressureLevel::Yellow => chunks.min(1),
            PressureLevel::Red => 0,
        };
        self.credit_chunk().saturating_mul(chunks)
    }

    /// Register an isolate.  The returned account is single-threaded (`!Sync`).
    pub fn register_isolate(&'static self, name: impl Into<Arc<str>>) -> IsolateAccount {
        let id = self.next_isolate_id.fetch_add(1, Ordering::Relaxed);
        let control = Arc::new(IsolateControl::new(id, name.into()));
        {
            let mut reg = self.registry.lock().unwrap_or_else(PoisonError::into_inner);
            reg.retain(|w| w.strong_count() > 0);
            reg.push(Arc::downgrade(&control));
        }
        self.isolates_registered.fetch_add(1, Ordering::Relaxed);
        IsolateAccount {
            governor: self,
            control,
            used: Cell::new(0),
            credit: Cell::new(0),
            published_credit: Cell::new(0),
            pending_allocs: Cell::new(0),
            pending_alloc_bytes: Cell::new(0),
        }
    }

    fn unregister(&self, control: &Arc<IsolateControl>) {
        let mut reg = self.registry.lock().unwrap_or_else(PoisonError::into_inner);
        reg.retain(|w| w.strong_count() > 0 && !std::ptr::eq(w.as_ptr(), Arc::as_ptr(control)));
    }

    /// Reserve `bytes` of `class` for an allocation with an independent
    /// lifetime.  Observe-only: the reservation always succeeds, and an
    /// over-limit reservation is recorded as a would-reject event.
    pub fn reserve_shared(
        &'static self,
        class: MemoryClass,
        bytes: usize,
    ) -> Result<MemoryCharge, MemoryLimitExceeded> {
        self.add(class, bytes);
        self.evaluate(None, bytes > 0);
        Ok(MemoryCharge {
            governor: self,
            class,
            bytes,
        })
    }

    /// Commit `bytes` of isolate credit.  Observe-only: never refused.
    fn grant(&self, bytes: usize) {
        self.add(MemoryClass::GcHeap, bytes);
        self.credit_refills.fetch_add(1, Ordering::Relaxed);
    }

    /// Uncommit `bytes` of isolate credit or used heap bytes.
    fn return_credit(&self, bytes: usize) {
        if bytes > 0 {
            self.sub(MemoryClass::GcHeap, bytes);
            self.credit_returned_bytes
                .fetch_add(bytes as u64, Ordering::Relaxed);
        }
    }

    fn add(&self, class: MemoryClass, bytes: usize) {
        if bytes > 0 {
            self.class_bytes[class.index()].fetch_add(bytes, Ordering::Relaxed);
            self.committed.fetch_add(bytes, Ordering::Relaxed);
        }
    }

    fn sub(&self, class: MemoryClass, bytes: usize) {
        if bytes == 0 {
            return;
        }
        checked_sub(&self.class_bytes[class.index()], bytes, class.name());
        checked_sub(&self.committed, bytes, "committed");
    }

    fn add_free_credit(&self, bytes: usize) {
        if bytes > 0 {
            self.free_credit.fetch_add(bytes, Ordering::Relaxed);
        }
    }

    fn sub_free_credit(&self, bytes: usize) {
        if bytes > 0 {
            checked_sub(&self.free_credit, bytes, "free-credit");
        }
    }

    /// Current bytes in `class`.
    pub fn class_bytes(&self, class: MemoryClass) -> usize {
        self.class_bytes[class.index()].load(Ordering::Relaxed)
    }

    /// Total committed bytes: used bytes plus unused isolate credit.
    pub fn committed_bytes(&self) -> usize {
        self.committed.load(Ordering::Relaxed)
    }

    /// Unused isolate credit, as last published by each isolate.
    pub fn free_credit_bytes(&self) -> usize {
        self.free_credit
            .load(Ordering::Relaxed)
            .min(self.committed_bytes())
    }

    /// Current pressure level.
    pub fn pressure(&self) -> PressureLevel {
        PressureLevel::from_u8(self.pressure.load(Ordering::Acquire))
    }

    /// Recompute pressure; on a rise, recall unused credit; request
    /// collection at `Yellow`/`Red`.
    fn evaluate(&self, requester: Option<&IsolateControl>, grew: bool) {
        self.ensure_configured();
        let committed = self.committed_bytes();
        let free = self.free_credit.load(Ordering::Relaxed).min(committed);
        raise_peak(&self.peak_committed, committed);
        raise_peak(&self.peak_used, committed - free);
        let soft = self.soft_limit.load(Ordering::Relaxed);
        let hard = self.hard_limit.load(Ordering::Relaxed);
        let level = PressureLevel::classify(committed, soft, hard);
        let prev = self.pressure();
        // Only a change writes the shared pressure word.
        if prev != level
            && self
                .pressure
                .compare_exchange(prev as u8, level as u8, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
        {
            self.pressure_transitions.fetch_add(1, Ordering::Relaxed);
            tracing::info!(
                target: "memory",
                from = %prev,
                to = %level,
                committed,
                soft_limit = soft,
                hard_limit = hard,
                "memory pressure transition"
            );
            if level > prev {
                self.recall_credit();
            }
        }
        if level == PressureLevel::Red && grew {
            self.over_limit_events.fetch_add(1, Ordering::Relaxed);
            self.over_limit_bytes
                .fetch_add((committed - hard) as u64, Ordering::Relaxed);
            tracing::debug!(
                target: "memory",
                committed,
                hard_limit = hard,
                isolate = requester.map(|c| c.id),
                "governed allocation over the hard limit (observe-only; not rejected)"
            );
        }
        if level == PressureLevel::Green {
            return;
        }
        if let Some(req) = requester
            && req.governor_request(level)
        {
            self.collection_requests.fetch_add(1, Ordering::Relaxed);
        }
        if grew && let Some(largest) = self.largest_isolate() {
            let same = requester.is_some_and(|r| r.id == largest.id);
            if !same && largest.governor_request(level) {
                self.collection_requests.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    /// Ask every registered isolate to return its unused credit.  Each one
    /// does so at its next refill, collection, or runtime service poll; an
    /// isolate that never polls keeps at most its retained chunks.
    fn recall_credit(&self) {
        let reg = self.registry.lock().unwrap_or_else(PoisonError::into_inner);
        for control in reg.iter().filter_map(Weak::upgrade) {
            control.request_recall();
        }
        drop(reg);
        self.credit_recalls.fetch_add(1, Ordering::Relaxed);
    }

    fn largest_isolate(&self) -> Option<Arc<IsolateControl>> {
        let reg = self.registry.lock().unwrap_or_else(PoisonError::into_inner);
        reg.iter()
            .filter_map(Weak::upgrade)
            .max_by_key(|c| c.used_bytes())
    }

    /// Point-in-time view of every counter.
    pub fn snapshot(&self) -> MemorySnapshot {
        let config = self.config();
        let mut class_bytes = [0usize; CLASS_COUNT];
        for (slot, c) in class_bytes.iter_mut().zip(self.class_bytes.iter()) {
            *slot = c.load(Ordering::Relaxed);
        }
        let committed = self.committed_bytes();
        let free = self.free_credit_bytes();
        let mut isolates: Vec<IsolateSnapshot> = {
            let reg = self.registry.lock().unwrap_or_else(PoisonError::into_inner);
            reg.iter()
                .filter_map(Weak::upgrade)
                .map(|c| c.snapshot())
                .collect()
        };
        isolates.sort_by_key(|i| i.id);
        MemorySnapshot {
            soft_limit: config.soft_limit,
            hard_limit: config.hard_limit,
            critical_reserve: config.critical_reserve,
            credit_chunk: config.credit_chunk,
            pressure: self.pressure(),
            committed_bytes: committed,
            used_bytes: committed - free,
            free_credit_bytes: free,
            peak_committed_bytes: self.peak_committed.load(Ordering::Relaxed),
            peak_used_bytes: self.peak_used.load(Ordering::Relaxed),
            class_bytes,
            isolates,
            isolates_registered: self.isolates_registered.load(Ordering::Relaxed),
            pressure_transitions: self.pressure_transitions.load(Ordering::Relaxed),
            collection_requests: self.collection_requests.load(Ordering::Relaxed),
            credit_refills: self.credit_refills.load(Ordering::Relaxed),
            credit_returned_bytes: self.credit_returned_bytes.load(Ordering::Relaxed),
            credit_recalls: self.credit_recalls.load(Ordering::Relaxed),
            over_limit_events: self.over_limit_events.load(Ordering::Relaxed),
            over_limit_bytes: self.over_limit_bytes.load(Ordering::Relaxed),
        }
    }
}

/// Raise `peak` to `value`.  Reads first, so the common no-change case does
/// not write the shared cache line.
fn raise_peak(peak: &AtomicUsize, value: usize) {
    if value > peak.load(Ordering::Relaxed) {
        peak.fetch_max(value, Ordering::Relaxed);
    }
}

/// Subtract `bytes` from `counter`, reporting underflow in debug builds.
fn checked_sub(counter: &AtomicUsize, bytes: usize, what: &str) {
    let prev = counter.fetch_sub(bytes, Ordering::Relaxed);
    if prev < bytes {
        // Restore the counter before reporting, so release builds keep a
        // sane (zero) value rather than a wrapped one.  Not atomic with the
        // `fetch_sub`: a concurrent `add` in between can still lose bytes.
        // Underflow is a bug, and this repair is best-effort.
        counter.fetch_add(bytes - prev, Ordering::Relaxed);
        debug_assert!(
            false,
            "{what} counter underflow: released {bytes} with {prev} committed"
        );
    }
}

impl Default for ProcessMemoryGovernor {
    fn default() -> Self {
        Self::new()
    }
}

static GOVERNOR: ProcessMemoryGovernor = ProcessMemoryGovernor::new();

/// The process governor shared by every runtime in the process.
pub fn governor() -> &'static ProcessMemoryGovernor {
    &GOVERNOR
}

// ── Snapshots ────────────────────────────────────────────────────────────────

/// Point-in-time per-isolate counters.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IsolateSnapshot {
    pub id: IsolateId,
    pub name: Arc<str>,
    pub used_bytes: usize,
    pub free_credit_bytes: usize,
    pub peak_used_bytes: usize,
    /// Used bytes at which the isolate requests its own collection.
    pub collection_target: usize,
    pub collections: u64,
    pub collection_requests: u64,
    pub credit_refills: u64,
    pub zero_yield_collections: u64,
    pub last_bytes_before: usize,
    pub last_bytes_after: usize,
    pub last_duration: Duration,
}

/// Point-in-time process counters.  Fields are individually atomic
/// observations; the snapshot as a whole is not a consistent cut.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemorySnapshot {
    pub soft_limit: usize,
    pub hard_limit: usize,
    pub critical_reserve: usize,
    pub credit_chunk: usize,
    pub pressure: PressureLevel,
    /// Used bytes plus unused isolate credit.  Exact.
    pub committed_bytes: usize,
    /// Committed bytes minus free credit.
    pub used_bytes: usize,
    /// Unused isolate credit as of each isolate's last refill, collection,
    /// or poll; it can overstate current free credit by up to one chunk per
    /// isolate.
    pub free_credit_bytes: usize,
    pub peak_committed_bytes: usize,
    pub peak_used_bytes: usize,
    class_bytes: [usize; CLASS_COUNT],
    /// Live registered isolates.
    pub isolates: Vec<IsolateSnapshot>,
    /// Isolates ever registered.
    pub isolates_registered: u64,
    pub pressure_transitions: u64,
    pub collection_requests: u64,
    /// Credit grants to isolates (each one is a governor round trip).
    pub credit_refills: u64,
    /// Bytes of credit and used heap returned by isolates.
    pub credit_returned_bytes: u64,
    /// Pressure rises that recalled unused credit from every isolate.
    pub credit_recalls: u64,
    /// Grants that ended above the hard limit.  Strict mode would
    /// have rejected each one.
    pub over_limit_events: u64,
    /// Sum of the overage at each over-limit event.
    pub over_limit_bytes: u64,
}

impl MemorySnapshot {
    /// Current bytes in `class`.
    pub fn class_bytes(&self, class: MemoryClass) -> usize {
        self.class_bytes[class.index()]
    }
}

impl fmt::Display for MemorySnapshot {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        writeln!(f, "Managed memory (observe-only):")?;
        writeln!(
            f,
            "  Limits:                soft {} / hard {} bytes (reserve {})",
            self.soft_limit, self.hard_limit, self.critical_reserve
        )?;
        writeln!(f, "  Pressure:              {}", self.pressure)?;
        writeln!(
            f,
            "  Committed:             {} bytes (peak {})",
            self.committed_bytes, self.peak_committed_bytes
        )?;
        writeln!(
            f,
            "  Used:                  {} bytes (peak {}), {} bytes free credit",
            self.used_bytes, self.peak_used_bytes, self.free_credit_bytes
        )?;
        for class in MemoryClass::ALL {
            let bytes = self.class_bytes(class);
            if bytes > 0 {
                writeln!(f, "    {:<20} {bytes} bytes", format!("{}:", class.name()))?;
            }
        }
        writeln!(
            f,
            "  Isolates:              {} live ({} registered)",
            self.isolates.len(),
            self.isolates_registered
        )?;
        for iso in &self.isolates {
            writeln!(
                f,
                "    #{} {}: {} bytes (peak {}, target {}), {} free credit, {} refills, {} collections, {} requested",
                iso.id,
                iso.name,
                iso.used_bytes,
                iso.peak_used_bytes,
                iso.collection_target,
                iso.free_credit_bytes,
                iso.credit_refills,
                iso.collections,
                iso.collection_requests
            )?;
        }
        writeln!(f, "  Pressure transitions:  {}", self.pressure_transitions)?;
        writeln!(f, "  Collection requests:   {}", self.collection_requests)?;
        writeln!(
            f,
            "  Credit:                {} refills, {} bytes returned, {} recalls",
            self.credit_refills, self.credit_returned_bytes, self.credit_recalls
        )?;
        write!(
            f,
            "  Over hard limit:       {} events ({} bytes over)",
            self.over_limit_events, self.over_limit_bytes
        )
    }
}

// ── Current-thread isolate ───────────────────────────────────────────────────

thread_local! {
    static CURRENT: OnceCell<IsolateAccount> = const { OnceCell::new() };
}

fn default_isolate_name() -> Arc<str> {
    let thread = std::thread::current();
    match thread.name() {
        Some(name) => Arc::from(name),
        None => Arc::from(format!("{:?}", thread.id())),
    }
}

/// Register the calling thread as an isolate named `name` (or rename it if
/// it is already registered).  Returns its control handle.
pub fn register_current_isolate(name: &str) -> Option<Arc<IsolateControl>> {
    CURRENT
        .try_with(|cell| {
            if let Some(account) = cell.get() {
                account.control.set_name(Arc::from(name));
            } else {
                let _ = cell.set(governor().register_isolate(name));
            }
            cell.get().map(|a| a.control.clone())
        })
        .ok()
        .flatten()
}

/// Run `f` against the calling thread's account, registering it under the
/// thread's name on first use.  `None` during thread-local teardown.
#[inline]
pub fn with_current_account<R>(f: impl FnOnce(&IsolateAccount) -> R) -> Option<R> {
    CURRENT
        .try_with(
            |cell| f(cell.get_or_init(|| governor().register_isolate(default_isolate_name()))),
        )
        .ok()
}

/// The calling thread's control handle.
pub fn current_control() -> Option<Arc<IsolateControl>> {
    with_current_account(|a| a.control.clone())
}

/// Flush the calling thread's account and return a process snapshot.
pub fn snapshot() -> MemorySnapshot {
    let _ = CURRENT.try_with(|cell| {
        if let Some(account) = cell.get() {
            account.publish();
        }
    });
    governor().snapshot()
}

#[cfg(test)]
mod tests {
    use super::*;

    fn leaked(soft: usize, hard: usize, chunk: usize) -> &'static ProcessMemoryGovernor {
        let mut config = MemoryConfig::with_limits(soft, hard).unwrap();
        config.credit_chunk = chunk;
        Box::leak(Box::new(
            ProcessMemoryGovernor::with_config(config).unwrap(),
        ))
    }

    fn report(before: usize, after: usize) -> CollectionReport {
        CollectionReport {
            bytes_before: before,
            bytes_after: after,
            bytes_returned: before - after,
            duration: Duration::from_micros(1),
        }
    }

    #[test]
    fn config_rejects_invalid_states() {
        assert_eq!(
            MemoryConfig::with_limits(0, 0),
            Err(MemoryConfigError::ZeroHardLimit)
        );
        assert!(matches!(
            MemoryConfig::with_limits(10 * MIB, 5 * MIB),
            Err(MemoryConfigError::SoftAboveHard { .. })
        ));
        let mut c = MemoryConfig::with_limits(MIB, 2 * MIB).unwrap();
        c.critical_reserve = 3 * MIB;
        assert!(matches!(
            c.validate(),
            Err(MemoryConfigError::ReserveAboveHard { .. })
        ));
        let mut c = MemoryConfig::with_limits(MIB, 2 * MIB).unwrap();
        c.credit_chunk = 0;
        assert_eq!(c.validate(), Err(MemoryConfigError::ZeroCreditChunk));
        let mut c = MemoryConfig::with_limits(MIB, 2 * MIB).unwrap();
        c.queue_limit = 3 * MIB;
        assert!(matches!(
            c.validate(),
            Err(MemoryConfigError::QueueAboveBudget { .. })
        ));
    }

    #[test]
    fn config_defaults() {
        let c = MemoryConfig::with_limits(300 * MIB, 400 * MIB).unwrap();
        assert_eq!(c.critical_reserve, 4 * MIB);
        assert_eq!(c.credit_chunk, DEFAULT_CREDIT_CHUNK);
        assert_eq!(c.queue_limit, DEFAULT_QUEUE_LIMIT);
        let c = MemoryConfig::with_limits(1, 2 * MIB).unwrap();
        assert_eq!(
            c.critical_reserve,
            2 * MIB,
            "reserve is capped at the hard limit"
        );
        assert_eq!(c.queue_limit, 2 * MIB);
        let big = MemoryConfig::with_limits(1, 1 << 40).unwrap();
        assert_eq!(big.critical_reserve, MAX_DEFAULT_CRITICAL_RESERVE);
        let c = MemoryConfig::from_optional_limits(None, Some(400 * MIB)).unwrap();
        assert_eq!(c.soft_limit, 300 * MIB);
        assert_eq!(
            default_soft_limit(1),
            1,
            "tiny hard limit keeps a nonzero soft limit"
        );
        assert_eq!(default_soft_limit(3), 3);
        assert_eq!(default_soft_limit(4), 3);
        let c = MemoryConfig::from_optional_limits(Some(usize::MAX / 2), None).unwrap();
        assert_eq!(
            c.soft_limit, c.hard_limit,
            "soft-only raises the hard limit"
        );
    }

    #[test]
    fn env_new_names_win_and_old_names_warn() {
        let vars = [
            (SOFT_LIMIT_ENV, "100"),
            (DEPRECATED_SOFT_LIMIT_ENV, "50"),
            (DEPRECATED_HARD_LIMIT_ENV, "200"),
            (CREDIT_ENV, "128"),
        ];
        let lookup = |k: &str| {
            vars.iter()
                .find(|(n, _)| *n == k)
                .map(|(_, v)| v.to_string())
        };
        let mut warnings = Vec::new();
        let c = MemoryConfig::from_lookup(lookup, |w| warnings.push(w)).unwrap();
        assert_eq!(c.soft_limit, 100 * MIB);
        assert_eq!(c.hard_limit, 200 * MIB);
        assert_eq!(c.credit_chunk, 128 * KIB);
        assert_eq!(warnings.len(), 2, "{warnings:?}");
        assert!(warnings[0].contains("ignored"));
        assert!(warnings[1].contains(DEPRECATED_HARD_LIMIT_ENV));
    }

    #[test]
    fn env_rejects_bad_values() {
        let err = MemoryConfig::from_lookup(
            |k| (k == HARD_LIMIT_ENV).then(|| "lots".to_string()),
            |_| {},
        )
        .unwrap_err();
        assert!(matches!(err, MemoryConfigError::InvalidNumber { .. }));
        let err = MemoryConfig::from_lookup(
            |k| match k {
                SOFT_LIMIT_ENV => Some("20".into()),
                HARD_LIMIT_ENV => Some("10".into()),
                _ => None,
            },
            |_| {},
        )
        .unwrap_err();
        assert!(matches!(err, MemoryConfigError::SoftAboveHard { .. }));
    }

    #[test]
    fn charge_refills_one_chunk_at_a_time() {
        let g = leaked(10 * MIB, 20 * MIB, 4 * KIB);
        let acc = g.register_isolate("a");
        acc.charge(KIB);
        assert_eq!(
            g.committed_bytes(),
            4 * KIB,
            "the first charge takes a chunk"
        );
        assert_eq!(acc.used_bytes(), KIB);
        assert_eq!(acc.free_credit_bytes(), 3 * KIB);
        acc.charge(3 * KIB);
        assert_eq!(g.committed_bytes(), 4 * KIB, "exact fit stays local");
        assert_eq!(acc.control().credit_refills(), 1);
        acc.charge(1);
        assert_eq!(g.committed_bytes(), 8 * KIB);
        assert_eq!(acc.control().credit_refills(), 2);
        assert_eq!(acc.control().used_bytes(), 4 * KIB + 1, "refill publishes");
        assert_eq!(acc.committed_bytes(), g.committed_bytes());
    }

    #[test]
    fn common_path_does_not_contact_the_governor() {
        let g = leaked(10 * MIB, 20 * MIB, 64 * KIB);
        let acc = g.register_isolate("a");
        for _ in 0..64 {
            acc.charge(KIB);
        }
        assert_eq!(g.snapshot().credit_refills, 1, "64 allocations, one refill");
        assert_eq!(g.committed_bytes(), 64 * KIB);
    }

    #[test]
    fn large_allocation_requests_its_own_size() {
        let g = leaked(10 * MIB, 20 * MIB, 4 * KIB);
        let acc = g.register_isolate("a");
        acc.charge(100 * KIB + 1);
        assert_eq!(
            g.committed_bytes(),
            104 * KIB,
            "rounded to the accounting unit"
        );
        assert_eq!(acc.free_credit_bytes(), 4 * KIB - 1);
        assert_eq!(acc.control().credit_refills(), 1);
    }

    #[test]
    fn sweep_returns_exact_bytes_and_trims_credit() {
        let g = leaked(10 * MIB, 20 * MIB, KIB);
        let acc = g.register_isolate("a");
        for _ in 0..100 {
            acc.charge(48);
        }
        assert_eq!(g.committed_bytes(), 5 * KIB);
        acc.release(48 * 60);
        assert_eq!(acc.used_bytes(), 48 * 40);
        assert_eq!(acc.free_credit_bytes(), 5 * KIB - 48 * 40);
        acc.record_collection(report(4800, 48 * 40));
        assert_eq!(acc.control().collection_epoch(), 1);
        assert_eq!(
            acc.free_credit_bytes(),
            2 * KIB,
            "keeps the retained chunks only"
        );
        let snap = g.snapshot();
        assert_eq!(snap.committed_bytes, 48 * 40 + 2 * KIB);
        assert_eq!(snap.class_bytes(MemoryClass::GcHeap), 48 * 40 + 2 * KIB);
        assert_eq!(snap.free_credit_bytes, 2 * KIB);
        assert_eq!(snap.used_bytes, 48 * 40);
        assert_eq!(
            snap.credit_returned_bytes,
            (5 * KIB - 48 * 40 - 2 * KIB) as u64
        );
    }

    #[test]
    fn shutdown_returns_used_bytes_and_free_credit() {
        let g = leaked(10 * MIB, 20 * MIB, KIB);
        let acc = g.register_isolate("a");
        acc.charge(10 * KIB + 100);
        acc.release(KIB);
        acc.publish();
        assert!(g.free_credit_bytes() > 0);
        assert_eq!(g.snapshot().isolates.len(), 1);
        drop(acc);
        let snap = g.snapshot();
        assert_eq!(snap.committed_bytes, 0);
        assert_eq!(snap.free_credit_bytes, 0);
        assert!(snap.isolates.is_empty());
        assert_eq!(snap.isolates_registered, 1);
    }

    #[test]
    fn shared_charge_returns_once() {
        let g = leaked(10 * MIB, 20 * MIB, KIB);
        let charge = Arc::new(g.reserve_shared(MemoryClass::SharedValue, 1000).unwrap());
        let clone = charge.clone();
        assert_eq!(g.class_bytes(MemoryClass::SharedValue), 1000);
        assert_eq!(g.committed_bytes(), 1000);
        drop(charge);
        assert_eq!(g.class_bytes(MemoryClass::SharedValue), 1000);
        drop(clone);
        assert_eq!(g.class_bytes(MemoryClass::SharedValue), 0);
        assert_eq!(g.committed_bytes(), 0);
    }

    #[test]
    fn green_does_not_request_collection() {
        let g = leaked(64 * MIB, 128 * MIB, 64 * KIB);
        let acc = g.register_isolate("a");
        acc.charge(32 * MIB);
        assert_eq!(g.pressure(), PressureLevel::Green);
        assert!(
            !acc.control().collection_requested(),
            "GcConfig owns the heap trigger"
        );
        assert_eq!(acc.control().collection_requests(), 0);
    }

    #[test]
    fn pressure_transitions_and_requests() {
        let g = leaked(4 * MIB, 8 * MIB, 64 * KIB);
        let acc = g.register_isolate("a");
        acc.charge(MIB);
        assert_eq!(g.pressure(), PressureLevel::Green);
        assert!(!acc.control().collection_requested());

        acc.charge(3 * MIB);
        assert_eq!(g.pressure(), PressureLevel::Yellow);
        assert!(
            acc.control().take_collection_request(),
            "Yellow requests collection"
        );

        // Rate limit: no second request in the same collection epoch.
        acc.charge(64 * KIB);
        assert!(!acc.control().collection_requested());

        // After a zero-yield collection the next request waits for growth.
        let used = acc.used_bytes();
        acc.record_collection(report(used, used));
        assert_eq!(acc.control().collection_target(), 2 * used);
        acc.charge(64 * KIB);
        assert!(
            !acc.control().collection_requested(),
            "Yellow honours the target"
        );

        // Red overrides the target.
        acc.charge(5 * MIB);
        assert_eq!(g.pressure(), PressureLevel::Red);
        assert!(acc.control().take_collection_request());
        let snap = g.snapshot();
        assert!(snap.over_limit_events >= 1);
        assert!(snap.pressure_transitions >= 2);

        acc.release(acc.used_bytes());
        acc.record_collection(report(9 * MIB, 0));
        assert_eq!(acc.free_credit_bytes(), 0, "Red retains no credit");
        assert_eq!(g.committed_bytes(), 0);
        assert_eq!(g.pressure(), PressureLevel::Green);
    }

    #[test]
    fn yellow_requests_largest_heap() {
        let g = leaked(4 * MIB, 8 * MIB, 64 * KIB);
        let big = g.register_isolate("big");
        let small = g.register_isolate("small");
        big.charge(3 * MIB);
        assert!(!big.control().collection_requested());
        small.charge(MIB);
        assert_eq!(g.pressure(), PressureLevel::Yellow);
        assert!(small.control().take_collection_request());
        assert!(
            big.control().take_collection_request(),
            "largest heap asked too"
        );
    }

    #[test]
    fn rising_pressure_recalls_free_credit() {
        let g = leaked(4 * MIB, 64 * MIB, 64 * KIB);
        let idle = g.register_isolate("idle");
        idle.charge(10 * 64 * KIB);
        idle.release(10 * 64 * KIB);
        idle.record_collection(report(10 * 64 * KIB, 0));
        assert_eq!(
            idle.free_credit_bytes(),
            2 * 64 * KIB,
            "Green keeps two chunks"
        );
        assert_eq!(g.committed_bytes(), 2 * 64 * KIB);

        let busy = g.register_isolate("busy");
        busy.charge(4 * MIB);
        assert_eq!(g.pressure(), PressureLevel::Yellow);
        assert!(idle.control().recall_requested());
        idle.poll();
        assert!(!idle.control().recall_requested());
        assert_eq!(idle.free_credit_bytes(), 64 * KIB, "Yellow keeps one chunk");

        busy.charge(60 * MIB);
        assert_eq!(g.pressure(), PressureLevel::Red);
        idle.poll();
        assert_eq!(idle.free_credit_bytes(), 0, "Red keeps none");
        assert!(g.snapshot().credit_recalls >= 2);
    }

    #[test]
    fn committed_equals_account_totals_across_threads() {
        use std::sync::Barrier;
        let g = leaked(512 * MIB, 1024 * MIB, 4 * KIB);
        let barrier = Arc::new(Barrier::new(9));
        let threads: Vec<_> = (0..8)
            .map(|i| {
                let barrier = barrier.clone();
                std::thread::spawn(move || {
                    let acc = g.register_isolate(format!("iso-{i}"));
                    for n in 0..2000usize {
                        // Mix small charges with an occasional large one.
                        acc.charge(if n % 97 == 0 {
                            20 * KIB + n
                        } else {
                            40 + n % 200
                        });
                        if n % 500 == 499 {
                            acc.release(acc.used_bytes() / 2);
                            let used = acc.used_bytes();
                            acc.record_collection(report(used * 2, used));
                        }
                    }
                    let committed = acc.committed_bytes();
                    barrier.wait(); // every account is quiescent
                    barrier.wait(); // main thread has read the governor
                    committed
                })
            })
            .collect();
        barrier.wait();
        let governor_committed = g.committed_bytes();
        barrier.wait();
        let sum: usize = threads.into_iter().map(|t| t.join().unwrap()).sum();
        assert_eq!(governor_committed, sum, "committed = used + free credit");
        assert_eq!(g.committed_bytes(), 0, "thread exit returns everything");
        assert_eq!(g.free_credit_bytes(), 0);
        assert!(g.snapshot().peak_committed_bytes >= sum);
    }
}