use std::collections::{HashMap, VecDeque};
use std::ffi::OsString;
use std::fs::File;
use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::os::unix::ffi::OsStringExt as _;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use lazaret_core::identity::{
    ContainerId, CoordinatorId, DeviceId, DomainId, FileDescriptor, GenerationId,
    GuestProcessGroupId, GuestProcessId, GuestProcessStartTimeTicks, InodeId,
    PidfsProcessIdentityV1, SessionId,
};
use lazaret_core::records::{
    GuestProcessIdentity, ParticipantOwnership, ParticipantRole, ParticipantState,
    PersistedHostProcessOwnership, ProcessOwnershipCommitAck, PtyIdentity, SessionExit,
    SessionPhase, SessionTerminal, SessionTerminalEvent, SessionTerminalFields,
    SessionTerminalMessage, SessionTerminalState, StagedProcessOwnership, TerminalOutput,
    TerminalResize, TerminalSignalKind,
};
use lazaret_linux::process::{
    ChildStandardIo, ProcessExit, ProcessOwnershipContext, RetainedProcess, TargetCommand,
    child_broker, current_process_identity,
};
use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl, fstat};
use rustix::pty::{OpenptFlags, grantpt, ioctl_tiocgptpeer, openpt, unlockpt};
use rustix::termios::{Winsize, tcsetwinsize};

use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
use crate::sandbox_init::{PRIVATE_SESSION_EXEC_ROLE, PRIVATE_SESSION_IDENTITY_ROLE};

const MAX_EVENT_BYTES: usize = 16 * 1024;
const MAX_INPUT_QUEUE_BYTES: usize = 64 * 1024;
const MAX_OUTPUT_BYTES: usize = 16 * 1024;
const MAX_DRAIN_BYTES: usize = 64 * 1024;
const POLL_INTERVAL: Duration = Duration::from_millis(5);
const IDENTITY_DEADLINE: Duration = Duration::from_secs(5);
const SIGNAL_GRACE: Duration = Duration::from_millis(250);
const DRAIN_DEADLINE: Duration = Duration::from_millis(250);
const SANDBOX_INIT_PATH: &str = "/run/lazaret/bin/lazaret-sandbox-init";

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TerminalRoute {
    Detached,
    Sandbox,
    Prompt,
    HostProcess,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TerminalLoss {
    ClientDisconnected,
    PtyLost,
    ExecClientLost,
    HostCoordinatorLost,
}

#[derive(Debug)]
pub enum TerminalError {
    Io(io::Error),
    Store(DomainStoreError),
    UnknownSession,
    DuplicateSession,
    StaleGeneration,
    InactiveRoute,
    Releasing,
    OversizedInput,
    Backpressure,
    GuestIdentityMismatch,
    GuestGroupStillPresent,
    InvalidExit,
    AuthorityMismatch,
}

impl std::fmt::Display for TerminalError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io(error) => write!(formatter, "Session Terminal I/O failed: {error}"),
            Self::Store(error) => {
                write!(formatter, "Session Terminal store update failed: {error}")
            }
            Self::UnknownSession => formatter.write_str("Session Terminal is unknown"),
            Self::DuplicateSession => formatter.write_str("Session Terminal already exists"),
            Self::StaleGeneration => formatter.write_str("Session Terminal generation is stale"),
            Self::InactiveRoute => formatter.write_str("Session Terminal route is not sandbox"),
            Self::Releasing => formatter.write_str("Session Terminal is releasing"),
            Self::OversizedInput => formatter.write_str("terminal input exceeds the event bound"),
            Self::Backpressure => formatter.write_str("terminal input is backpressured"),
            Self::GuestIdentityMismatch => {
                formatter.write_str("guest process-group identity changed")
            }
            Self::GuestGroupStillPresent => {
                formatter.write_str("guest process group remains present")
            }
            Self::InvalidExit => {
                formatter.write_str("exec client returned an invalid exit outcome")
            }
            Self::AuthorityMismatch => {
                formatter.write_str("Session Terminal runtime authority does not match")
            }
        }
    }
}

impl std::error::Error for TerminalError {}

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

impl From<DomainStoreError> for TerminalError {
    fn from(error: DomainStoreError) -> Self {
        Self::Store(error)
    }
}

#[allow(clippy::missing_errors_doc)]
pub trait PromptRouteCancellation: Send + Sync {
    fn cancel_for_close(&self, session_id: SessionId) -> Result<(), TerminalError>;
}

pub struct NoActivePrompt;

