lazaret/crates/lazaret-host-coordinator/src/terminal.rs

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::protocol::VerifiedSession;
use lazaret_core::records::{
    DeveloperDecision, DomainStore, DurableDecision, GuestProcessIdentity, InteractiveSessionBatch,
    InteractiveSessionEvent, InteractiveSessionMessage, InteractiveSessionRequest,
    ParticipantOwnership, ParticipantRole, ParticipantState, PersistedHostProcessOwnership,
    ProcessOwnershipCommitAck, ProtocolVersion, PtyIdentity, SessionExit, SessionPhase,
    SessionTerminal, SessionTerminalEvent, SessionTerminalFields, SessionTerminalMessage,
    SessionTerminalState, StagedProcessOwnership, TerminalOutput, TerminalPrompt, 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 sha2::{Digest, Sha256};

use crate::control::{ControlError, InteractiveControlSession, InteractiveSessionExecutor};
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(Clone, Copy)]
enum GuestReleaseAuthority {
    Validate,
    HostOnly,
}

enum GuestWaitOutcome {
    Absent,
    TimedOut,
    Revoked,
}

#[derive(Debug)]
pub enum TerminalError {
    Io(io::Error),
    Store(DomainStoreError),
    UnknownSession,
    DuplicateSession,
    StaleGeneration,
    InactiveRoute,
    Releasing,
    OversizedInput,
    Backpressure,
    GuestIdentityMismatch,
    GuestGroupStillPresent,
    InvalidExit,
    AuthorityMismatch,
    DuplicatePrompt,
    PromptValidation,
    AcquisitionCleanup {
        original: Box<TerminalError>,
        cleanup: Option<Box<TerminalError>>,
    },
}

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")
            }
            Self::DuplicatePrompt => formatter.write_str("Session already has an active prompt"),
            Self::PromptValidation => formatter.write_str("prompt authority validation failed"),
            Self::AcquisitionCleanup { original, cleanup } => {
                write!(formatter, "Session Terminal acquisition failed: {original}")?;
                if let Some(cleanup) = cleanup {
                    write!(formatter, "; partial terminal cleanup failed: {cleanup}")?;
                }
                Ok(())
            }
        }
    }
}

impl std::error::Error for TerminalError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Store(error) => Some(error),
            Self::AcquisitionCleanup { original, .. } => Some(original.as_ref()),
            Self::UnknownSession
            | Self::DuplicateSession
            | Self::StaleGeneration
            | Self::InactiveRoute
            | Self::Releasing
            | Self::OversizedInput
            | Self::Backpressure
            | Self::GuestIdentityMismatch
            | Self::GuestGroupStillPresent
            | Self::InvalidExit
            | Self::AuthorityMismatch
            | Self::DuplicatePrompt
            | Self::PromptValidation => None,
        }
    }
}

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(())
    }
}

struct PendingPrompt {
    authority: VerifiedSession,
    prompt: TerminalPrompt,
    emitted: bool,
    resumed_size: Option<TerminalResize>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RoutedPromptDecision {
    pub signed_decision: Vec<u8>,
    pub decision: DeveloperDecision,
    pub resumed_size: Option<TerminalResize>,
}

pub struct SessionPromptTransport {
    store: Arc<HostCoordinatorDomainStore>,
    pending: Mutex<HashMap<SessionId, PendingPrompt>>,
    routed: Mutex<HashMap<SessionId, RoutedPromptDecision>>,
}

impl SessionPromptTransport {
    #[must_use]
    pub fn new(store: Arc<HostCoordinatorDomainStore>) -> Self {
        Self {
            store,
            pending: Mutex::new(HashMap::new()),
            routed: Mutex::new(HashMap::new()),
        }
    }

