use std::cell::RefCell;
use std::collections::BTreeMap;
use std::ffi::{OsStr, OsString};
use std::fmt;
use std::io::{self, Read};
use std::os::fd::{AsFd, OwnedFd};
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::rc::Rc;
use std::time::{Duration, Instant};

use ed25519_dalek::SigningKey;
use lazaret_core::configuration::ResolvedDomainConfiguration;
use lazaret_core::identity::{RequestDigest, RequestId, SessionId, StreamDigest};
use lazaret_core::protocol::{
    MAX_RESULT_CHUNK_BYTES, ProtocolCodecError, ProtocolValidationError, RequestReplayGuard,
    ResultSenderWindow, ResultWindowError, VerifiedSession, encode_signed, result_chunk_file_name,
    result_file_name,
};
use lazaret_core::records::{
    DeveloperDecisionKind, EscalationOutcome, GenerationIdentity, HostProcessOwnershipSlot,
    MailboxRequest, MailboxResult, MailboxResultChunk, ProcessOwnershipCommitAck, ProtocolVersion,
    ResultStream, StagedProcessOwnership,
};
use lazaret_linux::filesystem::{MailboxEntryIdentity, RegularFileMailbox};
use lazaret_linux::process::{
    CapturedProcess, ProcessOwnershipContext, TargetCommand, child_broker, current_process_identity,
};
use rustix::fs::{FileType, Mode, OFlags, ResolveFlags, fchmod, fstat, mkdirat, openat, openat2};
use rustix::process::Signal;
use sha2::{Digest, Sha256};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ApprovalSource {
    AutoRule,
    Developer,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ApprovalAttribution {
    pub source: ApprovalSource,
    pub command_argv: Vec<Vec<u8>>,
}

pub struct GenerationMailbox {
    generation_directory: RegularFileMailbox,
    inbox: RegularFileMailbox,
    outbox: RegularFileMailbox,
}

impl GenerationMailbox {
    /// Creates the fixed generation path beneath an already-confined runtime Domain root.
    ///
    /// # Errors
    /// Refuses links, changed ownership or mode, existing identity, or incomplete publication.
    pub fn create(
        runtime_domain_root: impl AsFd,
        generation: &GenerationIdentity,
        invoking_uid: u32,
    ) -> Result<Self, EscalationError> {
        let mailbox =
            create_private_directory(runtime_domain_root, OsStr::new("mailbox"), invoking_uid)?;
        let generation_name =
            lazaret_core::protocol::mailbox_generation_name(generation.generation_id.get());
        let generation_directory =
            create_private_directory(&mailbox, &generation_name, invoking_uid)?;
        let identity = lazaret_core::codec::encode(generation)
            .map_err(|error| io::Error::other(error.to_string()))?;
        generation_directory.publish(OsStr::new("generation-identity.postcard"), &identity)?;
        let inbox =
            create_private_directory(&generation_directory, OsStr::new("inbox"), invoking_uid)?;
        let outbox =
            create_private_directory(&generation_directory, OsStr::new("outbox"), invoking_uid)?;
        Ok(Self {
            generation_directory,
            inbox,
            outbox,
        })
    }

    #[must_use]
    pub const fn inbox(&self) -> &RegularFileMailbox {
        &self.inbox
    }

    #[must_use]
    pub const fn outbox(&self) -> &RegularFileMailbox {
        &self.outbox
    }

    #[must_use]
    pub fn into_directories(self) -> (RegularFileMailbox, RegularFileMailbox) {
        let _ = self.generation_directory;
        (self.inbox, self.outbox)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum EscalationPhase {
    AwaitingApproval,
    Executing,
    Delivered(EscalationOutcome),
}

pub trait DeveloperExecutableResolver {
    /// Opens the executable named by element zero and returns authority over that exact object.
    ///
    /// # Errors
    /// Refuses an argv vector that cannot be resolved to one approved executable descriptor.
    fn resolve(&mut self, command_argv: &[Vec<u8>]) -> io::Result<OwnedFd>;
}

pub struct DirectDeveloperExecutableResolver {
    root: OwnedFd,
}

fn create_private_directory(
    parent: impl AsFd,
    name: &OsStr,
    invoking_uid: u32,
) -> io::Result<RegularFileMailbox> {
    let created = match mkdirat(parent.as_fd(), name, Mode::RUSR | Mode::WUSR | Mode::XUSR) {
        Ok(()) => true,
        Err(rustix::io::Errno::EXIST) => false,
        Err(error) => return Err(error.into()),
    };
    let directory = openat(
        parent.as_fd(),
        name,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
        Mode::empty(),
    )?;
    if created {
        fchmod(&directory, Mode::RUSR | Mode::WUSR | Mode::XUSR)?;
    }
    drop(directory);
    RegularFileMailbox::openat_private(parent, name, invoking_uid)
}

impl DirectDeveloperExecutableResolver {
    /// # Errors
    /// Returns an error when host root authority cannot be retained without following a link.
    pub fn open() -> io::Result<Self> {
        Ok(Self {
            root: openat(
                rustix::fs::CWD,
                "/",
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                Mode::empty(),
            )?,
        })
    }
}

impl DeveloperExecutableResolver for DirectDeveloperExecutableResolver {
    fn resolve(&mut self, command_argv: &[Vec<u8>]) -> io::Result<OwnedFd> {
        let executable = command_argv
            .first()
            .ok_or_else(|| io::Error::other("approved command vector is empty"))?;
        if executable.contains(&0) || !executable.starts_with(b"/") {
            return Err(io::Error::other(
                "approved executable is not a NUL-free absolute path",
            ));
        }
        let relative = std::ffi::OsStr::from_bytes(&executable[1..]);
        openat2(
            &self.root,
            relative,
            OFlags::RDONLY | OFlags::NONBLOCK | OFlags::CLOEXEC,
            Mode::empty(),
            ResolveFlags::BENEATH | ResolveFlags::NO_MAGICLINKS | ResolveFlags::NO_SYMLINKS,
        )
        .map_err(Into::into)
    }
}

pub trait ExecutionOwnershipRecorder {
    /// # Errors
    /// Refuses when exact staged ownership cannot be retained.
    fn stage(&mut self, ownership: StagedProcessOwnership) -> io::Result<()>;

    /// # Errors
    /// Refuses missing or mismatched exact ownership acknowledgement.
    fn acknowledge(
        &mut self,
        ownership: StagedProcessOwnership,
    ) -> io::Result<ProcessOwnershipCommitAck>;
}

#[derive(Debug)]
pub enum EscalationError {
    Io(io::Error),
    Protocol(ProtocolValidationError),
    ProtocolCodec(ProtocolCodecError),
    Window(ResultWindowError),
    DuplicateSessionRequest,
    UnknownRequest,
    InvalidPhase,
    InvalidApproval,
}

impl fmt::Display for EscalationError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => error.fmt(formatter),
            Self::Protocol(error) => error.fmt(formatter),
            Self::ProtocolCodec(error) => error.fmt(formatter),
            Self::Window(error) => error.fmt(formatter),
            Self::DuplicateSessionRequest => {
                formatter.write_str("session already has an active escalation request")
            }
            Self::UnknownRequest => formatter.write_str("escalation request is unknown"),
            Self::InvalidPhase => formatter.write_str("operation is invalid in the request phase"),
            Self::InvalidApproval => {
                formatter.write_str("developer approval does not authorize an executable vector")
            }
        }
    }
}