impl PromptRouteCancellation for NoActivePrompt {
    fn cancel_for_close(&self, _session_id: SessionId) -> Result<(), TerminalError> {
        Ok(())
    }
}

#[allow(clippy::missing_errors_doc)]
pub trait SessionTerminalProcess: Send {
    fn write_input(&mut self, bytes: &[u8]) -> io::Result<usize>;
    fn resize(&self, resize: TerminalResize) -> io::Result<()>;
    fn read_output(&mut self, bytes: &mut [u8]) -> io::Result<usize>;
    fn exec_exited(&self) -> io::Result<bool>;
    fn signal_group(&self, signal: &str) -> Result<(), TerminalError>;
    fn group_absent(&self) -> Result<bool, TerminalError>;
    fn terminate_exec(&self) -> io::Result<()>;
    fn reap_exec(&mut self) -> io::Result<ProcessExit>;
    fn close_pty(&mut self);
}

#[allow(clippy::missing_errors_doc)]
pub trait SessionTerminalFactory: Send + Sync {
    fn acquire(
        &self,
        request: &TerminalAcquisition,
        recorder: &HostCoordinatorTerminalRecorder,
    ) -> Result<(SessionTerminal, Box<dyn SessionTerminalProcess>), TerminalError>;
}

#[derive(Clone, Debug)]
pub struct TerminalAcquisition {
    pub domain_id: DomainId,
    pub generation_id: GenerationId,
    pub coordinator_id: CoordinatorId,
    pub session_id: SessionId,
    pub container_id: ContainerId,
    pub host_coordinator: PidfsProcessIdentityV1,
    pub runsc_program: OsString,
    pub runsc_global_arguments: Vec<OsString>,
    pub internal_pid_file: PathBuf,
    pub argv: Vec<Vec<u8>>,
}

pub struct HostCoordinatorTerminalRecorder {
    store: Arc<HostCoordinatorDomainStore>,
}

impl HostCoordinatorTerminalRecorder {
    #[must_use]
    pub fn new(store: Arc<HostCoordinatorDomainStore>) -> Self {
        Self { store }
    }

    fn put(&self, terminal: SessionTerminal) -> Result<(), TerminalError> {
        self.store
            .apply(&[DomainStoreMutation::PutSessionTerminal(terminal)])?;
        Ok(())
    }

    fn require_authority(&self, request: &TerminalAcquisition) -> Result<(), TerminalError> {
        let snapshot = self.store.snapshot()?;
        if snapshot.domain.domain_id != request.domain_id
            || !snapshot.generations.iter().any(|generation| {
                generation.generation_id == request.generation_id
                    && generation.coordinator_id == request.coordinator_id
            })
        {
            return Err(TerminalError::AuthorityMismatch);
        }
        let sandbox = snapshot
            .participants
            .iter()
            .find(|participant| {
                participant.generation_id() == request.generation_id
                    && participant.role() == ParticipantRole::Sandbox
                    && participant.state() == ParticipantState::Committed
            })
            .and_then(|participant| match participant.ownership() {
                Some(ParticipantOwnership::Sandbox(sandbox)) => Some(sandbox),
                _ => None,
            })
            .ok_or(TerminalError::AuthorityMismatch)?;
        if sandbox.container_id != Some(request.container_id)
            || sandbox.host_coordinator.serviceable_pidfs().ok() != Some(&request.host_coordinator)
        {
            return Err(TerminalError::AuthorityMismatch);
        }
        Ok(())
    }

    fn begin_release(&self, terminal: SessionTerminal) -> Result<(), TerminalError> {
        let snapshot = self.store.snapshot()?;
        let mut session = snapshot
            .sessions
            .into_iter()
            .find(|session| {
                session.generation_id == terminal.generation_id()
                    && session.session_id == terminal.session_id()
            })
            .ok_or(TerminalError::UnknownSession)?;
        session.session_phase = SessionPhase::Closed;
        self.store.apply(&[
            DomainStoreMutation::PutSession(session),
            DomainStoreMutation::PutSessionTerminal(terminal),
        ])?;
        Ok(())
    }

    fn clean(&self, terminal: SessionTerminal) -> Result<(), TerminalError> {
        self.store
            .apply(&[DomainStoreMutation::CleanSessionTerminalAfterAbsence(
                terminal,
            )])?;
        Ok(())
    }
}

struct ActiveTerminal {
    generation_id: GenerationId,
    route: TerminalRoute,
    record: SessionTerminal,
    process: Box<dyn SessionTerminalProcess>,
    releasing: bool,
    pending_input: VecDeque<u8>,
}

