//! Physical page authority, dynamic reservation, and exclusive lease ledger.
//! Initialization consumes DTB banks and a conventional-only firmware
//! intersection. Every mutation stages its map update before publishing it.
//! Caller-owned page backing is touched only after identity/lease validation.

use turkos_os_runtime_platform::ram::{Authority, RamBanks, Region};

pub const PAGE: u64 = 4096;
pub const MAX_REGIONS: usize = 64;
pub const MAX_LIVE: usize = 64;
pub const MAX_RESERVATIONS: usize = 64;
pub const MAX_DESCRIPTORS: usize = 512;
pub const CONVENTIONAL: u32 = 7;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Error {
    Invalid,
    UnknownFirmwareType(u32),
    FirmwareOverlap,
    Capacity,
    OutsideAuthority,
    Reserved,
    LiveReservation,
    NotFree,
    NoMemory,
    ForeignOwner,
    StaleLease,
    Generation,
    CountOverflow,
    Audit,
    BackingAddress,
    PageBytes,
    PageContents,
}
impl Error {
    pub const fn tag(self) -> &'static str {
        match self {
            Self::Invalid => "frame-invalid",
            Self::UnknownFirmwareType(_) => "frame-unknown-firmware-type",
            Self::FirmwareOverlap => "frame-firmware-overlap",
            Self::Capacity => "frame-capacity",
            Self::OutsideAuthority => "frame-outside-authority",
            Self::Reserved => "frame-reserved",
            Self::LiveReservation => "frame-live-reservation",
            Self::NotFree => "frame-not-free",
            Self::NoMemory => "frame-no-memory",
            Self::ForeignOwner => "frame-foreign-owner",
            Self::StaleLease => "frame-stale-lease",
            Self::Generation => "frame-generation",
            Self::CountOverflow => "frame-count-overflow",
            Self::Audit => "frame-audit",
            Self::BackingAddress => "frame-backing-address",
            Self::PageBytes => "frame-page-bytes",
            Self::PageContents => "frame-page-contents",
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Descriptor {
    pub kind: u32,
    pub region: Region,
}

fn platform(error: turkos_os_runtime_platform::ram::Error) -> Error {
    use turkos_os_runtime_platform::ram::Error as E;
    match error {
        E::Capacity => Error::Capacity,
        E::Overlap => Error::FirmwareOverlap,
        _ => Error::Invalid,
    }
}
fn full_pages(region: Region) -> Result<Option<Region>, Error> {
    if region.start >= region.end {
        return Err(Error::Invalid);
    }
    let start = region.start.checked_add(PAGE - 1).ok_or(Error::Invalid)? & !(PAGE - 1);
    let end = region.end & !(PAGE - 1);
    Ok((start < end).then_some(Region { start, end }))
}
fn touched_pages(region: Region) -> Result<Region, Error> {
    if region.start >= region.end {
        return Err(Error::Invalid);
    }
    Ok(Region {
        start: region.start & !(PAGE - 1),
        end: region.end.checked_add(PAGE - 1).ok_or(Error::Invalid)? & !(PAGE - 1),
    })
}
fn pages(regions: &[Region]) -> Result<u64, Error> {
    regions.iter().try_fold(0u64, |sum, region| {
        if region.start >= region.end
            || !region.start.is_multiple_of(PAGE)
            || !region.end.is_multiple_of(PAGE)
        {
            return Err(Error::Audit);
        }
        sum.checked_add((region.end - region.start) / PAGE)
            .ok_or(Error::CountOverflow)
    })
}

/// Not Clone: transfer this plan into exactly one allocator during runtime
/// initialization. Reclassification is a privileged boot operation, not an
/// API for clients to create another allocator over the same physical pages.
pub struct Plan {
    authority: Authority,
    ram_full_pages: u64,
    firmware_withheld: u64,
    boot_reserved: u64,
}
impl Plan {
    /// Privileged compatibility entry for an already finalized boot FreeMap.
    /// Its denominator is the admitted pool, not a claim about raw DTB RAM.
    /// Actual firmware classification must be completed before this call.
    pub fn already_admitted(regions: &[Region]) -> Result<Self, Error> {
        if regions.is_empty() {
            return Err(Error::NoMemory);
        }
        if regions.len() > MAX_REGIONS {
            return Err(Error::Capacity);
        }
        pages(regions)?;
        let first = regions
            .iter()
            .map(|region| region.start)
            .min()
            .ok_or(Error::Invalid)?;
        let last = regions
            .iter()
            .map(|region| region.end)
            .max()
            .ok_or(Error::Invalid)?;
        let banks = RamBanks::from_slice(&[Region {
            start: first,
            end: last,
        }])
        .map_err(platform)?;
        let authority = Authority::create(&banks, &[], Some(regions)).map_err(platform)?;
        let admitted = authority.pages();
        if admitted == 0 {
            return Err(Error::NoMemory);
        }
        Ok(Self {
            authority,
            ram_full_pages: admitted,
            firmware_withheld: 0,
            boot_reserved: 0,
        })
    }
    fn finish(
        banks: &RamBanks,
        reservations: &[Region],
        conventional: Option<&[Region]>,
    ) -> Result<Self, Error> {
        let authority = Authority::create(banks, reservations, conventional).map_err(platform)?;
        let ram_full_pages = banks.as_slice().iter().try_fold(0u64, |sum, bank| {
            let count = full_pages(*bank)?
                .map(|region| (region.end - region.start) / PAGE)
                .unwrap_or(0);
            sum.checked_add(count).ok_or(Error::CountOverflow)
        })?;
        // Count the unreserved intersection directly. Holding two complete
        // Authority maps plus the returned Plan overflows the 16-KiB boot
        // stack on the firmware path. The final Authority above validates the
        // same banks and conventional extents before this bounded count.
        let initial_pages = if let Some(conventional) = conventional {
            let mut total = 0u64;
            for bank in banks.as_slice() {
                for region in conventional {
                    let intersect = Region {
                        start: bank.start.max(region.start),
                        end: bank.end.min(region.end),
                    };
                    if intersect.start < intersect.end
                        && let Some(full) = full_pages(intersect)?
                    {
                        total = total
                            .checked_add((full.end - full.start) / PAGE)
                            .ok_or(Error::CountOverflow)?;
                    }
                }
            }
            total
        } else {
            ram_full_pages
        };
        if authority.pages() == 0 {
            return Err(Error::NoMemory);
        }
        Ok(Self {
            authority,
            ram_full_pages,
            firmware_withheld: ram_full_pages
                .checked_sub(initial_pages)
                .ok_or(Error::Audit)?,
            boot_reserved: initial_pages
                .checked_sub(authority.pages())
                .ok_or(Error::Audit)?,
        })
    }
    pub fn direct(banks: &RamBanks, reservations: &[Region]) -> Result<Self, Error> {
        Self::finish(banks, reservations, None)
    }
    pub fn firmware(
        banks: &RamBanks,
        reservations: &[Region],
        descriptors: &[Descriptor],
    ) -> Result<Self, Error> {
        if descriptors.is_empty() {
            return Err(Error::Invalid);
        }
        if descriptors.len() > MAX_DESCRIPTORS {
            return Err(Error::Capacity);
        }
        let mut conventional = [Region::EMPTY; MAX_REGIONS];
        let mut len = 0;
        for (index, descriptor) in descriptors.iter().enumerate() {
            // This boot contract explicitly supports type numbers 0..=14.
            // New/future values must be reviewed before being admitted.
            if descriptor.kind > 14 {
                return Err(Error::UnknownFirmwareType(descriptor.kind));
            }
            let region = descriptor.region;
            if region.start >= region.end
                || !region.start.is_multiple_of(PAGE)
                || !region.end.is_multiple_of(PAGE)
            {
                return Err(Error::Invalid);
            }
            if descriptors[..index]
                .iter()
                .any(|prior| prior.region.overlaps(region))
            {
                return Err(Error::FirmwareOverlap);
            }
            if descriptor.kind == CONVENTIONAL || descriptor.kind == 4 {
                if len == MAX_REGIONS {
                    return Err(Error::Capacity);
                }
                conventional[len] = region;
                len += 1;
            }
        }
        Self::finish(banks, reservations, Some(&conventional[..len]))
    }
    pub fn authority(&self) -> &Authority {
        &self.authority
    }
    pub fn matches_admitted(&self, regions: &[Region]) -> Result<(), Error> {
        let allowed = Map::from_authority(&self.authority);
        if pages(regions)? != self.authority.pages() {
            return Err(Error::Audit);
        }
        for (index, region) in regions.iter().enumerate() {
            if !allowed.contains(*region)
                || regions[..index].iter().any(|prior| prior.overlaps(*region))
            {
                return Err(Error::Audit);
            }
        }
        Ok(())
    }
    pub fn ram_full_pages(&self) -> u64 {
        self.ram_full_pages
    }
    pub fn boot_audit(&self) -> Audit {
        let authorized = self.authority.pages();
        Audit {
            ram_full_pages: self.ram_full_pages,
            firmware_withheld_pages: self.firmware_withheld,
            boot_reserved_pages: self.boot_reserved,
            authorized_pages: authorized,
            free_pages: authorized,
            reserved_pages: 0,
            live_pages: 0,
        }
    }
}

#[derive(Clone, Copy)]
struct Map {
    regions: [Region; MAX_REGIONS],
    len: usize,
}
impl Map {
    fn from_authority(authority: &Authority) -> Self {
        let mut regions = [Region::EMPTY; MAX_REGIONS];
        let len = authority.as_slice().len();
        regions[..len].copy_from_slice(authority.as_slice());
        Self { regions, len }
    }
    fn as_slice(&self) -> &[Region] {
        &self.regions[..self.len]
    }
    fn contains(&self, target: Region) -> bool {
        let mut cursor = target.start;
        for region in self.as_slice() {
            if region.end <= cursor {
                continue;
            }
            if region.start > cursor {
                return false;
            }
            cursor = region.end.min(target.end);
            if cursor == target.end {
                return true;
            }
        }
        false
    }
    fn subtract(&self, target: Region) -> Result<Self, Error> {
        let mut result = Self {
            regions: [Region::EMPTY; MAX_REGIONS],
            len: 0,
        };
        for region in self.as_slice() {
            if !region.overlaps(target) {
                result.push(*region)?;
                continue;
            }
            if region.start < target.start {
                result.push(Region {
                    start: region.start,
                    end: target.start.min(region.end),
                })?;
            }
            if region.end > target.end {
                result.push(Region {
                    start: target.end.max(region.start),
                    end: region.end,
                })?;
            }
        }
        Ok(result)
    }
    fn push(&mut self, region: Region) -> Result<(), Error> {
        if self.len == MAX_REGIONS {
            return Err(Error::Capacity);
        }
        self.regions[self.len] = region;
        self.len += 1;
        Ok(())
    }
    fn add(&self, target: Region) -> Result<Self, Error> {
        if self.as_slice().iter().any(|region| region.overlaps(target)) {
            return Err(Error::Audit);
        }
        let mut output = Self {
            regions: [Region::EMPTY; MAX_REGIONS],
            len: 0,
        };
        let mut merged = target;
        let mut inserted = false;
        for region in self.as_slice() {
            if region.end == merged.start {
                merged.start = region.start;
            } else if merged.end == region.start {
                merged.end = region.end;
            } else if region.start > merged.end {
                if !inserted {
                    output.push(merged)?;
                    inserted = true;
                }
                output.push(*region)?;
            } else {
                output.push(*region)?;
            }
        }
        if !inserted {
            output.push(merged)?;
        }
        output.regions[..output.len].sort_unstable_by_key(|region| region.start);
        Ok(output)
    }
}

#[derive(Clone, Copy)]
struct Live {
    address: u64,
    generation: u64,
    active: bool,
}
const EMPTY_LIVE: Live = Live {
    address: 0,
    generation: 0,
    active: false,
};
#[derive(Clone, Copy)]
struct Reserved {
    region: Region,
    generation: u64,
    active: bool,
}
const EMPTY_RESERVED: Reserved = Reserved {
    region: Region::EMPTY,
    generation: 0,
    active: false,
};

/// Non-copy proof of one allocator's live page and its allocation generation.
#[derive(Debug)]
pub struct Lease {
    address: u64,
    owner: u32,
    generation: u64,
    slot: usize,
}
impl Lease {
    pub fn address(&self) -> u64 {
        self.address
    }
    pub fn generation(&self) -> u64 {
        self.generation
    }
}
#[derive(Debug)]
pub struct Reservation {
    region: Region,
    owner: u32,
    generation: u64,
    slot: usize,
}
impl Reservation {
    pub fn region(&self) -> Region {
        self.region
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Audit {
    pub ram_full_pages: u64,
    pub firmware_withheld_pages: u64,
    pub boot_reserved_pages: u64,
    pub authorized_pages: u64,
    pub free_pages: u64,
    pub reserved_pages: u64,
    pub live_pages: u64,
}
pub struct Allocator {
    plan: Option<Plan>,
    owner: u32,
    free: Map,
    live: [Live; MAX_LIVE],
    reserved: [Reserved; MAX_RESERVATIONS],
    generation: u64,
}
impl Allocator {
    /// BSS storage can be initialized without constructing/copying the whole
    /// bounded lease ledger on the small bootstrap stack.
    pub const fn uninitialized() -> Self {
        Self {
            plan: None,
            owner: 0,
            free: Map {
                regions: [Region::EMPTY; MAX_REGIONS],
                len: 0,
            },
            live: [EMPTY_LIVE; MAX_LIVE],
            reserved: [EMPTY_RESERVED; MAX_RESERVATIONS],
            generation: 0,
        }
    }
    pub fn initialize(&mut self, plan: Plan, owner: u32) -> Result<(), Error> {
        if owner == 0 {
            return Err(Error::ForeignOwner);
        }
        if self.plan.is_some() {
            return Err(Error::NotFree);
        }
        self.free = Map::from_authority(&plan.authority);
        self.plan = Some(plan);
        self.owner = owner;
        self.live.fill(EMPTY_LIVE);
        self.reserved.fill(EMPTY_RESERVED);
        self.generation = 0;
        self.audit()?;
        Ok(())
    }
    /// Only an empty, fully returned bootstrap pool may be reused. A dropped
    /// live owner is retained in BSS and blocks another overlapping authority.
    pub fn reset_empty(&mut self) -> Result<(), Error> {
        let audit = self.audit()?;
        if audit.live_pages != 0 || audit.reserved_pages != 0 {
            return Err(Error::NotFree);
        }
        self.plan = None;
        self.owner = 0;
        self.free.len = 0;
        self.free.regions.fill(Region::EMPTY);
        Ok(())
    }
    fn plan(&self) -> &Plan {
        self.plan.as_ref().expect("initialized authority")
    }
    pub fn new(plan: Plan, owner: u32) -> Result<Self, Error> {
        if owner == 0 {
            return Err(Error::ForeignOwner);
        }
        let mut result = Self::uninitialized();
        result.initialize(plan, owner)?;
        Ok(result)
    }
    pub fn authority(&self) -> &Authority {
        &self.plan().authority
    }
    /// Inert evidence for the central pool's consumers. A metadata owner is a
    /// child scope of this pool; count each address once across consumer scopes.
    pub fn live_addresses(&self) -> impl Iterator<Item = u64> + '_ {
        self.live
            .iter()
            .filter_map(|record| record.active.then_some(record.address))
    }
    pub fn reserved_regions(&self) -> impl Iterator<Item = Region> + '_ {
        self.reserved
            .iter()
            .filter_map(|record| record.active.then_some(record.region))
    }

    /// Existing trusted kernel MMU code uses address leases. Untrusted clients
    /// must use generation-bearing Lease/metadata handles; this compatibility
    /// surface cannot distinguish an old numeric address from a current one.
    pub fn allocate_address(&mut self) -> Result<u64, Error> {
        self.allocate().map(|lease| lease.address)
    }
    pub fn allocate_address_at(&mut self, address: u64) -> Result<u64, Error> {
        self.allocate_at(address).map(|lease| lease.address)
    }
    /// Acquire a complete contiguous block after validating every physical
    /// page, ledger slot, map split and generation. No partial acquisition is
    /// observable on failure; callers retain the returned nonclone tokens.
    pub fn allocate_many_at(
        &mut self,
        address: u64,
        count: usize,
    ) -> Result<[Option<Lease>; 8], Error> {
        self.audit()?;
        if count == 0 || count > 8 {
            return Err(Error::Invalid);
        }
        self.page(address)?;
        let region = Region {
            start: address,
            end: address
                .checked_add(count as u64 * PAGE)
                .ok_or(Error::Invalid)?,
        };
        if !Map::from_authority(&self.plan().authority).contains(region) {
            return Err(Error::OutsideAuthority);
        }
        if !self.free.contains(region) {
            return Err(Error::NotFree);
        }
        let generation = self
            .generation
            .checked_add(count as u64)
            .ok_or(Error::Generation)?;
        let mut slots = [0; 8];
        let mut found = 0;
        for (slot, record) in self.live.iter().enumerate() {
            if !record.active {
                slots[found] = slot;
                found += 1;
                if found == count {
                    break;
                }
            }
        }
        if found != count {
            return Err(Error::Capacity);
        }
        let free = self.free.subtract(region)?;
        let mut leases = core::array::from_fn(|_| None);
        for index in 0..count {
            let live = Live {
                address: address + index as u64 * PAGE,
                generation: self.generation + index as u64 + 1,
                active: true,
            };
            self.live[slots[index]] = live;
            leases[index] = Some(Lease {
                address: live.address,
                owner: self.owner,
                generation: live.generation,
                slot: slots[index],
            });
        }
        self.free = free;
        self.generation = generation;
        Ok(leases)
    }
    /// Validate the entire typed owner set, including generations and free-map
    /// capacity, before publishing any returned page.
    pub fn release_many_leases(&mut self, leases: &[&Lease]) -> Result<(), Error> {
        self.audit()?;
        if leases.len() > MAX_LIVE {
            return Err(Error::Capacity);
        }
        let mut free = self.free;
        for (index, lease) in leases.iter().enumerate() {
            self.valid_lease(lease)?;
            if leases[..index].iter().any(|prior| prior.slot == lease.slot) {
                return Err(Error::StaleLease);
            }
            free = free.add(self.page(lease.address)?)?;
        }
        self.free = free;
        for lease in leases {
            self.live[lease.slot].active = false;
        }
        Ok(())
    }
    fn current_lease(&self, address: u64) -> Result<Lease, Error> {
        let (slot, record) = self
            .live
            .iter()
            .enumerate()
            .find(|(_, record)| record.active && record.address == address)
            .ok_or(Error::StaleLease)?;
        Ok(Lease {
            address,
            owner: self.owner,
            generation: record.generation,
            slot,
        })
    }
    pub fn release_address(&mut self, address: u64) -> Result<(), Error> {
        let lease = self.current_lease(address)?;
        self.release(&lease)
    }
    pub fn release_many_addresses(&mut self, addresses: &[u64]) -> Result<(), Error> {
        self.release_many_with(addresses, |_| Ok(()))
    }
    /// Validate the complete owner set before invoking an erasure callback.
    /// Callback failure retains every lease (although already erased contents
    /// cannot be restored). Free publication follows the final successful wipe.
    pub fn release_many_with(
        &mut self,
        addresses: &[u64],
        mut erase: impl FnMut(u64) -> Result<(), Error>,
    ) -> Result<(), Error> {
        self.audit()?;
        if addresses.len() > MAX_LIVE {
            return Err(Error::Capacity);
        }
        let mut free = self.free;
        let mut slots = [0; MAX_LIVE];
        for (index, address) in addresses.iter().enumerate() {
            if addresses[..index].contains(address) {
                return Err(Error::StaleLease);
            }
            let lease = self.current_lease(*address)?;
            free = free.add(self.page(*address)?)?;
            slots[index] = lease.slot;
        }
        for address in addresses {
            erase(*address)?;
        }
        self.free = free;
        for slot in &slots[..addresses.len()] {
            self.live[*slot].active = false;
        }
        Ok(())
    }
    pub fn prepare_address_page(
        &self,
        address: u64,
        bytes: &mut [u8],
        canary: u64,
    ) -> Result<(), Error> {
        self.prepare_page(&self.current_lease(address)?, bytes, canary)
    }
    pub fn verify_address_page(
        &self,
        address: u64,
        bytes: &[u8],
        canary: u64,
    ) -> Result<(), Error> {
        self.verify_page(&self.current_lease(address)?, bytes, canary)
    }
    pub fn free_regions(&self) -> &[Region] {
        self.free.as_slice()
    }
    pub fn remaining_pages(&self) -> u64 {
        pages(self.free.as_slice()).expect("validated free map")
    }
    fn page(&self, address: u64) -> Result<Region, Error> {
        if !address.is_multiple_of(PAGE) {
            return Err(Error::Invalid);
        }
        let target = Region {
            start: address,
            end: address.checked_add(PAGE).ok_or(Error::Invalid)?,
        };
        if !self.plan().authority.authorizes(address) {
            return Err(Error::OutsideAuthority);
        }
        Ok(target)
    }
    pub fn allocate(&mut self) -> Result<Lease, Error> {
        let address = self.free.as_slice().first().ok_or(Error::NoMemory)?.start;
        self.allocate_at(address)
    }
    pub fn allocate_at(&mut self, address: u64) -> Result<Lease, Error> {
        self.audit()?;
        let target = self.page(address)?;
        if self
            .reserved
            .iter()
            .any(|record| record.active && record.region.overlaps(target))
        {
            return Err(Error::Reserved);
        }
        if !self.free.contains(target) {
            return Err(Error::NotFree);
        }
        let slot = self
            .live
            .iter()
            .position(|record| !record.active)
            .ok_or(Error::Capacity)?;
        let generation = self.generation.checked_add(1).ok_or(Error::Generation)?;
        let free = self.free.subtract(target)?;
        self.free = free;
        self.live[slot] = Live {
            address,
            generation,
            active: true,
        };
        self.generation = generation;
        Ok(Lease {
            address,
            owner: self.owner,
            generation,
            slot,
        })
    }
    fn valid_lease(&self, lease: &Lease) -> Result<(), Error> {
        if lease.owner != self.owner {
            return Err(Error::ForeignOwner);
        }
        let record = self.live.get(lease.slot).ok_or(Error::StaleLease)?;
        if !record.active
            || record.address != lease.address
            || record.generation != lease.generation
        {
            return Err(Error::StaleLease);
        }
        Ok(())
    }
    pub fn validate_lease(&self, lease: &Lease) -> Result<(), Error> {
        self.audit()?;
        self.valid_lease(lease)
    }
    pub fn release(&mut self, lease: &Lease) -> Result<(), Error> {
        self.audit()?;
        self.valid_lease(lease)?;
        let target = self.page(lease.address)?;
        let free = self.free.add(target)?;
        self.free = free;
        self.live[lease.slot].active = false;
        Ok(())
    }
    pub fn reserve(&mut self, bytes: Region) -> Result<Reservation, Error> {
        self.audit()?;
        let region = touched_pages(bytes)?;
        if !Map::from_authority(&self.plan().authority).contains(region) {
            return Err(Error::OutsideAuthority);
        }
        if self.live.iter().any(|record| {
            record.active
                && region.overlaps(Region {
                    start: record.address,
                    end: record.address + PAGE,
                })
        }) {
            return Err(Error::LiveReservation);
        }
        if self
            .reserved
            .iter()
            .any(|record| record.active && region.overlaps(record.region))
        {
            return Err(Error::Reserved);
        }
        if !self.free.contains(region) {
            return Err(Error::NotFree);
        }
        let slot = self
            .reserved
            .iter()
            .position(|record| !record.active)
            .ok_or(Error::Capacity)?;
        let generation = self.generation.checked_add(1).ok_or(Error::Generation)?;
        let free = self.free.subtract(region)?;
        self.free = free;
        self.reserved[slot] = Reserved {
            region,
            generation,
            active: true,
        };
        self.generation = generation;
        Ok(Reservation {
            region,
            owner: self.owner,
            generation,
            slot,
        })
    }
    pub fn release_reservation(&mut self, token: &Reservation) -> Result<(), Error> {
        self.audit()?;
        if token.owner != self.owner {
            return Err(Error::ForeignOwner);
        }
        let record = self.reserved.get(token.slot).ok_or(Error::StaleLease)?;
        if !record.active || record.region != token.region || record.generation != token.generation
        {
            return Err(Error::StaleLease);
        }
        let free = self.free.add(token.region)?;
        self.free = free;
        self.reserved[token.slot].active = false;
        Ok(())
    }
    pub fn endpoints(&self, bank: Region) -> Result<(u64, u64), Error> {
        self.audit()?;
        let bank = full_pages(bank)?.ok_or(Error::NoMemory)?;
        let mut first = None;
        let mut last = None;
        for region in self.free.as_slice() {
            let start = region.start.max(bank.start);
            let end = region.end.min(bank.end);
            if start < end {
                first.get_or_insert(start);
                last = Some(end - PAGE);
            }
        }
        Ok((first.ok_or(Error::NoMemory)?, last.ok_or(Error::NoMemory)?))
    }
    pub fn prepare_page(&self, lease: &Lease, bytes: &mut [u8], canary: u64) -> Result<(), Error> {
        self.backing(lease, bytes)?;
        bytes.fill(0);
        if bytes.iter().any(|byte| *byte != 0) {
            return Err(Error::PageContents);
        }
        bytes[..8].copy_from_slice(&canary.to_le_bytes());
        bytes[PAGE as usize - 8..].copy_from_slice(&(!canary).to_le_bytes());
        self.verify_page(lease, bytes, canary)
    }
    fn backing(&self, lease: &Lease, bytes: &[u8]) -> Result<(), Error> {
        self.valid_lease(lease)?;
        if bytes.len() != PAGE as usize {
            return Err(Error::PageBytes);
        }
        if bytes.as_ptr() as u64 != lease.address {
            return Err(Error::BackingAddress);
        }
        Ok(())
    }
    pub fn verify_page(&self, lease: &Lease, bytes: &[u8], canary: u64) -> Result<(), Error> {
        self.backing(lease, bytes)?;
        if bytes[..8] != canary.to_le_bytes()
            || bytes[PAGE as usize - 8..] != (!canary).to_le_bytes()
            || bytes[8..PAGE as usize - 8].iter().any(|byte| *byte != 0)
        {
            return Err(Error::PageContents);
        }
        Ok(())
    }
    pub fn audit(&self) -> Result<Audit, Error> {
        if self.plan.is_none() || self.owner == 0 {
            return Err(Error::Invalid);
        }
        let authorized = self.plan().authority.pages();
        let allowed = Map::from_authority(&self.plan().authority);
        let free_pages = pages(self.free.as_slice())?;
        for (index, region) in self.free.as_slice().iter().enumerate() {
            if !allowed.contains(*region)
                || index > 0 && self.free.regions[index - 1].end > region.start
            {
                return Err(Error::Audit);
            }
        }
        let mut reserved_pages = 0u64;
        let mut live_pages = 0u64;
        for (index, record) in self
            .reserved
            .iter()
            .enumerate()
            .filter(|(_, record)| record.active)
        {
            if record.generation == 0
                || record.generation > self.generation
                || !allowed.contains(record.region)
                || self
                    .free
                    .as_slice()
                    .iter()
                    .any(|free| free.overlaps(record.region))
                || self.reserved[..index]
                    .iter()
                    .any(|prior| prior.active && prior.region.overlaps(record.region))
            {
                return Err(Error::Audit);
            }
            reserved_pages = reserved_pages
                .checked_add(pages(&[record.region])?)
                .ok_or(Error::CountOverflow)?;
        }
        for (index, record) in self
            .live
            .iter()
            .enumerate()
            .filter(|(_, record)| record.active)
        {
            let page = self.page(record.address).map_err(|_| Error::Audit)?;
            if record.generation == 0
                || record.generation > self.generation
                || self.free.as_slice().iter().any(|free| free.overlaps(page))
                || self
                    .reserved
                    .iter()
                    .any(|reserved| reserved.active && reserved.region.overlaps(page))
                || self.live[..index]
                    .iter()
                    .any(|prior| prior.active && prior.address == record.address)
            {
                return Err(Error::Audit);
            }
            live_pages = live_pages.checked_add(1).ok_or(Error::CountOverflow)?;
        }
        if free_pages
            .checked_add(reserved_pages)
            .and_then(|sum| sum.checked_add(live_pages))
            != Some(authorized)
            || self
                .plan()
                .firmware_withheld
                .checked_add(self.plan().boot_reserved)
                .and_then(|sum| sum.checked_add(authorized))
                != Some(self.plan().ram_full_pages)
        {
            return Err(Error::Audit);
        }
        Ok(Audit {
            ram_full_pages: self.plan().ram_full_pages,
            firmware_withheld_pages: self.plan().firmware_withheld,
            boot_reserved_pages: self.plan().boot_reserved,
            authorized_pages: authorized,
            free_pages,
            reserved_pages,
            live_pages,
        })
    }

    /// Cross-check an external consumer's physical inventory against this
    /// owner. Exact page membership, duplicates and gaps are checked, not
    /// merely caller-supplied totals. Split adjacent free extents are allowed.
    pub fn audit_observed(
        &self,
        free: &[Region],
        live: &[u64],
        reserved: &[Region],
    ) -> Result<Audit, Error> {
        let audit = self.audit()?;
        if free.len() > MAX_REGIONS || live.len() > MAX_LIVE || reserved.len() > MAX_RESERVATIONS {
            return Err(Error::Capacity);
        }
        for (index, region) in free.iter().enumerate() {
            if !self.free.contains(*region)
                || free[..index].iter().any(|prior| prior.overlaps(*region))
            {
                return Err(Error::Audit);
            }
        }
        if pages(free)? != audit.free_pages || live.len() as u64 != audit.live_pages {
            return Err(Error::Audit);
        }
        for (index, address) in live.iter().enumerate() {
            if live[..index].contains(address)
                || !self
                    .live
                    .iter()
                    .any(|record| record.active && record.address == *address)
            {
                return Err(Error::Audit);
            }
        }
        let mut reservation_map = Map {
            regions: [Region::EMPTY; MAX_REGIONS],
            len: 0,
        };
        for record in self.reserved.iter().filter(|record| record.active) {
            reservation_map = reservation_map.add(record.region)?;
        }
        for (index, region) in reserved.iter().enumerate() {
            if !reservation_map.contains(*region)
                || reserved[..index]
                    .iter()
                    .any(|prior| prior.overlaps(*region))
            {
                return Err(Error::Audit);
            }
        }
        if pages(reserved)? != audit.reserved_pages {
            return Err(Error::Audit);
        }
        Ok(audit)
    }
}