    /// Queues one sandbox-signed request for its exact registered Session.
    ///
    /// # Errors
    /// Refuses invalid authority, stale Session state, and concurrent prompts.
    pub fn queue_signed_request(
        &self,
        authority: VerifiedSession,
        signed_request: &[u8],
    ) -> Result<TerminalPrompt, TerminalError> {
        let (request, digest) = authority
            .verify_request(signed_request)
            .map_err(|_| TerminalError::PromptValidation)?;
        let binding = authority.session_binding();
        let anchor_id = lazaret_core::identity::SessionAnchorId::new(
            Sha256::digest(&binding.session_anchor_public_key).into(),
        );
        let snapshot = self.store.snapshot()?;
        let mut session = snapshot
            .sessions
            .into_iter()
            .find(|session| {
                session.generation_id == binding.generation_identity.generation_id
                    && session.session_id == binding.session_id
                    && session.cli_id == binding.cli_id
                    && session.session_anchor_id == anchor_id
                    && session.session_phase == SessionPhase::Attached
            })
            .ok_or(TerminalError::AuthorityMismatch)?;
        let prompt = TerminalPrompt {
            session_id: binding.session_id,
            request_id: request.request_id(),
            request_sha256: digest,
            agent_profile: binding.generation_identity.agent_profile.clone(),
            command_argv: request.command_argv().to_vec(),
            justification: request.justification().to_owned(),
        };
        let mut pending = self
            .pending
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?;
        if pending.contains_key(&binding.session_id) {
            return Err(TerminalError::DuplicatePrompt);
        }
        session.session_phase = SessionPhase::Prompting;
        self.store
            .apply(&[DomainStoreMutation::PutSession(session)])?;
        pending.insert(
            binding.session_id,
            PendingPrompt {
                authority,
                prompt: prompt.clone(),
                emitted: false,
                resumed_size: None,
            },
        );
        Ok(prompt)
    }

    fn prompt_for(&self, session_id: SessionId) -> Result<Option<TerminalPrompt>, TerminalError> {
        let mut pending = self
            .pending
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?;
        let Some(pending) = pending.get_mut(&session_id) else {
            return Ok(None);
        };
        if pending.emitted {
            return Ok(None);
        }
        pending.emitted = true;
        Ok(Some(pending.prompt.clone()))
    }

    fn route_decision(
        &self,
        session_id: SessionId,
        signed_decision: Vec<u8>,
    ) -> Result<RoutedPromptDecision, TerminalError> {
        let mut prompts = self
            .pending
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?;
        let pending = prompts
            .get(&session_id)
            .ok_or(TerminalError::InactiveRoute)?;
        let decision = pending
            .authority
            .verify_developer_decision(
                &signed_decision,
                pending.prompt.request_id,
                pending.prompt.request_sha256,
            )
            .map_err(|_| TerminalError::PromptValidation)?;
        let resumed_size = pending.resumed_size;
        prompts.remove(&session_id);
        Ok(RoutedPromptDecision {
            signed_decision,
            decision,
            resumed_size,
        })
    }

    fn retain_prompt_resize(&self, resize: TerminalResize) -> Result<bool, TerminalError> {
        let mut prompts = self
            .pending
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?;
        let Some(pending) = prompts.get_mut(&resize.session_id) else {
            return Ok(false);
        };
        if !pending.emitted {
            return Err(TerminalError::InactiveRoute);
        }
        pending.resumed_size = Some(resize);
        Ok(true)
    }

    fn publish_decision(
        &self,
        session_id: SessionId,
        decision: RoutedPromptDecision,
    ) -> Result<(), TerminalError> {
        self.routed
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?
            .insert(session_id, decision);
        Ok(())
    }

    fn restore_sandbox_route(&self, session_id: SessionId) -> Result<(), TerminalError> {
        let snapshot = self.store.snapshot()?;
        let mut session = snapshot
            .sessions
            .into_iter()
            .find(|session| session.session_id == session_id)
            .ok_or(TerminalError::UnknownSession)?;
        if session.session_phase != SessionPhase::Prompting {
            return Err(TerminalError::InactiveRoute);
        }
        session.session_phase = SessionPhase::Attached;
        self.store
            .apply(&[DomainStoreMutation::PutSession(session)])?;
        Ok(())
    }

    /// # Errors
    /// Refuses access after internal synchronization failure.
    pub fn take_decision(
        &self,
        session_id: SessionId,
    ) -> Result<Option<RoutedPromptDecision>, TerminalError> {
        Ok(self
            .routed
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?
            .remove(&session_id))
    }

    pub(crate) fn begin_host_execution(&self, session_id: SessionId) -> Result<(), TerminalError> {
        self.transition_session(session_id, SessionPhase::HostExecuting)
    }

