use std::fmt;
use std::fs::File;
use std::io::{self, Read};
use std::os::fd::AsFd;
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};

use lazaret_core::codec;
use lazaret_core::configuration::ResolvedDomainConfiguration;
use lazaret_core::identity::{ConfigurationFingerprint, DomainId, GenerationId, SessionId};
pub use lazaret_core::protocol::{
    BOOTSTRAP_WAITER_SOCKET_NAME, BootstrapWaitOutcome, BootstrapWaitRequest,
};
use lazaret_core::protocol::{read_waiter_frame, write_waiter_frame};
use lazaret_core::records::{ProtocolVersion, Session, SessionPhase};
use rustix::net::sockopt::socket_peercred;
use serde::Serialize;

use crate::domain_store::HostCoordinatorDomainStore;
use crate::lifecycle::{HostCoordinatorLifecycle, LifecycleAdapter, SessionLifecycleAdapter};

const MAX_FRAME_BYTES: usize = 4_096;
const MAX_WAITERS: usize = 64;

#[derive(Debug)]
pub struct BootstrapWaiterError(String);

impl fmt::Display for BootstrapWaiterError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

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

fn failure(error: impl fmt::Display) -> BootstrapWaiterError {
    BootstrapWaiterError(error.to_string())
}

#[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(),
        }
    }
}

struct PendingWaiter {
    stream: UnixStream,
    bytes: Vec<u8>,
}

struct LiveWaiter {
    stream: UnixStream,
    session_id: SessionId,
}

pub struct BootstrapWaiterFace {
    root: File,
    root_path: PathBuf,
    root_identity: FileIdentity,
    socket_path: PathBuf,
    socket_identity: FileIdentity,
    listener: UnixListener,
    pending: Vec<PendingWaiter>,
    waiters: Vec<LiveWaiter>,
    domain_id: DomainId,
    configuration_fingerprint: ConfigurationFingerprint,
    generation_id: GenerationId,
    uid: u32,
    withdrawn: bool,
}

impl BootstrapWaiterFace {
    /// Publishes one bounded invoking-user-only waiter admission socket.
    ///
    /// # Errors
    /// Refuses changed Runtime Domain authority, a mismatched Domain Store, or a live, replaced, or
    /// non-private waiter path. Recovers an inactive exact socket under the retained Domain lock.
    pub(crate) fn bind(
        configuration: &ResolvedDomainConfiguration,
        store: &HostCoordinatorDomainStore,
        generation_id: GenerationId,
        uid: u32,
    ) -> Result<Self, BootstrapWaiterError> {
        if store.snapshot().map_err(failure)?.domain.domain_id != configuration.domain_id() {
            return Err(failure(
                "Domain Store authority does not match the bootstrap waiter Domain",
            ));
        }
        let root = File::from(
            configuration
                .runtime_domain_root()
                .open_authority()
                .map_err(failure)?,
        );
        let root_path = PathBuf::from(std::ffi::OsString::from_vec(
            configuration
                .runtime_domain_root()
                .canonical_label()
                .as_bytes()
                .to_vec(),
        ));
        let root_metadata = root.metadata().map_err(failure)?;
        let path_metadata = std::fs::symlink_metadata(&root_path).map_err(failure)?;
        if FileIdentity::from_metadata(&root_metadata)
            != FileIdentity::from_metadata(&path_metadata)
            || !root_metadata.is_dir()
            || root_metadata.uid() != uid
            || root_metadata.mode() & 0o7777 != 0o700
        {
            return Err(failure(
                "Runtime Domain root is not exact invoking-user mode 0700",
            ));
        }
        let socket_path = root_path.join(BOOTSTRAP_WAITER_SOCKET_NAME);
        match std::fs::symlink_metadata(&socket_path) {
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Ok(metadata) => recover_stale_socket(&socket_path, &metadata, uid)?,
            Err(error) => return Err(failure(error)),
        }
        let listener = UnixListener::bind(&socket_path).map_err(failure)?;
        let socket_metadata = configure_bound_socket(&listener, &socket_path, uid)?;
        Ok(Self {
            root,
            root_path,
            root_identity: FileIdentity::from_metadata(&root_metadata),
            socket_path,
            socket_identity: FileIdentity::from_metadata(&socket_metadata),
            listener,
            pending: Vec::new(),
            waiters: Vec::new(),
            domain_id: configuration.domain_id(),
            configuration_fingerprint: configuration.configuration_fingerprint(),
            generation_id,
            uid,
            withdrawn: false,
        })
    }

