use std::collections::BTreeMap;
use std::ffi::{OsStr, OsString};
use std::fmt;
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::os::fd::{AsFd, AsRawFd};
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};

use ed25519_dalek::SigningKey;
use lazaret_core::codec::{CodecError, decode, encode};
use lazaret_core::configuration::ConfinedPathPlan;
use lazaret_core::identity::{
    CertificateDigest, ClientId, ContainerId, SessionAnchorId, SessionId, TrustManifestDigest,
};
use lazaret_core::protocol::{
    ProtocolValidationError, issue_protocol_credential, issue_session_capability,
    issue_trust_manifest, verify_tls_credential,
};
use lazaret_core::records::{
    ClientEnrollmentRequest, ClientEnrollmentResponse, ControlRequest, ControlResponse,
    DurableDecision, GenerationIdentity, NonInteractiveSessionMessage,
    NonInteractiveSessionRequest, ProtocolCredential, ProtocolCredentialRole,
    ProtocolPublicationIdentity, ProtocolSubjectId, ProtocolVersion, SandboxCredentialBundle,
    Session, SessionBinding, SessionExit, SessionPhase, SessionRegistrationRequest,
    SessionRegistrationResponse, TlsCredential, TlsRole, TlsSubjectId, TrustManifest,
};
use rcgen::{
    BasicConstraints, CertificateParams, CertificateSigningRequestParams, CertifiedIssuer,
    DistinguishedName, DnType, DnValue, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose,
    PKCS_ED25519,
};
use rustix::fs::{Mode, OFlags, fchmod, openat};
use rustix::net::sockopt::socket_peercred;
use rustls::pki_types::{
    CertificateDer, CertificateSigningRequestDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName,
};
use rustls::server::{NoServerSessionStorage, WebPkiClientVerifier};
use rustls::{ClientConfig, ClientConnection, RootCertStore, ServerConfig, ServerConnection};
use sha2::{Digest, Sha256};
use time::{Duration, OffsetDateTime};

use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
use crate::lifecycle::{
    HostCoordinatorLifecycle, LifecycleAction, SessionAdmissionIdentity, SessionLifecycleAdapter,
};
use lazaret_linux::process::{ProcessExit, TargetCommand, child_broker};
use lazaret_runtime_supervisor::sandbox::RunscCommand;

pub const CONTROL_ALPN: &[u8] = b"lazaret-control/1";
pub const CONTROL_SERVER_NAME: &str = "coordinator.lazaret.invalid";
pub const MAX_ENROLLMENT_BYTES: usize = 8_192;
const MAX_CONTROL_BYTES: usize = 65_536;
const CONTROL_SOCKET_NAME: &str = "c.sock";
const PUBLICATION_IDENTITY_NAME: &str = "protocol-identity.postcard";
const STAGED_CONTROL_SOCKET_NAME: &str = ".c.staged";
const STAGED_PUBLICATION_IDENTITY_NAME: &str = ".protocol-identity.staged";

#[derive(Debug)]
pub enum ControlError {
    Io(io::Error),
    Codec(CodecError),
    Certificate(rcgen::Error),
    Tls(rustls::Error),
    Protocol(ProtocolValidationError),
    DomainStore(DomainStoreError),
    InvalidPath(&'static str),
    InvalidPublication(&'static str),
    UnauthorizedPeer,
    InvalidCertificateProfile,
    StaleGeneration,
    DuplicateIdentity,
    InvalidRequest,
    OversizeRequest { length: usize, maximum: usize },
    LockPoisoned,
    Lifecycle(String),
}

impl fmt::Display for ControlError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => write!(formatter, "control I/O failed: {error}"),
            Self::Codec(error) => write!(formatter, "control record failed validation: {error}"),
            Self::Certificate(error) => write!(formatter, "certificate operation failed: {error}"),
            Self::Tls(error) => write!(formatter, "TLS failed: {error}"),
            Self::Protocol(error) => write!(formatter, "protocol validation failed: {error}"),
            Self::DomainStore(error) => write!(formatter, "session persistence failed: {error}"),
            Self::InvalidPath(reason) => write!(formatter, "invalid control path: {reason}"),
            Self::InvalidPublication(reason) => {
                write!(formatter, "invalid control publication: {reason}")
            }
            Self::UnauthorizedPeer => formatter.write_str("control peer UID is not authorized"),
            Self::InvalidCertificateProfile => {
                formatter.write_str("TLS certificate profile is invalid")
            }
            Self::StaleGeneration => formatter.write_str("control generation is stale"),
            Self::DuplicateIdentity => formatter.write_str("session identity is already active"),
            Self::InvalidRequest => {
                formatter.write_str("operation is not allowed on this connection")
            }
            Self::OversizeRequest { length, maximum } => {
                write!(
                    formatter,
                    "control request length {length} exceeds {maximum}"
                )
            }
            Self::LockPoisoned => formatter.write_str("control registry lock is poisoned"),
            Self::Lifecycle(detail) => write!(formatter, "control lifecycle failed: {detail}"),
        }
    }
}

impl std::error::Error for ControlError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Codec(error) => Some(error),
            Self::Certificate(error) => Some(error),
            Self::Tls(error) => Some(error),
            Self::Protocol(error) => Some(error),
            Self::DomainStore(error) => Some(error),
            _ => None,
        }
    }
}

macro_rules! from_error {
    ($source:ty, $variant:ident) => {
        impl From<$source> for ControlError {
            fn from(error: $source) -> Self {
                Self::$variant(error)
            }
        }
    };
}

from_error!(io::Error, Io);
from_error!(CodecError, Codec);
from_error!(rcgen::Error, Certificate);
from_error!(rustls::Error, Tls);
from_error!(ProtocolValidationError, Protocol);
from_error!(DomainStoreError, DomainStore);

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct FileIdentity {
    device: u64,
    inode: u64,
}

impl FileIdentity {
    fn from_metadata(metadata: &std::fs::Metadata) -> Self {
        Self {
            device: metadata.dev(),
            inode: metadata.ino(),
        }
    }
}

pub struct PendingClientEnrollment {
    request: ClientEnrollmentRequest,
    private_key: PrivatePkcs8KeyDer<'static>,
}

impl PendingClientEnrollment {
    /// Creates an Ed25519 CSR whose signed subject binds the exact generation and Client CLI ID.
    ///
    /// # Errors
    /// Returns an error when key generation, encoding, or CSR signing fails.
    pub fn create(
        generation_identity: GenerationIdentity,
        cli_id: ClientId,
    ) -> Result<Self, ControlError> {
        let binding = enrollment_binding(&generation_identity, cli_id)?;
        let mut params = CertificateParams::default();
        params.distinguished_name = DistinguishedName::new();
        params.distinguished_name.push(DnType::CommonName, binding);
        params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
        let key = KeyPair::generate_for(&PKCS_ED25519)?;
        let csr = params.serialize_request(&key)?;
        Ok(Self {
            request: ClientEnrollmentRequest {
                protocol_version: ProtocolVersion::V1,
                generation_identity,
                cli_id,
                certificate_signing_request_der: csr.der().to_vec(),
            },
            private_key: PrivatePkcs8KeyDer::from(key.serialize_der()),
        })
    }

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

    #[must_use]
    pub fn private_key(&self) -> PrivateKeyDer<'static> {
        PrivateKeyDer::Pkcs8(self.private_key.clone_key())
    }
}

pub struct TrustAnchor {
    generation_identity: GenerationIdentity,
    tls_ca: CertifiedIssuer<'static, KeyPair>,
    tls_ca_der: CertificateDer<'static>,
    coordinator_tls_credential: TlsCredential,
    coordinator_tls_private_key: PrivatePkcs8KeyDer<'static>,
    trust_manifest: TrustManifest,
    trust_manifest_digest: TrustManifestDigest,
    coordinator_protocol_credential: ProtocolCredential,
    sandbox_protocol_credential: ProtocolCredential,
    coordinator_protocol_key: SigningKey,
    sandbox_protocol_key: Option<SigningKey>,
}

impl TrustAnchor {
    #[must_use]
    pub const fn generation_identity(&self) -> &GenerationIdentity {
        &self.generation_identity
    }

