//! Observed scheduler fairness, owner chains and two distinct time ledgers.
//! No guest acceptance: actual assembly and one hour elapsed time need their
//! own later guest evidence. Tick ownership is independent of this task owner.
use core::{array, marker::PhantomData};
use turkos_os_runqueue::runqueue::Class;
use turkos_os_threads::threads::{self as thread, ThreadId};

pub const MAX_MEMBERS: usize = 32;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Error {
    Configuration,
    Context,
    Cpu,
    Capacity,
    Duplicate,
    Foreign,
    Stale,
    Exhausted,
    Tick,
    Clock,
    Guard,
    State,
    Fairness,
    OwnerChain,
    BusyLoop,
    Resource,
    Backend,
    Busy,
    Duration,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Execution {
    pub cpu: u32,
    pub masked: bool,
    pub in_irq: bool,
    pub current: Option<ThreadId>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Observation {
    pub thread: ThreadId,
    pub state: thread::State,
    pub class: Class,
    pub slice: u64,
    pub slices: u64,
    pub progress: u64,
    pub wait_epoch: u64,
    pub owner: Option<ThreadId>,
    pub pages: u64,
    pub owner_fingerprint: u64,
    pub guard_descriptor: u64,
    pub low_canary: u64,
    pub high_canary: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TickObservation {
    pub cpu: u32,
    pub source: u64,
    pub sequence: u64,
    pub counter: u64,
    pub queue_epoch: u64,
    pub current: Option<ThreadId>,
}
/// # Safety
/// This backend owns real nonclone membership pins for the frozen queue,
/// context and Table. observe validates the actual Table generation, queue
/// record, physical guard/canaries and private page-owner fingerprint. Owner
/// chains come from live mutex wait ownership, never a caller supplied ID.
/// Tick owns one actual independent timer ingress delivery. Its CPU, source,
/// sequence, counter, current ThreadId and queue epoch were captured without
/// borrowing this backend in IRQ. epoch and observe belong to the same current
/// task safe point with IRQs masked. consume_tick is infallible, consumes that
/// exact delivery once, and cannot mutate Table, queue or resource ownership.
/// Members have inert destructors; failed registration returns the actual pin.
/// No mutable backend/Policy loan spans assembly or an unmasked IRQ.
pub unsafe trait Backend {
    type Member;
    type Tick;
    fn execution(&self) -> Execution;
    fn epoch(&self) -> u64;
    fn source(&self) -> u64;
    fn frequency(&self) -> u64;
    fn counter(&self) -> u64;
    fn observe(&self, member: &Self::Member) -> Result<Observation, Error>;
    fn tick(&self, tick: &Self::Tick) -> Result<TickObservation, Error>;
    fn consume_tick(&mut self, tick: Self::Tick);
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Handle {
    identity: u64,
    slot: usize,
    generation: u64,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct TaskTime {
    pub running: u64,
    pub sleeping: u64,
    pub ready: u64,
    pub dead: u64,
}
impl TaskTime {
    pub fn total(self) -> Option<u64> {
        self.running
            .checked_add(self.sleeping)?
            .checked_add(self.ready)?
            .checked_add(self.dead)
    }
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Record {
    pub handle: Handle,
    pub thread: ThreadId,
    pub observation: Observation,
    pub time: TaskTime,
    pub first_tick: u64,
    pub eligible_since: u64,
    pub last_selected: u64,
    pub inversion_ticks: u64,
    pub last_chain: [Option<ThreadId>; MAX_MEMBERS],
    pub chain_len: usize,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Snapshot {
    pub members: usize,
    pub ticks: u64,
    pub cpu_running: u64,
    pub cpu_idle: u64,
    pub last_sequence: u64,
    pub last_counter: u64,
    pub queue_epoch: u64,
    pub maximum_counter_gap: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct RunMember {
    handle: Handle,
    thread: ThreadId,
    pages: u64,
    fingerprint: u64,
    progress: u64,
    running: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LongRun {
    identity: u64,
    episode: u64,
    start: u64,
    finish_after: u64,
    first_tick: u64,
    owner_pages: u64,
    generation_sum: u64,
    members: usize,
    identities: [Option<RunMember>; MAX_MEMBERS],
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct WorkloadMember {
    handle: Handle,
    thread: ThreadId,
    progress: u64,
    last_progress_counter: u64,
    progress_samples: u64,
    maximum_progress_gap: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct Workload<const N: usize> {
    episode: u64,
    members: [Option<WorkloadMember>; N],
    failed: Option<Error>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WorkloadStatus {
    pub episode: u64,
    pub members: usize,
    pub failed: Option<Error>,
    pub minimum_progress_samples: u64,
    pub maximum_progress_gap: u64,
}
struct Entry<M> {
    member: Option<M>,
    record: Option<Record>,
    generation: u64,
}
pub struct Policy<B: Backend, const N: usize> {
    backend: B,
    identity: u64,
    cpu: u32,
    source: u64,
    frequency: u64,
    entries: [Entry<B::Member>; N],
    snapshot: Snapshot,
    workload: Option<Workload<N>>,
    next_episode: u64,
    non_send: PhantomData<*mut ()>,
}
#[cfg(not(target_os = "none"))]
fn identity() -> Result<u64, Error> {
    use core::sync::atomic::{AtomicU64, Ordering};
    static NEXT: AtomicU64 = AtomicU64::new(1);
    NEXT.try_update(Ordering::Relaxed, Ordering::Relaxed, |v| v.checked_add(1))
        .map_err(|_| Error::Exhausted)
}
#[cfg(target_os = "none")]
fn identity() -> Result<u64, Error> {
    unsafe extern "C" {
        fn __os_k1100_policy_identity() -> u64;
    }
    let value = unsafe { __os_k1100_policy_identity() };
    if value == 0 {
        Err(Error::Exhausted)
    } else {
        Ok(value)
    }
}
impl<B: Backend, const N: usize> Policy<B, N> {
    pub fn new(backend: B) -> Result<Self, (Error, B)> {
        if N == 0 || N > MAX_MEMBERS {
            return Err((Error::Configuration, backend));
        }
        let e = backend.execution();
        if !e.masked || e.in_irq {
            return Err((Error::Context, backend));
        }
        let source = backend.source();
        let frequency = backend.frequency();
        if source == 0 || frequency == 0 {
            return Err((Error::Configuration, backend));
        }
        let id = match identity() {
            Ok(i) => i,
            Err(e) => return Err((e, backend)),
        };
        let snapshot = Snapshot {
            last_counter: backend.counter(),
            queue_epoch: backend.epoch(),
            ..Snapshot::default()
        };
        Ok(Self {
            backend,
            identity: id,
            cpu: e.cpu,
            source,
            frequency,
            entries: array::from_fn(|_| Entry {
                member: None,
                record: None,
                generation: 0,
            }),
            snapshot,
            workload: None,
            next_episode: 0,
            non_send: PhantomData,
        })
    }
    fn gate(&self) -> Result<Execution, Error> {
        let e = self.backend.execution();
        if e.cpu != self.cpu {
            return Err(Error::Cpu);
        }
        if !e.masked || e.in_irq {
            return Err(Error::Context);
        }
        if self.backend.source() != self.source || self.backend.frequency() != self.frequency {
            return Err(Error::Clock);
        }
        Ok(e)
    }
    fn valid(o: Observation) -> Result<(), Error> {
        if o.guard_descriptor != 0
            || o.low_canary != thread::LOW_CANARY
            || o.high_canary != thread::HIGH_CANARY
        {
            return Err(Error::Guard);
        }
        if o.pages == 0 || o.owner_fingerprint == 0 {
            return Err(Error::Resource);
        }
        if !matches!(
            o.state,
            thread::State::Ready
                | thread::State::Running
                | thread::State::Sleeping
                | thread::State::Dead
        ) {
            return Err(Error::State);
        }
        if o.owner.is_some() && o.wait_epoch == 0 {
            return Err(Error::OwnerChain);
        }
        Ok(())
    }
    fn observed_record(old: Observation, o: Observation, epoch: u64) -> Result<(), Error> {
        Self::valid(o)?;
        if o.thread != old.thread {
            return Err(Error::Stale);
        }
        if o.pages != old.pages || o.owner_fingerprint != old.owner_fingerprint {
            return Err(Error::Resource);
        }
        if o.class != old.class
            || o.slices < old.slices
            || o.slice < old.slice
            || o.progress < old.progress
            || o.wait_epoch < old.wait_epoch
            || (old.state == thread::State::Dead && o.state != thread::State::Dead)
        {
            return Err(Error::State);
        }
        if o.slice > epoch
            || (o.slices == 0) != (o.slice == 0)
            || (o.slices == old.slices && o.slice != old.slice)
            || (o.slices > old.slices
                && o.slice.checked_sub(old.slice).ok_or(Error::State)? < o.slices - old.slices)
        {
            return Err(Error::State);
        }
        Ok(())
    }
    fn live_records(&self) -> Result<[Option<Observation>; N], Error> {
        let e = self.gate()?;
        let epoch = self.backend.epoch();
        if epoch < self.snapshot.queue_epoch || self.backend.counter() < self.snapshot.last_counter
        {
            return Err(Error::Clock);
        }
        let mut records = [None; N];
        let mut running = None;
        for (i, entry) in self.entries.iter().enumerate() {
            if let Some(member) = entry.member.as_ref() {
                let old = entry.record.ok_or(Error::State)?.observation;
                let o = self.backend.observe(member)?;
                Self::observed_record(old, o, epoch)?;
                if o.state == thread::State::Running && running.replace(o.thread).is_some() {
                    return Err(Error::State);
                }
                records[i] = Some(o);
            }
        }
        if running != e.current {
            return Err(Error::State);
        }
        Ok(records)
    }
    pub fn register(&mut self, member: B::Member) -> Result<Handle, (Error, B::Member)> {
        if let Err(e) = self.gate() {
            return Err((e, member));
        }
        let o = match self.backend.observe(&member) {
            Ok(o) => o,
            Err(e) => return Err((e, member)),
        };
        if let Err(e) = Self::valid(o) {
            return Err((e, member));
        }
        if self
            .entries
            .iter()
            .any(|r| r.record.is_some_and(|r| r.thread == o.thread))
        {
            return Err((Error::Duplicate, member));
        }
        let Some(i) = self
            .entries
            .iter()
            .position(|e| e.member.is_none() && e.generation != u64::MAX)
        else {
            return Err((Error::Capacity, member));
        };
        let generation = self.entries[i].generation + 1;
        let handle = Handle {
            identity: self.identity,
            slot: i,
            generation,
        };
        let epoch = self.backend.epoch();
        if epoch < self.snapshot.queue_epoch || o.slice > epoch || (o.slices == 0) != (o.slice == 0)
        {
            return Err((Error::State, member));
        }
        self.entries[i] = Entry {
            member: Some(member),
            generation,
            record: Some(Record {
                handle,
                thread: o.thread,
                observation: o,
                time: TaskTime::default(),
                first_tick: self.snapshot.ticks,
                eligible_since: epoch,
                last_selected: epoch,
                inversion_ticks: 0,
                last_chain: [None; MAX_MEMBERS],
                chain_len: 0,
            }),
        };
        self.snapshot.members += 1;
        Ok(handle)
    }
    fn index(&self, h: Handle) -> Result<usize, Error> {
        if h.identity != self.identity {
            return Err(Error::Foreign);
        }
        let e = self.entries.get(h.slot).ok_or(Error::Stale)?;
        if e.generation != h.generation || e.member.is_none() {
            return Err(Error::Stale);
        }
        Ok(h.slot)
    }
    pub fn record(&self, h: Handle) -> Result<Record, Error> {
        self.entries[self.index(h)?].record.ok_or(Error::Stale)
    }
    pub fn snapshot(&self) -> Snapshot {
        self.snapshot
    }
    pub fn charge(&mut self, tick: B::Tick) -> Result<Snapshot, (Error, B::Tick)> {
        let e = match self.gate() {
            Ok(e) => e,
            Err(e) => return Err((e, tick)),
        };
        let t = match self.backend.tick(&tick) {
            Ok(t) => t,
            Err(e) => return Err((e, tick)),
        };
        let expected = match self.snapshot.last_sequence.checked_add(1) {
            Some(n) => n,
            None => return Err((Error::Exhausted, tick)),
        };
        if t.cpu != self.cpu
            || t.source != self.source
            || t.sequence != expected
            || t.current != e.current
            || t.queue_epoch != self.backend.epoch()
        {
            return Err((Error::Tick, tick));
        }
        if t.counter <= self.snapshot.last_counter
            || t.counter > self.backend.counter()
            || t.queue_epoch < self.snapshot.queue_epoch
        {
            return Err((Error::Clock, tick));
        }
        let result = self.preflight(t, e);
        let (records, snapshot) = match result {
            Ok(v) => v,
            Err(e) => return Err((e, tick)),
        };
        let workload = match self.observe_workload(&records, t.counter) {
            Ok(workload) => workload,
            Err(error) => return Err((error, tick)),
        };
        self.backend.consume_tick(tick);
        for (i, r) in records.into_iter().enumerate() {
            self.entries[i].record = r;
        }
        self.snapshot = snapshot;
        self.workload = workload;
        Ok(snapshot)
    }
    fn preflight(
        &self,
        t: TickObservation,
        e: Execution,
    ) -> Result<([Option<Record>; N], Snapshot), Error> {
        let mut records = array::from_fn(|i| self.entries[i].record);
        let mut running = None;
        for (i, entry) in self.entries.iter().enumerate() {
            let Some(member) = entry.member.as_ref() else {
                continue;
            };
            let o = self.backend.observe(member)?;
            let r = records[i].as_mut().ok_or(Error::State)?;
            let old = r.observation;
            Self::observed_record(old, o, t.queue_epoch)?;
            if o.slices > old.slices {
                r.last_selected = o.slice;
            }
            if matches!(o.state, thread::State::Ready | thread::State::Running)
                && !matches!(old.state, thread::State::Ready | thread::State::Running)
            {
                r.eligible_since = t.queue_epoch;
                r.last_selected = t.queue_epoch;
            }
            if o.state == thread::State::Running && running.replace(o.thread).is_some() {
                return Err(Error::State);
            }
            let count = match o.state {
                thread::State::Running => &mut r.time.running,
                thread::State::Sleeping => &mut r.time.sleeping,
                thread::State::Ready => &mut r.time.ready,
                thread::State::Dead => &mut r.time.dead,
                _ => return Err(Error::State),
            };
            *count = count.checked_add(1).ok_or(Error::Exhausted)?;
            r.observation = o;
        }
        if running != e.current {
            return Err(Error::State);
        }
        let eligible = records
            .iter()
            .flatten()
            .filter(|r| {
                matches!(
                    r.observation.state,
                    thread::State::Ready | thread::State::Running
                )
            })
            .count() as u64;
        for i in 0..N {
            let Some(mut r) = records[i] else {
                continue;
            };
            if matches!(
                r.observation.state,
                thread::State::Ready | thread::State::Running
            ) && t
                .queue_epoch
                .checked_sub(r.last_selected)
                .ok_or(Error::Clock)?
                > eligible + 1
            {
                return Err(Error::Fairness);
            }
            r.last_chain = [None; MAX_MEMBERS];
            r.chain_len = 0;
            let mut owner = r.observation.owner;
            let mut inversion = false;
            while let Some(id) = owner {
                if id == r.thread
                    || r.last_chain[..r.chain_len].contains(&Some(id))
                    || r.chain_len >= N
                {
                    return Err(Error::OwnerChain);
                }
                let o = records
                    .iter()
                    .flatten()
                    .find(|p| p.thread == id)
                    .ok_or(Error::OwnerChain)?
                    .observation;
                if o.state == thread::State::Dead {
                    return Err(Error::OwnerChain);
                }
                r.last_chain[r.chain_len] = Some(id);
                r.chain_len += 1;
                if r.observation.class == Class::Normal && o.class == Class::Low {
                    if o.state == thread::State::Sleeping
                        && r.observation.state == thread::State::Running
                    {
                        return Err(Error::BusyLoop);
                    }
                    inversion = true;
                }
                owner = o.owner;
            }
            if inversion {
                r.inversion_ticks = r.inversion_ticks.checked_add(1).ok_or(Error::Exhausted)?;
            }
            records[i] = Some(r);
        }
        let mut s = self.snapshot;
        s.ticks = s.ticks.checked_add(1).ok_or(Error::Exhausted)?;
        let count = if e.current.is_some() {
            &mut s.cpu_running
        } else {
            &mut s.cpu_idle
        };
        *count = count.checked_add(1).ok_or(Error::Exhausted)?;
        if s.cpu_running.checked_add(s.cpu_idle) != Some(s.ticks) {
            return Err(Error::State);
        }
        for r in records.iter().flatten() {
            if r.time.total() != s.ticks.checked_sub(r.first_tick) {
                return Err(Error::State);
            }
        }
        s.maximum_counter_gap = s.maximum_counter_gap.max(t.counter - s.last_counter);
        s.last_sequence = t.sequence;
        s.last_counter = t.counter;
        s.queue_epoch = t.queue_epoch;
        Ok((records, s))
    }
    fn live_workload_state(o: Observation) -> Result<(), Error> {
        if !matches!(
            o.state,
            thread::State::Ready | thread::State::Running | thread::State::Sleeping
        ) {
            return Err(Error::State);
        }
        Ok(())
    }
    fn observe_workload(
        &self,
        records: &[Option<Record>; N],
        counter: u64,
    ) -> Result<Option<Workload<N>>, Error> {
        let Some(mut workload) = self.workload else {
            return Ok(None);
        };
        let interval = self.frequency.checked_mul(60).ok_or(Error::Exhausted)?;
        for (slot, member) in workload.members.iter_mut().enumerate() {
            let Some(member) = member else {
                continue;
            };
            let Some(record) = records[slot] else {
                workload.failed.get_or_insert(Error::Resource);
                continue;
            };
            if record.handle != member.handle || record.thread != member.thread {
                workload.failed.get_or_insert(Error::Resource);
                continue;
            }
            if let Err(error) = Self::live_workload_state(record.observation) {
                workload.failed.get_or_insert(error);
            }
            let gap = counter
                .checked_sub(member.last_progress_counter)
                .ok_or(Error::Clock)?;
            member.maximum_progress_gap = member.maximum_progress_gap.max(gap);
            // Check the old deadline first. A late progress update cannot erase
            // an earlier interval with no observed useful work.
            if gap > interval {
                workload.failed.get_or_insert(Error::Fairness);
            }
            if record.observation.progress > member.progress {
                member.progress_samples = member
                    .progress_samples
                    .checked_add(1)
                    .ok_or(Error::Exhausted)?;
                member.progress = record.observation.progress;
                member.last_progress_counter = counter;
            }
        }
        Ok(Some(workload))
    }
    pub fn workload_status(&self, run: &LongRun) -> Result<WorkloadStatus, Error> {
        if run.identity != self.identity {
            return Err(Error::Foreign);
        }
        let workload = self.workload.as_ref().ok_or(Error::Stale)?;
        if run.episode != workload.episode {
            return Err(Error::Stale);
        }
        let mut minimum = u64::MAX;
        let mut maximum = 0;
        let mut members = 0;
        for m in workload.members.iter().flatten() {
            members += 1;
            minimum = minimum.min(m.progress_samples);
            maximum = maximum.max(m.maximum_progress_gap);
        }
        Ok(WorkloadStatus {
            episode: workload.episode,
            members,
            failed: workload.failed,
            minimum_progress_samples: minimum,
            maximum_progress_gap: maximum,
        })
    }
    pub fn unregister(&mut self, h: Handle) -> Result<B::Member, Error> {
        let e = self.gate()?;
        let i = self.index(h)?;
        let o = self
            .backend
            .observe(self.entries[i].member.as_ref().ok_or(Error::Stale)?)?;
        Self::observed_record(
            self.entries[i].record.ok_or(Error::Stale)?.observation,
            o,
            self.backend.epoch(),
        )?;
        if o.state != thread::State::Dead || o.owner.is_some() || e.current == Some(o.thread) {
            return Err(Error::Busy);
        }
        for (j, entry) in self.entries.iter().enumerate() {
            if j != i
                && let Some(member) = entry.member.as_ref()
            {
                let other = self.backend.observe(member)?;
                Self::observed_record(
                    entry.record.ok_or(Error::State)?.observation,
                    other,
                    self.backend.epoch(),
                )?;
                if other.owner == Some(o.thread) {
                    return Err(Error::Busy);
                }
            }
        }
        let member = self.entries[i].member.take().ok_or(Error::Stale)?;
        self.entries[i].record = None;
        self.snapshot.members -= 1;
        Ok(member)
    }
    pub fn begin_long_run(&mut self) -> Result<LongRun, Error> {
        self.gate()?;
        if self.snapshot.members < 3 {
            return Err(Error::Configuration);
        }
        let live = self.live_records()?;
        for observation in live.iter().flatten() {
            Self::live_workload_state(*observation)?;
        }
        let start = self.backend.counter();
        if start < self.snapshot.last_counter {
            return Err(Error::Clock);
        }
        let duration = self.frequency.checked_mul(3600).ok_or(Error::Exhausted)?;
        let finish_after = start.checked_add(duration).ok_or(Error::Exhausted)?;
        let (owner_pages, generation_sum) = self.inventory()?;
        let mut identities = [None; MAX_MEMBERS];
        for (i, entry) in self.entries.iter().enumerate() {
            if let Some(o) = live[i] {
                let r = entry.record.ok_or(Error::State)?;
                identities[i] = Some(RunMember {
                    handle: r.handle,
                    thread: o.thread,
                    pages: o.pages,
                    fingerprint: o.owner_fingerprint,
                    progress: o.progress,
                    running: r.time.running,
                });
            }
        }
        let episode = self.next_episode.checked_add(1).ok_or(Error::Exhausted)?;
        let workload = Workload {
            episode,
            failed: None,
            members: array::from_fn(|i| {
                identities[i].map(|member| WorkloadMember {
                    handle: member.handle,
                    thread: member.thread,
                    progress: member.progress,
                    last_progress_counter: start,
                    progress_samples: 0,
                    maximum_progress_gap: 0,
                })
            }),
        };
        self.workload = Some(workload);
        self.next_episode = episode;
        Ok(LongRun {
            identity: self.identity,
            episode,
            start,
            finish_after,
            first_tick: self.snapshot.ticks,
            owner_pages,
            generation_sum,
            members: self.snapshot.members,
            identities,
        })
    }
    fn inventory(&self) -> Result<(u64, u64), Error> {
        let mut pages = 0u64;
        let mut generations = 0u64;
        for entry in &self.entries {
            if let Some(member) = entry.member.as_ref() {
                let o = self.backend.observe(member)?;
                Self::valid(o)?;
                let r = entry.record.ok_or(Error::State)?;
                if o.thread != r.thread
                    || o.owner_fingerprint != r.observation.owner_fingerprint
                    || o.pages != r.observation.pages
                {
                    return Err(Error::Resource);
                }
                pages = pages.checked_add(o.pages).ok_or(Error::Exhausted)?;
                generations = generations
                    .checked_add(o.thread.generation())
                    .ok_or(Error::Exhausted)?;
            }
        }
        Ok((pages, generations))
    }
    pub fn finish_long_run(&self, run: &LongRun) -> Result<Snapshot, Error> {
        self.gate()?;
        if run.identity != self.identity {
            return Err(Error::Foreign);
        }
        let live = self.live_records()?;
        for observation in live.iter().flatten() {
            Self::live_workload_state(*observation)?;
        }
        let status = self.workload_status(run)?;
        let elapsed = self
            .backend
            .counter()
            .checked_sub(run.start)
            .ok_or(Error::Clock)?;
        if elapsed < self.frequency.checked_mul(3600).ok_or(Error::Exhausted)?
            || self.snapshot.last_counter < run.finish_after
            || self
                .snapshot
                .ticks
                .checked_sub(run.first_tick)
                .ok_or(Error::Clock)?
                < 60
            || self.snapshot.maximum_counter_gap
                > self.frequency.checked_mul(60).ok_or(Error::Exhausted)?
        {
            return Err(Error::Duration);
        }
        if let Some(error) = status.failed {
            return Err(error);
        }
        if status.minimum_progress_samples < 60
            || status.maximum_progress_gap
                > self.frequency.checked_mul(60).ok_or(Error::Exhausted)?
        {
            return Err(Error::Fairness);
        }
        let workload = self.workload.as_ref().ok_or(Error::Stale)?;
        let now = self.backend.counter();
        for member in workload.members.iter().flatten() {
            if now
                .checked_sub(member.last_progress_counter)
                .ok_or(Error::Clock)?
                > self.frequency.checked_mul(60).ok_or(Error::Exhausted)?
            {
                return Err(Error::Fairness);
            }
        }
        if self.snapshot.members != run.members
            || self.inventory()? != (run.owner_pages, run.generation_sum)
        {
            return Err(Error::Resource);
        }
        for (i, baseline) in run.identities.into_iter().enumerate() {
            if let Some(b) = baseline {
                let r = self
                    .entries
                    .get(i)
                    .and_then(|e| e.record)
                    .ok_or(Error::Resource)?;
                if r.handle != b.handle
                    || r.thread != b.thread
                    || r.observation.pages != b.pages
                    || r.observation.owner_fingerprint != b.fingerprint
                {
                    return Err(Error::Resource);
                }
                let current = live[i].ok_or(Error::Resource)?;
                if r.time.running <= b.running
                    || r.observation.progress <= b.progress
                    || current.progress <= b.progress
                {
                    return Err(Error::Fairness);
                }
            }
        }
        Ok(self.snapshot)
    }
    pub fn into_backend(self) -> Result<B, Self> {
        if self.gate().is_ok() && self.snapshot.members == 0 {
            Ok(self.backend)
        } else {
            Err(self)
        }
    }
}

#[cfg(test)]
mod private_tests {
    use super::*;
    use crate::test_model::*;
    #[test]
    fn same_source_independent_policies_reject_foreign_handle_and_run() {
        let (mut a, da, ha) = policy_setup();
        let (b, _, _) = policy_setup();
        assert_eq!(b.record(ha[0]), Err(Error::Foreign));
        let run = a.begin_long_run().unwrap();
        assert_eq!(b.finish_long_run(&run), Err(Error::Foreign));
        da.cancel(0);
        let m = a.unregister(ha[0]).unwrap();
        let new = a.register(m).ok().unwrap();
        assert_ne!(new, ha[0]);
        assert_eq!(a.record(ha[0]), Err(Error::Stale));
        assert!(a.record(new).is_ok());
    }
    #[test]
    fn generation_max_retires_slot_and_returns_actual_member() {
        let (b, d, mut members) = raw();
        let mut p = Policy::<_, 1>::new(b).ok().unwrap();
        p.entries[0].generation = u64::MAX;
        let (e, m) = p.register(members.remove(0)).err().unwrap();
        assert_eq!(e, Error::Capacity);
        assert_eq!(m.thread, d.threads[0]);
        assert_eq!(p.entries[0].generation, u64::MAX);
        assert_eq!(p.snapshot.members, 0);
    }
    #[test]
    fn tick_counter_overflow_refuses_before_consume_or_partial_commit() {
        let (mut p, d, _) = policy_setup();
        p.snapshot.ticks = u64::MAX;
        let before = p.snapshot;
        let t = d.issue(1);
        assert_eq!(p.charge(t).err().unwrap().0, Error::Exhausted);
        assert_eq!(p.snapshot, before);
        assert_eq!(d.source.borrow().consumed, 0);
    }
    #[test]
    fn per_thread_overflow_refuses_before_consume_or_partial_commit() {
        let (mut p, d, h) = policy_setup();
        p.entries[h[0].slot].record.as_mut().unwrap().time.ready = u64::MAX;
        let before = p.entries[h[0].slot].record;
        let t = d.issue(1);
        assert_eq!(p.charge(t).err().unwrap().0, Error::Exhausted);
        assert_eq!(p.entries[h[0].slot].record, before);
        assert_eq!(d.source.borrow().consumed, 0);
    }
    #[test]
    fn sequence_exhaustion_returns_actual_delivery_unchanged() {
        let (mut p, d, _) = policy_setup();
        p.snapshot.last_sequence = u64::MAX;
        let t = d.issue(1);
        let (e, t) = p.charge(t).err().unwrap();
        assert_eq!(e, Error::Exhausted);
        assert_eq!(d.source.borrow().live, Some(t.observation));
        assert_eq!(p.snapshot.ticks, 0);
    }
    #[test]
    fn full_long_run_policy_handle_generation_is_bound_even_with_same_page_sum() {
        let (mut p, d, _) = policy_setup();
        let mut run = p.begin_long_run().unwrap();
        for _ in 0..180 {
            d.advance(false);
            d.charge(&mut p, 20000);
        }
        run.identities[0].as_mut().unwrap().handle.generation += 1;
        assert_eq!(p.finish_long_run(&run), Err(Error::Resource));
    }
    #[test]
    fn begin_uses_fresh_actual_progress_baseline_and_finish_requires_new_progress() {
        let (mut p, d, _) = policy_setup();
        d.advance(false);
        d.source.borrow_mut().progress = [900, 901, 902, 0];
        let run = p.begin_long_run().unwrap();
        assert_eq!(run.identities[0].unwrap().progress, 900);
        for _ in 0..180 {
            d.advance(false);
            d.charge(&mut p, 20000);
        }
        assert!(p.finish_long_run(&run).is_ok());
        let mut frozen = run;
        frozen.identities[0].as_mut().unwrap().progress = p
            .record(p.entries[0].record.unwrap().handle)
            .unwrap()
            .observation
            .progress;
        assert_eq!(p.finish_long_run(&frozen), Err(Error::Fairness));
    }
    #[test]
    fn long_run_start_or_duration_overflow_does_not_change_any_owner() {
        let (mut p, d, _) = policy_setup();
        d.source.borrow_mut().counter = u64::MAX;
        assert_eq!(p.begin_long_run(), Err(Error::Exhausted));
        assert_eq!(p.snapshot.members, 3);
        assert_eq!(d.owners(), (3, 3, 12));
    }
    #[test]
    fn long_run_requires_sixty_new_samples_independently_of_elapsed_counter() {
        let (mut p, d, _) = policy_setup();
        let mut run = p.begin_long_run().unwrap();
        for _ in 0..180 {
            d.advance(false);
            d.charge(&mut p, 20000);
        }
        run.first_tick = 121;
        assert_eq!(p.finish_long_run(&run), Err(Error::Duration));
    }
    #[test]
    fn new_long_run_episode_refuses_older_same_policy_receipt() {
        let (mut p, _, _) = policy_setup();
        let old = p.begin_long_run().unwrap();
        let new = p.begin_long_run().unwrap();
        assert_ne!(old.episode, new.episode);
        assert_eq!(p.workload_status(&old), Err(Error::Stale));
        assert_eq!(p.finish_long_run(&old), Err(Error::Stale));
    }
    #[test]
    fn workload_progress_counter_overflow_returns_tick_before_any_commit() {
        let (mut p, d, _) = policy_setup();
        let run = p.begin_long_run().unwrap();
        p.workload.as_mut().unwrap().members[0]
            .as_mut()
            .unwrap()
            .progress_samples = u64::MAX;
        d.advance(false);
        let before = p.snapshot;
        let workload = p.workload;
        let t = d.issue(1);
        let (e, t) = p.charge(t).err().unwrap();
        assert_eq!(e, Error::Exhausted);
        assert_eq!(p.snapshot, before);
        assert_eq!(p.workload, workload);
        assert_eq!(d.source.borrow().live, Some(t.observation));
        assert_eq!(p.workload_status(&run).unwrap().failed, None);
    }
    #[test]
    fn episode_exhaustion_retains_previous_monitor_and_actual_backend() {
        let (mut p, d, _) = policy_setup();
        let run = p.begin_long_run().unwrap();
        let before = p.workload;
        p.next_episode = u64::MAX;
        assert_eq!(p.begin_long_run(), Err(Error::Exhausted));
        assert_eq!(p.workload, before);
        assert_eq!(p.workload_status(&run).unwrap().members, 3);
        assert_eq!(d.owners(), (3, 3, 12));
    }
    #[test]
    fn full_hour_needs_sixty_progress_observations_for_each_cohort_member() {
        let (mut p, d, _) = policy_setup();
        let run = p.begin_long_run().unwrap();
        d.hour(&mut p);
        assert!(p.finish_long_run(&run).is_ok());
        p.workload.as_mut().unwrap().members[0]
            .as_mut()
            .unwrap()
            .progress_samples = 59;
        assert_eq!(p.finish_long_run(&run), Err(Error::Fairness));
    }
    #[test]
    fn changed_full_cohort_handle_is_latched_without_discarding_tick_accounting() {
        let (mut p, d, _) = policy_setup();
        let run = p.begin_long_run().unwrap();
        p.workload.as_mut().unwrap().members[0]
            .as_mut()
            .unwrap()
            .handle
            .generation += 1;
        d.advance(false);
        d.charge(&mut p, 1);
        assert_eq!(p.snapshot.ticks, 1);
        assert_eq!(
            p.workload_status(&run).unwrap().failed,
            Some(Error::Resource)
        );
        assert_eq!(d.source.borrow().consumed, 1);
        assert_eq!(d.owners(), (3, 3, 12));
    }
}