    /// Admits every complete valid request currently available without blocking.
    ///
    /// # Errors
    /// Refuses changed socket authority, I/O failures, or durable lifecycle failures.
    pub fn admit_pending<A, S>(
        &mut self,
        lifecycle: &mut HostCoordinatorLifecycle<'_, A, S>,
    ) -> Result<(), BootstrapWaiterError>
    where
        A: LifecycleAdapter,
        A::Error: fmt::Display,
        S: SessionLifecycleAdapter<Error = A::Error>,
    {
        self.validate_paths()?;
        while self.pending.len() + self.waiters.len() < MAX_WAITERS {
            match self.listener.accept() {
                Ok((stream, _)) => {
                    let credentials = socket_peercred(stream.as_fd()).map_err(failure)?;
                    if credentials.uid.as_raw() != self.uid {
                        continue;
                    }
                    stream.set_nonblocking(true).map_err(failure)?;
                    self.pending.push(PendingWaiter {
                        stream,
                        bytes: Vec::new(),
                    });
                }
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                Err(error) => return Err(failure(error)),
            }
        }

        let mut index = 0;
        while index < self.pending.len() {
            match read_request(&mut self.pending[index]) {
                Ok(Some(request)) => {
                    let mut pending = self.pending.swap_remove(index);
                    if request.protocol_version != ProtocolVersion::V1
                        || request.domain_id != self.domain_id
                        || request.configuration_fingerprint != self.configuration_fingerprint
                    {
                        let _ = send_outcome(&mut pending.stream, BootstrapWaitOutcome::Refused);
                        continue;
                    }
                    let session = Session {
                        generation_id: self.generation_id,
                        session_id: request.session_id,
                        cli_id: request.cli_id,
                        session_anchor_id: request.session_anchor_id,
                        session_phase: SessionPhase::New,
                    };
                    if lifecycle.wait_for_domain(session).is_err() {
                        let _ = send_outcome(&mut pending.stream, BootstrapWaitOutcome::Refused);
                        continue;
                    }
                    if send_outcome(&mut pending.stream, BootstrapWaitOutcome::Waiting).is_err() {
                        lifecycle
                            .disconnect_session(request.session_id)
                            .map_err(failure)?;
                        continue;
                    }
                    self.waiters.push(LiveWaiter {
                        stream: pending.stream,
                        session_id: request.session_id,
                    });
                }
                Ok(None) => index += 1,
                Err(_) => {
                    self.pending.swap_remove(index);
                }
            }
        }
        Ok(())
    }

    /// Withdraws the exact waiter admission path.
    ///
    /// # Errors
    /// Refuses changed authority or failure to remove the exact socket.
    pub fn withdraw(&mut self) -> Result<(), BootstrapWaiterError> {
        if self.withdrawn {
            return Ok(());
        }
        self.validate_paths()?;
        std::fs::remove_file(&self.socket_path).map_err(failure)?;
        self.withdrawn = true;
        Ok(())
    }

    #[must_use]
    pub fn notify(&mut self, outcome: BootstrapWaitOutcome) -> Vec<SessionId> {
        if matches!(
            outcome,
            BootstrapWaitOutcome::Waiting | BootstrapWaitOutcome::Refused
        ) {
            return self
                .waiters
                .iter()
                .map(|waiter| waiter.session_id)
                .collect();
        }
        let mut failed = Vec::new();
        self.waiters.retain_mut(|waiter| {
            let retained = send_outcome(&mut waiter.stream, outcome).is_ok()
                && outcome == BootstrapWaitOutcome::Published
                && waiter.stream.set_nonblocking(true).is_ok();
            if !retained && outcome == BootstrapWaitOutcome::Published {
                failed.push(waiter.session_id);
            }
            retained
        });
        failed
    }