    pub(crate) fn finish_escalation(&self, session_id: SessionId) -> Result<(), TerminalError> {
        self.pending
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?
            .remove(&session_id);
        self.routed
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?
            .remove(&session_id);
        let snapshot = self.store.snapshot()?;
        let Some(session) = snapshot
            .sessions
            .iter()
            .find(|session| session.session_id == session_id)
        else {
            return Err(TerminalError::UnknownSession);
        };
        match session.session_phase {
            SessionPhase::Attached => return Ok(()),
            SessionPhase::Prompting | SessionPhase::HostExecuting => {}
            SessionPhase::New
            | SessionPhase::Waiting
            | SessionPhase::Registering
            | SessionPhase::Closed
            | SessionPhase::Failed => return Err(TerminalError::InactiveRoute),
        }
        self.transition_session(session_id, SessionPhase::Attached)
    }

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

impl PromptRouteCancellation for SessionPromptTransport {
    fn cancel_for_close(&self, session_id: SessionId) -> Result<(), TerminalError> {
        self.pending
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?
            .remove(&session_id);
        self.routed
            .lock()
            .map_err(|_| TerminalError::PromptValidation)?
            .remove(&session_id);
        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<dyn TerminalDomainStore>,
}

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

    #[cfg(test)]
    fn with_store(store: Arc<dyn TerminalDomainStore>) -> Self {
        Self { store }
    }

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

    fn require_authority(
        &self,
        request: &TerminalAcquisition,
    ) -> Result<TerminalRoute, TerminalError> {
        let snapshot = self.store.snapshot()?;
        let generation = snapshot
            .generations
            .iter()
            .find(|generation| generation.generation_id == request.generation_id);
        let current = snapshot
            .generations
            .iter()
            .all(|candidate| candidate.generation_id <= request.generation_id);
        if snapshot.domain.domain_id != request.domain_id
            || !current
            || generation.is_none_or(|generation| {
                generation.coordinator_id != request.coordinator_id
                    || generation.durable_decision != DurableDecision::Published
            })
        {
            return Err(TerminalError::AuthorityMismatch);
        }
        let route = snapshot
            .sessions
            .iter()
            .find(|session| {
                session.generation_id == request.generation_id
                    && session.session_id == request.session_id
            })
            .map(|session| match session.session_phase {
                SessionPhase::Attached => TerminalRoute::Sandbox,
                SessionPhase::Prompting => TerminalRoute::Prompt,
                SessionPhase::HostExecuting => TerminalRoute::HostProcess,
                SessionPhase::New
                | SessionPhase::Waiting
                | SessionPhase::Registering
                | SessionPhase::Closed
                | SessionPhase::Failed => TerminalRoute::Detached,
            })
            .ok_or(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(route)
    }

    fn require_guest_signal_authority(
        &self,
        request: &TerminalAcquisition,
        terminal: &SessionTerminal,
    ) -> Result<(), TerminalError> {
        let snapshot = self.store.snapshot()?;
        let current = snapshot
            .generations
            .iter()
            .all(|candidate| candidate.generation_id <= request.generation_id);
        let generation = snapshot.generations.iter().any(|generation| {
            generation.generation_id == request.generation_id
                && generation.coordinator_id == request.coordinator_id
                && generation.durable_decision == DurableDecision::Published
        });
        let session = snapshot.sessions.iter().any(|session| {
            session.generation_id == request.generation_id
                && session.session_id == request.session_id
                && matches!(
                    session.session_phase,
                    SessionPhase::Attached | SessionPhase::Closed
                )
        });
        let sandbox = snapshot.participants.iter().any(|participant| {
            participant.generation_id() == request.generation_id
                && participant.role() == ParticipantRole::Sandbox
                && participant.state() == ParticipantState::Committed
                && matches!(
                    participant.ownership(),
                    Some(ParticipantOwnership::Sandbox(sandbox))
                        if sandbox.container_id == Some(request.container_id)
                            && sandbox.host_coordinator.serviceable_pidfs().ok()
                                == Some(&request.host_coordinator)
                )
        });
        let terminal = snapshot.session_terminals.iter().any(|record| {
            record == terminal && record.state() == SessionTerminalState::ReleaseIntent
        });
        if snapshot.domain.domain_id == request.domain_id
            && current
            && generation
            && session
            && sandbox
            && terminal
        {
            Ok(())
        } else {
            Err(TerminalError::AuthorityMismatch)
        }
    }

    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)?;
        let mut mutations = Vec::with_capacity(2);
        if !matches!(
            session.session_phase,
            SessionPhase::Closed | SessionPhase::Failed
        ) {
            session.session_phase = SessionPhase::Closed;
            mutations.push(DomainStoreMutation::PutSession(session));
        }
        mutations.push(DomainStoreMutation::PutSessionTerminal(terminal));
        self.store.apply(&mutations)?;
        Ok(())
    }

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

trait TerminalDomainStore: Send + Sync {
    fn snapshot(&self) -> Result<DomainStore, DomainStoreError>;
    fn apply(&self, mutations: &[DomainStoreMutation]) -> Result<(), DomainStoreError>;
}

impl TerminalDomainStore for HostCoordinatorDomainStore {
    fn snapshot(&self) -> Result<DomainStore, DomainStoreError> {
        self.snapshot()
    }