pub struct SessionTerminalEngine {
    recorder: HostCoordinatorTerminalRecorder,
    factory: Arc<dyn SessionTerminalFactory>,
    cancellation: Arc<dyn PromptRouteCancellation>,
    active: HashMap<SessionId, ActiveTerminal>,
}

impl SessionTerminalEngine {
    #[must_use]
    pub fn new(
        store: Arc<HostCoordinatorDomainStore>,
        factory: Arc<dyn SessionTerminalFactory>,
        cancellation: Arc<dyn PromptRouteCancellation>,
    ) -> Self {
        Self {
            recorder: HostCoordinatorTerminalRecorder::new(store),
            factory,
            cancellation,
            active: HashMap::new(),
        }
    }

    /// # Errors
    /// Refuses duplicate identity or any durable, PTY, causal acquisition, or guest failure.
    pub fn acquire(&mut self, request: &TerminalAcquisition) -> Result<(), TerminalError> {
        if self.active.contains_key(&request.session_id) {
            return Err(TerminalError::DuplicateSession);
        }
        self.recorder.require_authority(request)?;
        let intent = terminal_snapshot(
            request,
            SessionTerminalState::Intent,
            PersistedHostProcessOwnership::unassigned(),
            None,
            PersistedHostProcessOwnership::unassigned(),
            None,
        )?;
        self.recorder.put(intent)?;
        let (record, process) = self.factory.acquire(request, &self.recorder)?;
        self.active.insert(
            request.session_id,
            ActiveTerminal {
                generation_id: request.generation_id,
                route: TerminalRoute::Sandbox,
                record,
                process,
                releasing: false,
                pending_input: VecDeque::new(),
            },
        );
        Ok(())
    }

    /// # Errors
    /// Refuses stale, unknown, detached, or releasing Session Terminals.
    pub fn set_route(
        &mut self,
        session_id: SessionId,
        generation_id: GenerationId,
        route: TerminalRoute,
    ) -> Result<(), TerminalError> {
        let terminal = self.active_mut(session_id, generation_id)?;
        if terminal.releasing || route == TerminalRoute::Detached {
            return Err(TerminalError::Releasing);
        }
        terminal.route = route;
        Ok(())
    }

    /// # Errors
    /// Refuses stale, unknown, inactive, releasing, oversized, or failed terminal events.
    pub fn handle(
        &mut self,
        generation_id: GenerationId,
        event: SessionTerminalEvent,
    ) -> Result<Vec<SessionTerminalMessage>, TerminalError> {
        let session_id = match &event {
            SessionTerminalEvent::Input(event) => event.session_id,
            SessionTerminalEvent::Resize(event) => event.session_id,
            SessionTerminalEvent::Signal(event) => event.session_id,
            SessionTerminalEvent::Close(event) => event.session_id,
        };
        if matches!(event, SessionTerminalEvent::Close(_)) {
            return self.close(session_id, generation_id, None, true);
        }
        let terminal = self.active_mut(session_id, generation_id)?;
        if terminal.releasing {
            return Err(TerminalError::Releasing);
        }
        if terminal.route != TerminalRoute::Sandbox {
            return Err(TerminalError::InactiveRoute);
        }
        match event {
            SessionTerminalEvent::Input(input) => {
                if input.bytes.len() > MAX_EVENT_BYTES {
                    return Err(TerminalError::OversizedInput);
                }
                if terminal.pending_input.len() + input.bytes.len() > MAX_INPUT_QUEUE_BYTES {
                    return Err(TerminalError::Backpressure);
                }
                terminal.pending_input.extend(input.bytes);
                flush_input(terminal)?;
                Ok(Vec::new())
            }
            SessionTerminalEvent::Resize(resize) => {
                terminal.process.resize(resize)?;
                terminal.process.signal_group("WINCH")?;
                Ok(Vec::new())
            }
            SessionTerminalEvent::Signal(signal) => {
                terminal.process.signal_group(match signal.signal {
                    TerminalSignalKind::Interrupt => "INT",
                    TerminalSignalKind::Quit => "QUIT",
                    TerminalSignalKind::Suspend => "TSTP",
                })?;
                Ok(Vec::new())
            }
            SessionTerminalEvent::Close(_) => unreachable!(),
        }
    }