impl std::error::Error for EscalationError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Protocol(error) => Some(error),
            Self::ProtocolCodec(error) => Some(error),
            Self::Window(error) => Some(error),
            _ => None,
        }
    }
}

impl From<io::Error> for EscalationError {
    fn from(error: io::Error) -> Self {
        Self::Io(error)
    }
}

impl From<ProtocolValidationError> for EscalationError {
    fn from(error: ProtocolValidationError) -> Self {
        Self::Protocol(error)
    }
}

impl From<ProtocolCodecError> for EscalationError {
    fn from(error: ProtocolCodecError) -> Self {
        Self::ProtocolCodec(error)
    }
}

impl From<ResultWindowError> for EscalationError {
    fn from(error: ResultWindowError) -> Self {
        Self::Window(error)
    }
}

struct RequestAuthority {
    session: VerifiedSession,
    request: MailboxRequest,
    digest: RequestDigest,
    admitted_at: Instant,
}

struct PublishedChunk {
    name: OsString,
    identity: MailboxEntryIdentity,
}

struct Execution {
    authority: RequestAuthority,
    process: CapturedProcess,
    window: ResultSenderWindow,
    chunks: BTreeMap<u64, PublishedChunk>,
    stdout: Sha256,
    stderr: Sha256,
    stdout_eof: bool,
    stderr_eof: bool,
    acknowledgment_deadline: Option<Instant>,
}

