//! Host↔guest vsock bridge: Unix domain sockets (macOS/Linux) vs named pipes (Windows WHPX).
//!
//! ## Platform Transport
//! - Unix (Linux/macOS): Unix domain sockets
//! - Windows: Named pipes (WHPX requires this for vsock emulation)

use color_eyre::eyre;
use color_eyre::Result;
use std::path::Path;
use std::time::Duration;

/// Default retry count for vsock bridge connection.
/// With 5ms delay, 300 retries gives ~1.5 seconds total wait time,
/// sufficient for VM startup and vsock pipe creation.
pub const VSOCK_BRIDGE_MAX_RETRIES: u32 = 300;

#[cfg(unix)]
pub fn setup_vsock_ready_listener(env_hash: &str) -> Result<Option<std::os::unix::net::UnixListener>> {
    let run_dir = &crate::models::dirs().epkg_run;
    // Clean up stale sockets from previous runs
    if let Ok(entries) = std::fs::read_dir(run_dir) {
        for entry in entries.flatten() {
            let name = entry.file_name().to_string_lossy().to_string();
            if name.starts_with("vsock-") && name.ends_with(".sock") {
                let _ = std::fs::remove_file(entry.path());
                log::trace!("libkrun: cleaned up stale socket {}", name);
            }
            if name.starts_with("ready-") && name.ends_with(".sock") {
                let _ = std::fs::remove_file(entry.path());
                log::trace!("libkrun: cleaned up stale socket {}", name);
            }
        }
    }

    let ready_path = run_dir.join(format!("ready-{}.sock", env_hash));
    let _ = std::fs::remove_file(&ready_path);

    log::debug!("libkrun: creating ready listener on {}", ready_path.display());
    let listener = std::os::unix::net::UnixListener::bind(&ready_path)
        .map_err(|e| eyre::eyre!("Failed to bind ready socket {}: {}", ready_path.display(), e))?;

    listener.set_nonblocking(true)
        .map_err(|e| eyre::eyre!("Failed to set non-blocking on ready socket: {}", e))?;

    Ok(Some(listener))
}

#[cfg(unix)]
pub fn wait_guest_ready_unix(
    listener: &std::os::unix::net::UnixListener,
    vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<()> {
    use std::os::unix::io::AsRawFd;
    use std::time::Instant;

    let listener_fd = listener.as_raw_fd();
    let start = Instant::now();
    // Windows/WSL2 needs longer timeout due to slower virtiofs and potential large init binary
    // With 195MB debug binary, guest takes ~217 seconds to boot. Allow 5 minutes.
    #[cfg(windows)]
    let total_timeout = Duration::from_secs(300);
    #[cfg(not(windows))]
    let total_timeout = Duration::from_secs(30);

    loop {
        // Check if VM start failed first
        if let Some(ref failed_rx) = vm_start_failed_rx {
            if failed_rx.try_recv().is_ok() {
                return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
            }
        }

        // Calculate remaining timeout
        let remaining = total_timeout.saturating_sub(start.elapsed());
        if remaining.is_zero() {
            log::error!("libkrun: timeout waiting for VM to become ready");
            return Err(eyre::eyre!("Timeout waiting for VM to start"));
        }
        let remaining_ms = (remaining.as_millis().min(u32::MAX as u128) as u32) as i32;

        let mut poll_fds = [libc::pollfd {
            fd: listener_fd,
            events: libc::POLLIN,
            revents: 0,
        }];

        let poll_result = unsafe { libc::poll(poll_fds.as_mut_ptr(), 1, remaining_ms) };

        match poll_result {
            0 => {
                // Timeout - loop back to check VM failure
            }
            n if n < 0 => {
                log::error!("libkrun: poll error on ready socket");
                return Err(eyre::eyre!("Poll error on ready socket"));
            }
            _ => {
                let (stream, _addr) = listener
                    .accept()
                    .map_err(|e| eyre::eyre!("Failed to accept on ready socket: {}", e))?;
                log::debug!("libkrun: guest connected to ready socket, guest is ready!");
                drop(stream);
                return Ok(());
            }
        }
    }
}

#[cfg(unix)]
pub fn connect_vsock_bridge(sock_path: &Path, max_retries: u32) -> Result<std::os::unix::net::UnixStream> {
    use std::os::unix::net::UnixStream;

    let mut retry_count = 0;
    let mut last_error = None;
    while retry_count < max_retries {
        match UnixStream::connect(sock_path) {
            Ok(stream) => {
                // Increase socket buffer sizes to handle large data transfers
                use std::os::unix::io::AsRawFd;
                super::set_socket_buffer_size(stream.as_raw_fd());
                return Ok(stream);
            }
            Err(e) => {
                last_error = Some(e);
                retry_count += 1;
                if retry_count >= max_retries {
                    break;
                }
                std::thread::sleep(Duration::from_millis(5));
            }
        }
    }
    Err(eyre::eyre!(
        "Failed to connect to Unix socket {} after {} retries: {}",
        sock_path.display(),
        max_retries,
        last_error.unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::Other, "connection failed"))
    ))
}