    /// Reconciles published waiter streams with their durable control-registration phase.
    ///
    /// # Errors
    /// Returns a durable lifecycle error when a vanished Waiting session cannot be disconnected.
    pub fn reconcile_published<A, S>(
        &mut self,
        lifecycle: &mut HostCoordinatorLifecycle<'_, A, S>,
    ) -> Result<(), BootstrapWaiterError>
    where
        A: LifecycleAdapter,
        A::Error: fmt::Display,
        S: SessionLifecycleAdapter<Error = A::Error>,
    {
        let mut index = 0;
        while index < self.waiters.len() {
            let closed = !matches!(
                self.waiters[index].stream.read(&mut [0_u8; 1]),
                Err(error) if error.kind() == io::ErrorKind::WouldBlock
            );
            if !closed {
                index += 1;
                continue;
            }
            let waiter = self.waiters.swap_remove(index);
            if lifecycle.session_phase(waiter.session_id) == Some(SessionPhase::Waiting) {
                lifecycle
                    .disconnect_session(waiter.session_id)
                    .map_err(failure)?;
                lifecycle.teardown_if_idle().map_err(failure)?;
            }
        }
        Ok(())
    }

    fn validate_paths(&self) -> Result<(), BootstrapWaiterError> {
        let root_metadata = self.root.metadata().map_err(failure)?;
        let path_metadata = std::fs::symlink_metadata(&self.root_path).map_err(failure)?;
        let socket_metadata = std::fs::symlink_metadata(&self.socket_path).map_err(failure)?;
        if FileIdentity::from_metadata(&root_metadata) != self.root_identity
            || FileIdentity::from_metadata(&path_metadata) != self.root_identity
            || !root_metadata.is_dir()
            || root_metadata.uid() != self.uid
            || root_metadata.mode() & 0o7777 != 0o700
            || FileIdentity::from_metadata(&socket_metadata) != self.socket_identity
            || !socket_metadata.file_type().is_socket()
            || socket_metadata.uid() != self.uid
            || socket_metadata.mode() & 0o7777 != 0o600
        {
            return Err(failure("bootstrap waiter authority changed"));
        }
        Ok(())
    }
}

fn configure_bound_socket(
    listener: &UnixListener,
    socket_path: &Path,
    uid: u32,
) -> Result<std::fs::Metadata, BootstrapWaiterError> {
    let initial_metadata = std::fs::symlink_metadata(socket_path).map_err(failure)?;
    let initial_identity = FileIdentity::from_metadata(&initial_metadata);
    listener.set_nonblocking(true).map_err(failure)?;
    let probe = UnixStream::connect(socket_path).map_err(failure)?;
    let (accepted, _) = listener.accept().map_err(failure)?;
    drop((probe, accepted));
    let configured = (|| {
        std::fs::set_permissions(socket_path, std::fs::Permissions::from_mode(0o600))
            .map_err(failure)?;
        let metadata = std::fs::symlink_metadata(socket_path).map_err(failure)?;
        if FileIdentity::from_metadata(&metadata) != initial_identity
            || !metadata.file_type().is_socket()
            || metadata.uid() != uid
            || metadata.mode() & 0o7777 != 0o600
        {
            return Err(failure(
                "bootstrap waiter socket is not exact invoking-user mode 0600",
            ));
        }
        Ok(metadata)
    })();
    if configured.is_err() {
        remove_socket_if_identity(socket_path, initial_identity);
    }
    configured
}

fn recover_stale_socket(
    socket_path: &Path,
    metadata: &std::fs::Metadata,
    uid: u32,
) -> Result<(), BootstrapWaiterError> {
    if !metadata.file_type().is_socket()
        || metadata.uid() != uid
        || metadata.mode() & 0o7777 != 0o600
    {
        return Err(failure("bootstrap waiter path already exists"));
    }
    match UnixStream::connect(socket_path) {
        Ok(_) => {
            return Err(failure(
                "bootstrap waiter socket is still accepting connections",
            ));
        }
        Err(error) if error.kind() == io::ErrorKind::ConnectionRefused => {}
        Err(error) => return Err(failure(format!("probing bootstrap waiter socket: {error}"))),
    }
    let current = std::fs::symlink_metadata(socket_path).map_err(failure)?;
    if FileIdentity::from_metadata(&current) != FileIdentity::from_metadata(metadata) {
        return Err(failure(
            "bootstrap waiter authority changed during recovery",
        ));
    }
    std::fs::remove_file(socket_path).map_err(failure)
}

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

