use core::{array, marker::PhantomData, mem};
pub use super::process_exit::{Fault, Outcome};
use crate::runtime::address_spaces as spaces;
use crate::runtime::frame_metadata as metadata;
use crate::runtime::process_threads as threads;

pub const MAX_PROCESSES: usize = 16;
pub const MAX_SPACES: usize = 2;
pub const MAX_HANDLES: usize = 8;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Error {
    Configuration,
    Context,
    Foreign,
    Stale,
    Capacity,
    Exhausted,
    State,
    Resource,
    Preparation,
    Publication,
    Busy,
    NoChild,
    Quarantined,
    Backend,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ProcessId {
    table: u64,
    slot: usize,
    generation: u64,
}
impl ProcessId {
    pub fn table(self) -> u64 {
        self.table
    }
    pub fn slot(self) -> usize {
        self.slot
    }
    pub fn generation(self) -> u64 {
        self.generation
    }
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Execution {
    pub cpu: u16,
    pub in_irq: bool,
    pub irq_masked: bool,
    pub current: Option<ProcessId>,
    pub kernel_bootstrap: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Spawn {
    pub handles: u8,
    pub code_entry: u64,
    pub user_stack_top: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Boundary {
    Before,
    After,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Failure {
    pub error: Error,
    pub boundary: Boundary,
}
pub struct BuildFailure<T> {
    pub failure: Failure,
    pub partial: Option<T>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BagObservation {
    pub process: ProcessId,
    pub resource_owner: u64,
    pub thread: threads::ThreadId,
    pub state: threads::State,
    pub spaces: [Option<spaces::Handle>; MAX_SPACES],
    pub handles: [Option<metadata::Handle>; MAX_HANDLES],
    pub pages: u64,
    pub image_ready: bool,
    pub stack_ready: bool,
    pub published: bool,
}
/// # Safety
/// Bag is a nonclone exclusive custodian of every partial resource. Observe
/// checks actual frozen Table/Spaces/Registry owners, not caller metadata.
/// Publish enqueues only a completely prepared real thread; exit makes that
/// thread Dead before returning. A Before error never mutates an owner.
/// Release proves inactive roots, no current SP/pin/IRQ user, erases and
/// returns all backing. Failure returns the identical Bag. Backend and Bag
/// Drop must not silently release resources. No mutable owner escapes.
pub unsafe trait Backend {
    type Bag;
    fn execution(&self) -> Execution;
    fn prepare(
        &mut self,
        id: ProcessId,
        spawn: Spawn,
    ) -> Result<Self::Bag, BuildFailure<Self::Bag>>;
    fn observe(&self, bag: &Self::Bag) -> Result<BagObservation, Error>;
    fn publish(&mut self, bag: &mut Self::Bag) -> Result<(), Failure>;
    fn exit(&mut self, bag: &mut Self::Bag, outcome: Outcome) -> Result<(), Failure>;
    fn release(&mut self, bag: Self::Bag) -> Result<(), (Error, Self::Bag)>;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Phase {
    Vacant,
    Building,
    Prepared,
    Published,
    Zombie,
    Quarantined,
}
#[derive(Debug, PartialEq, Eq)]
pub struct Reservation {
    table: u64,
    slot: usize,
    episode: u64,
}
impl Drop for Reservation {
    fn drop(&mut self) {}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Record {
    pub id: ProcessId,
    pub parent: Option<ProcessId>,
    /// Full identity of the parent whose successful exit transferred this child to INIT.
    pub orphaned_from: Option<ProcessId>,
    pub resources: BagObservation,
    pub exit_code: Option<i32>,
    pub outcome: Option<Outcome>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Snapshot {
    pub published: usize,
    pub constructing: usize,
    pub zombies: usize,
    pub quarantined: usize,
}
struct Entry<T> {
    generation: u64,
    episode: u64,
    phase: Phase,
    parent: Option<ProcessId>,
    orphaned_from: Option<ProcessId>,
    spawn: Option<Spawn>,
    bag: Option<T>,
    baseline: Option<BagObservation>,
    outcome: Option<Outcome>,
}
pub struct Table<B: Backend, const N: usize> {
    backend: Option<B>,
    identity: u64,
    cpu: u16,
    next_episode: u64,
    entries: [Entry<B::Bag>; N],
    _local: PhantomData<*mut ()>,
}
#[cfg(not(target_os = "none"))]
fn identity() -> Result<u64, Error> {
    use core::sync::atomic::{AtomicU64, Ordering};
    static NEXT: AtomicU64 = AtomicU64::new(0);
    NEXT.try_update(Ordering::AcqRel, Ordering::Acquire, |n| n.checked_add(1))
        .map(|n| n + 1)
        .map_err(|_| Error::Exhausted)
}
#[cfg(target_os = "none")]
fn identity() -> Result<u64, Error> {
    unsafe extern "C" {
        fn __os_k1120_process_identity() -> u64;
    }
    match unsafe { __os_k1120_process_identity() } {
        0 => Err(Error::Exhausted),
        n => Ok(n),
    }
}
impl<B: Backend, const N: usize> Table<B, N> {
    #[cfg(all(feature = "host-fixture", not(target_os = "none")))]
    pub fn process_test_backend(&mut self) -> &mut B {
        self.backend.as_mut().expect("owned test backend")
    }
    #[cfg(all(feature = "host-fixture", not(target_os = "none")))]
    pub fn process_test_bag(&self, slot: usize) -> Option<&B::Bag> {
        self.entries.get(slot).and_then(|entry| entry.bag.as_ref())
    }
    pub fn new(backend: B) -> Result<Self, (Error, B)> {
        let e = backend.execution();
        if N == 0 || N > MAX_PROCESSES {
            return Err((Error::Configuration, backend));
        }
        if e.cpu == u16::MAX || e.in_irq || !e.irq_masked || e.current.is_some() {
            return Err((Error::Context, backend));
        }
        let identity = match identity() {
            Ok(n) => n,
            Err(e) => return Err((e, backend)),
        };
        Ok(Self {
            backend: Some(backend),
            identity,
            cpu: e.cpu,
            next_episode: 0,
            entries: array::from_fn(|_| Entry {
                generation: 0,
                episode: 0,
                phase: Phase::Vacant,
                parent: None,
                orphaned_from: None,
                spawn: None,
                bag: None,
                baseline: None,
                outcome: None,
            }),
            _local: PhantomData,
        })
    }
    fn backend(&self) -> &B {
        self.backend.as_ref().expect("owned backend")
    }
    fn gate(&self) -> Result<Execution, Error> {
        let e = self.backend().execution();
        if e.cpu != self.cpu || e.in_irq || !e.irq_masked {
            return Err(Error::Context);
        }
        Ok(e)
    }
    fn public_index(&self, id: ProcessId) -> Result<usize, Error> {
        if id.table != self.identity {
            return Err(Error::Foreign);
        }
        let e = self.entries.get(id.slot).ok_or(Error::Stale)?;
        if e.generation != id.generation || id.generation == 0 {
            return Err(Error::Stale);
        }
        if !matches!(e.phase, Phase::Published | Phase::Zombie) {
            return Err(Error::Stale);
        }
        Ok(id.slot)
    }
    fn reservation_index(&self, r: &Reservation) -> Result<usize, Error> {
        if r.table != self.identity {
            return Err(Error::Foreign);
        }
        let e = self.entries.get(r.slot).ok_or(Error::Stale)?;
        if e.episode != r.episode || !matches!(e.phase, Phase::Building | Phase::Prepared) {
            return Err(if e.phase == Phase::Quarantined {
                Error::Quarantined
            } else {
                Error::Stale
            });
        }
        Ok(r.slot)
    }
    fn proposed(&self, i: usize) -> Result<ProcessId, Error> {
        Ok(ProcessId {
            table: self.identity,
            slot: i,
            generation: self.entries[i]
                .generation
                .checked_add(1)
                .ok_or(Error::Exhausted)?,
        })
    }
    pub fn begin(&mut self, spawn: Spawn) -> Result<Reservation, Error> {
        let e = self.gate()?;
        if spawn.handles == 0
            || usize::from(spawn.handles) > MAX_HANDLES
            || spawn.code_entry == 0
            || !spawn.code_entry.is_multiple_of(4)
            || spawn.user_stack_top == 0
            || !spawn.user_stack_top.is_multiple_of(16)
        {
            return Err(Error::Configuration);
        }
        let parent = if let Some(parent) = e.current {
            let r = self.record(parent)?;
            if r.outcome.is_some() || r.resources.state != threads::State::Running {
                return Err(Error::State);
            }
            Some(parent)
        } else {
            if !e.kernel_bootstrap {
                return Err(Error::Context);
            }
            if self
                .entries
                .iter()
                .any(|p| p.phase != Phase::Vacant && p.parent.is_none())
            {
                return Err(Error::Busy);
            }
            None
        };
        let i = self
            .entries
            .iter()
            .position(|p| p.phase == Phase::Vacant && p.generation != u64::MAX)
            .ok_or(Error::Capacity)?;
        let episode = self.next_episode.checked_add(1).ok_or(Error::Exhausted)?;
        self.entries[i].episode = episode;
        self.entries[i].phase = Phase::Building;
        self.entries[i].parent = parent;
        self.entries[i].orphaned_from = None;
        self.entries[i].spawn = Some(spawn);
        self.next_episode = episode;
        Ok(Reservation {
            table: self.identity,
            slot: i,
            episode,
        })
    }
    fn owner_context(&self, i: usize, e: Execution) -> Result<(), Error> {
        if let Some(parent) = self.entries[i].parent {
            if e.current != Some(parent) {
                return Err(Error::Foreign);
            }
            let r = self.record(parent)?;
            if r.resources.state != threads::State::Running || r.outcome.is_some() {
                return Err(Error::State);
            }
        } else if e.current.is_some() || !e.kernel_bootstrap {
            return Err(Error::Context);
        }
        Ok(())
    }
    fn validate_bag(&self, i: usize, o: BagObservation, prepared: bool) -> Result<(), Error> {
        let pid = if matches!(self.entries[i].phase, Phase::Published | Phase::Zombie) {
            ProcessId {
                table: self.identity,
                slot: i,
                generation: self.entries[i].generation,
            }
        } else {
            self.proposed(i)?
        };
        if o.process != pid || o.resource_owner == 0 || o.pages == 0 {
            return Err(Error::Resource);
        }
        let handles = usize::from(self.entries[i].spawn.ok_or(Error::State)?.handles);
        if prepared
            && (!o.image_ready
                || !o.stack_ready
                || o.spaces[0].is_none()
                || o.handles[..handles].iter().any(Option::is_none)
                || o.handles[handles..].iter().any(Option::is_some))
        {
            return Err(Error::Preparation);
        }
        for (a, h) in o.spaces.iter().enumerate() {
            if let Some(h) = h
                && (h.owner() != o.resource_owner
                    || h.generation() == 0
                    || o.spaces[..a].contains(&Some(*h)))
            {
                return Err(Error::Resource);
            }
        }
        for (a, h) in o.handles.iter().enumerate() {
            if let Some(h) = h
                && (h.generation() == 0 || o.handles[..a].contains(&Some(*h)))
            {
                return Err(Error::Resource);
            }
        }
        for (j, entry) in self.entries.iter().enumerate() {
            if j == i {
                continue;
            }
            if let Some(bag) = entry.bag.as_ref() {
                let other = self.backend().observe(bag)?;
                if other.thread == o.thread
                    || other.resource_owner == o.resource_owner
                    || o.spaces
                        .iter()
                        .flatten()
                        .any(|h| other.spaces.contains(&Some(*h)))
                    || o.handles
                        .iter()
                        .flatten()
                        .any(|h| other.handles.contains(&Some(*h)))
                {
                    return Err(Error::Resource);
                }
            }
        }
        Ok(())
    }
    fn clear_hidden(&mut self, i: usize) {
        let e = &mut self.entries[i];
        e.phase = Phase::Vacant;
        e.parent = None;
        e.orphaned_from = None;
        e.spawn = None;
        e.baseline = None;
        e.outcome = None;
    }
    fn release_owned(&mut self, i: usize) -> Result<(), Error> {
        let before = self.gate()?;
        if let Some(bag) = self.entries[i].bag.take() {
            match self.backend.as_mut().expect("owned backend").release(bag) {
                Ok(()) => (),
                Err((error, bag)) => {
                    self.entries[i].bag = Some(bag);
                    self.entries[i].phase = Phase::Quarantined;
                    return Err(error);
                }
            }
        }
        if self.backend().execution() != before {
            self.entries[i].phase = Phase::Quarantined;
            return Err(Error::Context);
        }
        Ok(())
    }
    pub fn prepare(&mut self, r: Reservation) -> Result<Reservation, (Error, Reservation)> {
        let perform = || -> Result<usize, Error> {
            let e = self.gate()?;
            let i = self.reservation_index(&r)?;
            self.owner_context(i, e)?;
            if self.entries[i].phase != Phase::Building || self.entries[i].bag.is_some() {
                return Err(Error::State);
            }
            Ok(i)
        };
        let i = match perform() {
            Ok(i) => i,
            Err(e) => return Err((e, r)),
        };
        let pid = match self.proposed(i) {
            Ok(p) => p,
            Err(e) => return Err((e, r)),
        };
        let spawn = self.entries[i].spawn.expect("reserved spawn");
        let before = self.backend().execution();
        match self
            .backend
            .as_mut()
            .expect("owned backend")
            .prepare(pid, spawn)
        {
            Ok(bag) => {
                self.entries[i].bag = Some(bag);
                let result = self
                    .backend()
                    .observe(self.entries[i].bag.as_ref().unwrap())
                    .and_then(|o| {
                        self.validate_bag(i, o, true)?;
                        if o.published || o.state != threads::State::Ready {
                            return Err(Error::State);
                        }
                        Ok(o)
                    });
                match result {
                    Ok(o) if self.backend().execution() == before => {
                        self.entries[i].baseline = Some(o);
                        self.entries[i].phase = Phase::Prepared;
                        Ok(r)
                    }
                    Ok(_) => {
                        self.entries[i].phase = Phase::Quarantined;
                        Err((Error::Context, r))
                    }
                    Err(e) => {
                        self.entries[i].phase = Phase::Quarantined;
                        Err((e, r))
                    }
                }
            }
            Err(f) => {
                self.entries[i].bag = f.partial;
                if f.failure.boundary == Boundary::After || self.backend().execution() != before {
                    self.entries[i].phase = Phase::Quarantined;
                    return Err((Error::Quarantined, r));
                }
                if let Err(e) = self.release_owned(i) {
                    return Err((e, r));
                }
                Err((f.failure.error, r))
            }
        }
    }
    pub fn abort(&mut self, r: Reservation) -> Result<(), (Error, Reservation)> {
        let result = || -> Result<usize, Error> {
            let e = self.gate()?;
            let i = self.reservation_index(&r)?;
            self.owner_context(i, e)?;
            Ok(i)
        };
        let i = match result() {
            Ok(i) => i,
            Err(e) => return Err((e, r)),
        };
        if let Err(e) = self.release_owned(i) {
            return Err((e, r));
        }
        self.clear_hidden(i);
        Ok(())
    }
    pub fn publish(&mut self, r: Reservation) -> Result<ProcessId, (Error, Reservation)> {
        let check = || -> Result<(usize, ProcessId, BagObservation), Error> {
            let e = self.gate()?;
            let i = self.reservation_index(&r)?;
            self.owner_context(i, e)?;
            if self.entries[i].phase != Phase::Prepared {
                return Err(Error::Preparation);
            }
            let o = self
                .backend()
                .observe(self.entries[i].bag.as_ref().ok_or(Error::Resource)?)?;
            self.validate_bag(i, o, true)?;
            if Some(o) != self.entries[i].baseline
                || o.published
                || o.state != threads::State::Ready
            {
                return Err(Error::Resource);
            }
            Ok((i, self.proposed(i)?, o))
        };
        let (i, pid, old) = match check() {
            Ok(v) => v,
            Err(e) => return Err((e, r)),
        };
        let before = self.backend().execution();
        let result = self
            .backend
            .as_mut()
            .expect("owned backend")
            .publish(self.entries[i].bag.as_mut().expect("prepared bag"));
        if let Err(f) = result {
            if f.boundary == Boundary::After || self.backend().execution() != before {
                self.entries[i].phase = Phase::Quarantined;
            }
            return Err((f.error, r));
        }
        let observation = self
            .backend()
            .observe(self.entries[i].bag.as_ref().unwrap());
        let mut expected = old;
        expected.published = true;
        if observation != Ok(expected) || self.backend().execution() != before {
            self.entries[i].phase = Phase::Quarantined;
            return Err((Error::Quarantined, r));
        }
        self.entries[i].generation = pid.generation;
        self.entries[i].phase = Phase::Published;
        self.entries[i].baseline = Some(expected);
        Ok(pid)
    }
    pub fn record(&self, pid: ProcessId) -> Result<Record, Error> {
        self.gate()?;
        let i = self.public_index(pid)?;
        let entry = &self.entries[i];
        let o = self
            .backend()
            .observe(entry.bag.as_ref().ok_or(Error::Resource)?)?;
        self.validate_bag(i, o, true)?;
        let old = entry.baseline.ok_or(Error::Resource)?;
        let mut stable = o;
        stable.state = old.state;
        if stable != old || !o.published {
            return Err(Error::Resource);
        }
        if !matches!(
            o.state,
            threads::State::Ready
                | threads::State::Running
                | threads::State::Sleeping
                | threads::State::Dead
        ) {
            return Err(Error::State);
        }
        if (entry.phase == Phase::Zombie) != (entry.outcome.is_some())
            || entry.outcome.is_some() && o.state != threads::State::Dead
        {
            return Err(Error::State);
        }
        Ok(Record {
            id: pid,
            parent: entry.parent,
            orphaned_from: entry.orphaned_from,
            resources: o,
            exit_code: entry.outcome.and_then(Outcome::exit_code),
            outcome: entry.outcome,
        })
    }
    fn finish_exit(&mut self, i: usize, outcome: Outcome) -> Result<(), Error> {
        let exiting = ProcessId {
            table: self.identity,
            slot: i,
            generation: self.entries[i].generation,
        };
        // Validate the entire adoption set before invoking the irreversible
        // backend exit. A hidden or quarantined child keeps its original owner.
        let mut adopt_to = None;
        if self.entries[i].parent.is_some() {
            let mut children = 0;
            for (slot, child) in self.entries.iter().enumerate() {
                if child.phase == Phase::Vacant || child.parent != Some(exiting) {
                    continue;
                }
                if !matches!(child.phase, Phase::Published | Phase::Zombie) {
                    return Err(Error::Busy);
                }
                let id = ProcessId {
                    table: self.identity,
                    slot,
                    generation: child.generation,
                };
                self.record(id)?;
                children += 1;
            }
            if children != 0 {
                let mut init = None;
                for (slot, root) in self.entries.iter().enumerate() {
                    if root.phase == Phase::Vacant || root.parent.is_some() {
                        continue;
                    }
                    if root.phase != Phase::Published || init.is_some() {
                        return Err(Error::State);
                    }
                    let id = ProcessId {
                        table: self.identity,
                        slot,
                        generation: root.generation,
                    };
                    let record = self.record(id)?;
                    if record.outcome.is_some()
                        || !matches!(record.resources.state,
                            threads::State::Ready | threads::State::Running)
                    {
                        return Err(Error::State);
                    }
                    init = Some(id);
                }
                adopt_to = Some(init.ok_or(Error::State)?);
            }
        }
        let before = self.backend().execution();
        let result = self
            .backend
            .as_mut()
            .expect("owned backend")
            .exit(self.entries[i].bag.as_mut().expect("published bag"), outcome);
        if let Err(f) = result {
            if f.boundary == Boundary::After || self.backend().execution() != before {
                self.entries[i].phase = Phase::Quarantined;
            }
            return Err(f.error);
        }
        let mut expected = self.entries[i].baseline.ok_or(Error::Resource)?;
        expected.state = threads::State::Dead;
        let mut expected_execution = before;
        if before.current == Some(expected.process) {
            expected_execution.current = None;
        }
        if self
            .backend()
            .observe(self.entries[i].bag.as_ref().unwrap())
            != Ok(expected)
            || self.backend().execution() != expected_execution
        {
            self.entries[i].phase = Phase::Quarantined;
            return Err(Error::Quarantined);
        }
        self.entries[i].baseline = Some(expected);
        self.entries[i].outcome = Some(outcome);
        self.entries[i].phase = Phase::Zombie;
        if let Some(init) = adopt_to {
            for child in &mut self.entries {
                if child.parent == Some(exiting) {
                    child.parent = Some(init);
                    child.orphaned_from = Some(exiting);
                }
            }
        }
        Ok(())
    }
    /// INIT observes orphans in publication order, including live children.
    /// A live oldest orphan blocks collection of a younger completed orphan.
    pub fn next_orphan(&self, init: ProcessId) -> Result<ProcessId, Error> {
        let execution = self.gate()?;
        if execution.current != Some(init) {
            return Err(Error::Foreign);
        }
        let root = self.record(init)?;
        if root.parent.is_some()
            || root.resources.state != threads::State::Running
            || root.outcome.is_some()
        {
            return Err(Error::State);
        }
        let oldest = self.entries.iter().enumerate()
            .filter(|(_, child)| {
                matches!(child.phase, Phase::Published | Phase::Zombie)
                    && child.parent == Some(init)
                    && child.orphaned_from.is_some()
            })
            .min_by_key(|(_, child)| child.episode)
            .map(|(slot, child)| ProcessId {
                table: self.identity,
                slot,
                generation: child.generation,
            })
            .ok_or(Error::NoChild)?;
        self.record(oldest)?;
        Ok(oldest)
    }
    pub fn exit_current(&mut self, code: i32) -> Result<(), Error> {
        let e = self.gate()?;
        let pid = e.current.ok_or(Error::Context)?;
        let r = self.record(pid)?;
        if r.resources.state != threads::State::Running || r.outcome.is_some() {
            return Err(Error::State);
        }
        let i = self.public_index(pid)?;
        self.finish_exit(i, Outcome::Exited(code))
    }
    pub fn fault_current(&mut self, fault: Fault) -> Result<(), Error> {
        let e = self.gate()?;
        let pid = e.current.ok_or(Error::Context)?;
        let r = self.record(pid)?;
        if r.resources.state != threads::State::Running || r.outcome.is_some() {
            return Err(Error::State);
        }
        let i = self.public_index(pid)?;
        self.finish_exit(i, Outcome::Fault(fault))
    }
    pub fn terminate_child(&mut self, child: ProcessId, code: i32) -> Result<(), Error> {
        let e = self.gate()?;
        let i = self.public_index(child)?;
        self.owner_context(i, e)?;
        let r = self.record(child)?;
        if r.outcome.is_some()
            || r.resources.state == threads::State::Running
            || e.current == Some(child)
        {
            return Err(Error::Busy);
        }
        self.finish_exit(i, Outcome::Exited(code))
    }
    pub fn wait_child(&mut self, child: ProcessId) -> Result<i32, Error> {
        let e = self.gate()?;
        let i = match self.public_index(child) {
            Ok(i) => i,
            Err(Error::Stale) => return Err(Error::NoChild),
            Err(e) => return Err(e),
        };
        self.owner_context(i, e)?;
        let r = self.record(child)?;
        if e.current == Some(child) || r.resources.state != threads::State::Dead {
            return Err(Error::Busy);
        }
        let code = match r.outcome {
            Some(Outcome::Exited(code)) => code,
            Some(Outcome::Fault(_)) => return Err(Error::State),
            None => return Err(Error::Busy),
        };
        if self
            .entries
            .iter()
            .any(|p| p.phase != Phase::Vacant && p.parent == Some(child))
        {
            return Err(Error::Busy);
        }
        self.release_owned(i)?;
        self.clear_hidden(i);
        Ok(code)
    }
    pub fn wait_child_outcome(&mut self, child: ProcessId) -> Result<Outcome, Error> {
        let e = self.gate()?;
        let i = match self.public_index(child) {
            Ok(i) => i,
            Err(Error::Stale) => return Err(Error::NoChild),
            Err(e) => return Err(e),
        };
        self.owner_context(i, e)?;
        let r = self.record(child)?;
        if e.current == Some(child) || r.resources.state != threads::State::Dead {
            return Err(Error::Busy);
        }
        let outcome = r.outcome.ok_or(Error::Busy)?;
        if self.entries.iter().any(|p| p.phase != Phase::Vacant && p.parent == Some(child)) {
            return Err(Error::Busy);
        }
        self.release_owned(i)?;
        self.clear_hidden(i);
        Ok(outcome)
    }
    pub fn snapshot(&self) -> Snapshot {
        let mut s = Snapshot::default();
        for e in &self.entries {
            match e.phase {
                Phase::Vacant => (),
                Phase::Building | Phase::Prepared => s.constructing += 1,
                Phase::Published => s.published += 1,
                Phase::Zombie => {
                    s.published += 1;
                    s.zombies += 1
                }
                Phase::Quarantined => s.quarantined += 1,
            }
        }
        s
    }
    pub fn published(&self) -> impl Iterator<Item = ProcessId> + '_ {
        self.entries
            .iter()
            .enumerate()
            .filter(|(_, e)| matches!(e.phase, Phase::Published | Phase::Zombie))
            .map(|(slot, e)| ProcessId {
                table: self.identity,
                slot,
                generation: e.generation,
            })
    }
    pub fn into_backend(mut self) -> Result<B, (Error, Self)> {
        match self.gate() {
            Ok(e) if e.current.is_none() => (),
            Ok(_) => return Err((Error::Context, self)),
            Err(e) => return Err((e, self)),
        }
        if self
            .entries
            .iter()
            .any(|e| e.phase != Phase::Vacant || e.bag.is_some())
        {
            return Err((Error::Busy, self));
        }
        Ok(self.backend.take().expect("owned backend"))
    }
}
impl<B: Backend, const N: usize> Drop for Table<B, N> {
    fn drop(&mut self) {
        for entry in &mut self.entries {
            if let Some(bag) = entry.bag.take() {
                mem::forget(bag);
            }
        }
        if let Some(backend) = self.backend.take() {
            mem::forget(backend);
        }
    }
}

#[cfg(test)]
mod private_tests {
    use super::*;
    struct Empty;
    unsafe impl Backend for Empty {
        type Bag = ();
        fn execution(&self) -> Execution {
            Execution {
                cpu: 0,
                in_irq: false,
                irq_masked: true,
                current: None,
                kernel_bootstrap: true,
            }
        }
        fn prepare(&mut self, _: ProcessId, _: Spawn) -> Result<(), BuildFailure<()>> {
            panic!("no acquisition expected")
        }
        fn observe(&self, _: &()) -> Result<BagObservation, Error> {
            panic!("no observation expected")
        }
        fn publish(&mut self, _: &mut ()) -> Result<(), Failure> {
            panic!("no publication expected")
        }
        fn exit(&mut self, _: &mut (), _: Outcome) -> Result<(), Failure> {
            panic!("no exit expected")
        }
        fn release(&mut self, _: ()) -> Result<(), (Error, ())> {
            panic!("no release expected")
        }
    }
    fn table() -> Table<Empty, 1> {
        Table::new(Empty).ok().unwrap()
    }
    fn spawn() -> Spawn {
        Spawn {
            handles: 1,
            code_entry: 0x10000,
            user_stack_top: 0x80000,
        }
    }
    #[test]
    fn generation_exhaustion_retires_vacant_slot_before_acquisition() {
        let mut t = table();
        t.entries[0].generation = u64::MAX;
        assert_eq!(t.begin(spawn()), Err(Error::Capacity));
        assert_eq!(t.entries[0].generation, u64::MAX);
        assert_eq!(t.snapshot(), Snapshot::default());
    }
    #[test]
    fn episode_exhaustion_does_not_publish_or_reserve_any_slot() {
        let mut t = table();
        t.next_episode = u64::MAX;
        assert_eq!(t.begin(spawn()), Err(Error::Exhausted));
        assert_eq!(t.snapshot(), Snapshot::default());
        assert_eq!(t.entries[0].episode, 0);
    }
    #[test]
    fn aborted_reservation_cannot_name_later_same_slot_generation_proposal() {
        let mut t = table();
        let r = t.begin(spawn()).unwrap();
        let old = Reservation {
            table: r.table,
            slot: r.slot,
            episode: r.episode,
        };
        t.abort(r).unwrap();
        let new = t.begin(spawn()).unwrap();
        assert_ne!(old.episode, new.episode);
        let before = t.snapshot();
        assert_eq!(t.abort(old).err().unwrap().0, Error::Stale);
        assert_eq!(t.snapshot(), before);
        t.abort(new).unwrap();
    }
    #[test]
    fn invalid_spawn_bounds_do_not_reserve_or_change_generation() {
        for s in [
            Spawn {
                handles: 0,
                ..spawn()
            },
            Spawn {
                handles: 9,
                ..spawn()
            },
            Spawn {
                code_entry: 0,
                ..spawn()
            },
            Spawn {
                code_entry: 0x10002,
                ..spawn()
            },
            Spawn {
                user_stack_top: 0,
                ..spawn()
            },
            Spawn {
                user_stack_top: 0x80008,
                ..spawn()
            },
        ] {
            let mut t = table();
            assert_eq!(t.begin(s), Err(Error::Configuration));
            assert_eq!(t.snapshot(), Snapshot::default());
            assert_eq!(t.entries[0].generation, 0);
        }
    }
}


impl<B: Backend, const N: usize> Table<B, N> {
    /// Construct directly in exclusive, aligned and uninitialized kernel storage.
    ///
    /// # Safety
    /// `target` must point to writable storage for one Self, be correctly
    /// aligned, and contain no live value. No other CPU or interrupt may use
    /// that storage during construction or for the resulting table lifetime.
    pub unsafe fn new_in_place(
        backend: B,
        target: *mut Self,
    ) -> Result<(), (Error, B)> {
        if target.is_null() || !(target as usize).is_multiple_of(core::mem::align_of::<Self>()) {
            return Err((Error::Configuration, backend));
        }
        let execution = backend.execution();
        if N == 0 || N > MAX_PROCESSES {
            return Err((Error::Configuration, backend));
        }
        if execution.cpu == u16::MAX || execution.in_irq || !execution.irq_masked
            || execution.current.is_some() {
            return Err((Error::Context, backend));
        }
        let owner = match identity() {
            Ok(owner) => owner,
            Err(error) => return Err((error, backend)),
        };
        unsafe {
            core::ptr::addr_of_mut!((*target).identity).write(owner);
            core::ptr::addr_of_mut!((*target).cpu).write(execution.cpu);
            core::ptr::addr_of_mut!((*target).next_episode).write(0);
            core::ptr::addr_of_mut!((*target)._local).write(PhantomData);
            let entries = core::ptr::addr_of_mut!((*target).entries).cast::<Entry<B::Bag>>();
            for index in 0..N {
                entries.add(index).write(Entry {
                    generation: 0,
                    episode: 0,
                    phase: Phase::Vacant,
                    parent: None,
                    orphaned_from: None,
                    spawn: None,
                    bag: None,
                    baseline: None,
                    outcome: None,
                });
            }
            core::ptr::addr_of_mut!((*target).backend).write(Some(backend));
        }
        Ok(())
    }
}

/// # Safety
/// Implementations derive the stack from a retained published owner. Landing
/// is registered only after an assembly switch to that stack. Departure is
/// recorded while still on it; neither call may invent a current PID.
pub unsafe trait LandingBackend: Backend {
    fn landing_stack(&self, id: ProcessId) -> Result<(u64, u64), Error>;
    fn suspended_child_stack(&self, child: ProcessId, parent: ProcessId,
        parent_owner: u64) -> Result<(u64, u64), Error>;
    unsafe fn register_actual_landing(&mut self, id: ProcessId) -> Result<(), Error>;
    unsafe fn release_actual_current_to_ready(&mut self) -> Result<(), Error>;
}

impl<B: LandingBackend, const N: usize> Table<B, N> {
    pub fn landing_stack(&self, id: ProcessId) -> Result<(u64, u64), Error> {
        let execution = self.gate()?;
        if execution.current.is_some() {
            return Err(Error::Context);
        }
        self.public_index(id)?;
        self.backend().landing_stack(id)
    }

    pub fn landing_stack_from_suspended_parent(&self,
        parent: ProcessId, child: ProcessId) -> Result<(u64, u64), Error> {
        let execution = self.gate()?;
        if execution.current.is_some() || execution.kernel_bootstrap {
            return Err(Error::Context);
        }
        let parent_record = self.record(parent)?;
        let child_record = self.record(child)?;
        if parent_record.resources.state != threads::State::Ready
            || child_record.resources.state != threads::State::Ready
            || child_record.parent != Some(parent)
            || parent_record.outcome.is_some() || child_record.outcome.is_some()
        { return Err(Error::State); }
        self.backend().suspended_child_stack(child, parent,
            parent_record.resources.resource_owner)
    }

    /// # Safety
    /// The caller must have switched to `id`'s retained stack first.
    pub unsafe fn register_actual_landing(&mut self, id: ProcessId) -> Result<(), Error> {
        self.public_index(id)?;
        unsafe {
            self.backend
                .as_mut()
                .expect("owned backend")
                .register_actual_landing(id)?;
        }
        if self.gate()?.current != Some(id) {
            return Err(Error::Context);
        }
        Ok(())
    }

    /// # Safety
    /// The caller must be on the actual current process stack and switch
    /// back to the bootstrap stack immediately after this returns.
    pub unsafe fn release_actual_current_to_ready(&mut self) -> Result<(), Error> {
        let execution = self.gate()?;
        if execution.current.is_none() {
            return Err(Error::Context);
        }
        unsafe {
            self.backend
                .as_mut()
                .expect("owned backend")
                .release_actual_current_to_ready()?;
        }
        if self.gate()?.current.is_some() {
            return Err(Error::Context);
        }
        Ok(())
    }
}

/// Read-only launch inputs from one published, actually landed process.
/// These scalars are not authority to switch roots; the single-CPU caller
/// must keep this Table and its retained Bag exclusive through EL0 return.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GuestLaunch {
    pub process: ProcessId,
    pub thread: threads::ThreadId,
    pub resource_owner: u64,
    pub user_root: u64,
    pub user_asid: u16,
    pub kernel_root: u64,
    pub code_frame: u64,
    pub data_frame: u64,
    pub stack_frame: u64,
    pub user_pc: u64,
    pub user_sp: u64,
    pub user_spsr: u64,
    pub kernel_stack_low: u64,
    pub kernel_stack_high: u64,
}

/// The backend validates real root descriptors, retained pins, frame owners,
/// thread stack and CPU context immediately before returning launch scalars.
pub trait GuestLaunchBackend: LandingBackend {
    fn current_launch(&self, bag: &Self::Bag, id: ProcessId)
        -> Result<GuestLaunch, Error>;
    fn target_digest(&self, bag: &Self::Bag, id: ProcessId) -> Result<u64, Error>;
}

pub trait GuestHeapBackend: GuestLaunchBackend {
    fn alloc_current_heap(&mut self, bag: &Self::Bag, id: ProcessId,
        bytes: usize, alignment: usize)
        -> Result<u64, turkos_os_syscall_abi::syscall_abi::Errno>;
    fn free_current_heap(&mut self, bag: &Self::Bag, id: ProcessId,
        address: u64)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno>;
}

/// A kernel-only two-page object. The backend owns physical frames while the
/// process table gates every map/unmap to the actual running process identity.
pub trait GuestSharedBackend: GuestLaunchBackend {
    type Shared;
    fn create_current_shared(&mut self, bag: &Self::Bag, id: ProcessId)
        -> Result<Self::Shared, turkos_os_syscall_abi::syscall_abi::Errno>;
    fn map_current_shared(&mut self, bag: &Self::Bag, id: ProcessId,
        shared: &Self::Shared, page: usize, slot: usize)
        -> Result<u64, turkos_os_syscall_abi::syscall_abi::Errno>;
    fn seal_current_shared(&mut self, bag: &Self::Bag, id: ProcessId,
        shared: &mut Self::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno>;
    fn request_writable_current_shared(&mut self, bag: &Self::Bag,
        id: ProcessId, shared: &Self::Shared, page: usize, slot: usize)
        -> Result<u64, turkos_os_syscall_abi::syscall_abi::Errno>;
    fn close_current_shared_handle(&mut self, bag: &Self::Bag,
        id: ProcessId, shared: &mut Self::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno>;
    fn probe_early_release_current_shared(&mut self, bag: &Self::Bag,
        id: ProcessId, shared: &Self::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno>;
    fn unmap_current_shared(&mut self, bag: &Self::Bag, id: ProcessId,
        shared: &Self::Shared, slot: usize)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno>;
    fn destroy_current_shared(&mut self, bag: &Self::Bag, id: ProcessId,
        shared: &mut Self::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno>;
}

impl<B: GuestSharedBackend, const N: usize> Table<B, N> {
    fn shared_current_index(&self) -> Result<(ProcessId, usize),
        turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let id = self.gate().map_err(|_| Errno::Access)?.current.ok_or(Errno::Access)?;
        let index = self.public_index(id).map_err(|_| Errno::Access)?;
        if self.record(id).map_err(|_| Errno::Access)?.resources.state
            != threads::State::Running { return Err(Errno::Access); }
        Ok((id, index))
    }

    pub fn create_current_shared(&mut self) -> Result<B::Shared,
        turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?.create_current_shared(bag, id)
    }

    pub fn map_current_shared(&mut self, shared: &B::Shared,
        page: usize, slot: usize) -> Result<u64,
        turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .map_current_shared(bag, id, shared, page, slot)
    }

    pub fn seal_current_shared(&mut self, shared: &mut B::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .seal_current_shared(bag, id, shared)
    }

    pub fn request_writable_current_shared(&mut self, shared: &B::Shared,
        page: usize, slot: usize) -> Result<u64,
        turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .request_writable_current_shared(bag, id, shared, page, slot)
    }

    pub fn close_current_shared_handle(&mut self, shared: &mut B::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .close_current_shared_handle(bag, id, shared)
    }

    pub fn probe_early_release_current_shared(&mut self, shared: &B::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .probe_early_release_current_shared(bag, id, shared)
    }

    pub fn unmap_current_shared(&mut self, shared: &B::Shared,
        slot: usize) -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .unmap_current_shared(bag, id, shared, slot)
    }

    pub fn destroy_current_shared(&mut self, shared: &mut B::Shared)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let (id, index) = self.shared_current_index()?;
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .destroy_current_shared(bag, id, shared)
    }
}

impl<B: GuestHeapBackend, const N: usize> Table<B, N> {
    pub fn alloc_current_heap(&mut self, bytes: usize, alignment: usize)
        -> Result<u64, turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let id = self.gate().map_err(|_| Errno::Access)?.current.ok_or(Errno::Access)?;
        let index = self.public_index(id).map_err(|_| Errno::Access)?;
        if self.record(id).map_err(|_| Errno::Access)?.resources.state
            != threads::State::Running { return Err(Errno::Access); }
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .alloc_current_heap(bag, id, bytes, alignment)
    }
    pub fn free_current_heap(&mut self, address: u64)
        -> Result<(), turkos_os_syscall_abi::syscall_abi::Errno> {
        use turkos_os_syscall_abi::syscall_abi::Errno;
        let id = self.gate().map_err(|_| Errno::Access)?.current.ok_or(Errno::Access)?;
        let index = self.public_index(id).map_err(|_| Errno::Access)?;
        if self.record(id).map_err(|_| Errno::Access)?.resources.state
            != threads::State::Running { return Err(Errno::Access); }
        let bag = self.entries[index].bag.as_ref().ok_or(Errno::Access)?;
        self.backend.as_mut().ok_or(Errno::Access)?
            .free_current_heap(bag, id, address)
    }
}

impl<B: GuestLaunchBackend, const N: usize> Table<B, N> {
    pub fn current_launch(&self, id: ProcessId) -> Result<GuestLaunch, Error> {
        let execution = self.gate()?;
        if execution.current != Some(id) || execution.kernel_bootstrap {
            return Err(Error::Context);
        }
        let index = self.public_index(id)?;
        if self.entries[index].phase != Phase::Published {
            return Err(Error::State);
        }
        let record = self.record(id)?;
        if record.resources.state != threads::State::Running {
            return Err(Error::State);
        }
        let bag = self.entries[index].bag.as_ref().ok_or(Error::Resource)?;
        let launch = self.backend().current_launch(bag, id)?;
        if launch.process != id || launch.thread != record.resources.thread
            || launch.resource_owner != record.resources.resource_owner
            || self.gate()? != execution
        {
            return Err(Error::Context);
        }
        Ok(launch)
    }
}


/// An actual current process may expose a bounded, owner-checked user range
/// to a kernel-owned buffer. No external output is part of this operation.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GuestCopyError {
    Fault,
    Context,
    Owner,
}

pub trait GuestCopyBackend: GuestLaunchBackend {
    fn copy_current_user_data(
        &self,
        bag: &Self::Bag,
        id: ProcessId,
        address: u64,
        output: &mut [u8],
    ) -> Result<(), GuestCopyError>;
    fn write_current_user_data(
        &self,
        bag: &Self::Bag,
        id: ProcessId,
        address: u64,
        input: &[u8],
    ) -> Result<(), GuestCopyError>;
}

impl<B: GuestCopyBackend, const N: usize> Table<B, N> {
    pub fn copy_current_user_data(
        &self,
        id: ProcessId,
        address: u64,
        output: &mut [u8],
    ) -> Result<(), GuestCopyError> {
        let before = self.gate().map_err(|_| GuestCopyError::Context)?;
        if before.current != Some(id) || before.kernel_bootstrap {
            return Err(GuestCopyError::Context);
        }
        self.current_launch(id)
            .map_err(|_| GuestCopyError::Context)?;
        let index = self.public_index(id).map_err(|_| GuestCopyError::Owner)?;
        let entry = &self.entries[index];
        if entry.phase != Phase::Published {
            return Err(GuestCopyError::Owner);
        }
        let bag = entry.bag.as_ref().ok_or(GuestCopyError::Owner)?;
        self.backend()
            .copy_current_user_data(bag, id, address, output)?;
        if self.gate() != Ok(before) {
            return Err(GuestCopyError::Context);
        }
        Ok(())
    }

    pub fn write_current_user_data(
        &self,
        id: ProcessId,
        address: u64,
        input: &[u8],
    ) -> Result<(), GuestCopyError> {
        let before = self.gate().map_err(|_| GuestCopyError::Context)?;
        if before.current != Some(id) || before.kernel_bootstrap {
            return Err(GuestCopyError::Context);
        }
        self.current_launch(id)
            .map_err(|_| GuestCopyError::Context)?;
        let index = self.public_index(id).map_err(|_| GuestCopyError::Owner)?;
        let entry = &self.entries[index];
        if entry.phase != Phase::Published {
            return Err(GuestCopyError::Owner);
        }
        let bag = entry.bag.as_ref().ok_or(GuestCopyError::Owner)?;
        self.backend()
            .write_current_user_data(bag, id, address, input)?;
        if self.gate() != Ok(before) {
            return Err(GuestCopyError::Context);
        }
        Ok(())
    }
}


impl<B: GuestLaunchBackend, const N: usize> Table<B, N> {
    /// Kernel-only digest of a published target's owned user backing.
    pub fn target_digest(&self, id: ProcessId) -> Result<u64, Error> {
        let execution = self.gate()?;
        let index = self.public_index(id)?;
        if self.entries[index].phase != Phase::Published { return Err(Error::State); }
        let record = self.record(id)?;
        if !record.resources.image_ready || !record.resources.published
            || record.resources.handles.iter().flatten().count() != 3
        { return Err(Error::Resource); }
        let bag = self.entries[index].bag.as_ref().ok_or(Error::Resource)?;
        let digest = self.backend().target_digest(bag, id)?;
        if self.gate()? != execution || self.record(id) != Ok(record)
        { return Err(Error::Context); }
        Ok(digest)
    }
}