    fn apply(&self, mutations: &[DomainStoreMutation]) -> Result<(), DomainStoreError> {
        self.apply(mutations)
    }
}

struct ActiveTerminal {
    generation_id: GenerationId,
    request: TerminalAcquisition,
    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>,
}

pub struct RunscInteractiveSessionExecutor {
    engine: Arc<Mutex<SessionTerminalEngine>>,
    prompt_transport: Arc<SessionPromptTransport>,
    domain_id: DomainId,
    generation_id: GenerationId,
    coordinator_id: CoordinatorId,
    container_id: ContainerId,
    host_coordinator: PidfsProcessIdentityV1,
    runsc_program: OsString,
    runsc_global_arguments: Vec<OsString>,
    runtime_root: PathBuf,
}

impl RunscInteractiveSessionExecutor {
    /// Creates the generation-bound control adapter for the coordinator-owned terminal engine.
    ///
    /// # Errors
    /// Refuses a run command without one exact `run`/bundle/container suffix.
    pub fn new(
        store: Arc<HostCoordinatorDomainStore>,
        generation: &lazaret_core::records::GenerationIdentity,
        container_id: ContainerId,
        host_coordinator: PidfsProcessIdentityV1,
        command: &lazaret_runtime_supervisor::sandbox::RunscCommand,
        runtime_root: PathBuf,
    ) -> Result<Self, TerminalError> {
        let run = command
            .arguments()
            .iter()
            .position(|argument| argument == "run")
            .ok_or(TerminalError::AuthorityMismatch)?;
        if run + 3 != command.arguments().len() {
            return Err(TerminalError::AuthorityMismatch);
        }
        let prompt_transport = Arc::new(SessionPromptTransport::new(Arc::clone(&store)));
        Ok(Self {
            engine: Arc::new(Mutex::new(SessionTerminalEngine::new(
                store,
                Arc::new(RunscSessionTerminalFactory),
                prompt_transport.clone(),
            ))),
            prompt_transport,
            domain_id: generation.domain_id,
            generation_id: generation.generation_id,
            coordinator_id: generation.coordinator_id,
            container_id,
            host_coordinator,
            runsc_program: command.program().to_owned(),
            runsc_global_arguments: command.arguments()[..run].to_vec(),
            runtime_root,
        })
    }