    /// Creates one ephemeral generation Trust Anchor with role-separated Ed25519 keys.
    ///
    /// # Errors
    /// Returns an error when key generation, certificate issuance, or protocol signing fails.
    pub fn create(generation_identity: GenerationIdentity) -> Result<Self, ControlError> {
        let now = OffsetDateTime::now_utc();
        let mut ca_params = CertificateParams::default();
        ca_params.distinguished_name = DistinguishedName::new();
        ca_params
            .distinguished_name
            .push(DnType::CommonName, "Lazaret ephemeral TLS CA");
        ca_params.not_before = now - Duration::minutes(1);
        ca_params.not_after = now + Duration::hours(24);
        ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
        ca_params.key_usages = vec![
            KeyUsagePurpose::DigitalSignature,
            KeyUsagePurpose::KeyCertSign,
            KeyUsagePurpose::CrlSign,
        ];
        let ca_key = KeyPair::generate_for(&PKCS_ED25519)?;
        let tls_ca = CertifiedIssuer::self_signed(ca_params, ca_key)?;
        let tls_ca_der = tls_ca.der().clone();
        let tls_ca_digest = certificate_digest(&tls_ca_der);

        let protocol_issuer = random_signing_key()?;
        let (trust_manifest, trust_manifest_digest) =
            issue_trust_manifest(generation_identity.clone(), tls_ca_digest, &protocol_issuer)?;
        let coordinator_protocol_key = random_signing_key()?;
        let sandbox_protocol_key = random_signing_key()?;
        let coordinator_protocol_credential = issue_protocol_credential(
            generation_identity.clone(),
            ProtocolCredentialRole::SessionCoordinator,
            ProtocolSubjectId::SessionCoordinator(generation_identity.coordinator_id),
            &coordinator_protocol_key.verifying_key(),
            &protocol_issuer,
        )?;
        let sandbox_protocol_credential = issue_protocol_credential(
            generation_identity.clone(),
            ProtocolCredentialRole::AgentSandbox,
            ProtocolSubjectId::AgentSandbox(generation_identity.sandbox_id),
            &sandbox_protocol_key.verifying_key(),
            &protocol_issuer,
        )?;

        let coordinator_key = KeyPair::generate_for(&PKCS_ED25519)?;
        let coordinator_certificate = leaf_parameters(
            CONTROL_SERVER_NAME,
            ExtendedKeyUsagePurpose::ServerAuth,
            now,
        )?
        .signed_by(&coordinator_key, &tls_ca)?;
        let coordinator_certificate_der = coordinator_certificate.der().to_vec();
        let coordinator_tls_credential = TlsCredential::session_coordinator(
            generation_identity.clone(),
            generation_identity.coordinator_id,
            coordinator_certificate_der.clone(),
            certificate_digest(coordinator_certificate.der()),
        );

        Ok(Self {
            generation_identity,
            tls_ca,
            tls_ca_der,
            coordinator_tls_credential,
            coordinator_tls_private_key: PrivatePkcs8KeyDer::from(coordinator_key.serialize_der()),
            trust_manifest,
            trust_manifest_digest,
            coordinator_protocol_credential,
            sandbox_protocol_credential,
            coordinator_protocol_key,
            sandbox_protocol_key: Some(sandbox_protocol_key),
        })
    }

    #[must_use]
    pub fn publication_identity(&self) -> ProtocolPublicationIdentity {
        ProtocolPublicationIdentity {
            protocol_version: ProtocolVersion::V1,
            generation_identity: self.generation_identity.clone(),
            tls_ca_certificate_der: self.tls_ca_der.to_vec(),
            trust_manifest_sha256: self.trust_manifest_digest,
            coordinator_tls_certificate_sha256: self.coordinator_tls_credential.certificate_sha256,
        }
    }

    #[must_use]
    pub fn coordinator_protocol_key(&self) -> SigningKey {
        self.coordinator_protocol_key.clone()
    }

    /// Transfers the sandbox credential package once without exposing coordinator or issuer keys.
    ///
    /// # Errors
    /// Returns an error after the sandbox private key has already been transferred.
    pub fn take_sandbox_credential_bundle(
        &mut self,
    ) -> Result<SandboxCredentialBundle, ControlError> {
        let private_key = self
            .sandbox_protocol_key
            .take()
            .ok_or(ControlError::InvalidRequest)?
            .to_bytes()
            .to_vec();
        SandboxCredentialBundle::new(
            self.generation_identity.clone(),
            self.trust_manifest.clone(),
            self.coordinator_protocol_credential.clone(),
            self.sandbox_protocol_credential.clone(),
            private_key,
        )
        .map_err(|_| ControlError::InvalidCertificateProfile)
    }

    fn issue_client_credential(
        &self,
        request: &ClientEnrollmentRequest,
    ) -> Result<TlsCredential, ControlError> {
        if request.protocol_version != ProtocolVersion::V1
            || request.generation_identity != self.generation_identity
        {
            return Err(ControlError::StaleGeneration);
        }
        let csr_der =
            CertificateSigningRequestDer::from(request.certificate_signing_request_der.as_slice());
        let csr = CertificateSigningRequestParams::from_der(&csr_der)?;
        if csr.public_key.algorithm() != &PKCS_ED25519
            || csr.params.distinguished_name.get(&DnType::CommonName)
                != Some(&DnValue::Utf8String(enrollment_binding(
                    &request.generation_identity,
                    request.cli_id,
                )?))
        {
            return Err(ControlError::InvalidCertificateProfile);
        }
        let now = OffsetDateTime::now_utc();
        let params = leaf_parameters(
            &format!("client-{}", short_identity(request.cli_id.as_bytes())),
            ExtendedKeyUsagePurpose::ClientAuth,
            now,
        )?;
        let constrained = CertificateSigningRequestParams {
            params,
            public_key: csr.public_key,
        };
        let certificate = constrained.signed_by(&self.tls_ca)?;
        let certificate_der = certificate.der().to_vec();
        Ok(TlsCredential::client_cli(
            self.generation_identity.clone(),
            request.cli_id,
            certificate_der,
            certificate_digest(certificate.der()),
        ))
    }

