use core::array;

pub const MAX_JOBS: usize = 8;
pub const MAX_REGISTRATIONS: usize = 8;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Error {
    Configuration,
    Context,
    Cpu,
    IrqMask,
    Foreign,
    Stale,
    Duplicate,
    Full,
    Draining,
    Busy,
    Generation,
    Overflow,
    Binding,
    Backend,
    Ownership,
    NotServiced,
    AlreadyServiced,
    Quarantined,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Context {
    Task,
    Interrupt,
}

/// Inert observations supplied by the trusted architecture ingress. This is
/// not a substitute for actually reading CPU, DAIF and exception/task state.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ObservedContext {
    pub cpu: u32,
    pub kind: Context,
    pub irq_masked: bool,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Operation {
    Register,
    Enqueue,
    Take,
    Service,
    Recover,
    Complete,
    Drain,
    Release,
}

/// Pure gate: call before forming mutable Queue or resource loans in ingress.
pub fn context(cpu: u32, observed: ObservedContext, operation: Operation) -> Result<(), Error> {
    if observed.cpu != cpu {
        return Err(Error::Cpu);
    }
    if operation != Operation::Enqueue && observed.kind != Context::Task {
        return Err(Error::Context);
    }
    if !matches!(operation, Operation::Service | Operation::Recover) && !observed.irq_masked {
        return Err(Error::IrqMask);
    }
    Ok(())
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct Binding {
    pub device: u64,
    pub epoch: u64,
    pub owner: u64,
    pub cpu: u32,
}

impl Binding {
    fn valid(self, cpu: u32) -> bool {
        self.device != 0 && self.epoch != 0 && self.owner != 0 && self.cpu == cpu
    }
}

/// Implementations own an actual immutable source/callback and backing pins.
/// `service` never borrows Queue or its Backend. A failure retains ownership
/// and enters quarantine; `recover` must quiesce partial I/O before retry.
/// The actual context observer reads trusted architecture/task state without
/// consulting a Registry/Backend loan that IRQ enqueue can mutate.
pub trait WorkResource {
    fn observed_context(&self) -> ObservedContext;
    fn identity(&self) -> u64;
    fn binding(&self) -> Binding;
    fn service(&mut self, completions: u32) -> Result<(), Error>;
    fn recover(&mut self) -> Result<(), Error>;
}

/// Every returned source and work resource is nonclone ownership in the
/// adapter, with inert destructors that never unpin/free resources. Err from
/// acquire/retain changes no owned inventory; return errors
/// carry the exact still-owned resource. `audit_owned` compares the complete
/// backend ledger to these inert records, including detached running leases.
/// `retain` in IRQ may only transfer preallocated private work metadata and
/// already held source/backing pins. Main acquires central backing before IRQ
/// enable; retain never borrows allocator, heap, Runtime or predecessor owners.
pub trait Backend {
    type Source;
    type Work: WorkResource;
    fn observed_context(&self) -> ObservedContext;
    fn acquire(&mut self, binding: Binding) -> Result<Self::Source, Error>;
    fn source_identity(&self, source: &Self::Source) -> u64;
    fn source_binding(&self, source: &Self::Source) -> Binding;
    fn retain(&mut self, source: &Self::Source) -> Result<Self::Work, Error>;
    fn return_work(&mut self, work: Self::Work) -> Result<(), (Error, Self::Work)>;
    fn return_source(&mut self, source: Self::Source) -> Result<(), (Error, Self::Source)>;
    fn audit_owned(&self, sources: &[SourceRecord], jobs: &[JobRecord]) -> Result<(), Error>;
}

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

    const TASK: ObservedContext = ObservedContext {
        cpu: 0,
        kind: Context::Task,
        irq_masked: true,
    };
    const IRQ: ObservedContext = ObservedContext {
        kind: Context::Interrupt,
        ..TASK
    };
    const BINDING: Binding = Binding {
        device: 9,
        epoch: 1,
        owner: 7,
        cpu: 0,
    };

    #[derive(Debug)]
    struct Token {
        identity: u64,
        binding: Binding,
    }
    impl WorkResource for Token {
        fn observed_context(&self) -> ObservedContext {
            TASK
        }
        fn identity(&self) -> u64 {
            self.identity
        }
        fn binding(&self) -> Binding {
            self.binding
        }
        fn service(&mut self, _: u32) -> Result<(), Error> {
            Ok(())
        }
        fn recover(&mut self) -> Result<(), Error> {
            Ok(())
        }
    }
    #[derive(Default)]
    struct Owner {
        identity: u64,
        sources: usize,
        works: usize,
        acquires: usize,
        retains: usize,
    }
    impl Backend for Owner {
        type Source = Token;
        type Work = Token;
        fn observed_context(&self) -> ObservedContext {
            TASK
        }
        fn acquire(&mut self, binding: Binding) -> Result<Token, Error> {
            self.identity += 1;
            self.sources += 1;
            self.acquires += 1;
            Ok(Token {
                identity: self.identity,
                binding,
            })
        }
        fn source_identity(&self, token: &Token) -> u64 {
            token.identity
        }
        fn source_binding(&self, token: &Token) -> Binding {
            token.binding
        }
        fn retain(&mut self, source: &Token) -> Result<Token, Error> {
            self.identity += 1;
            self.works += 1;
            self.retains += 1;
            Ok(Token {
                identity: self.identity,
                binding: source.binding,
            })
        }
        fn return_work(&mut self, _: Token) -> Result<(), (Error, Token)> {
            self.works -= 1;
            Ok(())
        }
        fn return_source(&mut self, _: Token) -> Result<(), (Error, Token)> {
            self.sources -= 1;
            Ok(())
        }
        fn audit_owned(&self, sources: &[SourceRecord], jobs: &[JobRecord]) -> Result<(), Error> {
            if sources
                .iter()
                .filter(|r| r.state != SourceState::Empty)
                .count()
                == self.sources
                && jobs.iter().filter(|r| r.state != JobState::Empty).count() == self.works
            {
                Ok(())
            } else {
                Err(Error::Ownership)
            }
        }
    }

    #[test]
    fn namespace_increment_never_wraps_or_reuses_zero() {
        assert_eq!(next_identity(u64::MAX), Err(Error::Generation));
        assert_eq!(next_identity(u64::MAX - 1), Ok(u64::MAX));
        assert_eq!(next_identity(1), Ok(2));
    }

    #[test]
    fn registration_generation_exhaustion_retires_slot_before_acquisition() {
        let mut q =
            Queue::<_, 1, 1>::new(Owner::default()).unwrap_or_else(|_| panic!("constructor"));
        q.sources[0].record.generation = u64::MAX;
        let before = q.source_records();
        assert_eq!(q.register(BINDING, TASK).unwrap_err(), Error::Generation);
        assert_eq!(q.source_records(), before);
        assert_eq!(q.backend.acquires, 0);
    }

    #[test]
    fn job_and_issue_generation_exhaustion_retains_exact_source_without_work_acquire() {
        let mut q =
            Queue::<_, 1, 1>::new(Owner::default()).unwrap_or_else(|_| panic!("constructor"));
        let r = q.register(BINDING, TASK).unwrap();
        q.jobs[0].record.generation = u64::MAX;
        let before = q.job_records();
        assert_eq!(q.enqueue(&r, 1, IRQ), Err(Error::Generation));
        assert_eq!(q.job_records(), before);
        assert_eq!(q.backend.retains, 0);
        q.jobs[0].record.generation = 0;
        q.sequence = u64::MAX;
        let before = q.job_records();
        assert_eq!(q.enqueue(&r, 1, IRQ), Err(Error::Generation));
        assert_eq!(q.job_records(), before);
        assert_eq!(q.backend.retains, 0);
        assert_eq!(q.snapshot().submitted, 0);
    }

    #[test]
    fn submitted_coalesced_and_loss_counter_overflow_preserve_all_records() {
        let mut q =
            Queue::<_, 1, 1>::new(Owner::default()).unwrap_or_else(|_| panic!("constructor"));
        let r = q.register(BINDING, TASK).unwrap();
        q.metrics.submitted = u64::MAX;
        q.metrics.completed = u64::MAX;
        let before = q.snapshot();
        assert_eq!(q.enqueue(&r, 1, IRQ), Err(Error::Overflow));
        assert_eq!(q.snapshot(), before);
        assert_eq!(q.backend.retains, 0);
        q.metrics.submitted = 0;
        q.metrics.completed = 0;
        q.enqueue(&r, 1, IRQ).unwrap();
        q.metrics.coalesced_jobs = u64::MAX;
        let before = (q.snapshot(), q.job_records());
        assert_eq!(q.enqueue(&r, 1, IRQ), Err(Error::Overflow));
        assert_eq!((q.snapshot(), q.job_records()), before);
        let _work = q.take(TASK).unwrap().unwrap();
        q.metrics.lost_jobs = u64::MAX;
        let before = (q.snapshot(), q.job_records());
        assert_eq!(q.enqueue(&r, 1, IRQ), Err(Error::Overflow));
        assert_eq!((q.snapshot(), q.job_records()), before);
        assert_eq!(q.backend.retains, 1);
    }

    #[test]
    fn forged_stale_registration_cannot_touch_current_lease() {
        let mut q =
            Queue::<_, 1, 1>::new(Owner::default()).unwrap_or_else(|_| panic!("constructor"));
        let r = q.register(BINDING, TASK).unwrap();
        let mut old = Registration {
            namespace: r.namespace,
            record: r.record,
        };
        old.record.generation += 1;
        let before = (q.source_records(), q.job_records(), q.snapshot());
        assert_eq!(q.enqueue(&old, 1, IRQ), Err(Error::Stale));
        assert_eq!(q.drain(&old, TASK), Err(Error::Stale));
        let (error, _) = q.release(old, TASK).unwrap_err();
        assert_eq!(error, Error::Stale);
        assert_eq!((q.source_records(), q.job_records(), q.snapshot()), before);
        assert_eq!(q.backend.retains, 0);
        q.drain(&r, TASK).unwrap();
        q.release(r, TASK).unwrap();
    }

    #[test]
    fn forged_running_token_cannot_credit_or_release_another_job() {
        let mut q =
            Queue::<_, 1, 1>::new(Owner::default()).unwrap_or_else(|_| panic!("constructor"));
        let r = q.register(BINDING, TASK).unwrap();
        q.enqueue(&r, 1, IRQ).unwrap();
        let mut work = q.take(TASK).unwrap().unwrap();
        work.service(TASK).unwrap();
        work.record.registration_generation += 1;
        let before = (q.job_records(), q.snapshot());
        let (error, mut work) = q.complete(work, TASK).unwrap_err();
        assert_eq!(error, Error::Binding);
        assert_eq!((q.job_records(), q.snapshot()), before);
        assert_eq!(q.backend.works, 1);
        work.record.registration_generation -= 1;
        q.complete(work, TASK).unwrap();
        q.drain(&r, TASK).unwrap();
        q.release(r, TASK).unwrap();
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SourceState {
    Empty,
    Open,
    Draining,
    Quarantined,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum JobState {
    Empty,
    Queued,
    Running,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WorkState {
    Ready,
    Serviced,
    Quarantined,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SourceRecord {
    pub state: SourceState,
    pub slot: usize,
    pub generation: u64,
    pub binding: Binding,
    pub resource: u64,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct JobRecord {
    pub state: JobState,
    pub slot: usize,
    pub generation: u64,
    pub registration_slot: usize,
    pub registration_generation: u64,
    pub binding: Binding,
    pub resource: u64,
    pub completions: u32,
    pub sequence: u64,
}

const EMPTY_BINDING: Binding = Binding {
    device: 0,
    epoch: 0,
    owner: 0,
    cpu: 0,
};

/// Nonclone authority to close one registration. Read-only records are inert.
#[derive(Debug)]
pub struct Registration {
    namespace: u64,
    record: SourceRecord,
}

impl Registration {
    pub fn record(&self) -> SourceRecord {
        self.record
    }
}

/// Owns the detached resource; deliberately has no lifetime borrowing Queue.
#[derive(Debug)]
pub struct Work<R: WorkResource> {
    namespace: u64,
    cpu: u32,
    record: JobRecord,
    resource: Option<R>,
    state: WorkState,
}

impl<R: WorkResource> Work<R> {
    pub fn record(&self) -> JobRecord {
        self.record
    }
    pub fn state(&self) -> WorkState {
        self.state
    }

    pub fn service(&mut self, observed: ObservedContext) -> Result<(), Error> {
        context(self.cpu, observed, Operation::Service)?;
        let actual = self
            .resource
            .as_ref()
            .ok_or(Error::Ownership)?
            .observed_context();
        context(self.cpu, actual, Operation::Service)?;
        match self.state {
            WorkState::Ready => {}
            WorkState::Serviced => return Err(Error::AlreadyServiced),
            WorkState::Quarantined => return Err(Error::Quarantined),
        }
        let resource = self.resource.as_mut().ok_or(Error::Ownership)?;
        if resource.identity() != self.record.resource || resource.binding() != self.record.binding
        {
            self.state = WorkState::Quarantined;
            return Err(Error::Binding);
        }
        if let Err(error) = resource.service(self.record.completions) {
            self.state = WorkState::Quarantined;
            return Err(error);
        }
        self.state = WorkState::Serviced;
        Ok(())
    }

    pub fn recover(&mut self, observed: ObservedContext) -> Result<(), Error> {
        context(self.cpu, observed, Operation::Recover)?;
        let actual = self
            .resource
            .as_ref()
            .ok_or(Error::Ownership)?
            .observed_context();
        context(self.cpu, actual, Operation::Recover)?;
        if self.state != WorkState::Quarantined {
            return Err(Error::Quarantined);
        }
        let resource = self.resource.as_mut().ok_or(Error::Ownership)?;
        if resource.identity() != self.record.resource || resource.binding() != self.record.binding
        {
            return Err(Error::Binding);
        }
        resource.recover()?;
        self.state = WorkState::Ready;
        Ok(())
    }
}

struct Source<S> {
    record: SourceRecord,
    resource: Option<S>,
}

struct Job<R> {
    record: JobRecord,
    resource: Option<R>,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct Snapshot {
    pub queued: usize,
    pub running: usize,
    pub open: usize,
    pub draining: usize,
    pub quarantined: usize,
    pub submitted: u64,
    pub completed: u64,
    pub coalesced_jobs: u64,
    pub coalesced_completions: u64,
    pub lost_jobs: u64,
    pub lost_completions: u64,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Enqueued {
    New(JobRecord),
    Coalesced(JobRecord),
}

pub struct Queue<B: Backend, const N: usize, const R: usize> {
    namespace: u64,
    cpu: u32,
    backend: B,
    sources: [Source<B::Source>; R],
    jobs: [Job<B::Work>; N],
    metrics: Snapshot,
    sequence: u64,
}

fn next_identity(current: u64) -> Result<u64, Error> {
    current.checked_add(1).ok_or(Error::Generation)
}

#[cfg(not(target_os = "none"))]
fn identity() -> Result<u64, Error> {
    use core::sync::atomic::{AtomicU64, Ordering};
    static NEXT: AtomicU64 = AtomicU64::new(1);
    let mut current = NEXT.load(Ordering::Relaxed);
    loop {
        let next = next_identity(current)?;
        match NEXT.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) {
            Ok(value) => return Ok(value),
            Err(value) => current = value,
        }
    }
}

#[cfg(target_os = "none")]
fn identity() -> Result<u64, Error> {
    unsafe extern "C" {
        fn __os_k1000_deferred_identity() -> u64;
    }
    // The separately source-bound runtime helper must issue a checked,
    // monotonic nonce with actual DAIF.I masked on the observed single CPU.
    let value = unsafe { __os_k1000_deferred_identity() };
    next_identity(value)?;
    if value == 0 {
        Err(Error::Generation)
    } else {
        Ok(value)
    }
}

impl<B: Backend, const N: usize, const R: usize> Queue<B, N, R> {
    pub fn new(backend: B) -> Result<Self, (Error, B)> {
        if N == 0 || N > MAX_JOBS || R == 0 || R > MAX_REGISTRATIONS {
            return Err((Error::Configuration, backend));
        }
        let actual = backend.observed_context();
        if let Err(error) = context(actual.cpu, actual, Operation::Register) {
            return Err((error, backend));
        }
        let namespace = match identity() {
            Ok(value) => value,
            Err(error) => return Err((error, backend)),
        };
        let cpu = actual.cpu;
        Ok(Self {
            namespace,
            cpu,
            backend,
            sources: array::from_fn(|slot| Source {
                record: SourceRecord {
                    state: SourceState::Empty,
                    slot,
                    generation: 0,
                    binding: EMPTY_BINDING,
                    resource: 0,
                },
                resource: None,
            }),
            jobs: array::from_fn(|slot| Job {
                record: JobRecord {
                    state: JobState::Empty,
                    slot,
                    generation: 0,
                    registration_slot: 0,
                    registration_generation: 0,
                    binding: EMPTY_BINDING,
                    resource: 0,
                    completions: 0,
                    sequence: 0,
                },
                resource: None,
            }),
            metrics: Snapshot::default(),
            sequence: 0,
        })
    }

    pub fn identity(&self) -> u64 {
        self.namespace
    }

    fn checked_context(
        &self,
        observed: ObservedContext,
        operation: Operation,
    ) -> Result<(), Error> {
        context(self.cpu, observed, operation)?;
        Ok(())
    }

    pub fn snapshot(&self) -> Snapshot {
        let mut result = self.metrics;
        for source in &self.sources {
            match source.record.state {
                SourceState::Open => result.open += 1,
                SourceState::Draining => result.draining += 1,
                SourceState::Quarantined => result.quarantined += 1,
                SourceState::Empty => {}
            }
        }
        for job in &self.jobs {
            match job.record.state {
                JobState::Queued => result.queued += 1,
                JobState::Running => result.running += 1,
                JobState::Empty => {}
            }
        }
        result
    }

    pub fn source_records(&self) -> [SourceRecord; R] {
        array::from_fn(|i| self.sources[i].record)
    }
    pub fn job_records(&self) -> [JobRecord; N] {
        array::from_fn(|i| self.jobs[i].record)
    }
    pub fn with_backend_observation<T>(&self, callback: impl FnOnce(&B) -> T) -> T {
        callback(&self.backend)
    }

    fn checked_registration(&self, registration: &Registration) -> Result<usize, Error> {
        if registration.namespace != self.namespace {
            return Err(Error::Foreign);
        }
        let slot = registration.record.slot;
        let source = self.sources.get(slot).ok_or(Error::Foreign)?;
        if source.record.state == SourceState::Empty
            || source.record.generation != registration.record.generation
        {
            return Err(Error::Stale);
        }
        if source.record.binding != registration.record.binding
            || source.record.resource != registration.record.resource
        {
            return Err(Error::Binding);
        }
        Ok(slot)
    }

    pub fn register(
        &mut self,
        binding: Binding,
        observed: ObservedContext,
    ) -> Result<Registration, Error> {
        self.checked_context(observed, Operation::Register)?;
        if !binding.valid(self.cpu) {
            return Err(Error::Binding);
        }
        if self
            .sources
            .iter()
            .any(|s| s.record.state != SourceState::Empty && s.record.binding == binding)
        {
            return Err(Error::Duplicate);
        }
        let slot = self
            .sources
            .iter()
            .position(|s| s.record.state == SourceState::Empty && s.record.generation != u64::MAX)
            .ok_or(
                if self
                    .sources
                    .iter()
                    .any(|s| s.record.state == SourceState::Empty)
                {
                    Error::Generation
                } else {
                    Error::Full
                },
            )?;
        let generation = self.sources[slot]
            .record
            .generation
            .checked_add(1)
            .ok_or(Error::Generation)?;
        let resource = self.backend.acquire(binding)?;
        let identity = self.backend.source_identity(&resource);
        // The backend's successful acquisition contract requires a valid,
        // uniquely owned exact binding; retain the source rather than discard
        // it if its observation violates that contract.
        let valid = identity != 0
            && self.backend.source_binding(&resource) == binding
            && !self
                .sources
                .iter()
                .any(|s| s.record.state != SourceState::Empty && s.record.resource == identity);
        let record = SourceRecord {
            state: if valid {
                SourceState::Open
            } else {
                SourceState::Quarantined
            },
            slot,
            generation,
            binding,
            resource: identity,
        };
        self.sources[slot] = Source {
            record,
            resource: Some(resource),
        };
        Ok(Registration {
            namespace: self.namespace,
            record,
        })
    }

    fn occupied(&self) -> usize {
        self.jobs
            .iter()
            .filter(|j| j.record.state != JobState::Empty)
            .count()
    }

    pub fn enqueue(
        &mut self,
        registration: &Registration,
        completions: u32,
        observed: ObservedContext,
    ) -> Result<Enqueued, Error> {
        self.checked_context(observed, Operation::Enqueue)?;
        let source_slot = self.checked_registration(registration)?;
        let source_record = self.sources[source_slot].record;
        if source_record.state != SourceState::Open {
            return Err(Error::Draining);
        }
        if completions == 0 {
            return Err(Error::Configuration);
        }
        let submitted = self
            .metrics
            .submitted
            .checked_add(u64::from(completions))
            .ok_or(Error::Overflow)?;
        if let Some(slot) = self.jobs.iter().position(|j| {
            j.record.state == JobState::Queued
                && j.record.registration_slot == source_slot
                && j.record.registration_generation == source_record.generation
                && j.record.binding == source_record.binding
        }) {
            let count = self.jobs[slot]
                .record
                .completions
                .checked_add(completions)
                .ok_or(Error::Overflow)?;
            let coalesced_jobs = self
                .metrics
                .coalesced_jobs
                .checked_add(1)
                .ok_or(Error::Overflow)?;
            let coalesced_completions = self
                .metrics
                .coalesced_completions
                .checked_add(u64::from(completions))
                .ok_or(Error::Overflow)?;
            self.jobs[slot].record.completions = count;
            self.metrics.submitted = submitted;
            self.metrics.coalesced_jobs = coalesced_jobs;
            self.metrics.coalesced_completions = coalesced_completions;
            return Ok(Enqueued::Coalesced(self.jobs[slot].record));
        }
        if self.occupied() == N {
            let lost_jobs = self
                .metrics
                .lost_jobs
                .checked_add(1)
                .ok_or(Error::Overflow)?;
            let lost_completions = self
                .metrics
                .lost_completions
                .checked_add(u64::from(completions))
                .ok_or(Error::Overflow)?;
            self.metrics.lost_jobs = lost_jobs;
            self.metrics.lost_completions = lost_completions;
            return Err(Error::Full);
        }
        let slot = self
            .jobs
            .iter()
            .position(|j| j.record.state == JobState::Empty && j.record.generation != u64::MAX)
            .ok_or(Error::Generation)?;
        let generation = self.jobs[slot]
            .record
            .generation
            .checked_add(1)
            .ok_or(Error::Generation)?;
        let sequence = self.sequence.checked_add(1).ok_or(Error::Generation)?;
        let resource = self.backend.retain(
            self.sources[source_slot]
                .resource
                .as_ref()
                .ok_or(Error::Ownership)?,
        )?;
        let record = JobRecord {
            state: JobState::Queued,
            slot,
            generation,
            registration_slot: source_slot,
            registration_generation: source_record.generation,
            binding: source_record.binding,
            resource: resource.identity(),
            completions,
            sequence,
        };
        self.jobs[slot] = Job {
            record,
            resource: Some(resource),
        };
        self.metrics.submitted = submitted;
        self.sequence = sequence;
        Ok(Enqueued::New(record))
    }

    pub fn take(&mut self, observed: ObservedContext) -> Result<Option<Work<B::Work>>, Error> {
        self.checked_context(observed, Operation::Take)?;
        self.audit()?;
        let Some(slot) = self
            .jobs
            .iter()
            .enumerate()
            .filter(|(_, j)| j.record.state == JobState::Queued)
            .min_by_key(|(_, j)| j.record.sequence)
            .map(|(i, _)| i)
        else {
            return Ok(None);
        };
        let resource = self.jobs[slot].resource.take().ok_or(Error::Ownership)?;
        self.jobs[slot].record.state = JobState::Running;
        Ok(Some(Work {
            namespace: self.namespace,
            cpu: self.cpu,
            record: self.jobs[slot].record,
            resource: Some(resource),
            state: WorkState::Ready,
        }))
    }

    pub fn complete(
        &mut self,
        mut work: Work<B::Work>,
        observed: ObservedContext,
    ) -> Result<(), (Error, Work<B::Work>)> {
        let preflight = (|| {
            self.checked_context(observed, Operation::Complete)?;
            if work.namespace != self.namespace || work.cpu != self.cpu {
                return Err(Error::Foreign);
            }
            let job = self.jobs.get(work.record.slot).ok_or(Error::Foreign)?;
            if job.record.state != JobState::Running
                || job.record.generation != work.record.generation
            {
                return Err(Error::Stale);
            }
            if job.record != work.record
                || work.resource.as_ref().is_none_or(|r| {
                    r.identity() != work.record.resource || r.binding() != work.record.binding
                })
            {
                return Err(Error::Binding);
            }
            match work.state {
                WorkState::Ready => return Err(Error::NotServiced),
                WorkState::Quarantined => return Err(Error::Quarantined),
                WorkState::Serviced => {}
            }
            self.audit()?;
            self.metrics
                .completed
                .checked_add(u64::from(work.record.completions))
                .ok_or(Error::Overflow)
        })();
        let completed = match preflight {
            Ok(value) => value,
            Err(error) => return Err((error, work)),
        };
        let resource = work.resource.take().expect("checked work owner");
        if let Err((error, resource)) = self.backend.return_work(resource) {
            work.resource = Some(resource);
            return Err((error, work));
        }
        let job = &mut self.jobs[work.record.slot];
        job.record.state = JobState::Empty;
        job.record.binding = EMPTY_BINDING;
        job.record.resource = 0;
        job.record.completions = 0;
        job.record.sequence = 0;
        self.metrics.completed = completed;
        Ok(())
    }

    pub fn drain(
        &mut self,
        registration: &Registration,
        observed: ObservedContext,
    ) -> Result<(), Error> {
        self.checked_context(observed, Operation::Drain)?;
        let slot = self.checked_registration(registration)?;
        if self.sources[slot].record.state == SourceState::Quarantined {
            return Err(Error::Quarantined);
        }
        self.sources[slot].record.state = SourceState::Draining;
        Ok(())
    }

    pub fn release(
        &mut self,
        registration: Registration,
        observed: ObservedContext,
    ) -> Result<(), (Error, Registration)> {
        let preflight = (|| {
            self.checked_context(observed, Operation::Release)?;
            let slot = self.checked_registration(&registration)?;
            if self.sources[slot].record.state == SourceState::Open {
                return Err(Error::Busy);
            }
            if self.jobs.iter().any(|j| {
                j.record.state != JobState::Empty
                    && j.record.registration_slot == slot
                    && j.record.registration_generation == registration.record.generation
            }) {
                return Err(Error::Busy);
            }
            self.audit()?;
            Ok(slot)
        })();
        let slot = match preflight {
            Ok(slot) => slot,
            Err(error) => return Err((error, registration)),
        };
        let source = self.sources[slot]
            .resource
            .take()
            .expect("checked source owner");
        if let Err((error, resource)) = self.backend.return_source(source) {
            self.sources[slot].resource = Some(resource);
            self.sources[slot].record.state = SourceState::Quarantined;
            return Err((error, registration));
        }
        self.sources[slot].record.state = SourceState::Empty;
        self.sources[slot].record.binding = EMPTY_BINDING;
        self.sources[slot].record.resource = 0;
        Ok(())
    }

    pub fn audit(&self) -> Result<(), Error> {
        let mut pending = 0u64;
        for source in &self.sources {
            if source.record.state == SourceState::Empty {
                if source.resource.is_some() || source.record.resource != 0 {
                    return Err(Error::Ownership);
                }
            } else if source.resource.as_ref().is_none_or(|r| {
                self.backend.source_identity(r) != source.record.resource
                    || self.backend.source_binding(r) != source.record.binding
            }) {
                return Err(Error::Binding);
            }
        }
        for (i, job) in self.jobs.iter().enumerate() {
            if job.record.state == JobState::Empty {
                if job.resource.is_some() || job.record.resource != 0 || job.record.completions != 0
                {
                    return Err(Error::Ownership);
                }
                continue;
            }
            let source = self
                .sources
                .get(job.record.registration_slot)
                .ok_or(Error::Ownership)?;
            if source.record.state == SourceState::Empty
                || source.record.generation != job.record.registration_generation
                || source.record.binding != job.record.binding
                || job.record.resource == 0
                || job.record.completions == 0
                || job.record.sequence == 0
                || self.jobs[..i].iter().any(|j| {
                    j.record.state != JobState::Empty
                        && (j.record.resource == job.record.resource
                            || j.record.sequence == job.record.sequence)
                })
            {
                return Err(Error::Ownership);
            }
            match job.record.state {
                JobState::Queued => {
                    if job.resource.as_ref().is_none_or(|r| {
                        r.identity() != job.record.resource || r.binding() != job.record.binding
                    }) {
                        return Err(Error::Binding);
                    }
                }
                JobState::Running => {
                    if job.resource.is_some() {
                        return Err(Error::Ownership);
                    }
                }
                JobState::Empty => {}
            }
            pending = pending
                .checked_add(u64::from(job.record.completions))
                .ok_or(Error::Overflow)?;
        }
        if self.metrics.completed.checked_add(pending) != Some(self.metrics.submitted) {
            return Err(Error::Ownership);
        }
        self.backend
            .audit_owned(&self.source_records(), &self.job_records())
    }

    pub fn into_backend(self) -> Result<B, Self> {
        if self
            .sources
            .iter()
            .all(|s| s.record.state == SourceState::Empty)
            && self.jobs.iter().all(|j| j.record.state == JobState::Empty)
            && self.audit().is_ok()
        {
            Ok(self.backend)
        } else {
            Err(self)
        }
    }
}