enum ActiveRequest {
    AwaitingApproval(Box<RequestAuthority>),
    Executing(Box<Execution>),
}

struct PublishedResult {
    session_id: SessionId,
    outcome: EscalationOutcome,
    name: OsString,
    identity: MailboxEntryIdentity,
}

pub struct EscalationCoordinator {
    coordinator_key: SigningKey,
    outbox: RegularFileMailbox,
    ownership: ProcessOwnershipContext,
    invoking_uid: u32,
    request_timeout: Duration,
    acknowledgment_timeout: Duration,
    replay: RequestReplayGuard,
    active_sessions: BTreeMap<SessionId, RequestId>,
    active: BTreeMap<RequestId, ActiveRequest>,
    approvals: BTreeMap<RequestId, ApprovalAttribution>,
    results: BTreeMap<RequestId, PublishedResult>,
}

impl EscalationCoordinator {
    #[must_use]
    pub fn new(
        coordinator_key: SigningKey,
        outbox: RegularFileMailbox,
        ownership: ProcessOwnershipContext,
        invoking_uid: u32,
        request_timeout: Duration,
        acknowledgment_timeout: Duration,
    ) -> Self {
        Self {
            coordinator_key,
            outbox,
            ownership,
            invoking_uid,
            request_timeout,
            acknowledgment_timeout,
            replay: RequestReplayGuard::default(),
            active_sessions: BTreeMap::new(),
            active: BTreeMap::new(),
            approvals: BTreeMap::new(),
            results: BTreeMap::new(),
        }
    }

    /// Admits one exact signed request and routes matching configuration rules automatically.
    ///
    /// # Errors
    /// Refuses replay, stale tuples, a second request for one session, or launch failure.
    pub fn receive_request<R: ExecutionOwnershipRecorder>(
        &mut self,
        session: VerifiedSession,
        signed_request: &[u8],
        configuration: &ResolvedDomainConfiguration,
        recorder: &mut R,
        now: Instant,
    ) -> Result<EscalationPhase, EscalationError> {
        let authority = self.admit(session, signed_request, now)?;
        let request_id = authority.request.request_id();
        for rule in configuration.auto_approval_rules() {
            if let Ok(executable) = rule.authorize(&authority.request) {
                let command_argv = authority.request.command_argv().to_vec();
                return self.start_execution(
                    authority,
                    ApprovalSource::AutoRule,
                    executable,
                    command_argv,
                    recorder,
                );
            }
        }
        self.active.insert(
            request_id,
            ActiveRequest::AwaitingApproval(Box::new(authority)),
        );
        Ok(EscalationPhase::AwaitingApproval)
    }

    /// Admits a request into the developer-decision path without composing prompt UI.
    ///
    /// # Errors
    /// Refuses replay, stale tuples, or a second request for one session.
    pub fn receive_for_developer_approval(
        &mut self,
        session: VerifiedSession,
        signed_request: &[u8],
        now: Instant,
    ) -> Result<EscalationPhase, EscalationError> {
        let authority = self.admit(session, signed_request, now)?;
        let request_id = authority.request.request_id();
        self.active.insert(
            request_id,
            ActiveRequest::AwaitingApproval(Box::new(authority)),
        );
        Ok(EscalationPhase::AwaitingApproval)
    }

    /// Admits an exact request after an external rule matcher supplies its already-open descriptor.
    ///
    /// # Errors
    /// Refuses changed argv, replay, stale tuples, duplicate session activity, or launch failure.
    pub fn receive_auto_approved<R: ExecutionOwnershipRecorder>(
        &mut self,
        session: VerifiedSession,
        signed_request: &[u8],
        executable: OwnedFd,
        command_argv: Vec<Vec<u8>>,
        recorder: &mut R,
        now: Instant,
    ) -> Result<EscalationPhase, EscalationError> {
        let authority = self.admit(session, signed_request, now)?;
        if authority.request.command_argv() != command_argv {
            let session_id = authority.request.session_binding().session_id;
            self.active_sessions.remove(&session_id);
            return Err(EscalationError::InvalidApproval);
        }
        self.start_execution(
            authority,
            ApprovalSource::AutoRule,
            executable,
            command_argv,
            recorder,
        )
    }