    fn server_config(&self) -> Result<ServerConfig, ControlError> {
        let mut roots = RootCertStore::empty();
        roots.add(self.tls_ca_der.clone())?;
        let verifier = WebPkiClientVerifier::builder(Arc::new(roots))
            .allow_unauthenticated()
            .build()
            .map_err(|_| ControlError::InvalidCertificateProfile)?;
        let mut config = ServerConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
            .with_client_cert_verifier(verifier)
            .with_single_cert(
                vec![CertificateDer::from(
                    self.coordinator_tls_credential.certificate_der.clone(),
                )],
                PrivateKeyDer::Pkcs8(self.coordinator_tls_private_key.clone_key()),
            )?;
        config.alpn_protocols = vec![CONTROL_ALPN.to_vec()];
        config.max_early_data_size = 0;
        config.send_tls13_tickets = 0;
        config.session_storage = Arc::new(NoServerSessionStorage {});
        Ok(config)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RegisteredSession {
    pub binding: SessionBinding,
    pub session_anchor_id: SessionAnchorId,
}

#[derive(Clone)]
pub struct ControlSessionAdmission {
    pub request: SessionRegistrationRequest,
    pub credential: TlsCredential,
}

pub type ControlHostCoordinatorLifecycle<'store, A> =
    HostCoordinatorLifecycle<'store, A, Arc<SessionRegistry>>;

#[derive(Default)]
struct RegistryState {
    enrolled: BTreeMap<CertificateDigest, TlsCredential>,
    connected: BTreeMap<SessionId, RegisteredSession>,
}

pub struct SessionRegistry {
    state: Mutex<RegistryState>,
    store: Arc<HostCoordinatorDomainStore>,
    generation_identity: GenerationIdentity,
}

impl SessionRegistry {
    #[must_use]
    pub fn new(
        store: Arc<HostCoordinatorDomainStore>,
        generation_identity: GenerationIdentity,
    ) -> Self {
        Self {
            state: Mutex::new(RegistryState::default()),
            store,
            generation_identity,
        }
    }

    /// Returns the exact set of connected, registered sessions.
    ///
    /// # Errors
    /// Returns an error if the registry lock is poisoned.
    pub fn connected_sessions(&self) -> Result<Vec<RegisteredSession>, ControlError> {
        Ok(self.lock()?.connected.values().cloned().collect())
    }

    fn lock(&self) -> Result<MutexGuard<'_, RegistryState>, ControlError> {
        self.state.lock().map_err(|_| ControlError::LockPoisoned)
    }

    fn enroll(&self, credential: TlsCredential) -> Result<(), ControlError> {
        self.lock()?
            .enrolled
            .insert(credential.certificate_sha256, credential);
        Ok(())
    }

    fn credential_for_leaf(&self, certificate: &[u8]) -> Result<TlsCredential, ControlError> {
        self.lock()?
            .enrolled
            .get(&certificate_digest(certificate))
            .cloned()
            .ok_or(ControlError::InvalidCertificateProfile)
    }

    fn register(
        &self,
        request: SessionRegistrationRequest,
        credential: &TlsCredential,
    ) -> Result<RegisteredSession, ControlError> {
        let binding = request.session_binding;
        if request.protocol_version != ProtocolVersion::V1
            || binding.generation_identity != self.generation_identity
        {
            return Err(ControlError::StaleGeneration);
        }
        let TlsSubjectId::ClientCli(certificate_cli_id) = credential.subject_id() else {
            return Err(ControlError::InvalidCertificateProfile);
        };
        if credential.role() != TlsRole::ClientCli || certificate_cli_id != binding.cli_id {
            return Err(ControlError::InvalidCertificateProfile);
        }
        verify_tls_credential(
            credential,
            &self.generation_identity,
            TlsRole::ClientCli,
            TlsSubjectId::ClientCli(binding.cli_id),
        )?;
        let anchor_bytes: [u8; 32] = binding
            .session_anchor_public_key
            .as_slice()
            .try_into()
            .map_err(|_| ControlError::InvalidRequest)?;
        ed25519_dalek::VerifyingKey::from_bytes(&anchor_bytes)
            .map_err(|_| ControlError::InvalidRequest)?;
        let session_anchor_id = SessionAnchorId::new(sha256(&anchor_bytes));

        self.require_serviceable_generation()?;
        let snapshot = self.store.snapshot()?;
        let state = self.lock()?;
        let duplicate_in_memory = state.connected.values().any(|registered| {
            registered.binding.session_id == binding.session_id
                || registered.binding.cli_id == binding.cli_id
                || registered.session_anchor_id == session_anchor_id
        });
        let resumable_waiter = snapshot.sessions.iter().any(|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 == session_anchor_id
                && session.session_phase == SessionPhase::Waiting
        });
        let duplicate_session = snapshot.sessions.iter().any(|session| {
            let same_session_allowed = resumable_waiter
                && session.session_id == binding.session_id
                && (session.session_phase == SessionPhase::Closed
                    || (session.generation_id == binding.generation_identity.generation_id
                        && session.cli_id == binding.cli_id
                        && session.session_anchor_id == session_anchor_id
                        && session.session_phase == SessionPhase::Waiting));
            !same_session_allowed
                && (session.session_id == binding.session_id
                    || (is_live_session(session.session_phase)
                        && (session.cli_id == binding.cli_id
                            || session.session_anchor_id == session_anchor_id)))
        });
        if duplicate_in_memory || duplicate_session {
            return Err(ControlError::DuplicateIdentity);
        }

        let registering = Session {
            generation_id: binding.generation_identity.generation_id,
            session_id: binding.session_id,
            cli_id: binding.cli_id,
            session_anchor_id,
            session_phase: SessionPhase::Registering,
        };
        self.store
            .apply(&[DomainStoreMutation::PutSession(registering.clone())])?;
        let mut attached = registering;
        attached.session_phase = SessionPhase::Attached;
        if let Err(error) = self
            .store
            .apply(&[DomainStoreMutation::PutSession(attached)])
        {
            let mut failed = Session {
                generation_id: binding.generation_identity.generation_id,
                session_id: binding.session_id,
                cli_id: binding.cli_id,
                session_anchor_id,
                session_phase: SessionPhase::Registering,
            };
            failed.session_phase = SessionPhase::Failed;
            let _ = self.store.apply(&[DomainStoreMutation::PutSession(failed)]);
            return Err(error.into());
        }
        let registered = RegisteredSession {
            binding,
            session_anchor_id,
        };
        drop(snapshot);
        drop(state);
        self.lock()?
            .connected
            .insert(registered.binding.session_id, registered.clone());
        Ok(registered)
    }

    fn disconnect(&self, session_id: SessionId) -> Result<(), ControlError> {
        let registered = self.lock()?.connected.remove(&session_id);
        let Some(registered) = registered else {
            return Ok(());
        };
        self.store
            .apply(&[DomainStoreMutation::PutSession(Session {
                generation_id: registered.binding.generation_identity.generation_id,
                session_id,
                cli_id: registered.binding.cli_id,
                session_anchor_id: registered.session_anchor_id,
                session_phase: SessionPhase::Closed,
            })])?;
        Ok(())
    }

    fn require_serviceable_generation(&self) -> Result<(), ControlError> {
        let decision = self.require_generation_authority()?;
        if decision != DurableDecision::Published {
            return Err(ControlError::StaleGeneration);
        }
        Ok(())
    }

    fn require_generation_authority(&self) -> Result<DurableDecision, ControlError> {
        let snapshot = self.store.snapshot()?;
        let generation = snapshot
            .generations
            .iter()
            .find(|generation| generation.generation_id == self.generation_identity.generation_id)
            .ok_or(ControlError::StaleGeneration)?;
        let is_latest = snapshot
            .generations
            .iter()
            .all(|candidate| candidate.generation_id <= self.generation_identity.generation_id);
        if !is_latest
            || snapshot.domain.domain_id != self.generation_identity.domain_id
            || snapshot.domain.profile_name != self.generation_identity.agent_profile
            || generation.domain_id != self.generation_identity.domain_id
            || generation.configuration_fingerprint
                != self.generation_identity.configuration_fingerprint
            || generation.coordinator_id != self.generation_identity.coordinator_id
            || generation.sandbox_id != self.generation_identity.sandbox_id
            || !matches!(
                generation.durable_decision,
                DurableDecision::Preparing | DurableDecision::Published
            )
        {
            return Err(ControlError::StaleGeneration);
        }
        Ok(generation.durable_decision)
    }
}

impl SessionLifecycleAdapter for Arc<SessionRegistry> {
    type Error = ControlError;
    type Admission = ControlSessionAdmission;

    fn admission_identity(&self, admission: &Self::Admission) -> SessionAdmissionIdentity {
        let binding = &admission.request.session_binding;
        SessionAdmissionIdentity {
            generation_id: binding.generation_identity.generation_id,
            session_id: binding.session_id,
            configuration_fingerprint: binding.generation_identity.configuration_fingerprint,
        }
    }

    fn register_session(
        &self,
        _store: &HostCoordinatorDomainStore,
        entry_action: LifecycleAction,
        admission: Self::Admission,
    ) -> Result<Session, Self::Error> {
        if !matches!(
            entry_action,
            LifecycleAction::StartReadySession | LifecycleAction::AdvanceWaitingSession
        ) {
            return Err(ControlError::InvalidRequest);
        }
        let registered = self.register(admission.request, &admission.credential)?;
        Ok(Session {
            generation_id: registered.binding.generation_identity.generation_id,
            session_id: registered.binding.session_id,
            cli_id: registered.binding.cli_id,
            session_anchor_id: registered.session_anchor_id,
            session_phase: SessionPhase::Attached,
        })
    }

    fn disconnect_session(
        &self,
        _store: &HostCoordinatorDomainStore,
        session: &Session,
    ) -> Result<(), Self::Error> {
        let mut state = self.lock()?;
        let registered = state.connected.get(&session.session_id);
        if let Some(registered) = registered
            && (registered.binding.generation_identity.generation_id != session.generation_id
                || registered.binding.cli_id != session.cli_id
                || registered.session_anchor_id != session.session_anchor_id)
        {
            return Err(ControlError::InvalidRequest);
        }
        if session.session_phase == SessionPhase::Attached && registered.is_none() {
            return Err(ControlError::InvalidRequest);
        }
        state.connected.remove(&session.session_id);
        drop(state);
        self.store
            .apply(&[DomainStoreMutation::PutSession(Session {
                session_phase: SessionPhase::Closed,
                ..session.clone()
            })])?;
        Ok(())
    }
}

struct ControlPaths {
    root: PathBuf,
    root_file: File,
    root_identity: FileIdentity,
    socket: PathBuf,
    identity: PathBuf,
    staged_socket: PathBuf,
    staged_identity: PathBuf,
}