    /// # Errors
    /// Refuses stale or unknown sessions and propagates terminal read or close failures.
    pub fn poll(
        &mut self,
        session_id: SessionId,
        generation_id: GenerationId,
    ) -> Result<Vec<SessionTerminalMessage>, TerminalError> {
        let terminal = self.active_mut(session_id, generation_id)?;
        if terminal.process.exec_exited()? {
            return self.close(session_id, generation_id, None, true);
        }
        if terminal.route != TerminalRoute::Sandbox || terminal.releasing {
            return Ok(Vec::new());
        }
        flush_input(terminal)?;
        let mut bytes = [0_u8; MAX_OUTPUT_BYTES];
        match terminal.process.read_output(&mut bytes) {
            Ok(0) => self.close(session_id, generation_id, Some(TerminalLoss::PtyLost), true),
            Ok(length) => Ok(vec![SessionTerminalMessage::Output(TerminalOutput {
                session_id,
                bytes: bytes[..length].to_vec(),
            })]),
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(Vec::new()),
            Err(_) => self.close(session_id, generation_id, Some(TerminalLoss::PtyLost), true),
        }
    }

    /// # Errors
    /// Refuses stale or unknown sessions and propagates ordered release failures.
    pub fn lose(
        &mut self,
        session_id: SessionId,
        generation_id: GenerationId,
        loss: TerminalLoss,
    ) -> Result<Vec<SessionTerminalMessage>, TerminalError> {
        if loss == TerminalLoss::HostCoordinatorLost {
            let mut terminal = self
                .active
                .remove(&session_id)
                .ok_or(TerminalError::UnknownSession)?;
            if terminal.generation_id != generation_id {
                self.active.insert(session_id, terminal);
                return Err(TerminalError::StaleGeneration);
            }
            terminal.process.close_pty();
            return Ok(Vec::new());
        }
        let connected = !matches!(
            loss,
            TerminalLoss::ClientDisconnected | TerminalLoss::HostCoordinatorLost
        );
        self.close(session_id, generation_id, Some(loss), connected)
    }

    #[must_use]
    pub fn contains(&self, session_id: SessionId) -> bool {
        self.active.contains_key(&session_id)
    }

    fn active_mut(
        &mut self,
        session_id: SessionId,
        generation_id: GenerationId,
    ) -> Result<&mut ActiveTerminal, TerminalError> {
        let terminal = self
            .active
            .get_mut(&session_id)
            .ok_or(TerminalError::UnknownSession)?;
        if terminal.generation_id != generation_id {
            return Err(TerminalError::StaleGeneration);
        }
        Ok(terminal)
    }

    fn close(
        &mut self,
        session_id: SessionId,
        generation_id: GenerationId,
        loss: Option<TerminalLoss>,
        connected: bool,
    ) -> Result<Vec<SessionTerminalMessage>, TerminalError> {
        let mut terminal = self
            .active
            .remove(&session_id)
            .ok_or(TerminalError::UnknownSession)?;
        if terminal.generation_id != generation_id {
            self.active.insert(session_id, terminal);
            return Err(TerminalError::StaleGeneration);
        }
        let result = (|| {
            terminal.releasing = true;
            terminal.route = TerminalRoute::Detached;
            let release =
                copy_terminal_state(&terminal.record, SessionTerminalState::ReleaseIntent)?;
            self.recorder.begin_release(release.clone())?;
            terminal.record = release;
            self.cancellation.cancel_for_close(session_id)?;

            let mut output = Vec::new();
            if !matches!(loss, Some(TerminalLoss::HostCoordinatorLost)) {
                for signal in ["HUP", "TERM", "KILL"] {
                    if terminal.process.group_absent()? {
                        break;
                    }
                    terminal.process.signal_group(signal)?;
                    wait_until(SIGNAL_GRACE, || terminal.process.group_absent())?;
                }
            }
            if !terminal.process.exec_exited()? {
                terminal.process.terminate_exec()?;
            }
            let exit = terminal.process.reap_exec()?;
            if connected {
                let deadline = Instant::now() + DRAIN_DEADLINE;
                let mut bytes = [0_u8; MAX_OUTPUT_BYTES];
                let mut drained = 0;
                while Instant::now() < deadline && drained < MAX_DRAIN_BYTES {
                    match terminal.process.read_output(&mut bytes) {
                        Ok(0) => break,
                        Ok(length) => {
                            let length = length.min(MAX_DRAIN_BYTES - drained);
                            drained += length;
                            output.push(SessionTerminalMessage::Output(TerminalOutput {
                                session_id,
                                bytes: bytes[..length].to_vec(),
                            }));
                        }
                        Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                            std::thread::sleep(POLL_INTERVAL);
                        }
                        Err(_) => break,
                    }
                }
            }
            terminal.process.close_pty();
            if !terminal.process.group_absent()? {
                return Err(TerminalError::GuestGroupStillPresent);
            }
            let clean = clean_terminal(generation_id, session_id)?;
            self.recorder.clean(clean)?;
            if connected {
                output.push(SessionTerminalMessage::Exit(exit_message(
                    session_id, loss, exit,
                )?));
            }
            Ok(output)
        })();
        if result.is_err() {
            self.active.insert(session_id, terminal);
        }
        result
    }
}