    fn admit(
        &mut self,
        session: VerifiedSession,
        signed_request: &[u8],
        now: Instant,
    ) -> Result<RequestAuthority, EscalationError> {
        let (request, _) = session.verify_request(signed_request)?;
        let session_id = request.session_binding().session_id;
        if self.active_sessions.contains_key(&session_id) {
            return Err(EscalationError::DuplicateSessionRequest);
        }
        let (request, digest) = self.replay.admit(&session, signed_request)?;
        let request_id = request.request_id();
        self.active_sessions.insert(session_id, request_id);
        Ok(RequestAuthority {
            session,
            request,
            digest,
            admitted_at: now,
        })
    }

    #[must_use]
    pub fn approval_source(&self, request_id: RequestId) -> Option<ApprovalSource> {
        self.approvals
            .get(&request_id)
            .map(|attribution| attribution.source)
    }

    #[must_use]
    pub fn approval_attribution(&self, request_id: RequestId) -> Option<&ApprovalAttribution> {
        self.approvals.get(&request_id)
    }

    /// Applies one exact Session Anchor decision without providing any prompt or PTY UI.
    ///
    /// # Errors
    /// Refuses stale, mismatched, unsigned, or phase-inconsistent decisions.
    pub fn apply_developer_decision<Resolver, Recorder>(
        &mut self,
        request_id: RequestId,
        signed_decision: &[u8],
        resolver: &mut Resolver,
        recorder: &mut Recorder,
    ) -> Result<EscalationPhase, EscalationError>
    where
        Resolver: DeveloperExecutableResolver,
        Recorder: ExecutionOwnershipRecorder,
    {
        let Some(ActiveRequest::AwaitingApproval(authority)) = self.active.get(&request_id) else {
            return Err(EscalationError::InvalidPhase);
        };
        let decision = authority.session.verify_developer_decision(
            signed_decision,
            request_id,
            authority.digest,
        )?;
        let executable = match decision.decision {
            DeveloperDecisionKind::Deny => None,
            DeveloperDecisionKind::Approve => {
                if decision.approved_command_argv.is_empty() {
                    return Err(EscalationError::InvalidApproval);
                }
                let executable = resolver.resolve(&decision.approved_command_argv)?;
                validate_executable(executable.as_fd(), self.invoking_uid)?;
                Some(executable)
            }
        };
        let Some(ActiveRequest::AwaitingApproval(authority)) = self.active.remove(&request_id)
        else {
            return Err(EscalationError::InvalidPhase);
        };
        match decision.decision {
            DeveloperDecisionKind::Deny => {
                self.finish(&authority, EscalationOutcome::Denied, None)?;
                Ok(EscalationPhase::Delivered(EscalationOutcome::Denied))
            }
            DeveloperDecisionKind::Approve => self.start_execution(
                *authority,
                ApprovalSource::Developer,
                executable.ok_or(EscalationError::InvalidApproval)?,
                decision.approved_command_argv,
                recorder,
            ),
        }
    }

    fn start_execution<R: ExecutionOwnershipRecorder>(
        &mut self,
        authority: RequestAuthority,
        approval_source: ApprovalSource,
        executable: OwnedFd,
        command_argv: Vec<Vec<u8>>,
        recorder: &mut R,
    ) -> Result<EscalationPhase, EscalationError> {
        let request_id = authority.request.request_id();
        self.approvals.insert(
            request_id,
            ApprovalAttribution {
                source: approval_source,
                command_argv: command_argv.clone(),
            },
        );
        let setup = (|| {
            validate_executable(executable.as_fd(), self.invoking_uid)?;
            let arguments = command_argv
                .into_iter()
                .skip(1)
                .map(OsString::from_vec)
                .collect::<Vec<_>>();
            let target = TargetCommand::from_executable(executable, arguments)?;
            let parent = current_process_identity()?;
            Ok::<_, io::Error>((target, parent))
        })();
        let Ok((target, parent)) = setup else {
            self.finish(&authority, EscalationOutcome::Failed, None)?;
            return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
        };
        let recorder = Rc::new(RefCell::new(recorder));
        let stage_recorder = Rc::clone(&recorder);
        let process = child_broker().acquire_captured(
            parent,
            ProcessOwnershipContext {
                slot: HostProcessOwnershipSlot::EscalationHostProcess(request_id),
                ..self.ownership
            },
            &target,
            move |staged| stage_recorder.borrow_mut().stage(staged),
            move |staged| recorder.borrow_mut().acknowledge(staged),
        );
        let Ok(process) = process else {
            self.finish(&authority, EscalationOutcome::Failed, None)?;
            return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
        };
        self.active.insert(
            request_id,
            ActiveRequest::Executing(Box::new(Execution {
                authority,
                process,
                window: ResultSenderWindow::default(),
                chunks: BTreeMap::new(),
                stdout: Sha256::new(),
                stderr: Sha256::new(),
                stdout_eof: false,
                stderr_eof: false,
                acknowledgment_deadline: None,
            })),
        );
        Ok(EscalationPhase::Executing)
    }