impl ControlPaths {
    fn open(root: &Path, uid: u32) -> Result<Self, ControlError> {
        if !root.is_absolute() {
            return Err(ControlError::InvalidPath(
                "runtime Domain root is not absolute",
            ));
        }
        let canonical = std::fs::canonicalize(root)?;
        if canonical != root {
            return Err(ControlError::InvalidPath(
                "runtime Domain root contains a symlink",
            ));
        }
        let metadata = std::fs::symlink_metadata(root)?;
        if !metadata.file_type().is_dir()
            || metadata.uid() != uid
            || metadata.mode() & 0o7777 != 0o700
        {
            return Err(ControlError::InvalidPath(
                "runtime Domain root is not invoking-user-owned mode 0700",
            ));
        }
        let root_file = OpenOptions::new().read(true).open(root)?;
        let root_identity = FileIdentity::from_metadata(&root_file.metadata()?);
        Ok(Self {
            root: root.to_path_buf(),
            root_file,
            root_identity,
            socket: root.join(CONTROL_SOCKET_NAME),
            identity: root.join(PUBLICATION_IDENTITY_NAME),
            staged_socket: root.join(STAGED_CONTROL_SOCKET_NAME),
            staged_identity: root.join(STAGED_PUBLICATION_IDENTITY_NAME),
        })
    }

    fn validate_root(&self, uid: u32) -> Result<(), ControlError> {
        let descriptor = self.root_file.metadata()?;
        let current = std::fs::symlink_metadata(&self.root)?;
        if FileIdentity::from_metadata(&descriptor) != self.root_identity
            || FileIdentity::from_metadata(&current) != self.root_identity
            || !current.file_type().is_dir()
            || current.uid() != uid
            || current.mode() & 0o7777 != 0o700
        {
            return Err(ControlError::InvalidPath("runtime Domain root changed"));
        }
        Ok(())
    }

    fn create_named_identity(
        &self,
        name: &str,
        bytes: &[u8],
        uid: u32,
    ) -> Result<(File, FileIdentity), ControlError> {
        let fd = openat(
            self.root_file.as_fd(),
            name,
            OFlags::CREATE | OFlags::EXCL | OFlags::RDWR | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::RUSR | Mode::WUSR,
        )
        .map_err(io::Error::from)?;
        let mut file = File::from(fd);
        let identity = FileIdentity::from_metadata(&file.metadata()?);
        let result = (|| {
            fchmod(file.as_fd(), Mode::RUSR | Mode::WUSR).map_err(io::Error::from)?;
            file.write_all(bytes)
                .map_err(|_| ControlError::InvalidPublication("identity write failed"))?;
            file.sync_all()
                .map_err(|_| ControlError::InvalidPublication("identity sync failed"))?;
            self.root_file
                .sync_all()
                .map_err(|_| ControlError::InvalidPublication("directory sync failed"))?;
            self.validate_identity_at(&self.root.join(name), &file, identity, uid, bytes.len())
        })();
        if let Err(error) = result {
            remove_if_identity_matches(&self.root.join(name), identity);
            return Err(error);
        }
        Ok((file, identity))
    }

    fn validate_identity(
        &self,
        file: &File,
        expected: FileIdentity,
        uid: u32,
        expected_length: usize,
    ) -> Result<(), ControlError> {
        self.validate_identity_at(&self.identity, file, expected, uid, expected_length)
    }

    fn validate_identity_at(
        &self,
        path: &Path,
        file: &File,
        expected: FileIdentity,
        uid: u32,
        expected_length: usize,
    ) -> Result<(), ControlError> {
        self.validate_root(uid)?;
        let descriptor = file.metadata()?;
        let current = std::fs::symlink_metadata(path)?;
        if FileIdentity::from_metadata(&descriptor) != expected
            || FileIdentity::from_metadata(&current) != expected
            || !descriptor.file_type().is_file()
            || !current.file_type().is_file()
            || descriptor.uid() != uid
            || current.uid() != uid
            || descriptor.mode() & 0o7777 != 0o600
            || current.mode() & 0o7777 != 0o600
            || usize::try_from(descriptor.len()).ok() != Some(expected_length)
            || current.len() != descriptor.len()
        {
            return Err(ControlError::InvalidPublication(
                "publication identity changed type, ownership, mode, or content length",
            ));
        }
        Ok(())
    }
}

pub struct PublishedControlFace {
    listener: std::os::unix::net::UnixListener,
    paths: ControlPaths,
    socket_identity: FileIdentity,
    identity_file: File,
    identity_file_identity: FileIdentity,
    publication_identity: ProtocolPublicationIdentity,
    publication_bytes: Vec<u8>,
    uid: u32,
    anchor: TrustAnchor,
    server_config: Arc<ServerConfig>,
    registry: Arc<SessionRegistry>,
    executor: Arc<dyn NonInteractiveSessionExecutor>,
}

pub trait NonInteractiveSessionExecutor: Send + Sync {
    /// # Errors
    /// Returns a runtime, stream, or exact-session execution failure.
    fn execute(
        &self,
        request: &NonInteractiveSessionRequest,
        emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
    ) -> Result<SessionExit, ControlError>;
}

struct ExecutionUnavailable;

impl NonInteractiveSessionExecutor for ExecutionUnavailable {
    fn execute(
        &self,
        _request: &NonInteractiveSessionRequest,
        _emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
    ) -> Result<SessionExit, ControlError> {
        Err(ControlError::InvalidRequest)
    }
}

pub struct RunscNonInteractiveExecutor {
    program: OsString,
    global_arguments: Vec<OsString>,
    container: OsString,
    runner: Arc<dyn NonInteractiveCommandRunner>,
}

trait NonInteractiveCommandRunner: Send + Sync {
    fn run(
        &self,
        target: &TargetCommand,
        session_id: SessionId,
        emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
    ) -> Result<ProcessExit, ControlError>;
}

struct ChildBrokerCommandRunner;

impl NonInteractiveCommandRunner for ChildBrokerCommandRunner {
    fn run(
        &self,
        target: &TargetCommand,
        session_id: SessionId,
        emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
    ) -> Result<ProcessExit, ControlError> {
        let mut child = child_broker().run_transient_captured(target)?;
        let mut stdout_done = false;
        let mut stderr_done = false;
        while !(stdout_done && stderr_done && child.process().has_exited()?) {
            let (stdout, stderr) = child.streams();
            stdout_done |= emit_available(stdout, session_id, true, emit)?;
            stderr_done |= emit_available(stderr, session_id, false, emit)?;
            if !(stdout_done && stderr_done) {
                std::thread::yield_now();
            }
        }
        child_broker()
            .reap_status(child.into_process())
            .map_err(Into::into)
    }
}

impl RunscNonInteractiveExecutor {
    /// Derives the exact `runsc exec` prefix from the generation's attached `runsc run` command.
    ///
    /// # Errors
    /// Refuses a non-runsc command or a command without one exact run/container suffix.
    pub fn from_run_command(
        command: &RunscCommand,
        container_id: ContainerId,
    ) -> Result<Self, ControlError> {
        if command.program() != OsStr::new("runsc") {
            return Err(ControlError::InvalidRequest);
        }
        let run = command
            .arguments()
            .iter()
            .position(|argument| argument == OsStr::new("run"))
            .ok_or(ControlError::InvalidRequest)?;
        if run + 3 != command.arguments().len()
            || !command.arguments()[run + 1]
                .as_encoded_bytes()
                .starts_with(b"-bundle=")
        {
            return Err(ControlError::InvalidRequest);
        }
        let container = OsString::from(hex_lower(container_id.as_bytes()));
        if command.arguments()[run + 2] != container {
            return Err(ControlError::InvalidRequest);
        }
        Ok(Self {
            program: command.program().to_owned(),
            global_arguments: command.arguments()[..run].to_vec(),
            container,
            runner: Arc::new(ChildBrokerCommandRunner),
        })
    }
}

impl NonInteractiveSessionExecutor for RunscNonInteractiveExecutor {
    fn execute(
        &self,
        request: &NonInteractiveSessionRequest,
        emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
    ) -> Result<SessionExit, ControlError> {
        let mut target = TargetCommand::new(self.program.clone());
        for argument in &self.global_arguments {
            target = target.argument(argument.clone());
        }
        target = target.argument("exec").argument(self.container.clone());
        for argument in &request.argv {
            target = target.argument(OsString::from_vec(argument.clone()));
        }
        let process_status = self.runner.run(&target, request.session_id, emit)?;
        match (process_status.exit_code, process_status.terminating_signal) {
            (Some(code), None) => Ok(SessionExit::exited(request.session_id, code)),
            (None, Some(_)) => Ok(SessionExit::terminated(request.session_id)),
            _ => Err(ControlError::InvalidRequest),
        }
    }
}