impl Drop for BootstrapWaiterFace {
    fn drop(&mut self) {
        if !self.withdrawn {
            remove_socket_if_identity(&self.socket_path, self.socket_identity);
        }
    }
}

fn read_request(
    pending: &mut PendingWaiter,
) -> Result<Option<BootstrapWaitRequest>, BootstrapWaiterError> {
    let mut buffer = [0_u8; 1_024];
    loop {
        match pending.stream.read(&mut buffer) {
            Ok(0) => return Err(failure("bootstrap waiter closed before one request")),
            Ok(length) => {
                pending.bytes.extend_from_slice(&buffer[..length]);
                if pending.bytes.len() > MAX_FRAME_BYTES + 4 {
                    return Err(failure("bootstrap waiter request is oversized"));
                }
            }
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
            Err(error) => return Err(failure(error)),
        }
    }
    if pending.bytes.len() < 4 {
        return Ok(None);
    }
    let length = u32::from_be_bytes(pending.bytes[..4].try_into().expect("four bytes")) as usize;
    if length > MAX_FRAME_BYTES {
        return Err(failure("bootstrap waiter request is oversized"));
    }
    if pending.bytes.len() < length + 4 {
        return Ok(None);
    }
    if pending.bytes.len() != length + 4 {
        return Err(failure("bootstrap waiter request has trailing bytes"));
    }
    codec::decode::<BootstrapWaitRequest>(&pending.bytes[4..])
        .map(Some)
        .map_err(failure)
}

fn send_outcome(
    stream: &mut UnixStream,
    outcome: BootstrapWaitOutcome,
) -> Result<(), BootstrapWaiterError> {
    stream.set_nonblocking(false).map_err(failure)?;
    write_frame(stream, &outcome)
}

/// # Errors
/// Refuses oversized or unencodable records and stream write failures.
pub fn write_frame<T: lazaret_core::records::SchemaType + Serialize>(
    stream: &mut UnixStream,
    record: &T,
) -> Result<(), BootstrapWaiterError> {
    write_waiter_frame(stream, record).map_err(failure)
}