    #[must_use]
    pub fn prompt_transport(&self) -> Arc<SessionPromptTransport> {
        Arc::clone(&self.prompt_transport)
    }
}

impl InteractiveSessionExecutor for RunscInteractiveSessionExecutor {
    fn start(
        &self,
        request: &InteractiveSessionRequest,
    ) -> Result<Box<dyn InteractiveControlSession>, ControlError> {
        if request.protocol_version != ProtocolVersion::V1
            || request.initial_size.session_id != request.session_id
        {
            return Err(ControlError::InvalidRequest);
        }
        let acquisition = TerminalAcquisition {
            domain_id: self.domain_id,
            generation_id: self.generation_id,
            coordinator_id: self.coordinator_id,
            session_id: request.session_id,
            container_id: self.container_id,
            host_coordinator: self.host_coordinator,
            runsc_program: self.runsc_program.clone(),
            runsc_global_arguments: self.runsc_global_arguments.clone(),
            internal_pid_file: self.runtime_root.join(format!(
                "session-{}.pid",
                hex_session(request.session_id.as_bytes())
            )),
            argv: request.argv.clone(),
        };
        let mut engine = self.engine.lock().map_err(|_| ControlError::LockPoisoned)?;
        engine
            .acquire(&acquisition)
            .map_err(|error| terminal_control_error(&error))?;
        if let Err(error) = engine.handle(
            self.generation_id,
            SessionTerminalEvent::Resize(request.initial_size),
        ) {
            let _ = engine.lose(
                request.session_id,
                self.generation_id,
                TerminalLoss::ClientDisconnected,
            );
            return Err(terminal_control_error(&error));
        }
        drop(engine);
        Ok(Box::new(RunscInteractiveControlSession {
            engine: Arc::clone(&self.engine),
            prompt_transport: Arc::clone(&self.prompt_transport),
            generation_id: self.generation_id,
            session_id: request.session_id,
        }))
    }
}

struct RunscInteractiveControlSession {
    engine: Arc<Mutex<SessionTerminalEngine>>,
    prompt_transport: Arc<SessionPromptTransport>,
    generation_id: GenerationId,
    session_id: SessionId,
}

impl InteractiveControlSession for RunscInteractiveControlSession {
    fn exchange(
        &mut self,
        event: InteractiveSessionEvent,
    ) -> Result<InteractiveSessionBatch, ControlError> {
        let mut engine = self.engine.lock().map_err(|_| ControlError::LockPoisoned)?;
        let mut messages = match event {
            InteractiveSessionEvent::Terminal(SessionTerminalEvent::Resize(resize)) => {
                if self
                    .prompt_transport
                    .retain_prompt_resize(resize)
                    .map_err(|error| terminal_control_error(&error))?
                {
                    Vec::new()
                } else {
                    engine
                        .handle(self.generation_id, SessionTerminalEvent::Resize(resize))
                        .map_err(|error| terminal_control_error(&error))?
                }
            }
            InteractiveSessionEvent::Terminal(event) => engine
                .handle(self.generation_id, event)
                .map_err(|error| terminal_control_error(&error))?,
            InteractiveSessionEvent::Poll { session_id } if session_id == self.session_id => {
                Vec::new()
            }
            InteractiveSessionEvent::PromptDecision {
                session_id,
                signed_decision,
            } if session_id == self.session_id => {
                let decision = self
                    .prompt_transport
                    .route_decision(session_id, signed_decision)
                    .map_err(|error| terminal_control_error(&error))?;
                if decision.decision.decision == lazaret_core::records::DeveloperDecisionKind::Deny
                {
                    self.prompt_transport
                        .restore_sandbox_route(session_id)
                        .map_err(|error| terminal_control_error(&error))?;
                    engine
                        .set_route(session_id, self.generation_id, TerminalRoute::Sandbox)
                        .map_err(|error| terminal_control_error(&error))?;
                    if let Some(resize) = decision.resumed_size {
                        engine
                            .handle(self.generation_id, SessionTerminalEvent::Resize(resize))
                            .map_err(|error| terminal_control_error(&error))?;
                    }
                }
                self.prompt_transport
                    .publish_decision(session_id, decision)
                    .map_err(|error| terminal_control_error(&error))?;
                Vec::new()
            }
            InteractiveSessionEvent::PromptDecision { .. }
            | InteractiveSessionEvent::Poll { .. } => return Err(ControlError::InvalidRequest),
        };
        if let Some(prompt) = self
            .prompt_transport
            .prompt_for(self.session_id)
            .map_err(|error| terminal_control_error(&error))?
        {
            engine
                .set_route(self.session_id, self.generation_id, TerminalRoute::Prompt)
                .map_err(|error| terminal_control_error(&error))?;
            return Ok(InteractiveSessionBatch {
                session_id: self.session_id,
                messages: vec![InteractiveSessionMessage::Prompt(prompt)],
            });
        }
        if !messages
            .iter()
            .any(|message| matches!(message, SessionTerminalMessage::Exit(_)))
        {
            messages.extend(
                engine
                    .poll(self.session_id, self.generation_id)
                    .map_err(|error| terminal_control_error(&error))?,
            );
        }
        Ok(InteractiveSessionBatch {
            session_id: self.session_id,
            messages: messages
                .into_iter()
                .map(InteractiveSessionMessage::Terminal)
                .collect(),
        })
    }
}

impl Drop for RunscInteractiveControlSession {
    fn drop(&mut self) {
        if let Ok(mut engine) = self.engine.lock()
            && engine.contains(self.session_id)
        {
            let _ = engine.lose(
                self.session_id,
                self.generation_id,
                TerminalLoss::ClientDisconnected,
            );
        }
    }
}

fn terminal_control_error(error: &TerminalError) -> ControlError {
    ControlError::Lifecycle(error.to_string())
}

fn hex_session(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
    })
}

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);
        }
        if self.recorder.require_authority(request)? != TerminalRoute::Sandbox {
            return Err(TerminalError::AuthorityMismatch);
        }
        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,
                request: request.clone(),
                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 authoritative = self.refresh_route(session_id, generation_id)?;
        if authoritative != route {
            return Err(TerminalError::InactiveRoute);
        }
        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,
                GuestReleaseAuthority::Validate,
            );
        }
        let authoritative = self.refresh_route(session_id, generation_id)?;
        if authoritative != TerminalRoute::Sandbox {
            return Err(TerminalError::InactiveRoute);
        }
        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 authoritative = self.refresh_route(session_id, generation_id)?;
        let terminal = self.active_mut(session_id, generation_id)?;
        if terminal.process.exec_exited()? {
            return self.close(
                session_id,
                generation_id,
                None,
                true,
                GuestReleaseAuthority::Validate,
            );
        }
        if authoritative != 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,
                GuestReleaseAuthority::Validate,
            ),
            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,
                GuestReleaseAuthority::Validate,
            ),
        }
    }

    /// # 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,
            GuestReleaseAuthority::Validate,
        )
    }

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

    fn refresh_route(
        &mut self,
        session_id: SessionId,
        generation_id: GenerationId,
    ) -> Result<TerminalRoute, TerminalError> {
        let active_generation = self
            .active
            .get(&session_id)
            .ok_or(TerminalError::UnknownSession)?
            .generation_id;
        if active_generation != generation_id {
            let cleanup = self.close(
                session_id,
                active_generation,
                Some(TerminalLoss::ExecClientLost),
                false,
                GuestReleaseAuthority::HostOnly,
            );
            return Err(with_cleanup(TerminalError::StaleGeneration, cleanup));
        }
        let request = {
            let terminal = self.active_mut(session_id, generation_id)?;
            terminal.request.clone()
        };
        match self.recorder.require_authority(&request) {
            Ok(TerminalRoute::Detached) => {
                let cleanup = self.close(
                    session_id,
                    generation_id,
                    Some(TerminalLoss::ExecClientLost),
                    false,
                    GuestReleaseAuthority::HostOnly,
                );
                Err(with_cleanup(TerminalError::AuthorityMismatch, cleanup))
            }
            Err(error) => {
                let cleanup = self.close(
                    session_id,
                    generation_id,
                    Some(TerminalLoss::ExecClientLost),
                    false,
                    GuestReleaseAuthority::HostOnly,
                );
                Err(with_cleanup(error, cleanup))
            }
            Ok(route) => {
                self.active_mut(session_id, generation_id)?.route = route;
                Ok(route)
            }
        }
    }

    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,
        authority: GuestReleaseAuthority,
    ) -> Result<Vec<SessionTerminalMessage>, TerminalError> {
        let active_generation = self
            .active
            .get(&session_id)
            .ok_or(TerminalError::UnknownSession)?
            .generation_id;
        if active_generation != generation_id {
            let cleanup = self.close(
                session_id,
                active_generation,
                loss,
                false,
                GuestReleaseAuthority::HostOnly,
            );
            return Err(with_cleanup(TerminalError::StaleGeneration, cleanup));
        }
        let mut terminal = self
            .active
            .remove(&session_id)
            .ok_or(TerminalError::UnknownSession)?;
        let was_serviceable = matches!(authority, GuestReleaseAuthority::Validate)
            && (matches!(
                self.recorder.require_authority(&terminal.request),
                Ok(TerminalRoute::Sandbox | TerminalRoute::Prompt | TerminalRoute::HostProcess)
            ) || (terminal.record.state() == SessionTerminalState::ReleaseIntent
                && self
                    .recorder
                    .require_guest_signal_authority(&terminal.request, &terminal.record)
                    .is_ok()));
        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 authorize_guest = || {
                was_serviceable
                    && self
                        .recorder
                        .require_guest_signal_authority(&terminal.request, &terminal.record)
                        .is_ok()
            };
            let (mut output, exit, guest_absent) = release_physical(
                terminal.process.as_mut(),
                session_id,
                connected,
                &mut authorize_guest,
            )?;
            if guest_absent {
                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 release_physical(
    process: &mut dyn SessionTerminalProcess,
    session_id: SessionId,
    connected: bool,
    authorize_guest: &mut dyn FnMut() -> bool,
) -> Result<(Vec<SessionTerminalMessage>, ProcessExit, bool), TerminalError> {
    let mut failure = None;
    let mut guest_authorized = authorize_guest();
    if guest_authorized {
        for signal in ["HUP", "TERM", "KILL"] {
            if !authorize_guest() {
                guest_authorized = false;
                break;
            }
            match process.group_absent() {
                Ok(true) => break,
                Ok(false) => {}
                Err(error) => {
                    failure.get_or_insert(error);
                    break;
                }
            }
            if !authorize_guest() {
                guest_authorized = false;
                break;
            }
            if let Err(error) = process.signal_group(signal) {
                failure.get_or_insert(error);
                break;
            }
            match wait_for_group_absence(process, authorize_guest) {
                Ok(GuestWaitOutcome::Absent) => break,
                Ok(GuestWaitOutcome::TimedOut) => {}
                Ok(GuestWaitOutcome::Revoked) => {
                    guest_authorized = false;
                    break;
                }
                Err(error) => {
                    failure.get_or_insert(error);
                    guest_authorized = false;
                    break;
                }
            }
        }
    }
    match process.exec_exited() {
        Ok(false) => {
            if let Err(error) = process.terminate_exec() {
                failure.get_or_insert(error.into());
            }
        }
        Ok(true) => {}
        Err(error) => {
            failure.get_or_insert(error.into());
        }
    }
    let exit = match process.reap_exec() {
        Ok(exit) => Some(exit),
        Err(error) => {
            failure.get_or_insert(error.into());
            None
        }
    };
    let output = drain_output(process, session_id, connected);
    process.close_pty();
    if guest_authorized && authorize_guest() {
        match process.group_absent() {
            Ok(true) if authorize_guest() => {}
            Ok(true) => guest_authorized = false,
            Ok(false) => {
                failure.get_or_insert(TerminalError::GuestGroupStillPresent);
            }
            Err(error) => {
                failure.get_or_insert(error);
            }
        }
    } else {
        guest_authorized = false;
    }
    if let Some(error) = failure {
        return Err(error);
    }
    Ok((
        output,
        exit.ok_or_else(|| io::Error::other("exec client exit is unavailable"))?,
        guest_authorized,
    ))
}

fn drain_output(
    process: &mut dyn SessionTerminalProcess,
    session_id: SessionId,
    connected: bool,
) -> Vec<SessionTerminalMessage> {
    let mut output = Vec::new();
    if !connected {
        return output;
    }
    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 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,
        }
    }
    output
}