#[cfg(windows)]
use std::os::windows::io::FromRawHandle;

#[cfg(windows)]
use windows::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE};

#[cfg(windows)]
use windows::Win32::Storage::FileSystem::{
    CreateFileW, FILE_ATTRIBUTE_NORMAL, FILE_SHARE_READ, FILE_SHARE_WRITE,
    OPEN_EXISTING, PIPE_ACCESS_DUPLEX,
};

#[cfg(windows)]
use windows::Win32::System::Pipes::{
    ConnectNamedPipe, CreateNamedPipeW, WaitNamedPipeA, PIPE_READMODE_BYTE, PIPE_TYPE_BYTE,
    PIPE_UNLIMITED_INSTANCES, PIPE_WAIT,
};

#[cfg(windows)]
use windows::Win32::Security::SECURITY_ATTRIBUTES;

#[cfg(windows)]
use windows::core::PCWSTR;

/// Stem used for `\\.\pipe\<stem>`; must match `krun_add_vsock_port_windows` on the host.
#[cfg(windows)]
pub(crate) fn pipe_name_from_sock_path(path: &Path) -> Result<String> {
    path.file_stem()
        .and_then(|s| s.to_str())
        .map(|s| s.to_string())
        .ok_or_else(|| eyre::eyre!("invalid vsock path (no file stem): {}", path.display()))
}

#[cfg(windows)]
fn to_wide_null(s: &str) -> Vec<u16> {
    s.encode_utf16().chain(std::iter::once(0)).collect()
}

#[cfg(windows)]
pub struct WindowsReadyPipe {
    handle: Option<HANDLE>,
}

/// Create SECURITY_ATTRIBUTES with a NULL DACL that grants access to everyone.
/// This is needed for named pipes to avoid ERROR_ACCESS_DENIED in some environments.
/// Returns the SECURITY_ATTRIBUTES and the allocated buffer size for cleanup.
#[cfg(windows)]
fn create_security_attributes() -> Result<(SECURITY_ATTRIBUTES, usize)> {
    use windows::Win32::Security::{InitializeSecurityDescriptor, SetSecurityDescriptorDacl, PSECURITY_DESCRIPTOR};
    use windows::Win32::Foundation::BOOL;

    unsafe {
        // Allocate security descriptor (use 256 bytes for safety)
        const SD_SIZE: usize = 256;
        let sd_ptr = std::alloc::alloc(std::alloc::Layout::from_size_align(SD_SIZE, 8).unwrap());
        if sd_ptr.is_null() {
            return Err(eyre::eyre!("Failed to allocate security descriptor"));
        }

        // Initialize security descriptor (revision 1)
        if InitializeSecurityDescriptor(PSECURITY_DESCRIPTOR(sd_ptr as *mut core::ffi::c_void), 1).is_err() {
            std::alloc::dealloc(sd_ptr, std::alloc::Layout::from_size_align(SD_SIZE, 8).unwrap());
            return Err(eyre::eyre!("InitializeSecurityDescriptor failed"));
        }

        // Set NULL DACL (allow access to everyone)
        if SetSecurityDescriptorDacl(PSECURITY_DESCRIPTOR(sd_ptr as *mut core::ffi::c_void), BOOL(1), None, BOOL(0)).is_err() {
            std::alloc::dealloc(sd_ptr, std::alloc::Layout::from_size_align(SD_SIZE, 8).unwrap());
            return Err(eyre::eyre!("SetSecurityDescriptorDacl failed"));
        }

        Ok((SECURITY_ATTRIBUTES {
            nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
            lpSecurityDescriptor: sd_ptr as *mut _,
            bInheritHandle: BOOL(0),
        }, SD_SIZE))
    }
}