    /// Advances nonblocking output while preserving the eight-chunk kernel backpressure bound.
    ///
    /// # Errors
    /// Returns an error when exact output publication, process observation, or reaping fails.
    pub fn progress(
        &mut self,
        request_id: RequestId,
        now: Instant,
    ) -> Result<EscalationPhase, EscalationError> {
        let Some(active) = self.active.remove(&request_id) else {
            return self
                .results
                .get(&request_id)
                .map(|result| EscalationPhase::Delivered(result.outcome))
                .ok_or(EscalationError::UnknownRequest);
        };
        let ActiveRequest::Executing(mut execution) = active else {
            self.active.insert(request_id, active);
            return Ok(EscalationPhase::AwaitingApproval);
        };
        let progress = self.read_available(&mut execution, now);
        if progress.is_err() {
            let authority = self.terminate(*execution)?;
            self.finish(&authority, EscalationOutcome::Failed, None)?;
            return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
        }
        if execution.process.process().has_exited()?
            && execution.stdout_eof
            && execution.stderr_eof
            && execution.window.outstanding() == 0
        {
            let execution = *execution;
            let authority = execution.authority;
            let stdout = StreamDigest::new(execution.stdout.finalize().into());
            let stderr = StreamDigest::new(execution.stderr.finalize().into());
            let chunk_count = execution.window.sent_count();
            let status = child_broker().reap_status(execution.process.into_process())?;
            self.finish_with_streams(
                &authority,
                EscalationOutcome::Completed,
                status.exit_code,
                chunk_count,
                stdout,
                stderr,
            )?;
            return Ok(EscalationPhase::Delivered(EscalationOutcome::Completed));
        }
        self.active
            .insert(request_id, ActiveRequest::Executing(execution));
        Ok(EscalationPhase::Executing)
    }

    fn read_available(
        &mut self,
        execution: &mut Execution,
        now: Instant,
    ) -> Result<(), EscalationError> {
        while execution.window.outstanding() < 8 {
            let mut made_progress = false;
            for stream in [ResultStream::Stdout, ResultStream::Stderr] {
                if execution.window.outstanding() >= 8 {
                    break;
                }
                let eof = match stream {
                    ResultStream::Stdout => &mut execution.stdout_eof,
                    ResultStream::Stderr => &mut execution.stderr_eof,
                };
                if *eof {
                    continue;
                }
                let (stdout, stderr) = execution.process.streams();
                let reader = match stream {
                    ResultStream::Stdout => stdout,
                    ResultStream::Stderr => stderr,
                };
                let mut bytes = vec![0_u8; MAX_RESULT_CHUNK_BYTES];
                match reader.read(&mut bytes) {
                    Ok(0) => {
                        *eof = true;
                        made_progress = true;
                    }
                    Ok(length) => {
                        bytes.truncate(length);
                        let sequence = execution.window.reserve(length)?;
                        let chunk = MailboxResultChunk {
                            protocol_version: ProtocolVersion::V1,
                            session_binding: execution.authority.session.session_binding().clone(),
                            request_id: execution.authority.request.request_id(),
                            request_sha256: execution.authority.digest,
                            coordinator_credential_id: execution
                                .authority
                                .session
                                .coordinator_credential_id(),
                            sequence,
                            stream,
                            bytes,
                        };
                        match stream {
                            ResultStream::Stdout => execution.stdout.update(&chunk.bytes),
                            ResultStream::Stderr => execution.stderr.update(&chunk.bytes),
                        }
                        let name = result_chunk_file_name(chunk.request_id, sequence);
                        let encoded = encode_signed(&chunk, &self.coordinator_key)?;
                        let identity = self.outbox.publish(&name, &encoded)?;
                        execution
                            .chunks
                            .insert(sequence, PublishedChunk { name, identity });
                        if execution.acknowledgment_deadline.is_none() {
                            execution.acknowledgment_deadline =
                                Some(now + self.acknowledgment_timeout);
                        }
                        made_progress = true;
                    }
                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
                    Err(error) => return Err(error.into()),
                }
            }
            if !made_progress {
                break;
            }
        }
        Ok(())
    }