fn with_cleanup<T>(original: TerminalError, cleanup: Result<T, TerminalError>) -> TerminalError {
    match cleanup {
        Ok(_) => original,
        Err(cleanup) => TerminalError::AcquisitionCleanup {
            original: Box::new(original),
            cleanup: Some(Box::new(cleanup)),
        },
    }
}

fn wait_for_group_absence(
    process: &dyn SessionTerminalProcess,
    authorize_guest: &mut dyn FnMut() -> bool,
) -> Result<GuestWaitOutcome, TerminalError> {
    let deadline = Instant::now() + SIGNAL_GRACE;
    while Instant::now() < deadline {
        if !authorize_guest() {
            return Ok(GuestWaitOutcome::Revoked);
        }
        if process.group_absent()? {
            return Ok(GuestWaitOutcome::Absent);
        }
        std::thread::sleep(POLL_INTERVAL);
    }
    Ok(GuestWaitOutcome::TimedOut)
}

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 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("-env")
            .argument(format!(
                "LAZARET_SESSION_ID={}",
                hex_lower(request.session_id.as_bytes())
            ))
            .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),
        )?;
        let mut terminal_process = RunscSessionTerminal {
            request: request.clone(),
            master: Some(master),
            exec: Some(process),
            exit: None,
            guest,
        };
        let exec_client = terminal_process
            .exec
            .as_ref()
            .expect("new terminal retains its exec client")
            .identity();
        commit_running_or_cleanup(
            recorder,
            request,
            running.clone(),
            pty,
            exec_client,
            guest,
            &mut terminal_process,
        )?;
        Ok((running, Box::new(terminal_process)))
    }
}