fn emit_available(
    stream: &mut File,
    session_id: SessionId,
    stdout: bool,
    emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
) -> Result<bool, ControlError> {
    let mut bytes = [0_u8; 16 * 1024];
    loop {
        match stream.read(&mut bytes) {
            Ok(0) => return Ok(true),
            Ok(length) => {
                let output = lazaret_core::records::NonInteractiveOutput {
                    session_id,
                    bytes: bytes[..length].to_vec(),
                };
                emit(if stdout {
                    NonInteractiveSessionMessage::Stdout(output)
                } else {
                    NonInteractiveSessionMessage::Stderr(output)
                })?;
            }
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(false),
            Err(error) => return Err(error.into()),
        }
    }
}

fn hex_lower(bytes: &[u8]) -> String {
    use fmt::Write as _;

    bytes.iter().fold(String::new(), |mut output, byte| {
        write!(output, "{byte:02x}").expect("writing to a String cannot fail");
        output
    })
}

pub struct StagedControlFace {
    listener: Option<std::os::unix::net::UnixListener>,
    paths: ControlPaths,
    staged_socket_identity: FileIdentity,
    identity_file: File,
    identity_file_identity: FileIdentity,
    publication_identity: ProtocolPublicationIdentity,
    publication_bytes: Vec<u8>,
    uid: u32,
    anchor: Option<TrustAnchor>,
    server_config: Arc<ServerConfig>,
    registry: Arc<SessionRegistry>,
    executor: Arc<dyn NonInteractiveSessionExecutor>,
}

impl StagedControlFace {
    /// Stages an undiscoverable private control face before durable publication.
    ///
    /// # Errors
    /// Refuses existing artifacts, changed paths, ownership/mode errors, or TLS failure.
    pub fn stage(
        runtime_domain_root: &Path,
        uid: u32,
        anchor: TrustAnchor,
        registry: Arc<SessionRegistry>,
    ) -> Result<Self, ControlError> {
        Self::stage_with_executor(
            runtime_domain_root,
            uid,
            anchor,
            registry,
            Arc::new(ExecutionUnavailable),
        )
    }

    /// Stages a control face with its generation-bound non-interactive executor.
    ///
    /// # Errors
    /// Refuses the same invalid staging conditions as [`Self::stage`].
    pub fn stage_with_executor(
        runtime_domain_root: &Path,
        uid: u32,
        anchor: TrustAnchor,
        registry: Arc<SessionRegistry>,
        executor: Arc<dyn NonInteractiveSessionExecutor>,
    ) -> Result<Self, ControlError> {
        if registry.generation_identity != anchor.generation_identity {
            return Err(ControlError::StaleGeneration);
        }
        let paths = ControlPaths::open(runtime_domain_root, uid)?;
        if [
            &paths.socket,
            &paths.identity,
            &paths.staged_socket,
            &paths.staged_identity,
        ]
        .into_iter()
        .any(|path| path.try_exists().unwrap_or(true))
        {
            return Err(ControlError::InvalidPublication(
                "publication path already exists",
            ));
        }
        let publication_identity = anchor.publication_identity();
        let publication_bytes = encode(&publication_identity)?;
        let server_config = Arc::new(anchor.server_config()?);
        let (identity_file, identity_file_identity) = paths.create_named_identity(
            STAGED_PUBLICATION_IDENTITY_NAME,
            &publication_bytes,
            uid,
        )?;
        let listener = std::os::unix::net::UnixListener::bind(&paths.staged_socket)
            .map_err(|_| ControlError::InvalidPublication("staged control socket bind failed"))?;
        std::fs::set_permissions(&paths.staged_socket, std::fs::Permissions::from_mode(0o600))?;
        let socket_metadata = std::fs::symlink_metadata(&paths.staged_socket)?;
        let staged_socket_identity = FileIdentity::from_metadata(&socket_metadata);
        if !socket_metadata.file_type().is_socket()
            || socket_metadata.uid() != uid
            || socket_metadata.mode() & 0o7777 != 0o600
        {
            return Err(ControlError::InvalidPublication(
                "staged control socket ownership or mode is invalid",
            ));
        }
        paths.root_file.sync_all()?;
        Ok(Self {
            listener: Some(listener),
            paths,
            staged_socket_identity,
            identity_file,
            identity_file_identity,
            publication_identity,
            publication_bytes,
            uid,
            anchor: Some(anchor),
            server_config,
            registry,
            executor,
        })
    }

    /// Exposes the staged face after the exact generation is durably Published.
    ///
    /// # Errors
    /// Refuses stale durable state, changed staged artifacts, or exposure failure.
    pub fn expose(mut self) -> Result<PublishedControlFace, ControlError> {
        self.registry.require_serviceable_generation()?;
        self.paths.validate_root(self.uid)?;
        let staged_socket = std::fs::symlink_metadata(&self.paths.staged_socket)?;
        if FileIdentity::from_metadata(&staged_socket) != self.staged_socket_identity {
            return Err(ControlError::InvalidPublication(
                "staged control socket changed",
            ));
        }
        self.paths.validate_identity_at(
            &self.paths.staged_identity,
            &self.identity_file,
            self.identity_file_identity,
            self.uid,
            self.publication_bytes.len(),
        )?;
        std::fs::rename(&self.paths.staged_socket, &self.paths.socket)?;
        if let Err(error) = std::fs::rename(&self.paths.staged_identity, &self.paths.identity) {
            remove_if_identity_matches(&self.paths.socket, self.staged_socket_identity);
            return Err(error.into());
        }
        self.paths.root_file.sync_all()?;
        let socket_identity =
            FileIdentity::from_metadata(&std::fs::symlink_metadata(&self.paths.socket)?);
        self.paths.validate_identity(
            &self.identity_file,
            self.identity_file_identity,
            self.uid,
            self.publication_bytes.len(),
        )?;
        let paths = ControlPaths::open(&self.paths.root, self.uid)?;
        Ok(PublishedControlFace {
            listener: self
                .listener
                .take()
                .ok_or(ControlError::InvalidPublication(
                    "staged listener is absent",
                ))?,
            paths,
            socket_identity,
            identity_file: self.identity_file.try_clone()?,
            identity_file_identity: self.identity_file_identity,
            publication_identity: self.publication_identity.clone(),
            publication_bytes: self.publication_bytes.clone(),
            uid: self.uid,
            anchor: self.anchor.take().ok_or(ControlError::InvalidPublication(
                "staged Trust Anchor is absent",
            ))?,
            server_config: Arc::clone(&self.server_config),
            registry: Arc::clone(&self.registry),
            executor: Arc::clone(&self.executor),
        })
    }
}

impl Drop for StagedControlFace {
    fn drop(&mut self) {
        remove_if_identity_matches(&self.paths.staged_socket, self.staged_socket_identity);
        remove_if_identity_matches(&self.paths.staged_identity, self.identity_file_identity);
    }
}

impl PublishedControlFace {
    /// Selects blocking or polling admission for the publication listener.
    ///
    /// # Errors
    /// Returns an operating-system failure changing the listener mode.
    pub fn set_nonblocking(&self, nonblocking: bool) -> Result<(), ControlError> {
        self.listener.set_nonblocking(nonblocking)?;
        Ok(())
    }