/// # Errors
/// Refuses oversized, malformed, or incomplete frames and stream read failures.
pub fn read_frame<T>(stream: &mut UnixStream) -> Result<T, BootstrapWaiterError>
where
    T: lazaret_core::records::SchemaType + serde::de::DeserializeOwned + Serialize,
{
    read_waiter_frame(stream).map_err(failure)
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::fs::{self, OpenOptions};
    use std::sync::Arc;
    use std::thread;

    use lazaret_core::configuration::{ConfigurationLoadRequest, load_trusted_configuration};
    use lazaret_core::identity::{
        ClientId, CoordinatorId, NetworkSidecarId, SandboxId, SessionAnchorId,
    };
    use lazaret_core::records::{Domain, DurableDecision, Generation};
    use lazaret_linux::filesystem::try_exclusive_flock;

    use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
    use crate::lifecycle::{LifecycleAction, SessionAdmissionIdentity};

    use super::*;

    struct Adapter;

    impl LifecycleAdapter for Adapter {
        type Error = &'static str;

        fn execute(
            &mut self,
            _action: LifecycleAction,
            _generation: &Generation,
        ) -> Result<(), Self::Error> {
            Ok(())
        }
    }

    #[derive(Clone)]
    struct Admission(Session, ConfigurationFingerprint);

    struct Sessions;

    impl SessionLifecycleAdapter for Sessions {
        type Error = &'static str;
        type Admission = Admission;

        fn admission_identity(&self, admission: &Admission) -> SessionAdmissionIdentity {
            SessionAdmissionIdentity {
                generation_id: admission.0.generation_id,
                session_id: admission.0.session_id,
                configuration_fingerprint: admission.1,
            }
        }

        fn register_session(
            &self,
            store: &HostCoordinatorDomainStore,
            _entry_action: LifecycleAction,
            admission: Admission,
        ) -> Result<Session, Self::Error> {
            let registering = Session {
                session_phase: SessionPhase::Registering,
                ..admission.0
            };
            let attached = Session {
                session_phase: SessionPhase::Attached,
                ..registering.clone()
            };
            store
                .apply(&[
                    DomainStoreMutation::PutSession(registering),
                    DomainStoreMutation::PutSession(attached.clone()),
                ])
                .map_err(|_| "register")?;
            Ok(attached)
        }

        fn disconnect_session(
            &self,
            store: &HostCoordinatorDomainStore,
            session: &Session,
        ) -> Result<(), Self::Error> {
            store
                .apply(&[DomainStoreMutation::PutSession(Session {
                    session_phase: SessionPhase::Closed,
                    ..session.clone()
                })])
                .map_err(|_| "disconnect")
        }
    }

    #[test]
    fn waiter_is_durable_before_notification_and_advances_after_publish() {
        let tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let store = store(&configuration);
        let mut lifecycle = HostCoordinatorLifecycle::new(&store, Adapter, Sessions);
        lifecycle
            .start_new_domain(generation(&configuration), session(1))
            .unwrap();
        let mut face = BootstrapWaiterFace::bind(
            &configuration,
            &store,
            GenerationId::new(1),
            rustix::process::getuid().as_raw(),
        )
        .unwrap();
        let socket = face.socket_path.clone();
        let request = BootstrapWaitRequest {
            protocol_version: ProtocolVersion::V1,
            domain_id: configuration.domain_id(),
            configuration_fingerprint: configuration.configuration_fingerprint(),
            cli_id: ClientId::new([34; 32]),
            session_id: SessionId::new([2; 32]),
            session_anchor_id: SessionAnchorId::new([66; 32]),
        };
        let client_request = request.clone();
        let client = thread::spawn(move || {
            let mut stream = UnixStream::connect(socket).unwrap();
            write_frame(&mut stream, &client_request).unwrap();
            assert_eq!(
                read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
                BootstrapWaitOutcome::Waiting
            );
            assert_eq!(
                read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
                BootstrapWaitOutcome::Published
            );
        });
        for _ in 0..10_000 {
            face.admit_pending(&mut lifecycle).unwrap();
            if lifecycle.session_phase(request.session_id).is_some() {
                break;
            }
            thread::yield_now();
        }
        assert!(lifecycle.session_phase(request.session_id).is_some());
        assert_eq!(
            store
                .snapshot()
                .unwrap()
                .sessions
                .iter()
                .find(|session| session.session_id == request.session_id)
                .unwrap()
                .session_phase,
            SessionPhase::Waiting
        );
        complete_startup(&mut lifecycle);
        face.withdraw().unwrap();
        assert!(face.notify(BootstrapWaitOutcome::Published).is_empty());
        lifecycle
            .admit_session(Admission(
                Session {
                    generation_id: GenerationId::new(1),
                    session_id: request.session_id,
                    cli_id: request.cli_id,
                    session_anchor_id: request.session_anchor_id,
                    session_phase: SessionPhase::Waiting,
                },
                configuration.configuration_fingerprint(),
            ))
            .unwrap();
        assert_eq!(
            lifecycle.session_phase(request.session_id),
            Some(SessionPhase::Attached)
        );
        client.join().unwrap();
        face.reconcile_published(&mut lifecycle).unwrap();
        assert_eq!(
            lifecycle.session_phase(request.session_id),
            Some(SessionPhase::Attached)
        );
    }

    #[test]
    fn vanished_published_waiter_releases_its_durable_interest() {
        let tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let store = store(&configuration);
        let mut lifecycle = HostCoordinatorLifecycle::new(&store, Adapter, Sessions);
        lifecycle
            .start_new_domain(generation(&configuration), session(1))
            .unwrap();
        let mut face = BootstrapWaiterFace::bind(
            &configuration,
            &store,
            GenerationId::new(1),
            rustix::process::getuid().as_raw(),
        )
        .unwrap();
        let request = BootstrapWaitRequest {
            protocol_version: ProtocolVersion::V1,
            domain_id: configuration.domain_id(),
            configuration_fingerprint: configuration.configuration_fingerprint(),
            cli_id: ClientId::new([35; 32]),
            session_id: SessionId::new([3; 32]),
            session_anchor_id: SessionAnchorId::new([67; 32]),
        };
        let socket = face.socket_path.clone();
        let client_request = request.clone();
        let client = thread::spawn(move || {
            let mut stream = UnixStream::connect(socket).unwrap();
            write_frame(&mut stream, &client_request).unwrap();
            assert_eq!(
                read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
                BootstrapWaitOutcome::Waiting
            );
            assert_eq!(
                read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
                BootstrapWaitOutcome::Published
            );
        });
        for _ in 0..10_000 {
            face.admit_pending(&mut lifecycle).unwrap();
            if lifecycle.session_phase(request.session_id) == Some(SessionPhase::Waiting) {
                break;
            }
            thread::yield_now();
        }
        complete_startup(&mut lifecycle);
        face.withdraw().unwrap();
        assert!(face.notify(BootstrapWaitOutcome::Published).is_empty());
        client.join().unwrap();
        for _ in 0..10_000 {
            face.reconcile_published(&mut lifecycle).unwrap();
            if lifecycle.session_phase(request.session_id) == Some(SessionPhase::Closed) {
                break;
            }
            thread::yield_now();
        }
        assert_eq!(
            lifecycle.session_phase(request.session_id),
            Some(SessionPhase::Closed)
        );
    }

    #[test]
    fn replacement_cannot_receive_or_be_removed_as_waiter_authority() {
        let tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let store = store(&configuration);
        let mut lifecycle = HostCoordinatorLifecycle::new(&store, Adapter, Sessions);
        lifecycle
            .start_new_domain(generation(&configuration), session(1))
            .unwrap();
        let mut face = BootstrapWaiterFace::bind(
            &configuration,
            &store,
            GenerationId::new(1),
            rustix::process::getuid().as_raw(),
        )
        .unwrap();
        let replacement = face.socket_path.with_extension("old");
        fs::rename(&face.socket_path, replacement).unwrap();
        let replacement_listener = UnixListener::bind(&face.socket_path).unwrap();
        fs::set_permissions(&face.socket_path, fs::Permissions::from_mode(0o600)).unwrap();
        assert!(face.admit_pending(&mut lifecycle).is_err());
        let path = face.socket_path.clone();
        drop(face);
        assert!(path.exists());
        drop(replacement_listener);
    }

    #[test]
    fn inactive_waiter_socket_is_recovered_under_matching_domain_lock() {
        let tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let store = store(&configuration);
        let socket = runtime_socket_path(&configuration);
        let listener = UnixListener::bind(&socket).unwrap();
        fs::set_permissions(&socket, fs::Permissions::from_mode(0o600)).unwrap();
        drop(listener);

        let face = BootstrapWaiterFace::bind(
            &configuration,
            &store,
            GenerationId::new(1),
            rustix::process::getuid().as_raw(),
        )
        .unwrap();

        UnixStream::connect(&socket).unwrap();
        drop(face);
        assert!(!socket.exists());
    }

    #[test]
    fn live_waiter_socket_is_not_replaced() {
        let tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let store = store(&configuration);
        let socket = runtime_socket_path(&configuration);
        let listener = UnixListener::bind(&socket).unwrap();
        fs::set_permissions(&socket, fs::Permissions::from_mode(0o600)).unwrap();

        assert!(
            BootstrapWaiterFace::bind(
                &configuration,
                &store,
                GenerationId::new(1),
                rustix::process::getuid().as_raw(),
            )
            .is_err()
        );
        UnixStream::connect(&socket).unwrap();
        drop(listener);
    }

    #[test]
    fn non_socket_waiter_path_is_not_removed() {
        let tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let store = store(&configuration);
        let socket = runtime_socket_path(&configuration);
        fs::write(&socket, b"occupied").unwrap();

        assert!(
            BootstrapWaiterFace::bind(
                &configuration,
                &store,
                GenerationId::new(1),
                rustix::process::getuid().as_raw(),
            )
            .is_err()
        );
        assert_eq!(fs::read(&socket).unwrap(), b"occupied");
    }

    fn complete_startup(lifecycle: &mut HostCoordinatorLifecycle<'_, Adapter, Sessions>) {
        lifecycle.create_trust_anchor().unwrap();
        lifecycle.start_runtime_supervisor().unwrap();
        lifecycle.lease_prepared_lower().unwrap();
        lifecycle.mount_state_anchor().unwrap();
        lifecycle.start_sidecar().unwrap();
        lifecycle.start_connectivity_helper().unwrap();
        lifecycle.arm_runtime_causal_edge().unwrap();
        lifecycle.start_sandbox().unwrap();
        lifecycle.mark_network_ready().unwrap();
        lifecycle.publish_domain().unwrap();
    }

    fn store(configuration: &ResolvedDomainConfiguration) -> Arc<HostCoordinatorDomainStore> {
        let root = PathBuf::from(std::ffi::OsString::from_vec(
            configuration
                .persistent_domain_root()
                .canonical_label()
                .as_bytes()
                .to_vec(),
        ));
        let lock = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(root.join("domain.lock"))
            .unwrap();
        Arc::new(
            HostCoordinatorDomainStore::open(
                try_exclusive_flock(lock).unwrap(),
                &root.join("domain.redb"),
                &Domain {
                    domain_id: configuration.domain_id(),
                    profile_name: "agent".into(),
                    canonical_workspace: configuration.workspace().canonical_label().clone(),
                },
            )
            .unwrap(),
        )
    }

    fn runtime_socket_path(configuration: &ResolvedDomainConfiguration) -> PathBuf {
        PathBuf::from(std::ffi::OsString::from_vec(
            configuration
                .runtime_domain_root()
                .canonical_label()
                .as_bytes()
                .to_vec(),
        ))
        .join(BOOTSTRAP_WAITER_SOCKET_NAME)
    }

    fn configuration(root: &std::path::Path) -> ResolvedDomainConfiguration {
        let configuration_directory = root.join("configuration");
        let workspace = root.join("workspace");
        let runtime = root.join("r");
        fs::create_dir(&configuration_directory).unwrap();
        fs::create_dir(&workspace).unwrap();
        fs::create_dir(&runtime).unwrap();
        let configuration_path = configuration_directory.join("lazaret.toml");
        fs::write(
            &configuration_path,
            format!(
                "schema = \"lazaret-trusted-host-config-v1\"\n[storage]\ndata_root = \"{}\"\n[network]\ndefault_allow = [\"192.0.2.10/32\"]\n[profiles.agent]\nimage = \"example.invalid/image:v1\"
",
                root.join("data").display()
            ),
        )
        .unwrap();
        load_trusted_configuration(ConfigurationLoadRequest {
            explicit_configuration_path: Some(configuration_path),
            selected_profile: "agent".into(),
            workspace,
            home: root.join("home"),
            xdg_config_home: None,
            xdg_data_home: None,
            xdg_runtime_dir: runtime,
            environment: BTreeMap::new(),
            sandbox_layout: lazaret_runtime_supervisor::sandbox::default_sandbox_layout().unwrap(),
        })
        .unwrap()
    }

    fn generation(configuration: &ResolvedDomainConfiguration) -> Generation {
        Generation {
            domain_id: configuration.domain_id(),
            generation_id: GenerationId::new(1),
            configuration_fingerprint: configuration.configuration_fingerprint(),
            coordinator_id: CoordinatorId::new([3; 32]),
            network_sidecar_id: NetworkSidecarId::new([4; 32]),
            sandbox_id: SandboxId::new([5; 32]),
            durable_decision: DurableDecision::None,
        }
    }

    fn session(seed: u8) -> Session {
        Session {
            generation_id: GenerationId::new(1),
            session_id: SessionId::new([seed; 32]),
            cli_id: ClientId::new([seed + 32; 32]),
            session_anchor_id: SessionAnchorId::new([seed + 64; 32]),
            session_phase: SessionPhase::New,
        }
    }
}