    /// Accepts a highest-contiguous signed acknowledgement and removes only matching chunks.
    ///
    /// # Errors
    /// Refuses tuple mismatch, replay, regression, or acknowledgement beyond sent output.
    pub fn acknowledge(
        &mut self,
        request_id: RequestId,
        signed_acknowledgment: &[u8],
        now: Instant,
    ) -> Result<(), EscalationError> {
        let Some(ActiveRequest::Executing(execution)) = self.active.get_mut(&request_id) else {
            return Err(EscalationError::InvalidPhase);
        };
        let ack = execution.authority.session.verify_result_ack(
            signed_acknowledgment,
            request_id,
            execution.authority.digest,
        )?;
        execution
            .window
            .acknowledge(ack.highest_contiguous_sequence)?;
        let acknowledged = execution
            .chunks
            .range(..=ack.highest_contiguous_sequence)
            .map(|(sequence, _)| *sequence)
            .collect::<Vec<_>>();
        for sequence in acknowledged {
            if let Some(chunk) = execution.chunks.get(&sequence) {
                self.outbox.remove_if_matches(&chunk.name, chunk.identity)?;
                execution.chunks.remove(&sequence);
            }
        }
        execution.acknowledgment_deadline = if execution.window.outstanding() == 0 {
            None
        } else {
            Some(now + self.acknowledgment_timeout)
        };
        Ok(())
    }

    /// Cancels an exact live request after verifying the sandbox signature and complete tuple.
    ///
    /// # Errors
    /// Refuses unknown, terminal, unsigned, or mismatched cancellations.
    pub fn cancel(
        &mut self,
        request_id: RequestId,
        signed_cancellation: &[u8],
    ) -> Result<EscalationPhase, EscalationError> {
        let Some(active) = self.active.get(&request_id) else {
            return Err(EscalationError::UnknownRequest);
        };
        match active {
            ActiveRequest::AwaitingApproval(authority) => {
                authority
                    .session
                    .verify_cancellation(signed_cancellation, request_id)?;
            }
            ActiveRequest::Executing(execution) => {
                execution
                    .authority
                    .session
                    .verify_cancellation(signed_cancellation, request_id)?;
            }
        }
        let active = self
            .active
            .remove(&request_id)
            .ok_or(EscalationError::UnknownRequest)?;
        let authority = match active {
            ActiveRequest::AwaitingApproval(authority) => *authority,
            ActiveRequest::Executing(execution) => self.terminate(*execution)?,
        };
        self.finish(&authority, EscalationOutcome::Cancelled, None)?;
        Ok(EscalationPhase::Delivered(EscalationOutcome::Cancelled))
    }

    /// Enforces request and acknowledgement deadlines with exact pidfd termination and reaping.
    ///
    /// # Errors
    /// Returns an error when exact cleanup or terminal publication fails.
    pub fn expire(&mut self, now: Instant) -> Result<Vec<RequestId>, EscalationError> {
        let expired = self
            .active
            .iter()
            .filter_map(|(request_id, active)| {
                let expired = match active {
                    ActiveRequest::AwaitingApproval(authority) => {
                        now.saturating_duration_since(authority.admitted_at) >= self.request_timeout
                    }
                    ActiveRequest::Executing(execution) => {
                        now.saturating_duration_since(execution.authority.admitted_at)
                            >= self.request_timeout
                            || execution
                                .acknowledgment_deadline
                                .is_some_and(|deadline| now >= deadline)
                    }
                };
                expired.then_some(*request_id)
            })
            .collect::<Vec<_>>();
        for request_id in &expired {
            let active = self
                .active
                .remove(request_id)
                .ok_or(EscalationError::UnknownRequest)?;
            let authority = match active {
                ActiveRequest::AwaitingApproval(authority) => *authority,
                ActiveRequest::Executing(execution) => self.terminate(*execution)?,
            };
            self.finish(&authority, EscalationOutcome::TimedOut, None)?;
        }
        Ok(expired)
    }