    /// Discovers one publication through the retained runtime Domain authority.
    ///
    /// # Errors
    /// Refuses changed root authority or absent, replaced, non-private, oversized, or malformed
    /// publication objects.
    pub fn discover_confined(
        runtime_domain_root: &ConfinedPathPlan,
        uid: u32,
    ) -> Result<(File, PathBuf, ProtocolPublicationIdentity), ControlError> {
        let root = File::from(
            runtime_domain_root
                .open_authority()
                .map_err(|_| ControlError::InvalidPath("runtime Domain authority changed"))?,
        );
        let root_metadata = root.metadata()?;
        if !root_metadata.is_dir()
            || root_metadata.uid() != uid
            || root_metadata.mode() & 0o7777 != 0o700
        {
            return Err(ControlError::InvalidPath(
                "runtime Domain root is not invoking-user-owned mode 0700",
            ));
        }
        let descriptor = openat(
            root.as_fd(),
            PUBLICATION_IDENTITY_NAME,
            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(io::Error::from)?;
        let mut identity = File::from(descriptor);
        let identity_metadata = identity.metadata()?;
        if !identity_metadata.file_type().is_file()
            || identity_metadata.uid() != uid
            || identity_metadata.mode() & 0o7777 != 0o600
            || identity_metadata.nlink() != 1
            || identity_metadata.len() > MAX_CONTROL_BYTES as u64
        {
            return Err(ControlError::InvalidPublication(
                "publication identity is not an exact private regular file",
            ));
        }
        let socket = File::from(
            openat(
                root.as_fd(),
                CONTROL_SOCKET_NAME,
                OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
            )
            .map_err(io::Error::from)?,
        );
        let socket_metadata = socket.metadata()?;
        if !socket_metadata.file_type().is_socket()
            || socket_metadata.uid() != uid
            || socket_metadata.mode() & 0o7777 != 0o600
        {
            return Err(ControlError::InvalidPublication(
                "control socket is not an invoking-user-private socket",
            ));
        }
        let mut bytes = Vec::new();
        identity.read_to_end(&mut bytes)?;
        let socket_path = PathBuf::from(format!(
            "/proc/self/fd/{}/{}",
            root.as_raw_fd(),
            CONTROL_SOCKET_NAME
        ));
        Ok((root, socket_path, decode(&bytes)?))
    }

    /// Discovers one exact current control publication beneath a validated runtime Domain root.
    ///
    /// # Errors
    /// Refuses absent, replaced, non-private, oversized, or malformed publication objects.
    pub fn discover(
        runtime_domain_root: &Path,
        uid: u32,
    ) -> Result<(PathBuf, ProtocolPublicationIdentity), ControlError> {
        let paths = ControlPaths::open(runtime_domain_root, uid)?;
        let descriptor = openat(
            paths.root_file.as_fd(),
            PUBLICATION_IDENTITY_NAME,
            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(io::Error::from)?;
        let mut identity = File::from(descriptor);
        let descriptor_metadata = identity.metadata()?;
        let path_metadata = std::fs::symlink_metadata(&paths.identity)?;
        let expected = FileIdentity::from_metadata(&descriptor_metadata);
        if FileIdentity::from_metadata(&path_metadata) != expected
            || !descriptor_metadata.file_type().is_file()
            || descriptor_metadata.uid() != uid
            || descriptor_metadata.mode() & 0o7777 != 0o600
            || descriptor_metadata.nlink() != 1
            || descriptor_metadata.len() > MAX_CONTROL_BYTES as u64
        {
            return Err(ControlError::InvalidPublication(
                "publication identity is not an exact private regular file",
            ));
        }
        let socket = std::fs::symlink_metadata(&paths.socket)?;
        if !socket.file_type().is_socket() || socket.uid() != uid || socket.mode() & 0o7777 != 0o600
        {
            return Err(ControlError::InvalidPublication(
                "control socket is not an invoking-user-private socket",
            ));
        }
        let mut bytes = Vec::new();
        identity.read_to_end(&mut bytes)?;
        Ok((paths.socket, decode(&bytes)?))
    }

    /// Publishes one generation control face in the fixed host-private runtime path.
    ///
    /// # Errors
    /// Refuses stale generations, existing publication objects, path/ownership/mode errors, and
    /// TLS configuration failures.
    pub fn publish(
        runtime_domain_root: &Path,
        uid: u32,
        anchor: TrustAnchor,
        store: Arc<HostCoordinatorDomainStore>,
    ) -> Result<Self, ControlError> {
        Self::publish_with_executor(
            runtime_domain_root,
            uid,
            anchor,
            store,
            Arc::new(ExecutionUnavailable),
        )
    }

    /// Publishes one generation control face with non-interactive session execution.
    ///
    /// # Errors
    /// Refuses the same invalid publication conditions as [`Self::publish`].
    pub fn publish_with_executor(
        runtime_domain_root: &Path,
        uid: u32,
        anchor: TrustAnchor,
        store: Arc<HostCoordinatorDomainStore>,
        executor: Arc<dyn NonInteractiveSessionExecutor>,
    ) -> Result<Self, ControlError> {
        let registry = Arc::new(SessionRegistry::new(
            store,
            anchor.generation_identity.clone(),
        ));
        StagedControlFace::stage_with_executor(
            runtime_domain_root,
            uid,
            anchor,
            registry,
            executor,
        )?
        .expose()
    }

    #[must_use]
    pub fn publication_identity(&self) -> &ProtocolPublicationIdentity {
        &self.publication_identity
    }

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

    #[must_use]
    pub fn socket_path(&self) -> &Path {
        &self.paths.socket
    }

    #[must_use]
    pub fn identity_path(&self) -> &Path {
        &self.paths.identity
    }

    /// Withdraws exact discovery and connection paths while retaining live accepted streams.
    ///
    /// # Errors
    /// Refuses changed artifacts, removal failures, or artifacts that remain after removal.
    pub fn withdraw(&self) -> Result<(), ControlError> {
        remove_exact_if_identity_matches(&self.paths.socket, self.socket_identity)?;
        remove_exact_if_identity_matches(&self.paths.identity, self.identity_file_identity)
    }

    /// Accepts and serves one TLS connection. Registered connections remain live until EOF.
    ///
    /// # Errors
    /// Refuses peer UID, publication, TLS, ALPN, certificate, framing, tuple, duplicate, and
    /// persistence failures.
    pub fn serve_next(&self) -> Result<(), ControlError> {
        self.serve_next_coordinated(
            |admission| {
                self.registry
                    .register(admission.request, &admission.credential)
                    .map(|_| ())
            },
            |session_id| self.registry.disconnect(session_id),
        )
    }

    /// Serves one connection while routing authenticated session state through its lifecycle.
    ///
    /// # Errors
    /// Refuses the same conditions as [`Self::serve_next`] plus lifecycle admission or disconnect
    /// failure.
    pub fn serve_next_coordinated<Admit, Disconnect>(
        &self,
        mut admit: Admit,
        mut disconnect: Disconnect,
    ) -> Result<(), ControlError>
    where
        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
    {
        let (socket, _) = self.listener.accept()?;
        let peer = socket_peercred(&socket).map_err(io::Error::from)?;
        require_peer_uid(peer.uid.as_raw(), self.uid)?;
        self.validate_publication()?;
        let mut connection = ServerConnection::new(Arc::clone(&self.server_config))?;
        let mut socket = socket;
        while connection.is_handshaking() {
            connection.complete_io(&mut socket)?;
        }
        if connection.alpn_protocol() != Some(CONTROL_ALPN) {
            return Err(ControlError::InvalidRequest);
        }
        let peer_certificate = connection.peer_certificates().map(|certificates| {
            certificates
                .iter()
                .map(|certificate| certificate.to_vec())
                .collect::<Vec<_>>()
        });
        let mut stream = rustls::StreamOwned::new(connection, socket);
        match peer_certificate {
            None => self.serve_enrollment(&mut stream),
            Some(certificates) => {
                self.serve_registration(&mut stream, &certificates, &mut admit, &mut disconnect)
            }
        }
    }

    fn serve_enrollment(
        &self,
        stream: &mut rustls::StreamOwned<ServerConnection, std::os::unix::net::UnixStream>,
    ) -> Result<(), ControlError> {
        let request: ControlRequest = read_frame(stream, MAX_ENROLLMENT_BYTES)?;
        let ControlRequest::Enroll(request) = request else {
            return Err(ControlError::InvalidRequest);
        };
        let credential = self.anchor.issue_client_credential(&request)?;
        self.registry.enroll(credential.clone())?;
        write_frame(
            stream,
            &ControlResponse::Enrolled(ClientEnrollmentResponse {
                protocol_version: ProtocolVersion::V1,
                credential,
            }),
        )
    }

    fn serve_registration<Admit, Disconnect>(
        &self,
        stream: &mut rustls::StreamOwned<ServerConnection, std::os::unix::net::UnixStream>,
        certificates: &[Vec<u8>],
        admit: &mut Admit,
        disconnect: &mut Disconnect,
    ) -> Result<(), ControlError>
    where
        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
    {
        let [leaf] = certificates else {
            return Err(ControlError::InvalidCertificateProfile);
        };
        let credential = self.registry.credential_for_leaf(leaf)?;
        let request: ControlRequest = read_frame(stream, MAX_CONTROL_BYTES)?;
        let ControlRequest::Register(request) = request else {
            return Err(ControlError::InvalidRequest);
        };
        let capability = issue_session_capability(
            request.session_binding.clone(),
            self.anchor.coordinator_protocol_credential.credential_id,
            self.anchor.sandbox_protocol_credential.credential_id,
            &self.anchor.coordinator_protocol_key,
        )?;
        let session_id = request.session_binding.session_id;
        admit(ControlSessionAdmission {
            request,
            credential,
        })?;
        let result = (|| {
            write_frame(
                stream,
                &ControlResponse::Registered(SessionRegistrationResponse {
                    protocol_version: ProtocolVersion::V1,
                    session_capability: capability,
                }),
            )?;
            let request: ControlRequest = match read_frame(stream, MAX_CONTROL_BYTES) {
                Ok(request) => request,
                Err(ControlError::Io(error))
                    if matches!(
                        error.kind(),
                        io::ErrorKind::UnexpectedEof | io::ErrorKind::ConnectionReset
                    ) =>
                {
                    return Ok(());
                }
                Err(error) => return Err(error),
            };
            let ControlRequest::Execute(request) = request else {
                return Err(ControlError::InvalidRequest);
            };
            if request.session_id != session_id
                || request.argv.is_empty()
                || request.argv[0].is_empty()
                || request.argv.iter().any(|argument| argument.contains(&0))
            {
                return Err(ControlError::InvalidRequest);
            }
            let mut send_message = |message: NonInteractiveSessionMessage| {
                if matches!(message, NonInteractiveSessionMessage::Exit(_)) {
                    return Err(ControlError::InvalidRequest);
                }
                require_non_interactive_session(&message, request.session_id)?;
                write_frame(stream, &message)
            };
            let outcome = self.executor.execute(&request, &mut send_message)?;
            if outcome.session_id() != request.session_id {
                return Err(ControlError::InvalidRequest);
            }
            require_non_interactive_session(
                &NonInteractiveSessionMessage::Exit(outcome),
                request.session_id,
            )?;
            write_frame(stream, &NonInteractiveSessionMessage::Exit(outcome))
        })();
        let disconnected = disconnect(session_id);
        result.and(disconnected)
    }

    fn validate_publication(&self) -> Result<(), ControlError> {
        self.paths.validate_root(self.uid)?;
        let socket = std::fs::symlink_metadata(&self.paths.socket)?;
        let identity = std::fs::symlink_metadata(&self.paths.identity)?;
        if FileIdentity::from_metadata(&socket) != self.socket_identity
            || !socket.file_type().is_socket()
            || socket.uid() != self.uid
            || socket.mode() & 0o7777 != 0o600
            || FileIdentity::from_metadata(&identity) != self.identity_file_identity
            || !identity.file_type().is_file()
            || identity.uid() != self.uid
            || identity.mode() & 0o7777 != 0o600
            || usize::try_from(self.identity_file.metadata()?.len()).ok()
                != Some(self.publication_bytes.len())
        {
            return Err(ControlError::InvalidPublication(
                "publication objects changed identity, ownership, or mode",
            ));
        }
        let mut exact_identity = self.identity_file.try_clone()?;
        exact_identity.seek(SeekFrom::Start(0))?;
        let mut bytes = Vec::new();
        exact_identity.read_to_end(&mut bytes)?;
        if bytes != self.publication_bytes {
            return Err(ControlError::InvalidPublication(
                "publication identity bytes changed",
            ));
        }
        self.registry.require_serviceable_generation()
    }

    /// Creates a TLS client configuration pinned to this publication identity.
    ///
    /// # Errors
    /// Refuses malformed CA or client certificate/key material.
    pub fn client_config(
        publication: &ProtocolPublicationIdentity,
        client: Option<(&TlsCredential, PrivateKeyDer<'static>)>,
    ) -> Result<ClientConfig, ControlError> {
        let mut roots = RootCertStore::empty();
        roots.add(CertificateDer::from(
            publication.tls_ca_certificate_der.clone(),
        ))?;
        let builder = ClientConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
            .with_root_certificates(roots);
        let mut config = if let Some((credential, private_key)) = client {
            verify_tls_credential(
                credential,
                &publication.generation_identity,
                TlsRole::ClientCli,
                credential.subject_id(),
            )?;
            builder.with_client_auth_cert(
                vec![CertificateDer::from(credential.certificate_der.clone())],
                private_key,
            )?
        } else {
            builder.with_no_client_auth()
        };
        config.alpn_protocols = vec![CONTROL_ALPN.to_vec()];
        config.enable_early_data = false;
        config.resumption = rustls::client::Resumption::disabled();
        Ok(config)
    }

    /// Connects a rustls stream to the fixed control socket using an existing configuration.
    ///
    /// # Errors
    /// Returns path, connection, or TLS initialization failures.
    pub fn connect(
        socket_path: &Path,
        config: ClientConfig,
        publication: &ProtocolPublicationIdentity,
    ) -> Result<rustls::StreamOwned<ClientConnection, std::os::unix::net::UnixStream>, ControlError>
    {
        let socket = std::os::unix::net::UnixStream::connect(socket_path)?;
        let server_name = ServerName::try_from(CONTROL_SERVER_NAME)
            .map_err(|_| ControlError::InvalidCertificateProfile)?;
        let mut connection = ClientConnection::new(Arc::new(config), server_name)?;
        let mut socket = socket;
        while connection.is_handshaking() {
            connection.complete_io(&mut socket)?;
        }
        let [certificate] = connection.peer_certificates().unwrap_or_default() else {
            return Err(ControlError::InvalidCertificateProfile);
        };
        if connection.protocol_version() != Some(rustls::ProtocolVersion::TLSv1_3)
            || connection.alpn_protocol() != Some(CONTROL_ALPN)
            || certificate_digest(certificate) != publication.coordinator_tls_certificate_sha256
        {
            return Err(ControlError::InvalidCertificateProfile);
        }
        Ok(rustls::StreamOwned::new(connection, socket))
    }

    /// Writes one length-delimited control record.
    ///
    /// # Errors
    /// Returns encoding or TLS I/O failures.
    pub fn send<Request: serde::Serialize>(
        stream: &mut impl Write,
        request: &Request,
    ) -> Result<(), ControlError> {
        write_frame(stream, request)
    }

    /// Reads one length-delimited control record.
    ///
    /// # Errors
    /// Returns framing, size, decoding, or TLS I/O failures.
    pub fn receive<Response: serde::de::DeserializeOwned + serde::Serialize>(
        stream: &mut impl Read,
    ) -> Result<Response, ControlError> {
        read_frame(stream, MAX_CONTROL_BYTES)
    }
}

impl Drop for PublishedControlFace {
    fn drop(&mut self) {
        remove_if_identity_matches(&self.paths.socket, self.socket_identity);
        remove_if_identity_matches(&self.paths.identity, self.identity_file_identity);
    }
}

fn remove_exact_if_identity_matches(
    path: &Path,
    expected: FileIdentity,
) -> Result<(), ControlError> {
    let metadata = match std::fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
        Err(error) => return Err(error.into()),
    };
    if FileIdentity::from_metadata(&metadata) != expected {
        return Err(ControlError::InvalidPublication(
            "publication artifact identity changed before withdrawal",
        ));
    }
    std::fs::remove_file(path)?;
    match std::fs::symlink_metadata(path) {
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.into()),
        Ok(_) => Err(ControlError::InvalidPublication(
            "publication artifact remained after withdrawal",
        )),
    }
}

fn remove_if_identity_matches(path: &Path, expected: FileIdentity) {
    if std::fs::symlink_metadata(path)
        .ok()
        .is_some_and(|metadata| FileIdentity::from_metadata(&metadata) == expected)
    {
        let _ = std::fs::remove_file(path);
    }
}

fn require_non_interactive_session(
    message: &NonInteractiveSessionMessage,
    expected: SessionId,
) -> Result<(), ControlError> {
    let actual = match message {
        NonInteractiveSessionMessage::Stdout(output)
        | NonInteractiveSessionMessage::Stderr(output) => output.session_id,
        NonInteractiveSessionMessage::Exit(exit) => exit.session_id(),
    };
    if actual == expected {
        Ok(())
    } else {
        Err(ControlError::InvalidRequest)
    }
}

fn leaf_parameters(
    common_name: &str,
    usage: ExtendedKeyUsagePurpose,
    now: OffsetDateTime,
) -> Result<CertificateParams, ControlError> {
    let mut params = CertificateParams::new(vec![common_name.to_owned()])?;
    params.distinguished_name = DistinguishedName::new();
    params
        .distinguished_name
        .push(DnType::CommonName, common_name);
    params.not_before = now - Duration::minutes(1);
    params.not_after = now + Duration::hours(1);
    params.is_ca = IsCa::NoCa;
    params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
    params.extended_key_usages = vec![usage];
    Ok(params)
}

fn random_signing_key() -> Result<SigningKey, ControlError> {
    let mut bytes = [0_u8; 32];
    getrandom::fill(&mut bytes).map_err(|_| ControlError::InvalidCertificateProfile)?;
    Ok(SigningKey::from_bytes(&bytes))
}

fn enrollment_binding(
    generation_identity: &GenerationIdentity,
    cli_id: ClientId,
) -> Result<String, ControlError> {
    let encoded = encode(&(generation_identity, cli_id))?;
    Ok(format!("lazaret-client-{}", hex(&sha256(&encoded))))
}

fn certificate_digest(certificate: impl AsRef<[u8]>) -> CertificateDigest {
    CertificateDigest::new(sha256(certificate.as_ref()))
}

fn sha256(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

fn short_identity(bytes: &[u8; 32]) -> String {
    hex(&bytes[..8])
}

fn hex(bytes: &[u8]) -> String {
    let mut encoded = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        use fmt::Write as _;
        let _ = write!(encoded, "{byte:02x}");
    }
    encoded
}

fn require_peer_uid(actual: u32, expected: u32) -> Result<(), ControlError> {
    if actual == expected {
        Ok(())
    } else {
        Err(ControlError::UnauthorizedPeer)
    }
}

fn is_live_session(phase: SessionPhase) -> bool {
    matches!(
        phase,
        SessionPhase::Registering
            | SessionPhase::Attached
            | SessionPhase::Prompting
            | SessionPhase::HostExecuting
    )
}

fn read_frame<Record: serde::de::DeserializeOwned + serde::Serialize>(
    reader: &mut impl Read,
    maximum: usize,
) -> Result<Record, ControlError> {
    let mut length = [0_u8; 4];
    reader.read_exact(&mut length)?;
    let length = u32::from_be_bytes(length) as usize;
    if length > maximum {
        return Err(ControlError::OversizeRequest { length, maximum });
    }
    let mut bytes = vec![0_u8; length];
    reader.read_exact(&mut bytes)?;
    decode(&bytes).map_err(Into::into)
}

fn write_frame<Record: serde::Serialize>(
    writer: &mut impl Write,
    record: &Record,
) -> Result<(), ControlError> {
    let bytes = encode(record)?;
    let length = u32::try_from(bytes.len())
        .map_err(|_| ControlError::OversizeRequest {
            length: bytes.len(),
            maximum: u32::MAX as usize,
        })?
        .to_be_bytes();
    writer.write_all(&length)?;
    writer.write_all(&bytes)?;
    writer.flush()?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::ffi::OsString;
    use std::sync::Arc;
    use std::thread;

    use lazaret_core::identity::{
        ConfigurationFingerprint, CoordinatorId, DomainId, GenerationId, SandboxId,
    };
    use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName};