#[cfg(windows)]
impl Drop for WindowsReadyPipe {
    fn drop(&mut self) {
        if let Some(h) = self.handle.take() {
            unsafe {
                let _ = CloseHandle(h);
            }
        }
    }
}

#[cfg(windows)]
impl WindowsReadyPipe {
    /// Take ownership of the handle, preventing Drop from closing it.
    /// Returns the raw handle value.
    fn take_handle(&mut self) -> Option<std::os::windows::io::RawHandle> {
        self.handle.take().map(|h| h.0 as std::os::windows::io::RawHandle)
    }
}

#[cfg(windows)]
pub fn setup_vsock_ready_listener(env_hash: &str) -> Result<Option<WindowsReadyPipe>> {
    let run_dir = &crate::models::dirs().epkg_run;
    let _ = std::fs::create_dir_all(run_dir);
    let ready_path = run_dir.join(format!("ready-{}.sock", env_hash));
    let pipe_name = pipe_name_from_sock_path(&ready_path)?;
    let full = format!("\\\\.\\pipe\\{}", pipe_name);
    let wide = to_wide_null(&full);

    unsafe {
        use windows::Win32::Foundation::BOOL;

        // Create security attributes with NULL DACL to allow access from guest VM
        let (security_attrs, sd_buffer_size) = create_security_attributes()
            .unwrap_or_else(|e| {
                crate::debug_epkg!("libkrun_bridge: warning - failed to create security attributes: {}, using default", e);
                (SECURITY_ATTRIBUTES {
                    nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
                    lpSecurityDescriptor: std::ptr::null_mut(),
                    bInheritHandle: BOOL(0),
                }, 0usize)
            });

        let h = CreateNamedPipeW(
            PCWSTR(wide.as_ptr()),
            PIPE_ACCESS_DUPLEX,  // Removed FILE_FLAG_OVERLAPPED for synchronous operation
            PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
            PIPE_UNLIMITED_INSTANCES,
            4096,
            4096,
            0,
            Some(&security_attrs),
        );

        // Clean up security descriptor if we allocated one
        if !security_attrs.lpSecurityDescriptor.is_null() {
            std::alloc::dealloc(
                security_attrs.lpSecurityDescriptor as *mut u8,
                std::alloc::Layout::from_size_align(sd_buffer_size, 8).unwrap()
            );
        }

        if h == INVALID_HANDLE_VALUE {
            return Err(eyre::eyre!(
                "CreateNamedPipeW failed: {}",
                std::io::Error::last_os_error()
            ));
        }
        Ok(Some(WindowsReadyPipe { handle: Some(h) }))
    }
}