fn flush_input(terminal: &mut ActiveTerminal) -> Result<(), TerminalError> {
    while !terminal.pending_input.is_empty() {
        let (first, second) = terminal.pending_input.as_slices();
        let bytes = if first.is_empty() { second } else { first };
        match terminal.process.write_input(bytes) {
            Ok(0) => {
                return Err(io::Error::new(io::ErrorKind::WriteZero, "PTY input closed").into());
            }
            Ok(length) => {
                terminal.pending_input.drain(..length);
            }
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(()),
            Err(error) => return Err(error.into()),
        }
    }
    Ok(())
}

fn wait_until(
    duration: Duration,
    mut predicate: impl FnMut() -> Result<bool, TerminalError>,
) -> Result<(), TerminalError> {
    let deadline = Instant::now() + duration;
    while Instant::now() < deadline {
        if predicate()? {
            return Ok(());
        }
        std::thread::sleep(POLL_INTERVAL);
    }
    Ok(())
}

fn exit_message(
    session_id: SessionId,
    loss: Option<TerminalLoss>,
    exit: ProcessExit,
) -> Result<SessionExit, TerminalError> {
    if loss.is_some() {
        return Ok(SessionExit::runtime_lost(session_id));
    }
    match (exit.exit_code, exit.terminating_signal) {
        (Some(code), None) => Ok(SessionExit::exited(session_id, code)),
        (None, Some(_)) => Ok(SessionExit::terminated(session_id)),
        _ => Err(TerminalError::InvalidExit),
    }
}

pub struct RunscSessionTerminalFactory;