    use super::*;

    fn generation() -> GenerationIdentity {
        GenerationIdentity {
            agent_profile: "trusted".into(),
            domain_id: DomainId::new([1; 32]),
            generation_id: GenerationId::new(7),
            configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
            coordinator_id: CoordinatorId::new([3; 32]),
            sandbox_id: SandboxId::new([4; 32]),
        }
    }

    fn client_config_with_certificate(
        anchor: &TrustAnchor,
        certificate: CertificateDer<'static>,
        private_key: PrivatePkcs8KeyDer<'static>,
    ) -> ClientConfig {
        let mut roots = RootCertStore::empty();
        roots.add(anchor.tls_ca_der.clone()).unwrap();
        let mut config = ClientConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
            .with_root_certificates(roots)
            .with_client_auth_cert(vec![certificate], PrivateKeyDer::Pkcs8(private_key))
            .unwrap();
        config.alpn_protocols = vec![CONTROL_ALPN.to_vec()];
        config.resumption = rustls::client::Resumption::disabled();
        config
    }

    fn handshake_pair(
        server_config: ServerConfig,
        client_config: ClientConfig,
    ) -> (Result<(), rustls::Error>, Result<(), rustls::Error>) {
        let (mut server_socket, mut client_socket) =
            std::os::unix::net::UnixStream::pair().unwrap();
        let server = thread::spawn(move || {
            let mut connection = ServerConnection::new(Arc::new(server_config)).unwrap();
            while connection.is_handshaking() {
                connection
                    .complete_io(&mut server_socket)
                    .map_err(|error| rustls::Error::General(error.to_string()))?;
            }
            Ok(())
        });
        let mut connection = ClientConnection::new(
            Arc::new(client_config),
            ServerName::try_from(CONTROL_SERVER_NAME).unwrap(),
        )
        .unwrap();
        let client = (|| {
            while connection.is_handshaking() {
                connection
                    .complete_io(&mut client_socket)
                    .map_err(|error| rustls::Error::General(error.to_string()))?;
            }
            Ok(())
        })();
        (server.join().unwrap(), client)
    }