#[cfg(windows)]
pub fn wait_guest_ready_windows(
    pipe: &WindowsReadyPipe,
    vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<()> {
    use std::sync::mpsc;
    use std::thread;

    let handle_raw = pipe.handle.as_ref().map(|h| h.0 as usize).ok_or_else(|| eyre::eyre!("Pipe handle already taken"))?;
    let (tx, rx) = mpsc::channel();
    let jh = thread::spawn(move || {
        let handle = HANDLE(handle_raw as *mut _);
        let r = unsafe { ConnectNamedPipe(handle, None) };
        let _ = tx.send(r);
    });

    let start = std::time::Instant::now();
    // Windows/WSL2 needs longer timeout due to slower virtiofs and potential large init binary
    // With 195MB debug binary, guest takes ~217 seconds to boot. Allow 5 minutes.
    #[cfg(windows)]
    let timeout = Duration::from_secs(300);
    #[cfg(not(windows))]
    let timeout = Duration::from_secs(30);
    // Use smaller poll interval for faster response
    let poll_interval = Duration::from_millis(10);

    loop {
        // Check if VM start failed
        if let Some(ref failed_rx) = vm_start_failed_rx {
            if failed_rx.try_recv().is_ok() {
                // Note: Don't join the pipe thread here - ConnectNamedPipe is blocking
                // and will never complete since VM failed. The thread will be cleaned
                // up when the process exits.
                return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
            }
        }

        // Use recv_timeout to wait efficiently
        match rx.recv_timeout(poll_interval) {
            Ok(Ok(())) => {
                log::debug!("libkrun: guest connected to ready pipe, guest is ready!");
                let _ = jh.join();
                return Ok(());
            }
            Ok(Err(e)) => {
                let _ = jh.join();
                return Err(eyre::eyre!("ConnectNamedPipe failed: {}", e));
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {
                // Not ready yet, check timeout and continue
            }
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                let _ = jh.join();
                return Err(eyre::eyre!("Pipe thread disconnected unexpectedly"));
            }
        }

        if start.elapsed() >= timeout {
            log::error!("libkrun: timeout waiting for VM to become ready");
            // Note: Don't join the pipe thread - ConnectNamedPipe is blocking
            return Err(eyre::eyre!("Timeout waiting for VM to start"));
        }
    }
}

#[cfg(windows)]
pub fn connect_vsock_bridge(sock_path: &Path, max_retries: u32) -> Result<std::fs::File> {
    let pipe_name = pipe_name_from_sock_path(sock_path)?;
    let full = format!("\\\\.\\pipe\\{}", pipe_name);
    crate::debug_epkg!("libkrun_bridge: connecting to named pipe: {}", full);
    let c_path = std::ffi::CString::new(full.as_bytes())
        .map_err(|_| eyre::eyre!("invalid pipe path"))?;

    let mut retry_count = 0;
    let mut last_error = None;
    crate::debug_epkg!("libkrun_bridge: starting connection retry loop (max={})", max_retries);
    while retry_count < max_retries {
        unsafe {
            crate::debug_epkg!("libkrun_bridge: attempt {} - waiting for named pipe...", retry_count);
            if WaitNamedPipeA(
                windows::core::PCSTR(c_path.as_ptr() as *const u8),
                30_000,
            )
            .is_err()
            {
                last_error = Some(std::io::Error::last_os_error());
                crate::debug_epkg!("libkrun_bridge: WaitNamedPipeA failed: {:?}", last_error);
                retry_count += 1;
                if retry_count >= max_retries {
                    break;
                }
                std::thread::sleep(Duration::from_millis(5));
                continue;
            }
            crate::debug_epkg!("libkrun_bridge: named pipe is available, connecting...");

            // Use synchronous mode for reliable operation with std::fs::File
            // FILE_FLAG_OVERLAPPED causes issues with synchronous I/O
            // GENERIC_READ = 0x80000000, GENERIC_WRITE = 0x40000000
            let access: u32 = 0x80000000u32 | 0x40000000u32;
            let handle = CreateFileW(
                PCWSTR(to_wide_null(&full).as_ptr()),
                access,
                FILE_SHARE_READ | FILE_SHARE_WRITE,
                None,
                OPEN_EXISTING,
                FILE_ATTRIBUTE_NORMAL,  // Removed FILE_FLAG_OVERLAPPED
                None,
            );

            match handle {
                Ok(h) if h != INVALID_HANDLE_VALUE => {
                    crate::debug_epkg!("libkrun_bridge: successfully connected to named pipe");
                    let file = std::fs::File::from_raw_handle(h.0);
                    return Ok(file);
                }
                Ok(_) => {
                    crate::debug_epkg!("libkrun_bridge: CreateFileW returned INVALID_HANDLE_VALUE");
                    last_error = Some(std::io::Error::last_os_error());
                }
                Err(e) => {
                    crate::debug_epkg!("libkrun_bridge: CreateFileW failed: {}", e);
                    last_error = Some(std::io::Error::last_os_error());
                }
            }
        }
        retry_count += 1;
        if retry_count < max_retries {
            std::thread::sleep(Duration::from_millis(5));
        }
    }
    crate::debug_epkg!("libkrun_bridge: failed to connect after {} retries: {:?}", max_retries, last_error);

    Err(eyre::eyre!(
        "Failed to connect to named pipe for {} after {} retries: {}",
        sock_path.display(),
        max_retries,
        last_error.unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::Other, "connection failed"))
    ))
}