impl SessionTerminalFactory for RunscSessionTerminalFactory {
    #[allow(clippy::too_many_lines)]
    fn acquire(
        &self,
        request: &TerminalAcquisition,
        recorder: &HostCoordinatorTerminalRecorder,
    ) -> Result<(SessionTerminal, Box<dyn SessionTerminalProcess>), TerminalError> {
        if request.argv.is_empty()
            || request.argv[0].is_empty()
            || request.argv.iter().any(|argument| argument.contains(&0))
            || request.runsc_global_arguments.iter().any(|argument| {
                argument
                    .as_encoded_bytes()
                    .windows(14)
                    .any(|part| part == b"console-socket")
            })
        {
            return Err(io::Error::other("interactive runsc request is invalid").into());
        }
        if current_process_identity()? != request.host_coordinator {
            return Err(TerminalError::AuthorityMismatch);
        }
        let (master, slave, pty) = allocate_pty()?;
        remove_pid_file(&request.internal_pid_file)?;
        let mut target = TargetCommand::new(request.runsc_program.clone());
        for argument in &request.runsc_global_arguments {
            target = target.argument(argument.clone());
        }
        target = target
            .argument("exec")
            .argument(format!(
                "-internal-pid-file={}",
                request.internal_pid_file.display()
            ))
            .argument(hex_lower(request.container_id.as_bytes()))
            .argument(SANDBOX_INIT_PATH)
            .argument(PRIVATE_SESSION_EXEC_ROLE)
            .argument("--");
        for argument in &request.argv {
            target = target.argument(OsString::from_vec(argument.clone()));
        }
        target = match slave
            .try_clone()
            .and_then(|stdin| target.with_target_stdin(stdin))
        {
            Ok(target) => target,
            Err(error) => {
                drop(master);
                drop(slave);
                recorder.clean(clean_terminal(request.generation_id, request.session_id)?)?;
                return Err(error.into());
            }
        };
        let stage_request = request.clone();
        let stage_pty = pty;
        let staged_identity = Arc::new(Mutex::new(None));
        let callback_identity = Arc::clone(&staged_identity);
        let standard_io = match ChildStandardIo::terminal(slave) {
            Ok(standard_io) => standard_io,
            Err(error) => {
                drop(master);
                recorder.clean(clean_terminal(request.generation_id, request.session_id)?)?;
                return Err(error.into());
            }
        };
        let acquired = child_broker().acquire_with_stdio(
            request.host_coordinator,
            ProcessOwnershipContext {
                domain_id: request.domain_id,
                generation_id: request.generation_id,
                coordinator_id: request.coordinator_id,
                slot: lazaret_core::records::HostProcessOwnershipSlot::SessionTerminalExecClient(
                    request.session_id,
                ),
            },
            &target,
            standard_io,
            |staged| {
                let terminal = staged_terminal(&stage_request, stage_pty, staged)
                    .map_err(|error| terminal_io(&error))?;
                recorder
                    .put(terminal)
                    .map_err(|error| terminal_io(&error))?;
                *callback_identity
                    .lock()
                    .map_err(|_| io::Error::other("staged terminal lock poisoned"))? =
                    Some(staged.identity());
                Ok(())
            },
            |staged| Ok(ProcessOwnershipCommitAck::v1(staged)),
        );
        let process = match acquired {
            Ok(process) => process,
            Err(error) => {
                drop(master);
                clean_failed_blocked_acquisition(
                    recorder,
                    request,
                    pty,
                    *staged_identity
                        .lock()
                        .map_err(|_| io::Error::other("staged terminal lock poisoned"))?,
                )?;
                return Err(error.into());
            }
        };
        let guest = match wait_for_guest_identity(request, &process) {
            Ok(guest) => guest,
            Err(error) => {
                let release = partial_terminal_release(request, pty, process.identity())?;
                recorder.put(release)?;
                if !process.has_exited()? {
                    process.send_signal(rustix::process::Signal::KILL)?;
                }
                child_broker().reap(process)?;
                return Err(error);
            }
        };
        let running = terminal_snapshot(
            request,
            SessionTerminalState::Running,
            PersistedHostProcessOwnership::committed(request.host_coordinator),
            Some(pty),
            PersistedHostProcessOwnership::committed(process.identity()),
            Some(guest),
        )?;
        recorder.put(running.clone())?;
        Ok((
            running,
            Box::new(RunscSessionTerminal {
                request: request.clone(),
                master: Some(master),
                exec: Some(process),
                guest,
            }),
        ))
    }
}

fn clean_failed_blocked_acquisition(
    recorder: &HostCoordinatorTerminalRecorder,
    request: &TerminalAcquisition,
    pty: PtyIdentity,
    staged: Option<PidfsProcessIdentityV1>,
) -> Result<(), TerminalError> {
    if let Some(staged) = staged {
        recorder.put(partial_terminal_release(request, pty, staged)?)?;
    }
    recorder.clean(clean_terminal(request.generation_id, request.session_id)?)
}

fn partial_terminal_release(
    request: &TerminalAcquisition,
    pty: PtyIdentity,
    exec_client: PidfsProcessIdentityV1,
) -> Result<SessionTerminal, TerminalError> {
    terminal_snapshot(
        request,
        SessionTerminalState::ReleaseIntent,
        PersistedHostProcessOwnership::committed(request.host_coordinator),
        Some(pty),
        PersistedHostProcessOwnership::staged(exec_client),
        None,
    )
}

struct RunscSessionTerminal {
    request: TerminalAcquisition,
    master: Option<File>,
    exec: Option<RetainedProcess>,
    guest: GuestProcessIdentity,
}

impl SessionTerminalProcess for RunscSessionTerminal {
    fn write_input(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.master
            .as_mut()
            .ok_or_else(|| io::Error::other("PTY is closed"))?
            .write(bytes)
    }

    fn resize(&self, resize: TerminalResize) -> io::Result<()> {
        let master = self
            .master
            .as_ref()
            .ok_or_else(|| io::Error::other("PTY is closed"))?;
        tcsetwinsize(
            master,
            Winsize {
                ws_row: resize.rows.get(),
                ws_col: resize.columns.get(),
                ws_xpixel: resize.pixel_width,
                ws_ypixel: resize.pixel_height,
            },
        )?;
        Ok(())
    }

    fn read_output(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
        self.master
            .as_mut()
            .ok_or_else(|| io::Error::other("PTY is closed"))?
            .read(bytes)
    }

    fn exec_exited(&self) -> io::Result<bool> {
        self.exec
            .as_ref()
            .ok_or_else(|| io::Error::other("exec client was already reaped"))?
            .has_exited()
    }