    #[test]
    fn exact_peer_uid_is_required() {
        assert!(require_peer_uid(1_000, 1_000).is_ok());
        assert!(matches!(
            require_peer_uid(1_001, 1_000),
            Err(ControlError::UnauthorizedPeer)
        ));
    }

    struct RecordingRunner {
        observed: Mutex<Option<(OsString, Vec<OsString>)>>,
        exit: ProcessExit,
    }

    impl NonInteractiveCommandRunner for RecordingRunner {
        fn run(
            &self,
            target: &TargetCommand,
            _session_id: SessionId,
            _emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
        ) -> Result<ProcessExit, ControlError> {
            *self.observed.lock().unwrap() =
                Some((target.program().to_owned(), target.arguments().to_vec()));
            Ok(self.exit)
        }
    }

    #[test]
    fn runsc_exec_uses_the_command_as_the_first_post_container_argument() {
        let runner = Arc::new(RecordingRunner {
            observed: Mutex::new(None),
            exit: ProcessExit {
                exit_code: Some(23),
                terminating_signal: None,
            },
        });
        let executor = RunscNonInteractiveExecutor {
            program: "runsc".into(),
            global_arguments: vec!["--root=/runtime".into()],
            container: "container-id".into(),
            runner: runner.clone(),
        };
        let session_id = SessionId::new([9; 32]);
        let exit = executor
            .execute(
                &NonInteractiveSessionRequest {
                    protocol_version: ProtocolVersion::V1,
                    session_id,
                    argv: vec![b"printf".to_vec(), b"hello".to_vec()],
                },
                &mut |_| Ok(()),
            )
            .unwrap();

        assert_eq!(exit.exit_code(), Some(23));
        assert_eq!(
            runner.observed.lock().unwrap().as_ref().unwrap(),
            &(
                OsString::from("runsc"),
                ["--root=/runtime", "exec", "container-id", "printf", "hello"]
                    .into_iter()
                    .map(OsString::from)
                    .collect()
            )
        );
    }

    #[test]
    fn rustls_refuses_client_leaf_with_server_role() {
        let anchor = TrustAnchor::create(generation()).unwrap();
        let key = KeyPair::generate_for(&PKCS_ED25519).unwrap();
        let certificate = leaf_parameters(
            "wrong-client-role.invalid",
            ExtendedKeyUsagePurpose::ServerAuth,
            OffsetDateTime::now_utc(),
        )
        .unwrap()
        .signed_by(&key, &anchor.tls_ca)
        .unwrap();
        let client_config = client_config_with_certificate(
            &anchor,
            certificate.der().clone(),
            PrivatePkcs8KeyDer::from(key.serialize_der()),
        );
        let (server, client) = handshake_pair(anchor.server_config().unwrap(), client_config);
        assert!(server.is_err() || client.is_err());
    }

    #[test]
    fn rustls_refuses_server_leaf_with_client_role() {
        let anchor = TrustAnchor::create(generation()).unwrap();
        let key = KeyPair::generate_for(&PKCS_ED25519).unwrap();
        let certificate = leaf_parameters(
            CONTROL_SERVER_NAME,
            ExtendedKeyUsagePurpose::ClientAuth,
            OffsetDateTime::now_utc(),
        )
        .unwrap()
        .signed_by(&key, &anchor.tls_ca)
        .unwrap();
        let server_config =
            ServerConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
                .with_no_client_auth()
                .with_single_cert(
                    vec![certificate.der().clone()],
                    PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key.serialize_der())),
                )
                .unwrap();
        let client_config =
            PublishedControlFace::client_config(&anchor.publication_identity(), None).unwrap();
        let (server, client) = handshake_pair(server_config, client_config);
        assert!(server.is_err() || client.is_err());
    }
}