// =============================================================================
// Reverse vsock mode: Guest connects to Host
// =============================================================================
// In reverse mode, the Host listens on a socket/pipe and waits for the Guest
// to connect. This avoids the vsock handshake timing issues on Windows/WHPX.

/// Set up a reverse listener for Guest to connect to.
/// In reverse mode, Host listens and Guest initiates the connection.
#[cfg(unix)]
#[allow(dead_code)]
pub fn setup_reverse_listener(sock_path: &Path) -> Result<std::os::unix::net::UnixListener> {
    // Clean up any stale socket
    let _ = std::fs::remove_file(sock_path);

    log::debug!("libkrun: creating reverse listener on {}", sock_path.display());
    let listener = std::os::unix::net::UnixListener::bind(sock_path)
        .map_err(|e| eyre::eyre!("Failed to bind reverse socket {}: {}", sock_path.display(), e))?;

    // Set non-blocking for timeout support
    listener.set_nonblocking(true)
        .map_err(|e| eyre::eyre!("Failed to set non-blocking on reverse socket: {}", e))?;

    Ok(listener)
}

/// Accept a connection from Guest in reverse mode.
/// Uses poll() for efficient waiting without busy-looping.
#[cfg(unix)]
#[allow(dead_code)]
pub fn accept_reverse_connection(
    listener: &std::os::unix::net::UnixListener,
    vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<std::os::unix::net::UnixStream> {
    use std::os::unix::io::AsRawFd;
    use std::time::Instant;

    let start = Instant::now();
    let timeout = Duration::from_secs(30);
    let listener_fd = listener.as_raw_fd();

    loop {
        // Check if VM start failed first
        if let Some(ref failed_rx) = vm_start_failed_rx {
            if failed_rx.try_recv().is_ok() {
                return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
            }
        }

        // Calculate remaining timeout
        let remaining = timeout.saturating_sub(start.elapsed());
        if remaining.is_zero() {
            return Err(eyre::eyre!("Timeout waiting for Guest to connect (reverse mode)"));
        }
        let remaining_ms = (remaining.as_millis().min(u32::MAX as u128) as u32) as i32;

        // Use poll() to wait for connection efficiently
        let mut poll_fds = [libc::pollfd {
            fd: listener_fd,
            events: libc::POLLIN,
            revents: 0,
        }];

        let poll_result = unsafe { libc::poll(poll_fds.as_mut_ptr(), 1, remaining_ms) };

        match poll_result {
            0 => {
                // Timeout - check VM failure and loop
            }
            n if n < 0 => {
                let errno = std::io::Error::last_os_error();
                if errno.raw_os_error() == Some(libc::EINTR) {
                    // Interrupted by signal, check conditions and retry
                    continue;
                }
                return Err(eyre::eyre!("Poll error on reverse listener: {}", errno));
            }
            _ => {
                // Socket is ready, accept the connection
                match listener.accept() {
                    Ok((stream, _addr)) => {
                        log::debug!("libkrun: Guest connected to reverse listener");
                        // Set blocking mode
                        stream.set_nonblocking(false)
                            .map_err(|e| eyre::eyre!("Failed to set blocking mode on reverse stream: {}", e))?;
                        return Ok(stream);
                    }
                    Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                        // Spurious wakeup, continue
                    }
                    Err(e) => {
                        return Err(eyre::eyre!("Failed to accept reverse connection: {}", e));
                    }
                }
            }
        }
    }
}