    fn signal_group(&self, signal: &str) -> Result<(), TerminalError> {
        let observed = observe_guest_identity(&self.request, self.guest.leader_pid.get())?;
        if observed != Some(self.guest) {
            return Err(TerminalError::GuestIdentityMismatch);
        }
        let mut target = runsc_target(&self.request);
        target = target
            .argument("kill")
            .argument("-pgid")
            .argument(self.guest.process_group_id.get().to_string())
            .argument(hex_lower(self.request.container_id.as_bytes()))
            .argument(signal);
        child_broker().run_transient(&target)?;
        Ok(())
    }

    fn group_absent(&self) -> Result<bool, TerminalError> {
        Ok(observe_guest_identity(&self.request, self.guest.leader_pid.get())?.is_none())
    }

    fn terminate_exec(&self) -> io::Result<()> {
        self.exec
            .as_ref()
            .ok_or_else(|| io::Error::other("exec client was already reaped"))?
            .send_signal(rustix::process::Signal::KILL)
    }

    fn reap_exec(&mut self) -> io::Result<ProcessExit> {
        child_broker().reap_status(
            self.exec
                .take()
                .ok_or_else(|| io::Error::other("exec client was already reaped"))?,
        )
    }

    fn close_pty(&mut self) {
        self.master.take();
    }
}

fn allocate_pty() -> io::Result<(File, File, PtyIdentity)> {
    let master = openpt(OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC)?;
    grantpt(&master)?;
    unlockpt(&master)?;
    let slave = ioctl_tiocgptpeer(
        &master,
        OpenptFlags::RDWR | OpenptFlags::NOCTTY | OpenptFlags::CLOEXEC,
    )?;
    let flags = fcntl_getfl(&master)?;
    fcntl_setfl(&master, flags | OFlags::NONBLOCK)?;
    let stat = fstat(&master)?;
    let pty = PtyIdentity {
        master_fd: FileDescriptor::new(master.as_raw_fd()),
        device: DeviceId::new(stat.st_dev),
        inode: InodeId::new(stat.st_ino),
    };
    Ok((File::from(master), File::from(slave), pty))
}

fn staged_terminal(
    request: &TerminalAcquisition,
    pty: PtyIdentity,
    staged: StagedProcessOwnership,
) -> Result<SessionTerminal, TerminalError> {
    terminal_snapshot(
        request,
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::committed(request.host_coordinator),
        Some(pty),
        PersistedHostProcessOwnership::staged(staged.identity()),
        None,
    )
}

fn terminal_snapshot(
    request: &TerminalAcquisition,
    state: SessionTerminalState,
    host_coordinator: PersistedHostProcessOwnership,
    pty_master: Option<PtyIdentity>,
    exec_client: PersistedHostProcessOwnership,
    guest_process: Option<GuestProcessIdentity>,
) -> Result<SessionTerminal, TerminalError> {
    SessionTerminal::snapshot(SessionTerminalFields {
        generation_id: request.generation_id,
        session_id: request.session_id,
        state,
        container_id: (state != SessionTerminalState::Intent || pty_master.is_some())
            .then_some(request.container_id),
        host_coordinator,
        pty_master,
        exec_client,
        guest_process,
    })
    .map_err(|_| io::Error::other("Session Terminal snapshot is invalid").into())
}

fn copy_terminal_state(
    record: &SessionTerminal,
    state: SessionTerminalState,
) -> Result<SessionTerminal, TerminalError> {
    SessionTerminal::snapshot(SessionTerminalFields {
        generation_id: record.generation_id(),
        session_id: record.session_id(),
        state,
        container_id: record.container_id(),
        host_coordinator: *record.host_coordinator(),
        pty_master: record.pty_master(),
        exec_client: *record.exec_client(),
        guest_process: record.guest_process(),
    })
    .map_err(|_| io::Error::other("Session Terminal release snapshot is invalid").into())
}

fn clean_terminal(
    generation_id: GenerationId,
    session_id: SessionId,
) -> Result<SessionTerminal, TerminalError> {
    SessionTerminal::snapshot(SessionTerminalFields {
        generation_id,
        session_id,
        state: SessionTerminalState::Clean,
        container_id: None,
        host_coordinator: PersistedHostProcessOwnership::unassigned(),
        pty_master: None,
        exec_client: PersistedHostProcessOwnership::unassigned(),
        guest_process: None,
    })
    .map_err(|_| io::Error::other("Session Terminal clean snapshot is invalid").into())
}