fn commit_running_or_cleanup(
    recorder: &HostCoordinatorTerminalRecorder,
    request: &TerminalAcquisition,
    running: SessionTerminal,
    pty: PtyIdentity,
    exec_client: PidfsProcessIdentityV1,
    guest: GuestProcessIdentity,
    process: &mut dyn SessionTerminalProcess,
) -> Result<(), TerminalError> {
    if let Err(original) = recorder.put(running) {
        let cleanup =
            cleanup_failed_running_commit(recorder, request, pty, exec_client, guest, process)
                .err()
                .map(Box::new);
        return Err(TerminalError::AcquisitionCleanup {
            original: Box::new(original),
            cleanup,
        });
    }
    Ok(())
}

fn cleanup_failed_running_commit(
    recorder: &HostCoordinatorTerminalRecorder,
    request: &TerminalAcquisition,
    pty: PtyIdentity,
    exec_client: PidfsProcessIdentityV1,
    guest: GuestProcessIdentity,
    process: &mut dyn SessionTerminalProcess,
) -> Result<(), TerminalError> {
    let observed = terminal_snapshot(
        request,
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::committed(request.host_coordinator),
        Some(pty),
        PersistedHostProcessOwnership::committed(exec_client),
        Some(guest),
    )?;
    let release = copy_terminal_state(&observed, SessionTerminalState::ReleaseIntent)?;
    let durable = recorder
        .put(observed)
        .and_then(|()| recorder.put(release.clone()));
    let mut authorize_guest = || {
        recorder
            .require_guest_signal_authority(request, &release)
            .is_ok()
    };
    let physical = release_physical(process, request.session_id, false, &mut authorize_guest)
        .map(|(_, _, guest_absent)| guest_absent);
    match (durable, physical) {
        (Err(durable), Err(physical)) => {
            return Err(TerminalError::AcquisitionCleanup {
                original: Box::new(durable),
                cleanup: Some(Box::new(physical)),
            });
        }
        (Err(error), Ok(_)) | (Ok(()), Err(error)) => return Err(error),
        (Ok(()), Ok(false)) => return Ok(()),
        (Ok(()), Ok(true)) => {}
    }
    recorder.clean(clean_terminal(request.generation_id, request.session_id)?)
}

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>,
    exit: Option<ProcessExit>,
    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()
            .map_or(Ok(true), RetainedProcess::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> {
        if observe_guest_identity(&self.request, self.guest.leader_pid.get())?
            .is_some_and(|observed| observed != self.guest)
        {
            return Err(TerminalError::GuestIdentityMismatch);
        }
        observe_group_absent(&self.request, self.guest.process_group_id.get())
    }

    fn terminate_exec(&self) -> io::Result<()> {
        match self.exec.as_ref() {
            Some(exec) => exec.send_signal(rustix::process::Signal::KILL),
            None => Ok(()),
        }
    }

    fn reap_exec(&mut self) -> io::Result<ProcessExit> {
        let exit = if let Some(exit) = self.exit {
            exit
        } else {
            let exit = child_broker().reap_status(
                self.exec
                    .take()
                    .ok_or_else(|| io::Error::other("exec client exit is unavailable"))?,
            )?;
            self.exit = Some(exit);
            exit
        };
        remove_pid_file(&self.request.internal_pid_file)?;
        Ok(exit)
    }

    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 stdout = run_guest_identity_query(request, &[OsString::from(pid.to_string())])?;
    if stdout == b"absent\n" {
        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 observe_group_absent(request: &TerminalAcquisition, group: u32) -> Result<bool, TerminalError> {
    let stdout = run_guest_identity_query(
        request,
        &[OsString::from("--group"), OsString::from(group.to_string())],
    )?;
    match stdout.as_slice() {
        b"absent\n" => Ok(true),
        b"present\n" => Ok(false),
        _ => Err(io::Error::other("guest group response is malformed").into()),
    }
}

fn run_guest_identity_query(
    request: &TerminalAcquisition,
    arguments: &[OsString],
) -> Result<Vec<u8>, 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);
    for argument in arguments {
        target = target.argument(argument.clone());
    }
    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() {
        let stderr = String::from_utf8_lossy(&stderr);
        return Err(io::Error::other(format!("guest identity query failed: {stderr}")).into());
    }
    Ok(stdout)
}

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;