/// Windows reverse listener setup.
#[cfg(windows)]
pub fn setup_reverse_listener(sock_path: &Path) -> Result<WindowsReadyPipe> {
    let pipe_name = pipe_name_from_sock_path(sock_path)?;
    let full = format!("\\\\.\\pipe\\{}", pipe_name);
    let wide = to_wide_null(&full);

    unsafe {
        use windows::Win32::Foundation::BOOL;

        // Create security attributes with NULL DACL to allow access from guest VM
        let (security_attrs, sd_buffer_size) = create_security_attributes()
            .unwrap_or_else(|e| {
                crate::debug_epkg!("libkrun_bridge: warning - failed to create security attributes: {}, using default", e);
                (SECURITY_ATTRIBUTES {
                    nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
                    lpSecurityDescriptor: std::ptr::null_mut(),
                    bInheritHandle: BOOL(0),
                }, 0usize)
            });

        // CreateNamedPipeW with PIPE_ACCESS_DUPLEX for bidirectional communication
        let h = CreateNamedPipeW(
            PCWSTR(wide.as_ptr()),
            PIPE_ACCESS_DUPLEX,
            PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
            PIPE_UNLIMITED_INSTANCES,
            4096,
            4096,
            0,
            Some(&security_attrs),
        );

        // Clean up security descriptor if we allocated one
        if !security_attrs.lpSecurityDescriptor.is_null() {
            std::alloc::dealloc(
                security_attrs.lpSecurityDescriptor as *mut u8,
                std::alloc::Layout::from_size_align(sd_buffer_size, 8).unwrap()
            );
        }

        if h == INVALID_HANDLE_VALUE {
            return Err(eyre::eyre!(
                "CreateNamedPipeW failed for reverse listener: {}",
                std::io::Error::last_os_error()
            ));
        }
        log::debug!("libkrun: reverse listener created on pipe {}", full);
        Ok(WindowsReadyPipe { handle: Some(h) })
    }
}

/// Accept a connection from Guest in reverse mode (Windows).
/// Takes ownership of the pipe to prevent double-close of the handle.
/// Uses recv_timeout() for efficient waiting without busy-looping.
#[cfg(windows)]
pub fn accept_reverse_connection(
    mut pipe: WindowsReadyPipe,
    vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<std::fs::File> {
    use std::sync::mpsc;
    use std::thread;
    use std::time::Instant;

    let handle_raw = pipe.take_handle()
        .ok_or_else(|| eyre::eyre!("Pipe handle already taken"))? as usize;
    let (tx, rx) = mpsc::channel();

    crate::debug_epkg!("libkrun_bridge: spawning ConnectNamedPipe thread...");
    // Spawn thread to wait for connection
    let jh = thread::spawn(move || {
        let handle = HANDLE(handle_raw as *mut _);
        let r = unsafe { ConnectNamedPipe(handle, None) };
        let _ = tx.send(r);
    });

    let start = Instant::now();
    let timeout = Duration::from_secs(30);
    // Use smaller poll interval for faster response while still checking VM failure
    let poll_interval = Duration::from_millis(10);

    crate::debug_epkg!("libkrun_bridge: waiting for Guest connection or VM failure...");

    loop {
        // Check if VM start failed
        if let Some(ref failed_rx) = vm_start_failed_rx {
            match failed_rx.try_recv() {
                Ok(_) => {
                    crate::debug_epkg!("libkrun_bridge: VM start failure detected!");
                    return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
                }
                Err(mpsc::TryRecvError::Empty) => {
                    // No failure yet, continue
                }
                Err(mpsc::TryRecvError::Disconnected) => {
                    crate::debug_epkg!("libkrun_bridge: VM failure channel disconnected");
                }
            }
        }

        // Use recv_timeout to wait for pipe connection efficiently
        match rx.recv_timeout(poll_interval) {
            Ok(Ok(())) => {
                log::debug!("libkrun: Guest connected to reverse pipe");
                crate::debug_epkg!("libkrun_bridge: Guest connected successfully!");
                let _ = jh.join();
                // Return the pipe handle as a File
                let file = std::fs::File::from(unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(handle_raw as std::os::windows::io::RawHandle) });
                return Ok(file);
            }
            Ok(Err(e)) => {
                crate::debug_epkg!("libkrun_bridge: ConnectNamedPipe failed: {:?}", e);
                let _ = jh.join();
                return Err(eyre::eyre!("ConnectNamedPipe failed: {:?}", e));
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {
                // Not ready yet, check timeout and continue
            }
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                crate::debug_epkg!("libkrun_bridge: Pipe thread disconnected unexpectedly");
                let _ = jh.join();
                return Err(eyre::eyre!("Pipe thread disconnected unexpectedly"));
            }
        }

        if start.elapsed() >= timeout {
            crate::debug_epkg!("libkrun_bridge: Timeout waiting for Guest connection");
            return Err(eyre::eyre!("Timeout waiting for Guest to connect (reverse mode)"));
        }
    }
}