fn wait_for_guest_identity(
    request: &TerminalAcquisition,
    process: &RetainedProcess,
) -> Result<GuestProcessIdentity, TerminalError> {
    let deadline = Instant::now() + IDENTITY_DEADLINE;
    while Instant::now() < deadline {
        if process.has_exited()? {
            return Err(io::Error::other("exec client exited before guest identity").into());
        }
        match std::fs::read_to_string(&request.internal_pid_file) {
            Ok(value) => {
                let pid = value
                    .trim()
                    .parse::<u32>()
                    .ok()
                    .filter(|pid| *pid > 0)
                    .ok_or_else(|| io::Error::other("runsc internal PID file is malformed"))?;
                if let Some(identity) = observe_guest_identity(request, pid)? {
                    if identity.leader_pid.get() != identity.process_group_id.get() {
                        return Err(TerminalError::GuestIdentityMismatch);
                    }
                    return Ok(identity);
                }
            }
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
        std::thread::sleep(POLL_INTERVAL);
    }
    Err(io::Error::other("guest identity deadline expired").into())
}

fn observe_guest_identity(
    request: &TerminalAcquisition,
    pid: u32,
) -> Result<Option<GuestProcessIdentity>, TerminalError> {
    let mut target = runsc_target(request);
    target = target
        .argument("exec")
        .argument(hex_lower(request.container_id.as_bytes()))
        .argument(SANDBOX_INIT_PATH)
        .argument(PRIVATE_SESSION_IDENTITY_ROLE)
        .argument(pid.to_string());
    let mut child = child_broker().run_transient_captured(&target)?;
    let deadline = Instant::now() + IDENTITY_DEADLINE;
    let mut stdout = Vec::new();
    let mut stderr = Vec::new();
    while Instant::now() < deadline && !child.process().has_exited()? {
        read_available(child.streams().0, &mut stdout)?;
        read_available(child.streams().1, &mut stderr)?;
        std::thread::sleep(POLL_INTERVAL);
    }
    if !child.process().has_exited()? {
        child.process().send_signal(rustix::process::Signal::KILL)?;
        child_broker().reap_status(child.into_process())?;
        return Err(io::Error::other("guest identity observation deadline expired").into());
    }
    read_available(child.streams().0, &mut stdout)?;
    read_available(child.streams().1, &mut stderr)?;
    let exit = child_broker().reap_status(child.into_process())?;
    if exit.exit_code != Some(0) || exit.terminating_signal.is_some() {
        return Ok(None);
    }
    let fields = std::str::from_utf8(&stdout)
        .ok()
        .map(str::split_whitespace)
        .into_iter()
        .flatten()
        .collect::<Vec<_>>();
    let [leader, start, group] = fields.as_slice() else {
        return Err(io::Error::other("guest identity response is malformed").into());
    };
    Ok(Some(GuestProcessIdentity {
        leader_pid: GuestProcessId::new(
            leader
                .parse()
                .map_err(|_| io::Error::other("guest leader PID is malformed"))?,
        ),
        leader_start_time: GuestProcessStartTimeTicks::new(
            start
                .parse()
                .map_err(|_| io::Error::other("guest start time is malformed"))?,
        ),
        process_group_id: GuestProcessGroupId::new(
            group
                .parse()
                .map_err(|_| io::Error::other("guest process group is malformed"))?,
        ),
    }))
}

fn runsc_target(request: &TerminalAcquisition) -> TargetCommand {
    request.runsc_global_arguments.iter().cloned().fold(
        TargetCommand::new(request.runsc_program.clone()),
        TargetCommand::argument,
    )
}

fn read_available(file: &mut File, output: &mut Vec<u8>) -> io::Result<()> {
    let mut bytes = [0_u8; 4096];
    loop {
        match file.read(&mut bytes) {
            Ok(0) => return Ok(()),
            Ok(length) => output.extend_from_slice(&bytes[..length]),
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(()),
            Err(error) => return Err(error),
        }
    }
}

fn remove_pid_file(path: &std::path::Path) -> io::Result<()> {
    match std::fs::remove_file(path) {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error),
    }
}

fn terminal_io(error: &TerminalError) -> io::Error {
    io::Error::other(error.to_string())
}

fn hex_lower(bytes: &[u8]) -> String {
    use std::fmt::Write as _;
    bytes.iter().fold(String::new(), |mut output, byte| {
        write!(output, "{byte:02x}").expect("writing to a String cannot fail");
        output
    })
}

#[cfg(test)]
mod tests;