    /// Removes all exact result files and exact-child authority for a disconnected session.
    ///
    /// # Errors
    /// Returns an error when exact cleanup cannot complete.
    pub fn session_lost(&mut self, session_id: SessionId) -> Result<(), EscalationError> {
        if let Some(request_id) = self.active_sessions.remove(&session_id)
            && let Some(active) = self.active.remove(&request_id)
        {
            match active {
                ActiveRequest::AwaitingApproval(_) => {}
                ActiveRequest::Executing(execution) => {
                    self.terminate(*execution)?;
                }
            }
        }
        let result_ids = self
            .results
            .iter()
            .filter_map(|(request_id, result)| {
                (result.session_id == session_id).then_some(*request_id)
            })
            .collect::<Vec<_>>();
        for request_id in result_ids {
            if let Some(result) = self.results.remove(&request_id) {
                self.outbox
                    .remove_if_matches(&result.name, result.identity)?;
            }
        }
        Ok(())
    }

    fn terminate(&mut self, execution: Execution) -> Result<RequestAuthority, EscalationError> {
        execution.process.process().send_signal(Signal::KILL)?;
        let authority = execution.authority;
        child_broker().reap(execution.process.into_process())?;
        for chunk in execution.chunks.into_values() {
            self.outbox.remove_if_matches(&chunk.name, chunk.identity)?;
        }
        Ok(authority)
    }

    fn finish(
        &mut self,
        authority: &RequestAuthority,
        outcome: EscalationOutcome,
        exit_code: Option<i32>,
    ) -> Result<(), EscalationError> {
        self.finish_with_streams(
            authority,
            outcome,
            exit_code,
            0,
            StreamDigest::new(Sha256::digest([]).into()),
            StreamDigest::new(Sha256::digest([]).into()),
        )
    }

    fn finish_with_streams(
        &mut self,
        authority: &RequestAuthority,
        outcome: EscalationOutcome,
        exit_code: Option<i32>,
        chunk_count: u64,
        stdout_sha256: StreamDigest,
        stderr_sha256: StreamDigest,
    ) -> Result<(), EscalationError> {
        let request_id = authority.request.request_id();
        let session_id = authority.request.session_binding().session_id;
        let result = MailboxResult {
            protocol_version: ProtocolVersion::V1,
            session_binding: authority.session.session_binding().clone(),
            request_id,
            request_sha256: authority.digest,
            coordinator_credential_id: authority.session.coordinator_credential_id(),
            outcome,
            exit_code,
            chunk_count,
            stdout_sha256,
            stderr_sha256,
        };
        let encoded = encode_signed(&result, &self.coordinator_key)?;
        let name = result_file_name(request_id);
        let identity = self.outbox.publish(&name, &encoded)?;
        self.active_sessions.remove(&session_id);
        self.results.insert(
            request_id,
            PublishedResult {
                session_id,
                outcome,
                name,
                identity,
            },
        );
        Ok(())
    }
}

fn validate_executable(executable: impl AsFd, invoking_uid: u32) -> io::Result<()> {
    let stat = fstat(executable)?;
    let file_type = FileType::from_raw_mode(stat.st_mode);
    let mode = Mode::from_raw_mode(stat.st_mode);
    if !file_type.is_file()
        || stat.st_nlink != 1
        || (stat.st_uid != 0 && stat.st_uid != invoking_uid)
        || mode.intersects(Mode::WGRP | Mode::WOTH)
        || !mode.contains(Mode::XUSR)
    {
        return Err(io::Error::other(
            "approved executable changed type, links, ownership, mode, or executability",
        ));
    }
    Ok(())
}

#[derive(Default)]
pub struct LocalExecutionOwnership {
    staged: Vec<StagedProcessOwnership>,
}

impl ExecutionOwnershipRecorder for LocalExecutionOwnership {
    fn stage(&mut self, ownership: StagedProcessOwnership) -> io::Result<()> {
        if self.staged.contains(&ownership) {
            Err(io::Error::other("execution ownership was already staged"))
        } else {
            self.staged.push(ownership);
            Ok(())
        }
    }

    fn acknowledge(
        &mut self,
        ownership: StagedProcessOwnership,
    ) -> io::Result<ProcessOwnershipCommitAck> {
        if let Some(index) = self.staged.iter().position(|staged| *staged == ownership) {
            self.staged.swap_remove(index);
            Ok(ProcessOwnershipCommitAck::v1(ownership))
        } else {
            Err(io::Error::other("execution ownership is not staged"))
        }
    }
}